浅念
322 lines
13 KiB
Python
322 lines
13 KiB
Python
"""询盘推送 UniService 服务(设计 2026-09-28-lead-push-to-uniservice)。
|
||
|
||
三个入口:
|
||
- dispatch_lead_push: 触发点调用(发信分支进入处),写 pending/skipped 日志。
|
||
红线: 整体吞异常,绝不影响邮件主流程。
|
||
- fire_pending_pushes: 主事务 commit 之后调用,create_task 派生后台任务。
|
||
- sweep_stale_pending: lifespan 启动清扫(崩溃遗留的 pending → failed)。
|
||
|
||
后台任务红线(设计 8.1): 不读库只写结果,payload 内存传递;
|
||
结果写入用独立 session(from app.main import async_session,照 chat_rest 模式)。
|
||
"""
|
||
import asyncio
|
||
import logging
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import httpx
|
||
from sqlalchemy import func, select, update
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.engine.config import settings
|
||
from app.main import async_session
|
||
from app.engine.models import (
|
||
Conversation, CustomerProfile, Evaluation, IntentLevel, LeadPushLog,
|
||
Message, MessageRole,
|
||
)
|
||
from app.engine.services.cc_completion import cc_state_for, completion_note, outbound_e164
|
||
from app.engine.services.ip_service import resolve_country
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 对方契约(docs/AI_INQUIRY_PUSH_API.md): phone/whatsapp/email 至少一个有效
|
||
_TRIO = ("phone", "whatsapp", "email")
|
||
_CST = timezone(timedelta(hours=8))
|
||
_INTENT_MAP = {
|
||
"interested": "high",
|
||
"pending": "medium",
|
||
"not_interested": "low",
|
||
}
|
||
|
||
|
||
def _fmt_cst(dt: datetime) -> str:
|
||
"""UTC(-aware 或 naive)→ +08:00 ISO 串。对方明确拒收 UTC(契约 §2)。"""
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
dt = dt.replace(microsecond=0) # 归一整秒: 契约示例为整秒,亚秒精度无益且冒严格解析器拒收风险
|
||
return dt.astimezone(_CST).isoformat()
|
||
|
||
|
||
def map_intent(level: str | None, evaluated: bool) -> str:
|
||
"""我方 IntentLevel → 对方 high/medium/low/unknown。
|
||
|
||
evaluated=False(<3 条客户消息,D2)→ unknown: 未评估不是"待确认",
|
||
与邮件标签 [未评估]/[待确认] 的区分同源。
|
||
"""
|
||
if not evaluated or not level:
|
||
return "unknown"
|
||
return _INTENT_MAP.get(level, "unknown")
|
||
|
||
|
||
_sleep = asyncio.sleep # 模块级绑定,测试 monkeypatch 记录退避序列用
|
||
|
||
|
||
async def _build_payload(
|
||
db: AsyncSession, ev: Evaluation, conv: Conversation,
|
||
profile: CustomerProfile, confirmed: dict, push_type: str,
|
||
evaluation: dict, evaluated: bool,
|
||
) -> dict:
|
||
"""组装契约报文(设计 7.2 映射表)。仅读当前事务,不产生副作用。"""
|
||
# 7.1 event 规则: 仅更正可能 update,且必须同会话此前推送成功过
|
||
# (不依赖对方"update 未知键"的未定义行为)
|
||
event = "new"
|
||
if push_type == "correction":
|
||
prior = await db.execute(
|
||
select(func.count()).select_from(LeadPushLog).where(
|
||
LeadPushLog.conversation_id == conv.id,
|
||
LeadPushLog.status == "success",
|
||
))
|
||
if (prior.scalar() or 0) > 0:
|
||
event = "update"
|
||
|
||
result = await db.execute(
|
||
select(Message)
|
||
.where(Message.conversation_id == conv.id)
|
||
.order_by(Message.created_at)
|
||
)
|
||
msgs = result.scalars().all()
|
||
|
||
raw_info = profile.raw_info or {}
|
||
client_info = conv.client_info or {}
|
||
# Task 6(spec §5 先消费后过滤,顺序钉死): 先读 raw_info 的 _cc 派生——
|
||
# contact 的 phone/whatsapp 换成对应渠道 e164(首封即补全号,确认与否都给;
|
||
# 无 e164 态用原文),标注进"号码补全说明"派生键;之后才做 _ 前缀过滤,
|
||
# _cc 本体绝不进 payload,派生键(无 _ 前缀)存活。canon 不匹配的残留态
|
||
# 在 cc_state_for 内判 None——旧补全永不跟随新号。
|
||
cc_map = raw_info.get("_cc") or {}
|
||
contact_out: dict = {}
|
||
cc_notes: list[str] = []
|
||
for k, v in (confirmed or {}).items():
|
||
if k in ("phone", "whatsapp") and v:
|
||
state = cc_state_for(cc_map, k, str(v))
|
||
contact_out[k] = outbound_e164(str(v), state)
|
||
note = completion_note(str(v), state)
|
||
if note:
|
||
cc_notes.append(note)
|
||
else:
|
||
contact_out[k] = v
|
||
# _ 前缀内部状态(一次性计数等)不进推送画像(spec §4.3——渲染侧
|
||
# 三处有过滤(enrich/reminder/email_service),推送端曾整体拷贝泄漏)
|
||
profile_out = {k: v for k, v in raw_info.items()
|
||
if not str(k).startswith("_")}
|
||
profile_out["country"] = resolve_country(client_info.get("ip", ""))
|
||
profile_out["source"] = conv.source.value
|
||
if client_info.get("lang"):
|
||
profile_out["语言"] = client_info["lang"]
|
||
if cc_notes:
|
||
profile_out["号码补全说明"] = "；".join(cc_notes)
|
||
|
||
reasoning = (evaluation.get("reasoning") or "") if evaluated else ""
|
||
|
||
return {
|
||
"event": event,
|
||
"external_key": f"{settings.PUSH_EXTERNAL_KEY_PREFIX}-{conv.id}",
|
||
"occurred_at": _fmt_cst(datetime.now(timezone.utc)),
|
||
"contact": {
|
||
"name": None, # 我方 contact_info 快照无称呼字段,契约可空
|
||
"phone": contact_out.get("phone"),
|
||
"whatsapp": contact_out.get("whatsapp"),
|
||
"email": contact_out.get("email"),
|
||
"wechat": contact_out.get("wechat"),
|
||
"skype": None,
|
||
},
|
||
"intent_level": map_intent(evaluation.get("intent_level"), evaluated),
|
||
"intent_reason": reasoning or None,
|
||
"profile": profile_out,
|
||
"conversation": [
|
||
{
|
||
"role": "customer" if m.role == MessageRole.user else "ai",
|
||
"text": m.content,
|
||
"at": _fmt_cst(m.created_at),
|
||
}
|
||
for m in msgs if m.role in (MessageRole.user, MessageRole.assistant)
|
||
],
|
||
}
|
||
|
||
|
||
async def dispatch_lead_push(
|
||
db: AsyncSession, ev: Evaluation, conv: Conversation,
|
||
profile: CustomerProfile, confirmed: dict, push_type: str,
|
||
evaluation: dict, evaluated: bool,
|
||
) -> LeadPushLog | None:
|
||
"""触发点唯一入口: 返回待派发的 pending 行(加入当前事务,随主事务提交)。
|
||
|
||
红线(设计 6.1): 整体吞异常——任何失败只记日志,绝不影响邮件主流程;
|
||
触发条件 = 调用方进入发信分支,与 SMTP 成败解耦。
|
||
DB 层失败用 SAVEPOINT 隔离: 只回滚推送写入,不毒化调用方事务
|
||
(否则 flush 失败 → PendingRollbackError → email_sent 落不了库 → 重复发邮件)。
|
||
"""
|
||
if not settings.PUSH_ENABLED:
|
||
return None
|
||
try:
|
||
async with db.begin_nested():
|
||
# 三选一预检(设计 §4 skipped 语义): 对方必拒,不发 HTTP 只留审计痕迹
|
||
if not any(confirmed.get(k) for k in _TRIO):
|
||
db.add(LeadPushLog(
|
||
evaluation_id=ev.id, conversation_id=conv.id,
|
||
push_type=push_type, status="skipped", attempts=0,
|
||
error_message="no valid contact (phone/whatsapp/email)",
|
||
))
|
||
return None
|
||
payload = await _build_payload(
|
||
db, ev, conv, profile, confirmed, push_type, evaluation, evaluated)
|
||
log = LeadPushLog(
|
||
evaluation_id=ev.id, conversation_id=conv.id,
|
||
push_type=push_type, status="pending", attempts=0,
|
||
request_payload=payload,
|
||
)
|
||
db.add(log)
|
||
await db.flush() # 取 id 供后台任务写结果
|
||
return log
|
||
except Exception:
|
||
logger.exception("dispatch_lead_push failed (type=%s, conv=%s)",
|
||
push_type, conv.id)
|
||
return None
|
||
|
||
|
||
async def _do_post(payload: dict) -> tuple[httpx.Response | None, str | None]:
|
||
"""实际 HTTP 调用(独立函数便于测试 monkeypatch)。
|
||
|
||
返回 (resp, transport_error): 传输层异常不抛出,转成错误串由重试矩阵消化。
|
||
"""
|
||
try:
|
||
async with httpx.AsyncClient(
|
||
timeout=settings.PUSH_TIMEOUT_SECONDS,
|
||
follow_redirects=True) as client: # 307/308 不落终态失败
|
||
resp = await client.post(
|
||
settings.PUSH_API_URL,
|
||
json={"messages": [payload]}, # 契约 §2: 批量结构,我方单条
|
||
headers={"X-Api-Key": settings.PUSH_API_KEY},
|
||
)
|
||
return resp, None
|
||
except httpx.HTTPError as exc:
|
||
return None, repr(exc)
|
||
|
||
|
||
def _retryable(resp: httpx.Response | None) -> bool:
|
||
"""契约 §3: 5xx/超时可重试;401/400/其他 4xx 重试无益。200 不在此判定。"""
|
||
return resp is None or resp.status_code >= 500
|
||
|
||
|
||
async def _update_log(push_log_id: str, **fields) -> None:
|
||
"""后台任务写结果——独立 session + 小事务(设计 6.3)。"""
|
||
async with async_session() as session:
|
||
await session.execute(
|
||
update(LeadPushLog)
|
||
.where(LeadPushLog.id == push_log_id)
|
||
.values(**fields)
|
||
)
|
||
await session.commit()
|
||
|
||
|
||
async def _run_push(push_log_id: str, payload: dict) -> None:
|
||
"""后台任务体: 分类重试 + 200 逐条结果解析(设计 6.3/§10)。
|
||
|
||
红线: 全套 try/except,异常只记日志绝不逃逸(fire-and-forget)。
|
||
"""
|
||
try:
|
||
attempts = 0
|
||
while True:
|
||
attempts += 1
|
||
resp, transport_error = await _do_post(payload)
|
||
if _retryable(resp) and attempts < settings.PUSH_MAX_ATTEMPTS:
|
||
await _update_log(
|
||
push_log_id, attempts=attempts,
|
||
error_message=transport_error
|
||
or (f"http {resp.status_code}" if resp is not None else "unknown"),
|
||
)
|
||
await _sleep(
|
||
settings.PUSH_RETRY_BACKOFF_BASE * 2 ** (attempts - 1))
|
||
continue
|
||
break
|
||
|
||
if resp is not None and resp.status_code == 200:
|
||
try:
|
||
raw = resp.json()
|
||
except ValueError:
|
||
# 非 JSON 的 200(如网关重定向到 HTML 登录页)——
|
||
# 无法确认对方已接受,诚实落 failed(幂等键保证补推安全)
|
||
await _update_log(push_log_id, attempts=attempts, status="failed",
|
||
response_status=200,
|
||
error_message="non-JSON 200 body")
|
||
return
|
||
data = raw.get("data") if isinstance(raw, dict) else None
|
||
data = data if isinstance(data, dict) else {}
|
||
failed_mine = [
|
||
f for f in (data.get("failed") or [])
|
||
if f.get("external_key") == payload["external_key"]
|
||
]
|
||
if failed_mine:
|
||
await _update_log(
|
||
push_log_id, attempts=attempts, status="failed",
|
||
response_status=200,
|
||
error_message=str(failed_mine[0].get("reason") or "rejected"),
|
||
)
|
||
else:
|
||
# 单条推送: accepted/duplicated 计数即本条结论
|
||
outcome = ("duplicated"
|
||
if (data.get("duplicated") or 0) > 0 else "accepted")
|
||
await _update_log(
|
||
push_log_id, attempts=attempts, status="success",
|
||
response_status=200, outcome=outcome,
|
||
)
|
||
return
|
||
|
||
# 终态失败: 重试耗尽(5xx/传输)或不可重试(401/400/其他)
|
||
if resp is not None and resp.status_code in (400, 401):
|
||
logger.error("lead push rejected (log_id=%s, http %s) — "
|
||
"密钥/报文问题需人工介入,不重试",
|
||
push_log_id, resp.status_code)
|
||
await _update_log(
|
||
push_log_id, attempts=attempts, status="failed",
|
||
response_status=resp.status_code if resp is not None else None,
|
||
error_message=transport_error
|
||
or (f"http {resp.status_code}" if resp is not None else "unknown"),
|
||
)
|
||
except Exception:
|
||
logger.exception("lead push task crashed (log_id=%s)", push_log_id)
|
||
|
||
|
||
# 后台任务引用集: 防 GC(asyncio 官方要求),done 即移除
|
||
_tasks: set[asyncio.Task] = set()
|
||
|
||
|
||
def fire_pending_pushes(pushes: list[LeadPushLog]) -> None:
|
||
"""主事务 commit 之后调用(设计 8.1): payload 内存传递,任务不读库。
|
||
|
||
同步函数、纳秒级返回——这就是"不阻塞"的派发点。
|
||
"""
|
||
for log in pushes:
|
||
if log.status != "pending":
|
||
continue
|
||
task = asyncio.create_task(_run_push(log.id, log.request_payload))
|
||
_tasks.add(task)
|
||
task.add_done_callback(_tasks.discard)
|
||
|
||
|
||
async def sweep_stale_pending(stale_minutes: int = 30) -> int:
|
||
"""启动清扫(设计 8.2): 崩溃遗留的 pending → failed。
|
||
|
||
已知限制(设计 §14): 推送成功但结果写回失败也会被标 failed——
|
||
幂等键保证将来补推安全(对方返回 duplicated),概率极低,接受。
|
||
"""
|
||
cutoff = datetime.now(timezone.utc) - timedelta(minutes=stale_minutes)
|
||
async with async_session() as session:
|
||
result = await session.execute(
|
||
update(LeadPushLog).where(
|
||
LeadPushLog.status == "pending",
|
||
LeadPushLog.updated_at < cutoff,
|
||
).values(status="failed", error_message="interrupted by restart")
|
||
)
|
||
await session.commit()
|
||
return result.rowcount or 0
|