协商中再问按后台超时小时顺延并可强制清跟进时刻;工单与账号支持按筛选导出 Excel。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
linmianxun
2026-10-13 12:34:03 +08:00
co-authored by Cursor
parent 97f0219dfd
commit 0b2f0b76c4
36 changed files with 867 additions and 64 deletions
@@ -50,6 +50,10 @@ def _reask_one(row: dict[str, Any], *, settings: Settings, ledger: Any = None) -
no = str(row.get("work_order_no") or row.get("workOrderNo") or "").strip()
if not no:
return False
# 进程/双 Worker 防重:同一工单同一天只发一轮(主账 next_followup 清不掉时曾连发 WO024)
if _already_sent_reask(no, settings):
logger.warning("协商再问去重跳过 wo=%s", no)
return False
chat_id = str(row.get("collab_chat_id") or row.get("collabChatId") or "").strip()
sales = str(row.get("sales_wecom_id") or row.get("salesWecomId") or "").strip()
name = str(row.get("sales_name") or row.get("salesName") or "").strip() or "销售"
@@ -69,6 +73,7 @@ def _reask_one(row: dict[str, Any], *, settings: Settings, ledger: Any = None) -
if not ok:
logger.warning("协商再问原群失败且不补发私聊 wo=%s chat=%s", no, chat_id)
return False
_mark_sent_reask(no, settings)
_start_deal_wait(ledger, no)
return True
if not sales:
@@ -87,10 +92,47 @@ def _reask_one(row: dict[str, Any], *, settings: Settings, ledger: Any = None) -
if not ok:
logger.warning("协商再问私聊入队失败 wo=%s info=%s", no, info)
return False
_mark_sent_reask(no, settings)
_start_deal_wait(ledger, no)
return True
def _already_sent_reask(work_order_no: str, settings: Settings) -> bool:
"""
跨进程去重:Redis SET NX。
返回 True=今天已发过应跳过;False=抢到发送权。
主账 next_followup 若因 updateById 清不掉 null,没有这层会每分钟连发。
"""
from datetime import datetime
from agent.policy.deal_followup import SHANGHAI
from agent.redis_coord import get_redis_runtime
from agent.redis_coord.idempotency import try_acquire
day = datetime.now(SHANGHAI).strftime("%Y-%m-%d")
try:
runtime = get_redis_runtime(settings)
client = getattr(runtime, "client", None)
if client is None:
return False
first = try_acquire(
client,
scope="deal_reask",
idempotency_key=f"{work_order_no}:{day}",
ttl_seconds=86400,
)
return not first
except Exception:
logger.exception("协商再问 Redis 去重失败 wo=%s", work_order_no)
return False
def _mark_sent_reask(work_order_no: str, settings: Settings) -> None:
"""兼容旧调用点;去重键已在发送前 SET NX。"""
_ = (work_order_no, settings)
def _start_deal_wait(ledger: Any, work_order_no: str) -> None:
"""再问发出后重新盯成交跟进。ledger 可空。"""
from agent.policy.event_exception import WAIT_DEAL, start_wait
+61 -4
View File
@@ -330,12 +330,32 @@ class HttpLedger:
return out if out.get("ok") else {"ok": False, "error": out.get("error") or "patch_lost_reason_failed"}
def claim_due_negotiate(self) -> list[dict[str, Any]]:
"""认领到期协商再问,主账会清空再问时刻防重发。"""
"""
认领到期协商再问。
主账应用 UpdateWrapper 清空 next_followup_at;若旧包仍用 updateById
写不进 null,这里对每单再调 clearNextFollowup 强制清掉,避免每分钟连发。
"""
out = self._post("/inquiry/agent/ticket/claimDueNegotiate", {})
rows = out.get("tickets") or out.get("result") or []
if isinstance(rows, list):
return [dict(x) for x in rows if isinstance(x, dict)]
return []
if not isinstance(rows, list):
return []
claimed: list[dict[str, Any]] = []
for raw in rows:
if not isinstance(raw, dict):
continue
row = dict(raw)
no = str(row.get("work_order_no") or row.get("workOrderNo") or "").strip()
if no:
try:
self._post(
"/inquiry/agent/ticket/clearNextFollowup",
{"workOrderNo": no, "work_order_no": no},
)
except Exception:
logger.exception("强制清空再问时刻失败 wo=%s", no)
claimed.append(row)
return claimed
def match_sea_staff(self, *, origin: str, destination: str) -> list[dict[str, Any]]:
"""主账按线路返回启用的海运产品岗。"""
@@ -451,6 +471,43 @@ class HttpLedger:
body["airline"] = airline.strip().upper()
if (flight_no or "").strip():
body["flightNo"] = flight_no.strip().upper()
# #region agent log
try:
import json as _json
import time as _time
from pathlib import Path as _Path
_line = _json.dumps(
{
"sessionId": "e38e4c",
"runId": "pre-fix",
"hypothesisId": "H-E",
"location": "http_ledger.py:lock_cabin",
"message": "ledger_lock_called",
"data": {
"wo": work_order_no,
"optionNo": option_no,
"airline": (airline or "").strip(),
"flightNo": (flight_no or "").strip(),
"instruction_head": (instruction_text or "")[:80],
},
"timestamp": int(_time.time() * 1000),
},
ensure_ascii=False,
) + "\n"
for _p in (
_Path(r"d:\own\友通达询价智能体(新)\debug-e38e4c.log"),
_Path(r"C:/Users/ytd_extdev_0903/incoming/debug-e38e4c.log"),
):
try:
_p.parent.mkdir(parents=True, exist_ok=True)
with _p.open("a", encoding="utf-8") as _fh:
_fh.write(_line)
except Exception:
pass
except Exception:
pass
# #endregion
return self._post(
"/inquiry/agent/tms/lockCabin",
body,
+2 -1
View File
@@ -109,7 +109,8 @@ class MemoryLedger:
self._staff_by_wecom: dict[str, dict[str, Any]] = {}
self._staff_by_name: dict[str, dict[str, Any]] = {}
self._exception_notify_users: list[dict[str, Any]] = []
self._timeouts: dict[str, float] = {}
# 单测默认带协商再问 48 工作小时,与主账 deal_negotiate 种子一致
self._timeouts: dict[str, float] = {"deal_negotiate": 48.0}
def _next_no(self) -> str:
self._seq += 1
+6 -3
View File
@@ -885,13 +885,16 @@ _SKIP_FEE_NAMES = frozenset(
{"usd", "cny", "rmb", "hkd", "eur", "jpy", "gbp", "美金", "美元", "人民币", "港币", "港元", "元"}
)
# 航线常写「空运费 单价CNY99,数量100,金额CNY199」。单价和金额不是同一个数,必须认金额。
# WO202610130003:协同句常见「空运费, KG, 单价CNY99.00, 数量100, 金额CNY9900」——费名与单价之间夹单位;
# 数量/金额前也可能只有空格无逗号。漏匹配会落到 _RE_FEE_LINE,把单价 99 当成行金额、数量被补成 1。
_RE_FEE_PRICED = re.compile(
r"(?P<name>[\u4e00-\u9fffA-Za-z][\u4e00-\u9fffA-Za-z0-9/()()++\--]{0,24}费)"
r"\s*单价\s*"
r"(?:\s*[,,、]\s*(?:KG|KGS|公斤|千克|TON|CBM|柜|票|板|点|[A-Za-z]{1,8}))?"
r"\s*[,,、]?\s*单价\s*"
rf"(?:(?P<unit_ccy>{_CCY_TOKEN})\s*)?"
r"(?P<unit>\d+(?:\.\d+)?)"
r"(?:\s*[,,、]\s*数量\s*(?P<qty>\d+(?:\.\d+)?))?"
r"(?:\s*[,,、]\s*金额\s*"
r"(?:\s*[,,、]?\s*数量\s*(?P<qty>\d+(?:\.\d+)?))?"
r"(?:\s*[,,、]?\s*金额\s*"
rf"(?:(?P<amt_ccy>{_CCY_TOKEN})\s*)?"
r"(?P<amount>\d+(?:\.\d+)?))?",
re.I,
@@ -48,6 +48,8 @@ _PENDING_LOCK = "__pending_air_lock"
_PENDING_LOCK_OPTION = "__pending_lock_option"
_PENDING_LOCK_AIRLINE = "__pending_lock_airline"
_PENDING_LOCK_FLIGHT = "__pending_lock_flight"
_LOCK_ID = "__air_lock_id"
_LOCK_OPTION = "__air_lock_option"
_LOCK_QUOTE = "__air_lock_quote"
@@ -861,6 +863,7 @@ def apply_air_deal(
no,
str(result.get("next_followup_at") or ""),
lost_reason=ack_reason,
followup_hours=result.get("negotiate_hours"),
),
)
return phase
@@ -1518,11 +1521,13 @@ def _resolve_lock_params(
air = _normalize_airline(airline or said_air or str(facts.get(_PENDING_LOCK_AIRLINE) or ""))
flight = _normalize_flight(flight_no or said_flight or str(facts.get(_PENDING_LOCK_FLIGHT) or ""))
chosen = _option_row(quote, option)
single_auto = False
if chosen is None and not option:
options = _air_options(quote)
if len(options) == 1:
chosen = options[0]
option = str(chosen.get("optionNo") or chosen.get("option_no") or "").strip()
single_auto = True
if not _airline_ready(air) and chosen is not None:
air = _normalize_airline(
str(chosen.get("airlineCode") or chosen.get("airline") or "")
@@ -2007,6 +2007,7 @@ class AirTextInquiryFlow:
sess.work_order_no,
str(result.get("next_followup_at") or ""),
lost_reason=ack_reason,
followup_hours=result.get("negotiate_hours"),
)
)
return phase
+45 -5
View File
@@ -88,7 +88,7 @@ def shift_workdays(start: date, count: int) -> date:
"""
从 start 当天之后数 count 个工作日。
成交选协商中:顺延 3 个工作日后,第四个工作日再问,故 count=4。
仍给单测与节假日校验用;协商再问已改走 add_workday_hours。
"""
if count < 1:
return start
@@ -101,11 +101,51 @@ def shift_workdays(start: date, count: int) -> date:
return cur
def next_followup_at(now: datetime | None = None) -> datetime:
"""第四个工作日 Asia/Shanghai 09:00。"""
# 协商中再问:主账 timeoutCode;缺配置时默认 48 工作小时。
TIMEOUT_CODE_DEAL_NEGOTIATE = "deal_negotiate"
DEFAULT_NEGOTIATE_HOURS = 48.0
def add_workday_hours(start: datetime, hours: float) -> datetime:
"""
从 start 起累加 hours 个工作小时;非工作日整天跳过。
与 workday_elapsed_hours 互逆:工作日当天 0–24 都算。
起点落在休息日:从下一个工作日 00:00 起算。
hours≤0 时原样返回(调用方应不排期)。
"""
base = start.astimezone(SHANGHAI) if start.tzinfo else start.replace(tzinfo=SHANGHAI)
remaining = float(hours)
if remaining <= 0:
return base
cur = base
if not is_workday(cur.date()):
cur = datetime.combine(cur.date() + timedelta(days=1), time(0, 0), tzinfo=SHANGHAI)
while not is_workday(cur.date()):
cur = datetime.combine(cur.date() + timedelta(days=1), time(0, 0), tzinfo=SHANGHAI)
while remaining > 0:
if not is_workday(cur.date()):
cur = datetime.combine(cur.date() + timedelta(days=1), time(0, 0), tzinfo=SHANGHAI)
continue
day_end = datetime.combine(cur.date() + timedelta(days=1), time(0, 0), tzinfo=SHANGHAI)
room = (day_end - cur).total_seconds() / 3600.0
if remaining <= room + 1e-9:
return cur + timedelta(hours=remaining)
remaining -= room
cur = day_end
return cur
def next_followup_at(now: datetime | None = None, *, hours: float | None = None) -> datetime:
"""
距今 hours 个工作小时后的再问时刻(默认 48)。
非工作日不计入;提醒时刻不再固定 09:00。
hours≤0 时返回起点(调用方应把 0 当作不排再问)。
"""
base = now.astimezone(SHANGHAI) if now else datetime.now(SHANGHAI)
day = shift_workdays(base.date(), 4)
return datetime.combine(day, time(9, 0), tzinfo=SHANGHAI)
h = DEFAULT_NEGOTIATE_HOURS if hours is None else float(hours)
return add_workday_hours(base, h)
def workday_elapsed_hours(start: datetime, now: datetime) -> float:
+42 -2
View File
@@ -13,7 +13,12 @@ from typing import Any, Callable, Optional
from agent.config import get_settings
from agent.policy import inquiry_copy as copy
from agent.policy.deal_followup import next_followup_at, pick_leaders
from agent.policy.deal_followup import (
DEFAULT_NEGOTIATE_HOURS,
TIMEOUT_CODE_DEAL_NEGOTIATE,
next_followup_at,
pick_leaders,
)
logger = logging.getLogger(__name__)
@@ -156,8 +161,14 @@ def finish_deal_outcome(
status = (outcome or "").strip()
extra: dict[str, Any] = {}
summary = ""
negotiate_hours = 0.0
if status == "协商中":
extra["next_followup_at"] = next_followup_at().strftime("%Y-%m-%d %H:%M:%S")
# 小时读后台超时响应 deal_negotiate;0=不排再问;缺配置默认 48
negotiate_hours = _negotiate_followup_hours(ledger)
if negotiate_hours > 0:
extra["next_followup_at"] = next_followup_at(hours=negotiate_hours).strftime(
"%Y-%m-%d %H:%M:%S"
)
if status == "已成交":
staff: list[dict[str, Any]] = []
lister = getattr(ledger, "list_staff_by_role", None)
@@ -242,9 +253,38 @@ def finish_deal_outcome(
"notify_summary": summary,
"status": status,
"next_followup_at": str(extra.get("next_followup_at") or ""),
"negotiate_hours": negotiate_hours,
}
def _negotiate_followup_hours(ledger: Any) -> float:
"""
读主账 deal_negotiate 超时小时;找不到或异常则用默认 48。
允许小数;负数当 0(不排再问)。不在回调线程打主账以外的慢活。
"""
lookup = getattr(ledger, "get_timeout_by_code", None)
if not callable(lookup):
return DEFAULT_NEGOTIATE_HOURS
try:
row = lookup(timeout_code=TIMEOUT_CODE_DEAL_NEGOTIATE) or {}
except Exception:
logger.exception("读协商再问小时失败,用默认 %s", DEFAULT_NEGOTIATE_HOURS)
return DEFAULT_NEGOTIATE_HOURS
if not row.get("found"):
return DEFAULT_NEGOTIATE_HOURS
raw = row.get("timeoutHours")
if raw is None:
raw = row.get("timeout_hours")
try:
hours = float(raw)
except (TypeError, ValueError):
return DEFAULT_NEGOTIATE_HOURS
if hours < 0:
return 0.0
return hours
def _transition_deal_status(
writer: Callable[..., dict[str, Any]],
*,
+22 -5
View File
@@ -2775,6 +2775,7 @@ def deal_ack(
work_order_no: str,
next_followup_at: str = "",
lost_reason: str = "",
followup_hours: float | None = None,
) -> str:
"""成交三态回执。后台 deal.recorded_won / lost.ask_reason / negotiating.recorded。"""
no = (work_order_no or "").strip()
@@ -2789,26 +2790,42 @@ def deal_ack(
if reason:
return f"已记录为「未成交」。{lost_reason_ack(reason, work_order_no=no)}"
return _r("lost.ask_reason", "已记录为「未成交」。请补充未成交原因。", ticket_no=no)
when = _followup_month_day(next_followup_at) or "第四个工作日"
from agent.policy.event_exception import format_timeout_hours
# 小时缺省按后台默认 48;0 表示运营关掉再问,话术仍说明已记录协商中
hours_val = 48.0 if followup_hours is None else float(followup_hours)
hours_text = format_timeout_hours(hours_val if hours_val > 0 else 48.0)
when = _followup_when_text(next_followup_at) or f"{hours_text} 个工作小时后"
return _r(
"negotiating.recorded",
f"已记录为「协商中」。已顺延三个工作日,{when}的9:00再次发起成交状态询问。",
f"已记录为「协商中」。已顺延 {hours_text} 个工作小时,将于 {when} 再次发起成交状态询问。",
followup_when=when,
followup_hours=hours_text,
ticket_no=no,
)
def _followup_month_day(raw: str) -> str:
"""把主账再问时刻收成「9月24日」。"""
"""把主账再问时刻收成「9月24日」(旧模板兼容)。"""
when = _followup_when_text(raw)
if "日" in when:
return when.split("日", 1)[0] + "日"
return when
def _followup_when_text(raw: str) -> str:
"""把主账再问时刻收成「9月24日15:30」。"""
from datetime import datetime
text = (raw or "").strip()
if not text:
return ""
for fmt, n in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d", 10)):
for fmt, n in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M", 16), ("%Y-%m-%d", 10)):
try:
dt = datetime.strptime(text[:n], fmt)
return f"{dt.month}月{dt.day}日"
if n <= 10:
return f"{dt.month}月{dt.day}日"
return f"{dt.month}月{dt.day}日{dt.hour}:{dt.minute:02d}"
except ValueError:
continue
return ""
+7 -1
View File
@@ -1305,7 +1305,13 @@ def apply_group_deal(
send_group_text(
client,
chat_id,
copy.deal_ack(outcome, no, str(result.get("next_followup_at") or ""), lost_reason=ack_reason),
copy.deal_ack(
outcome,
no,
str(result.get("next_followup_at") or ""),
lost_reason=ack_reason,
followup_hours=result.get("negotiate_hours"),
),
)
return phase
+5 -4
View File
@@ -75,13 +75,14 @@ class BotReplyRenderTests(unittest.TestCase):
override_catalog(
{
"negotiating.recorded": (
"已记录为「协商中」。已顺延三个工作日,{{followup_when}}的9:00再次发起成交状态询问。"
"已记录为「协商中」。已顺延 {{followup_hours}} 个工作小时,"
"将于 {{followup_when}} 再次发起成交状态询问。"
)
}
)
text = copy.deal_ack("协商中", "WO1", next_followup_at="")
self.assertIn("第四个工作日", text)
self.assertIn("9:00", text)
text = copy.deal_ack("协商中", "WO1", next_followup_at="", followup_hours=48)
self.assertIn("48", text)
self.assertIn("工作小时", text)
def test_group_add_fail_empty_names_reads_admin(self) -> None:
override_catalog(
+12 -5
View File
@@ -10,6 +10,7 @@ from zoneinfo import ZoneInfo
from agent.policy.deal_followup import (
SHANGHAI,
add_workday_hours,
air_route_keys,
is_workday,
leader_matches,
@@ -41,12 +42,18 @@ class TestWorkdays(unittest.TestCase):
self.assertTrue(is_workday(date(2026, 9, 20)))
self.assertFalse(is_workday(date(2026, 10, 7)))
def test_next_followup_is_9am(self) -> None:
when = next_followup_at(datetime(2026, 9, 18, 15, 30, tzinfo=SHANGHAI))
self.assertEqual(when.hour, 9)
self.assertEqual(when.minute, 0)
def test_next_followup_48_work_hours_skips_weekend(self) -> None:
# 周五 15:30 + 48 工作小时 → 跳过周末 → 周二 15:30
when = next_followup_at(
datetime(2026, 11, 13, 15, 30, tzinfo=SHANGHAI), hours=48
)
self.assertEqual(when.tzinfo, SHANGHAI)
self.assertEqual(when.date(), date(2026, 9, 23))
self.assertEqual(when, datetime(2026, 11, 17, 15, 30, tzinfo=SHANGHAI))
def test_add_workday_hours_from_rest_day(self) -> None:
# 周六起点:从下周一 00:00 起算 10 小时 → 周一 10:00
when = add_workday_hours(datetime(2026, 11, 14, 10, 0, tzinfo=SHANGHAI), 10)
self.assertEqual(when, datetime(2026, 11, 16, 10, 0, tzinfo=SHANGHAI))
def test_skip_weekend_keep_weekday_night(self) -> None:
start = datetime(2026, 11, 13, 17, 0, tzinfo=SHANGHAI) # 周五
+26 -2
View File
@@ -277,15 +277,22 @@ class TestDealOutcome(unittest.TestCase):
self.assertNotIn("请补充未成交原因", joined)
def test_negotiate_ack_names_followup_day(self) -> None:
text = copy.deal_ack("协商中", "WO202609190007", "2026-09-24 09:00:00")
text = copy.deal_ack(
"协商中",
"WO202609190007",
"2026-09-24 15:30:00",
followup_hours=48,
)
self.assertEqual(
text,
"已记录为「协商中」。已顺延三个工作日,9月24日的9:00再次发起成交状态询问。",
"已记录为「协商中」。已顺延 48 个工作小时,将于 9月24日15:30 再次发起成交状态询问。",
)
self.assertNotIn("跟进周期已顺延", text)
self.assertNotIn("三个工作日", text)
def test_negotiate_keeps_wait_deal(self) -> None:
self.ledger.start_event_wait(work_order_no=self.wo, wait_kind="deal")
self.ledger._timeouts["deal_negotiate"] = 48
out = finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
@@ -295,6 +302,7 @@ class TestDealOutcome(unittest.TestCase):
business_line="SEA",
)
self.assertEqual(out.get("phase"), "wait_deal")
self.assertEqual(out.get("negotiate_hours"), 48)
ticket = self.ledger.get_ticket(work_order_no=self.wo)
self.assertEqual(ticket.status, "协商中")
self.assertEqual(ticket.event_wait_kind, "")
@@ -313,6 +321,22 @@ class TestDealOutcome(unittest.TestCase):
self.assertEqual(self.ledger.get_ticket(work_order_no=self.wo).status, "协商中")
self.assertTrue(again.get("next_followup_at"))
def test_negotiate_zero_hours_skips_schedule(self) -> None:
"""后台把 deal_negotiate 设为 0:仍记协商中,但不排再问时刻。"""
self.ledger._timeouts["deal_negotiate"] = 0
out = finish_deal_outcome(
ledger=self.ledger,
work_order_no=self.wo,
outcome="协商中",
facts={},
quote={},
business_line="SEA",
)
self.assertEqual(out.get("phase"), "wait_deal")
self.assertEqual(out.get("negotiate_hours"), 0)
self.assertEqual(out.get("next_followup_at"), "")
self.assertEqual(self.ledger.get_ticket(work_order_no=self.wo).next_followup_at, "")
def test_private_next_text_is_lost_reason(self) -> None:
flow = AirTextInquiryFlow(self.ledger)
sess = FlowSession(
@@ -1376,8 +1376,24 @@ class SeaGroupCollabFieldTests(unittest.TestCase):
self.assertEqual(str(rows["空运费"].get("amount") or ""), "199.00")
self.assertEqual(str(rows["空运费"].get("currency") or ""), "CNY")
self.assertEqual(str(rows["空运费"].get("unit_price") or ""), "99.00")
self.assertEqual(str(rows["空运费"].get("quantity") or ""), "100")
self.assertEqual(str(quote.get("total") or ""), "199.0")
def test_air_fee_priced_with_unit_between_name_and_unit_price(self) -> None:
"""WO003:空运费, KG, 单价…数量…金额… 中间夹单位时不能把单价当行金额。"""
from agent.llm.extract_text import extract_product_quote
quote = extract_product_quote(
"空运段: 空运费, KG, 单价CNY99.00, 数量100, 金额CNY9900.00@询价小助手",
business_line="AIR",
)
rows = {str(x.get("name") or ""): x for x in (quote.get("fee_rows") or [])}
self.assertIn("空运费", rows)
self.assertEqual(str(rows["空运费"].get("amount") or ""), "9900.00")
self.assertEqual(str(rows["空运费"].get("unit_price") or ""), "99.00")
self.assertEqual(str(rows["空运费"].get("quantity") or ""), "100")
self.assertEqual(str(quote.get("total") or ""), "9900.0")
def test_fee_line_text_is_a_product_quote(self) -> None:
"""产品按行改价:码头操作费100CNY / 报关费改为300CNY,必须抽出费用名,不能只当海运费合计。"""
from agent.llm.extract_text import extract_product_quote