Files

T

浅念 49c0e5559c chore: 从汽车AI客服项目拷贝骨架并清理(blog页/redis)

Co-Authored-By: Claude Code <noreply@anthropic.com>

2026-09-29 14:15:12 +08:00

70 KiB

Raw Blame History

联系方式驱动的询盘邮件（Contact-Driven Lead Email）实施计划 v1.1

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 联系方式成为发询盘邮件的唯一条件——统一管道检测到联系方式立刻发邮件（不看消息数、不看意向评级）；无联系方式永不发邮件、永不结束会话；客户已给的联系方式绝不丢失、绝不重复索要。

Architecture: 新建确定性"统一联系方式管道"（app/services/contact_pipeline.py）：extractor 提取（多语言 LLM 源）+ 正则高置信扫描（确定性辅源）→ key 归一化 → 深合并（同 key 新值覆盖）→ 格式校验（confirmed/suspect 分级）。管道结果作为唯一事实源，同时供邮件触发器（去重 + 更正 + 24h 跨会话去重）、定向提醒（禁止重复索要）、会话结束守卫（全来源）三处消费。邮件触发从路由层（ended && interested）整体迁移到 ConversationManager 内部，与结束逻辑彻底解耦。

Tech Stack: Python 3.13 / FastAPI / SQLAlchemy 2.0 async + SQLite(aiosqlite) / pytest(asyncio_mode=auto)

1. 背景与问题（为什么做）

线上痛点：大量无联系方式的询盘邮件推送至销售邮箱（无效询盘 + 有意向但无法联系的客户）。根因链（已代码验证）：

widget/embedded/standalone 来源的 should_end 没有联系方式守卫（_has_real_contact 仅 app 生效，conversation.py:230）；
evaluator.md 要求"无联系方式 info_sufficient 必须 false"，但 GLM 不遵守（conversation.py:253 注释自证）；
判定 interested 即结束会话锁死客户（conversation.py:350），连补联系方式的机会都没有；
extractor 某轮返回空 联系方式 时浅合并整层覆盖，已提取的联系方式被静默抹掉（conversation.py:119）。

2. 已定决策（本计划的 spec，评审以此为准）

#	决策	内容
D1	邮件唯一条件	管道 confirmed 联系方式 → 立即发邮件。≥3轮、interested、info_sufficient 一律不参与邮件门槛。第 1 条消息含联系方式也发
D2	意向降级为标签	邮件主题打标签：interested→[意向强]、pending→[待确认]、not_interested→[无意向]、未评估(<3轮未跑评估)→[未评估]。评估被跳过时邮件 intent_level 参数传空串（→[未评估]），但发给客户端的 evaluated 事件仍报 `pending`（客户端语义不变）——两者不得混用
D3	会话永不自动结束（2026-08-18 决策 A，取代原"全来源守卫"）	邮件与结束解耦后，自动结束只剩收尾UX/停止计费两个弱价值，却会把还没问完的客户锁死（刷新丢上下文+销售响应空窗）。任何情况都不自动结束：会话保持 active 直到客户自然沉默，关闭交给超时清理（健康轨）。evaluated 事件 `ended` 恒 False（App 端需知晓）；`_finalize` 删除，D10 的 upsert 语义由"每轮评估落库"承担
D4	防重发	同会话 email_sent 标记，全量只发一次；联系方式变化(改号/新增key/无提示换值)→轻量[更正]邮件，每会话≤3封（2026-08-18 用户测试反馈放宽）；达到上限后"锁邮件不锁快照"——Evaluation.contact_info 仍刷新为最新（admin 永远看到客户最新号码）
D5	24h 跨会话去重	同一联系方式值(规范化比对)24h 内已发过全量 → 新会话只发一行"该客户又来聊了"轻提醒
D6	不重复索要	管道 confirmed → 系统提醒注入"禁止再索要，确认收到后继续解答"；suspect(校验不过) → "礼貌确认，禁止重新索要"。suspect 不触发邮件、不参与结束
D7	评估前置	用户消息 <3 条跳过 GLM evaluate 调用（结果必然 pending），纯省成本；邮件标签用[未评估]
D8	卡单标记	interested + 无联系方式 → upsert Evaluation（admin 后续可筛，人工兜底的数据基础）
D9	旧触发路径拆除	`chat.py:94-97` 与 `chat_rest.py:140-142` 的 ended&&interested→`_try_send_email` 整体删除，与 D1 触发器同一个 PR（两套并存=双发邮件）。批次 1 全部提交在同一分支、合并后一次部署——严禁部署分支中间提交：Task 6（新触发器生效）与 Task 7（旧路径拆除）之间存在新旧双触发窗口，该窗口只允许存在于开发分支上
D10	Evaluation upsert 语义	`_get_or_init_evaluation` 是唯一建行入口（绝不 INSERT 已存在记录，唯一约束会 IntegrityError）；每轮评估落库只更新 intent/reasoning/快照，永不触碰 email_sent/correction_sent（原 `_finalize` 已随决策 A 删除）
D11	话术（system.md）	①价值交换固定模板：只承诺"完整报价单"，禁止自由发挥价格/时限；②先聊后引：聊天内要两次未果才引导点 WhatsApp 按钮；③拿到联系方式后确认一句、继续解答，不急着收尾
D12	邮件转义（S2 提前）	email_service 全量 `html.escape`（邮件触发频率上升，注入面同步放大，不等健康轨）
D13	行为变化已批准	本计划推翻"不改 widget 行为"红线（2026-08-17 产品决策）。上线时同步清理 `conversation.py:131-134、226-229` 的 app-only 注释与 `chat_rest.py` 模块 docstring 第 12-13 行

多值/晋升细则（S1/S3/S4 钉死）：

S1 suspect 只用于"防重复索要"提醒，永不自动晋升；客户后续重发或改发新值时由 extractor/正则重新捕获、重新校验（新值覆盖）。
S3 24h 去重多值规则：confirmed 中任一值在 24h 内已发过 → 轻提醒（新值会显示在提醒主题里，信息不丢；全部值都未见 → 全量）。
S4 正则每轮扫描全部客户消息历史（非仅当轮），兜住 GLM 失败轮次漏掉的联系方式。

明确不做（防扩散）： 范式全量重构、超时清理、UI 留资卡片、WS 限流、会话持久化、admin 筛选视图 → 批次 2/3 或健康轨（见 §6）。process_message（conversation.py:70，grep 确认零调用方的死代码）在 Task 7 随守卫改造一并删除（若保留，其对 _has_real_contact 的调用会成为悬空引用）；Task 7 之前不碰它。

3. 文件结构

文件	动作	职责
`app/services/contact_pipeline.py`	新建	纯函数管道：归一化/深合并/校验分级/正则扫描/规范化比对
`app/services/conversation.py`	修改	接线管道、邮件触发器、全来源结束守卫、卡单标记、评估前置、提醒注入、finalize UPDATE、深合并修复
`app/services/email_service.py`	修改	转义、标签、中文标签列、轻提醒/更正邮件、"对话进行中"提示
`app/models/database.py`	修改	Evaluation 加 `correction_sent` 列
`app/main.py`	修改	`_migrate` 加 correction_sent 的 PRAGMA 迁移
`app/api/routes/chat.py`	修改	删除 `_try_send_email` 与触发块
`app/api/routes/chat_rest.py`	修改	删除触发块与 lazy import，更新 docstring
`app/prompts/system.md`	修改	D11 话术
`scripts/baseline_stats.py`	新建	批次 0.5 基线统计（只读）
`tests/test_contact_pipeline.py`	新建	管道单元测试
`tests/test_lead_email_trigger.py`	新建	触发器/结束守卫/卡单集成测试（FakeAI + async_db）
`tests/test_chat_rest_message.py`	修改	删除钉死旧触发路径的 3 个测试（被新测试取代）

⚠️ 行号锚点使用规则：文中行号是计划撰写时（当前 HEAD）的定位辅助。Task 5/6 会向 conversation.py 插入代码，后续任务（Task 7/8）引用的行号会漂移约 20 行——实施时一律以任务中引用的代码内容定位（如 should_end、_has_real_contact 方法名），禁止盲改行号区域。

批次 0.5：基线统计

Task 0: 基线脚本

Files:

Create: scripts/baseline_stats.py
Step 0.1: 写脚本（只读，不写库）

"""批次0.5 基线统计：上线前存档，上线后同口径对比。只读。"""
import re
import sqlite3
import sys

DB = sys.argv[1] if len(sys.argv) > 1 else "./data/cc_ai_kefu.db"

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
INTL_RE = re.compile(r"\+\d{6,15}")


def main() -> None:
    conn = sqlite3.connect(DB)
    cur = conn.cursor()

    # 1) 历史已发邮件中无联系方式占比（按 source）
    rows = cur.execute(
        """SELECT c.source,
                  COUNT(*) AS sent,
                  SUM(CASE WHEN COALESCE(e.contact_info, '{}') IN ('{}', 'null', '')
                           OR e.contact_info IS NULL THEN 1 ELSE 0 END) AS no_contact
           FROM evaluations e JOIN conversations c ON c.id = e.conversation_id
           WHERE e.email_sent = 1 GROUP BY c.source"""
    ).fetchall()
    print("== 已发邮件联系方式缺失（按来源）==")
    for source, sent, no_contact in rows:
        pct = (no_contact / sent * 100) if sent else 0
        print(f"  {source}: 发送 {sent}, 缺失 {no_contact} ({pct:.0f}%)")

    # 2) 用户消息中含联系方式样式的会话占比（新策略邮件量上界预估）
    convs = cur.execute("SELECT id FROM conversations").fetchall()
    with_contact = early_contact = total = 0
    for (cid,) in convs:
        msgs = cur.execute(
            "SELECT role, content FROM messages WHERE conversation_id=? ORDER BY created_at",
            (cid,),
        ).fetchall()
        user_msgs = [m for m in msgs if m[0] == "user"]
        if not user_msgs:
            continue
        total += 1
        hit_idx = next(
            (i for i, m in enumerate(user_msgs)
             if EMAIL_RE.search(m[1]) or INTL_RE.search(m[1])),
            None,
        )
        if hit_idx is not None:
            with_contact += 1
            if hit_idx < 2:
                early_contact += 1
    print("== 新策略邮件量预估 ==")
    print(f"  有用户消息的会话: {total}")
    print(f"  消息中含联系方式样式: {with_contact}"
          f" ({with_contact / total * 100 if total else 0:.0f}%) ← 全量邮件上界")
    print(f"  联系方式出现在前2条消息内: {early_contact}"
          f" ({early_contact / total * 100 if total else 0:.0f}%) ← 即时触发增量")

    conn.close()


if __name__ == "__main__":
    main()

Step 0.2: 运行并存档

Run: python scripts/baseline_stats.py | tee docs/superpowers/plans/2026-08-17-baseline.txt Expected: 打印三组数字（存档文件用于上线后对比）

Step 0.3: Commit

git add scripts/baseline_stats.py docs/superpowers/plans/2026-08-17-baseline.txt
git commit -m "chore(baseline): 询盘邮件改造前基线统计"

批次 1：核心实现

Task 1: contact_pipeline — 归一化 / 深合并 / 校验分级

Files:

Create: app/services/contact_pipeline.py
Test: tests/test_contact_pipeline.py
Step 1.1: 写失败测试

from app.services.contact_pipeline import (
    merge_contacts, normalize_contacts, split_validate,
)


def test_normalize_chinese_and_english_keys():
    raw = {"手机": "+8613800138000", "邮箱": "a@b.com", "微信": "wxid_1",
           "WhatsApp号码": "+254712345678", "空": ""}
    out = normalize_contacts(raw)
    assert out == {
        "phone": "+8613800138000", "email": "a@b.com",
        "wechat": "wxid_1", "whatsapp": "+254712345678",
    }


def test_normalize_later_alias_wins_on_same_key():
    # 同 key 别名冲突时后写者覆盖(上游 extractor 单轮内一般不会同时给 手机 与 tel,
    # 但语义必须被钉死并被测试)
    assert normalize_contacts({"手机": "1111111", "tel": "2222222"}) == {"phone": "2222222"}

def test_merge_new_value_wins_same_key():
    old = {"phone": "+254700000000"}
    new = {"手机": "+254711111111"}  # 客户改号
    assert merge_contacts(old, new) == {"phone": "+254711111111"}


def test_merge_empty_new_keeps_old():
    # T1 场景: extractor 某轮返回空联系方式,不得抹掉已有值
    old = {"phone": "+254700000000"}
    assert merge_contacts(old, {}) == {"phone": "+254700000000"}
    assert merge_contacts(old, None) == {"phone": "+254700000000"}


def test_split_validate_confirmed_and_suspect():
    confirmed, suspect = split_validate({
        "email": "a@b.com",          # 合法邮箱
        "phone": "+254 712 345 678", # 数字 12 位 → confirmed
        "whatsapp": "12345",         # 数字 5 位 → suspect
    })
    assert confirmed == {"email": "a@b.com", "phone": "+254 712 345 678"}
    assert suspect == {"whatsapp": "12345"}


def test_split_validate_bad_email_is_suspect():
    confirmed, suspect = split_validate({"email": "not-an-email"})
    assert confirmed == {} and suspect == {"email": "not-an-email"}


def test_split_validate_wechat_id_is_confirmed():
    # 微信号无数字格式(wxid_x 等),非空≥4字符即 confirmed——否则只留微信的
    # 客户永远无法触发邮件
    confirmed, suspect = split_validate({"wechat": "wxid_1"})
    assert confirmed == {"wechat": "wxid_1"} and suspect == {}


def test_split_validate_wechat_short_is_suspect():
    confirmed, suspect = split_validate({"wechat": "abc"})
    assert confirmed == {} and suspect == {"wechat": "abc"}


def test_normalize_coerces_int_values():
    # extractor 偶尔会把号码输出成 int,不得崩溃且要能归一化
    assert normalize_contacts({"手机": 254712345678}) == {"phone": "254712345678"}


def test_merge_none_old_keeps_new():
    assert merge_contacts(None, {"email": "a@b.com"}) == {"email": "a@b.com"}

Step 1.2: 运行确认失败

Run: pytest tests/test_contact_pipeline.py -v Expected: FAIL（ModuleNotFoundError）

Step 1.3: 实现 contact_pipeline.py（归一化/合并/校验部分）

"""统一联系方式管道：联系方式检测的唯一事实源。

三个动作：
- 归一化：中英文 key 别名统一为 phone/whatsapp/email/wechat
- 深合并：同 key 新值覆盖旧值（客户改号场景），空值不覆盖
- 校验分级：confirmed（格式通过，可触发邮件/结束）/ suspect（疑似，
  仅用于提醒 AI 礼貌确认，防止重复索要——D6/S1）

上游约定：raw_info 中 "_" 前缀 key 为内部状态，禁止进入提示词渲染。
"""
import re

KEY_ALIASES = {
    "phone": "phone", "手机": "phone", "电话": "phone", "电话号码": "phone",
    "tel": "phone", "telephone": "phone", "phone_number": "phone",
    "phone number": "phone", "mobile": "phone", "mobile_number": "phone",
    "mobile number": "phone",
    "whatsapp": "whatsapp", "wa": "whatsapp", "whats app": "whatsapp",
    "whatsapp号码": "whatsapp", "whatsapp number": "whatsapp",
    "email": "email", "邮箱": "email", "e-mail": "email", "mail": "email",
    "电子邮件": "email",
    "wechat": "wechat", "微信": "wechat", "wx": "wechat", "weixin": "wechat",
}


def normalize_contacts(raw: dict | None) -> dict:
    """key 归一化 + 去空值。同 key 别名冲突时后写者覆盖。"""
    out: dict[str, str] = {}
    for k, v in (raw or {}).items():
        key = KEY_ALIASES.get(str(k).strip().lower())
        val = str(v or "").strip()
        if key and val:
            out[key] = val
    return out


def merge_contacts(old: dict | None, new: dict | None) -> dict:
    """同 key 新值覆盖旧值；空/缺省不覆盖（堵浅合并抹数据 bug）。"""
    return {**normalize_contacts(old), **normalize_contacts(new)}


def digits_only(s: str) -> str:
    """提取纯数字（split_validate 位数判断与后续 canon 复用）。"""
    return re.sub(r"\D", "", s)


def split_validate(contacts: dict) -> tuple[dict, dict]:
    """返回 (confirmed, suspect)。电话/WhatsApp 取数字 7-15 位为合格；
    微信号无数字格式，取非空 ≥4 字符为合格（否则 wxid_x 类真实微信号
    会永远滞留 suspect，只留微信的客户无法触发邮件——违背 D1）。"""
    confirmed: dict[str, str] = {}
    suspect: dict[str, str] = {}
    for k, v in normalize_contacts(contacts).items():
        if k == "email":
            ok = re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", v) is not None
        elif k == "wechat":
            ok = len(v.strip()) >= 4
        else:
            ok = 7 <= len(digits_only(v)) <= 15
        (confirmed if ok else suspect)[k] = v
    return confirmed, suspect

（canon() 的定义已移至 Task 6——它只被去重逻辑消费，与本任务的纯函数无关。）

Step 1.4: 运行确认通过

Run: pytest tests/test_contact_pipeline.py -v Expected: 7 PASS（6 项基础用例 + wechat 修正用例）

Step 1.5: Commit

git add app/services/contact_pipeline.py tests/test_contact_pipeline.py
git commit -m "feat(pipeline): 联系方式归一化/深合并/校验分级"

Task 2: contact_pipeline — 正则扫描（高置信，S4）

Files:

Modify: app/services/contact_pipeline.py
Test: tests/test_contact_pipeline.py
Step 2.1: 写失败测试（含误报反例——二手车语境全是数字）

from app.services.contact_pipeline import scan_user_messages


def m(*texts):
    return [{"role": "user", "content": t} for t in texts] + [
        {"role": "assistant", "content": "reply"}]


def test_scan_email():
    found = scan_user_messages(m("my email is john.doe@x.co send quotes"))
    assert found == {"email": "john.doe@x.co"}


def test_scan_intl_phone():
    found = scan_user_messages(m("you can reach me +254712345678"))
    assert found == {"phone": "+254712345678"}


def test_scan_whatsapp_context_word():
    found = scan_user_messages(m("whatsapp: 0712 345 678"))
    assert found.get("whatsapp") == "0712 345 678"


def test_scan_wa_me_link():
    found = scan_user_messages(m("chat https://wa.me/254712345678"))
    assert found == {"whatsapp": "254712345678"}


def test_scan_ignores_prices_years_and_ai_messages():
    texts = m(
        "budget $25,000 for 2024 BYD Tang",   # 价格+年份
        "50,000 km mileage",                   # 里程
        "I want 5 units",                      # 数量
    )
    # 上一条含 assistant 消息,确保只扫 user
    found = scan_user_messages(texts)
    assert found == {}


def test_scan_full_history_not_only_last():
    # S4: 扫全量历史,早轮的联系方式也要能捞到
    msgs = [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hey"},
        {"role": "user", "content": "mail me a@b.com"},
        {"role": "assistant", "content": "got it"},
        {"role": "user", "content": "how much is tang"},
    ]
    assert scan_user_messages(msgs) == {"email": "a@b.com"}


def test_scan_context_word_does_not_grab_unrelated_numbers():
    # S1: 上下文词必须与数字邻接(同一匹配内),消息任意位置的7位数字不得被抓
    assert scan_user_messages(m("call me on delivery, order 1234567")) == {}
    assert scan_user_messages(m("车架号码是 LFV3A23C8K3123456 吗")) == {}
    assert scan_user_messages(m("engine number 2ZRFE1234567?")) == {}
    assert scan_user_messages(m("我手机坏了 budget 1500000 ksh")) == {}


def test_scan_adjacent_context_phone():
    assert scan_user_messages(m("my phone number is 0712 345 678")) == {
        "phone": "0712 345 678"}
    assert scan_user_messages(m("call me at +86 138 0013 8000")) == {
        "phone": "+86 138 0013 8000"}

Step 2.2: 运行确认失败

Run: pytest tests/test_contact_pipeline.py -v Expected: 新增用例 FAIL（scan_user_messages 未定义）

Step 2.3: 实现扫描器（追加到 contact_pipeline.py）

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
INTL_PHONE_RE = re.compile(r"\+\d{6,15}")
# 邻接绑定：上下文词与数字必须在同一匹配内（仅空白衔接）——禁止"消息任意位置
# 有上下文词 + 任意位置有 7 位数字"的消息级误绑（S1：车架号/订单号/先令价格）
WHATSAPP_CTX_RE = re.compile(
    r"(?:whats\s*app|wa\.?me)\D{0,12}(\+?\d[\d\s\-().]{5,17}\d)",
    re.IGNORECASE,
)
PHONE_CTX_RE = re.compile(
    r"(?:my\s+phone\s+(?:number\s+)?is|my\s+number\s+is|"
    r"call\s+me\s+(?:at|on)|reach\s+me\s+(?:at|on)|"
    r"phone\s*[:：]|number\s*[:：]|号码\s*[:：]|手机号?\s*[:：]|电话\s*[:：])\s*"
    r"(\+?\d[\d\s\-().]{5,17}\d)",
    re.IGNORECASE,
)


def scan_user_messages(messages: list[dict]) -> dict:
    """S4: 扫描全部客户消息,返回高置信捕获(email / +国际号 / whatsapp·wa.me /
    显式电话上下文邻接数字)。裸数字(价格/年份/里程/车架号)一律不收。"""
    found: dict[str, str] = {}
    for msg in messages:
        if msg.get("role") != "user":
            continue
        text = msg.get("content", "")
        if "email" not in found:
            hit = EMAIL_RE.search(text)
            if hit:
                found["email"] = hit.group(0)
        if "phone" not in found:
            hit = INTL_PHONE_RE.search(text)
            if hit:
                found["phone"] = hit.group(0)
        if "whatsapp" not in found:
            hit = WHATSAPP_CTX_RE.search(text)
            if hit:
                found["whatsapp"] = hit.group(1)
        if "phone" not in found:
            hit = PHONE_CTX_RE.search(text)
            if hit:
                found["phone"] = hit.group(1)
    return found

Step 2.4: 运行确认通过

Run: pytest tests/test_contact_pipeline.py -v Expected: 19 PASS（11 存量 + 8 本任务，含 S1 邻接反例与邻接正向用例）

Step 2.5: Commit

git add app/services/contact_pipeline.py tests/test_contact_pipeline.py
git commit -m "feat(pipeline): 高置信正则扫描(仅客户消息/全量历史/价格年份不误捕)"

Task 3: Evaluation 加 correction_sent 列 + 迁移（D4）

Files:

Modify: app/models/database.py:109-126
Modify: app/main.py:16-22
Step 3.1: database.py 的 Evaluation 类加列（email_sent 行后）

    correction_sent: Mapped[bool] = mapped_column(Boolean, default=False)

Step 3.2: main.py 的 _migrate 追加迁移（page_token 块之后）

    result = await conn.execute(text("PRAGMA table_info(evaluations)"))
    columns = {row[1] for row in result}
    if "correction_sent" not in columns:
        await conn.execute(text(
            "ALTER TABLE evaluations ADD COLUMN correction_sent BOOLEAN DEFAULT 0"
        ))

Step 3.3: 验证未破坏

Run: pytest tests/test_smoke.py -v Expected: PASS（create_all 在全新库上含新列；PRAGMA ALTER 分支与既有 page_token 迁移一样无测试覆盖，属既有模式，接受——不要把本步当作 ALTER 分支的验证）

Step 3.4: Commit

git add app/models/database.py app/main.py
git commit -m "feat(db): evaluations.correction_sent 列 + PRAGMA 迁移"

Task 4: email_service — 转义 / 标签 / 轻提醒 / 更正邮件（D2/D4/D12）

Files:

Modify: app/services/email_service.py
Test: tests/test_lead_email_trigger.py（本 Task 只改实现，集成验证在 Task 7）
Step 4.0: 写失败测试（先红）

Create tests/test_email_rendering.py（monkeypatch 模块内 _send，不触网；断言主题标签、HTML 转义、内部 key 过滤）:

"""邮件渲染单测: 标签主题/HTML转义/内部key过滤(D2/D12)。不触网。"""
import asyncio

from app.services import email_service


def _capture_send(monkeypatch):
    captured = {}

    async def fake_send(msg):
        captured["msg"] = msg
        return True

    monkeypatch.setattr(email_service, "_send", fake_send)
    return captured


def _html(msg) -> str:
    return msg.get_payload()[0].get_payload(decode=True).decode("utf-8")


def test_lead_email_subject_tag_and_escaped_body(monkeypatch):
    captured = _capture_send(monkeypatch)
    ok = asyncio.run(email_service.send_lead_email(
        customer_profile={"目标国家": "肯尼亚", "_internal": "x"},
        contact_info={"phone": "+254712345678"},
        reasoning="<b>强意向</b>",
        conversation_messages=[
            {"role": "user", "content": "<script>alert(1)</script>"},
            {"role": "assistant", "content": "ok"},
        ],
        intent_level="interested",
    ))
    assert ok is True
    subject = str(captured["msg"]["Subject"])
    assert "[新询盘·意向强]" in subject and "肯尼亚" in subject
    body = _html(captured["msg"])
    assert "<script>" not in body          # D12: 转义生效
    assert "&lt;script&gt;" in body
    assert "_internal" not in body         # 内部 key 不渲染


def test_lead_email_unevaluated_tag(monkeypatch):
    captured = _capture_send(monkeypatch)
    asyncio.run(email_service.send_lead_email(
        customer_profile={}, contact_info={"email": "a@b.com"},
        reasoning="", conversation_messages=[], intent_level="",
    ))
    assert "[新询盘·未评估]" in str(captured["msg"]["Subject"])


def test_returning_and_correction_emails(monkeypatch):
    captured = _capture_send(monkeypatch)
    assert asyncio.run(
        email_service.send_returning_lead_email({"whatsapp": "+254700000000"})
    ) is True
    assert "[线索回访]" in str(captured["msg"]["Subject"])
    assert asyncio.run(
        email_service.send_correction_email({"email": "a@b.com"})
    ) is True
    assert "[询盘·更正]" in str(captured["msg"]["Subject"])

Step 4.1: 重写 email_service.py

import html
import logging
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

import aiosmtplib

from app.config import settings

logger = logging.getLogger(__name__)

TAG_LABELS = {
    "interested": "意向强", "pending": "待确认",
    "not_interested": "无意向", "": "未评估",
}
CONTACT_LABELS = {"phone": "手机", "whatsapp": "WhatsApp", "email": "邮箱", "wechat": "微信"}


def _esc(v) -> str:
    return html.escape(str(v)) if v else ""


def _contact_rows(contacts: dict) -> str:
    return "".join(
        f"<tr><td style='padding:6px 12px;font-weight:bold;'>{html.escape(CONTACT_LABELS.get(k, k))}</td>"
        f"<td style='padding:6px 12px;'>{_esc(v)}</td></tr>"
        for k, v in (contacts or {}).items() if v
    )


async def _send(msg: MIMEMultipart) -> bool:
    if not all([settings.SMTP_HOST, settings.SMTP_USER, settings.NOTIFY_EMAIL]):
        logger.warning("SMTP not configured, skipping email")
        return False
    try:
        await aiosmtplib.send(
            msg, hostname=settings.SMTP_HOST, port=settings.SMTP_PORT,
            username=settings.SMTP_USER, password=settings.SMTP_PASS,
            use_tls=True,
        )
        logger.info("Lead email sent to %s", settings.NOTIFY_EMAIL)
        return True
    except Exception:
        logger.exception("Failed to send lead email")
        return False


async def send_lead_email(
    customer_profile: dict,
    contact_info: dict,
    reasoning: str,
    conversation_messages: list[dict],
    intent_level: str = "",
) -> bool:
    tag = TAG_LABELS.get(intent_level, "未评估")
    contacts = dict(contact_info or {})
    country = (customer_profile or {}).get("目标国家", "")
    if not country:
        country = (customer_profile or {}).get("target_country", "未知")

    msg = MIMEMultipart("alternative")
    contact_parts = [
        f"{CONTACT_LABELS.get(k, k)}: {v}" for k, v in contacts.items() if v
    ]
    subject = f"[新询盘·{tag}] {country}"
    if contact_parts:
        subject += " | " + ", ".join(contact_parts)
    msg["Subject"] = subject
    msg["From"] = f"【XYH】 <{settings.SMTP_USER}>"
    msg["To"] = settings.NOTIFY_EMAIL

    info_rows = "".join(
        f"<tr><td style='padding:6px 12px;font-weight:bold;'>{_esc(k)}</td>"
        f"<td style='padding:6px 12px;'>{_esc(v)}</td></tr>"
        for k, v in (customer_profile or {}).items() if not str(k).startswith("_")
    )
    chat_lines = "".join(
        f"<div style='margin:4px 0;'><b>{'客户' if m.get('role') == 'user' else 'AI'}:</b> {_esc(m.get('content', ''))}</div>"
        for m in conversation_messages or []
    )

    html_body = f"""
    <html><body style="font-family:sans-serif;color:#333;">
      <h2>🎉 新询盘（{html.escape(tag)}）</h2>
      <h3>联系方式</h3>
      <table style="border-collapse:collapse;border:1px solid #ddd;">{_contact_rows(contacts)}</table>
      <h3>客户画像</h3>
      <table style="border-collapse:collapse;border:1px solid #ddd;">{info_rows}</table>
      <h3>评估理由</h3><p>{_esc(reasoning)}</p>
      <h3>完整对话记录</h3>
      <div style="background:#f5f5f5;padding:12px;border-radius:6px;">{chat_lines}</div>
      <p style="color:#999;font-size:12px;">注：邮件为触发时快照，对话可能仍在进行，最新记录请以后台为准。</p>
    </body></html>
    """
    msg.attach(MIMEText(html_body, "html", "utf-8"))
    return await _send(msg)


async def send_returning_lead_email(contact_info: dict) -> bool:
    """D5: 24h 内同联系方式再访,只发一行轻提醒。"""
    contact_parts = [
        f"{CONTACT_LABELS.get(k, k)}: {v}" for k, v in (contact_info or {}).items() if v
    ]
    msg = MIMEMultipart("alternative")
    msg["Subject"] = f"[线索回访] {' | '.join(contact_parts) or '已知客户'}再次来访"
    msg["From"] = f"【XYH】 <{settings.SMTP_USER}>"
    msg["To"] = settings.NOTIFY_EMAIL
    msg.attach(MIMEText(
        f"<html><body><p>该联系方式 24h 内已发过完整询盘，客户又开启了新对话，"
        f"详情请查后台。</p></body></html>", "html", "utf-8"))
    return await _send(msg)


async def send_correction_email(contact_info: dict) -> bool:
    """D4: 联系方式变化(改号/新增)后的轻量更正邮件。"""
    msg = MIMEMultipart("alternative")
    msg["Subject"] = "[询盘·更正] 客户联系方式已更新"
    msg["From"] = f"【XYH】 <{settings.SMTP_USER}>"
    msg["To"] = settings.NOTIFY_EMAIL
    msg.attach(MIMEText(
        f"<html><body><p>客户更新了联系方式，请以本封为准：</p>"
        f"<table>{_contact_rows(contact_info)}</table>"
        f"<p style='color:#999;font-size:12px;'>每个会话最多发送一次更正。</p>"
        f"</body></html>", "html", "utf-8"))
    return await _send(msg)

Step 4.2: 确认测试转绿 + 导入无损

Run: python -m pytest tests/test_email_rendering.py -v → 3 PASS Run: python -m pytest -q → 39 passed Run: python -c "from app.services import email_service" → 无输出（成功）

Step 4.3: Commit

git add app/services/email_service.py
git commit -m "feat(email): 标签/转义/中文标签列/回访轻提醒/更正邮件"

Task 5: conversation.py — 管道接线 + 深合并修复 + 评估前置（T1/T3/D7）

Files:

Modify: app/services/conversation.py（仅 stream_process_message，154-246 行区域）
Test: tests/test_lead_email_trigger.py
Step 5.1: 写失败测试（FakeAI + async_db 模式）

"""批次1 集成测试：管道接线/邮件触发/结束守卫/卡单标记。"""
import asyncio

from app.models.database import ConversationSource, ConversationStatus
from app.services.conversation import ConversationManager
from app.services import email_service


class FakeAI:
    """替代 GLM 的确定性桩：构造参数即每轮返回值。"""

    def __init__(self, extract=None, evaluate=None):
        self._extract_seq = extract if isinstance(extract, list) else [extract or {}]
        self._evaluate = evaluate or {}
        self.eval_calls = 0

    async def chat(self, messages, context=""):
        return "ok reply"

    async def stream_chat(self, messages, context=""):
        for token in ("ok ", "reply"):
            yield token

    async def extract_info(self, messages, existing_info):
        # 第 N 轮客户消息 → seq[N-1]（messages 为 [u,a,u,a...] 全量历史）
        idx = min(max(len(messages) // 2 - 1, 0), len(self._extract_seq) - 1)
        return self._extract_seq[idx]

    async def evaluate(self, profile, recent_messages):
        self.eval_calls += 1
        return dict(self._evaluate)


INTERESTED = {
    "intent_level": "interested", "reasoning": "预算国家车型齐全",
    "contact_info": {}, "info_sufficient": True,
}


async def _send(manager, sid, text):
    async for _ in manager.stream_process_message(sid, text):
        pass


def _setup(async_db, monkeypatch, source=ConversationSource.widget, ai=None):
    sent = []

    async def fake_lead(customer_profile, contact_info, reasoning,
                        conversation_messages, intent_level=""):
        sent.append(("lead", dict(contact_info), intent_level))
        return True

    async def fake_nudge(contact_info):
        sent.append(("nudge", dict(contact_info), ""))
        return True

    async def fake_fix(contact_info):
        sent.append(("fix", dict(contact_info), ""))
        return True

    monkeypatch.setattr(email_service, "send_lead_email", fake_lead)
    monkeypatch.setattr(email_service, "send_returning_lead_email", fake_nudge)
    monkeypatch.setattr(email_service, "send_correction_email", fake_fix)

    async def run():
        async with async_db() as db:
            manager = ConversationManager(db, ai_service=ai or FakeAI())
            conv = await manager.create_conversation(source=source)
            return manager, conv
    return asyncio.run(run()), sent


def test_extractor_empty_round_preserves_contact(async_db, monkeypatch):
    # T1: extractor 第2轮返回空联系方式,不得抹掉第1轮已提取的
    (manager, conv), _ = _setup(async_db, monkeypatch, ai=FakeAI(
        extract=[{"联系方式": {"手机": "+8613800138000"}}, {"联系方式": {}}],
    ))
    asyncio.run(_send(manager, conv.session_id, "my phone is +8613800138000"))
    asyncio.run(_send(manager, conv.session_id, "how much"))
    profile = asyncio.run(manager._get_profile(conv.id))
    assert (profile.raw_info or {}).get("联系方式", {}).get("phone") == "+8613800138000"


def test_evaluation_skipped_below_3_user_messages(async_db, monkeypatch):
    # D7: 前2条客户消息不调 evaluate
    ai = FakeAI()
    (manager, conv), _ = _setup(async_db, monkeypatch, ai=ai)
    asyncio.run(_send(manager, conv.session_id, "hi"))
    asyncio.run(_send(manager, conv.session_id, "price?"))
    assert ai.eval_calls == 0
    asyncio.run(_send(manager, conv.session_id, "kenya, 20k"))
    assert ai.eval_calls == 1

（注：T3"正则兜住 extractor 漏提取 → 仍触发邮件"的用例依赖 Task 6 的触发器，移至 Task 6 Step 6.1 一并转绿，本 Task 不含。）

Step 5.2: 运行确认失败

Run: pytest tests/test_lead_email_trigger.py -v Expected: 2 用例 FAIL（T1 抹数据 / D7 评估未跳过）

Step 5.3: 修改 stream_process_message

在 stream_process_message 中，将现有 Round 2（extract）与 Round 3（evaluate）区域（conversation.py:206-224）替换为：

        # Round 2: Extract info (background)
        full_history = history + [{"role": "assistant", "content": ai_reply}]
        profile = await self._get_profile(conv.id)

        new_info = await self.ai.extract_info(
            full_history, existing_info=profile.raw_info or {}
        )
        logger.info("Extractor result: %s", new_info)
        if new_info:
            raw = dict(profile.raw_info or {})
            for k, v in new_info.items():
                # 联系方式深合并(同key新值覆盖),其余顶层覆盖 —— 堵浅合并抹数据
                if k == "联系方式":
                    raw[k] = merge_contacts(raw.get(k), v)
                else:
                    raw[k] = v
            profile.raw_info = raw
            self.db.add(profile)

        # Round 2.5: 联系方式正则辅源(S4: 扫全量客户消息历史)
        # 🔴 合并方向: 扫描器只填空,extractor 已有值优先——否则扫描器的
        # 首个历史值会压掉 extractor 后续捕获的改号, D4 更正邮件失效
        scanned = scan_user_messages(full_history)
        if scanned:
            raw = dict(profile.raw_info or {})
            raw["联系方式"] = merge_contacts(scanned, raw.get("联系方式"))
            profile.raw_info = raw
            self.db.add(profile)

        # Round 3: Evaluate (仅 ≥3 条客户消息才有意义, 前两轮必然 pending —— D7)
        user_msg_count = sum(1 for m in full_history if m["role"] == "user")
        evaluation_ran = user_msg_count >= 3
        if evaluation_ran:
            evaluation = await self.ai.evaluate(
                profile=profile.raw_info or {},
                recent_messages=full_history,
            )
        else:
            # 客户端 evaluated 事件语义保持 pending;邮件标签由 evaluation_ran 决定(D2)
            evaluation = {"intent_level": "pending", "reasoning": "",
                          "contact_info": {}, "info_sufficient": False}
        logger.info("Evaluation (stream): profile=%s | eval=%s",
                    profile.raw_info, evaluation)

文件顶部 import 区追加（Task 6 会再补 normalize_contacts, split_validate, canon——本任务只导入用到的，避免未用导入）：

from app.services.contact_pipeline import merge_contacts, scan_user_messages

注意：旧代码中 user_msg_count 原在 evaluate 之后计算（conversation.py:225），现在提前——旧的 should_end 区域在 Task 7 重写，本 Task 不动结束逻辑。evaluation_ran 在 Task 6 的触发器中使用（D2：跳过评估时邮件标签必须是[未评估]而不是[待确认]）。

Step 5.4: 运行确认通过

Run: pytest tests/test_lead_email_trigger.py -v Expected: 2 PASS（T3 用例在 Task 6 引入）

Step 5.5: Commit

git add app/services/conversation.py tests/test_lead_email_trigger.py
git commit -m "feat(pipeline): 接入统一管道/深合并修复/正则辅源/评估前置"

Task 6: 邮件触发器 + 卡单标记 + finalize UPDATE（D1/D4/D5/D8/D10）

Files:

Modify: app/services/conversation.py
Test: tests/test_lead_email_trigger.py
Step 6.1: 写失败测试（追加）

测试基建两条卫生规则（Task 5 质量评审采纳项，随本任务落地）：

本文件每个用例结尾若持有 manager，追加 asyncio.run(manager.db.close())（含既有 T1/D7 两例）——消除悬挂读事务的 GC 警告；
FakeAI.evaluate 按 _normalize_evaluation 的生产行为做默认补全/非法值矫正（4 个 key 全兜底），防止后续用例传部分 dict 时与生产行为分叉。

from app.models.database import Evaluation, IntentLevel
from sqlalchemy import select


def _mk_ai(extract, evaluate=None):
    return FakeAI(extract=extract, evaluate=evaluate)


def test_regex_catches_extractor_miss(async_db, monkeypatch):
    # T3(自 Task 5 移入): extractor 全程漏提取,正则从消息里捞到 email → 仍然触发邮件
    (manager, conv), sent = _setup(async_db, monkeypatch, ai=FakeAI(extract={}))
    asyncio.run(_send(manager, conv.session_id, "my email is a@b.com"))
    assert len(sent) == 1 and sent[0][0] == "lead"
    assert sent[0][1]["email"] == "a@b.com"


def test_first_message_contact_sends_email_with_unevaluated_tag(async_db, monkeypatch):
    # D1: 第1条消息含联系方式即发,标签[未评估](评估未跑 → intent_level 传空,D2)
    (manager, conv), sent = _setup(async_db, monkeypatch,
                                   ai=_mk_ai({"联系方式": {"手机": "+8613800138000"}}))
    asyncio.run(_send(manager, conv.session_id, "wa me +8613800138000"))
    assert len(sent) == 1 and sent[0][0] == "lead"
    assert sent[0][2] == ""  # 评估跳过 → 邮件 intent_level 参数为空串 → 标签[未评估]


def test_no_contact_never_emails_and_marks_stuck(async_db, monkeypatch):
    # D8: interested+无联系方式 → 不发邮件,upsert Evaluation 卡单记录
    (manager, conv), sent = _setup(async_db, monkeypatch,
                                   ai=_mk_ai({}, evaluate=INTERESTED))
    for text in ("hi", "kenya", "20k budget", "byd tang"):
        asyncio.run(_send(manager, conv.session_id, text))
    assert sent == []

    async def check():
        ev = (await manager.db.execute(
            select(Evaluation).where(Evaluation.conversation_id == conv.id)
        )).scalar_one()
        return ev
    ev = asyncio.run(check())
    assert ev.intent_level == IntentLevel.interested
    assert (ev.contact_info or {}) == {}
    assert ev.email_sent is False


def test_email_sent_once_per_conversation(async_db, monkeypatch):
    # D4: 联系方式持续存在,后续消息不得重发
    (manager, conv), sent = _setup(async_db, monkeypatch,
                                   ai=_mk_ai({"联系方式": {"email": "a@b.com"}}))
    asyncio.run(_send(manager, conv.session_id, "mail me a@b.com"))
    asyncio.run(_send(manager, conv.session_id, "another question"))
    asyncio.run(_send(manager, conv.session_id, "more"))
    assert len(sent) == 1


def test_correction_email_on_change_once(async_db, monkeypatch):
    # D4: 改号 → 1封更正;再改 → 不再发(cap 必须被真正跑到第二次改号)
    ai = FakeAI(extract=[{"联系方式": {"手机": "+8613800138000"}},
                         {"联系方式": {"手机": "+8613800138000"}},
                         {"联系方式": {"手机": "+8613800139999"}},
                         {"联系方式": {"手机": "+8613800139999"}},
                         {"联系方式": {"手机": "+8613800122222"}}])
    (manager, conv), sent = _setup(async_db, monkeypatch, ai=ai)
    asyncio.run(_send(manager, conv.session_id, "my number +8613800138000"))
    asyncio.run(_send(manager, conv.session_id, "wrong, use +8613800139999"))
    asyncio.run(_send(manager, conv.session_id, "again +8613800122222"))
    asyncio.run(_send(manager, conv.session_id, "fourth message"))
    asyncio.run(_send(manager, conv.session_id, "fifth message"))
    kinds = [s[0] for s in sent]
    assert kinds == ["lead", "fix"]  # 第5轮 contact 再变(C),correction_sent 已锁 → 不发


def test_suspect_contact_never_triggers_email(async_db, monkeypatch):
    # D1 边界: 校验不过的疑似联系方式(5位 whatsapp)绝不触发邮件
    (manager, conv), sent = _setup(async_db, monkeypatch,
                                   ai=_mk_ai({"联系方式": {"whatsapp": "12345"}}))
    for text in ("hi", "second", "third"):
        asyncio.run(_send(manager, conv.session_id, text))
    assert sent == []


def test_24h_dedup_sends_nudge(async_db, monkeypatch):
    # D5: 另一会话24h内已对同值发过全量 → 本会话只发轻提醒
    from datetime import datetime, timezone
    ai1 = _mk_ai({"联系方式": {"email": "same@x.com"}})
    (manager, conv1), sent1 = _setup(async_db, monkeypatch, ai=ai1)
    asyncio.run(_send(manager, conv1.session_id, "mail same@x.com"))
    assert [s[0] for s in sent1] == ["lead"]

    ai2 = _mk_ai({"联系方式": {"email": "same@x.com"}})
    (manager, conv2), sent2 = _setup(async_db, monkeypatch, ai=ai2)
    asyncio.run(_send(manager, conv2.session_id, "im back same@x.com"))
    assert [s[0] for s in sent2] == ["nudge"]

Step 6.2: 运行确认失败

Run: pytest tests/test_lead_email_trigger.py -v Expected: 新增 6 用例 FAIL（含自 Task 5 移入的 T3 用例）

Step 6.3: 实现（conversation.py）

(a0) 先在 app/services/contact_pipeline.py 追加 canon（含单测，先红后绿）：

def canon(value: str) -> str:
    """去重比对用的规范形：邮箱小写，其余取纯数字（S3/R3 best-effort）。"""
    value = str(value or "").strip()
    if "@" in value:
        return value.lower()
    return digits_only(value)

def test_canon_normalizes_for_dedup():
    from app.services.contact_pipeline import canon
    assert canon("A@B.COM") == "a@b.com"
    assert canon("+254 712-345 678") == "254712345678"

(a) 类内新增三个方法：

    async def _get_or_init_evaluation(self, conv_id: str) -> Evaluation:
        result = await self.db.execute(
            select(Evaluation).where(Evaluation.conversation_id == conv_id)
        )
        ev = result.scalar_one_or_none()
        if not ev:
            ev = Evaluation(
                conversation_id=conv_id,
                intent_level=IntentLevel.pending,
                contact_info={},
            )
            self.db.add(ev)
            await self.db.flush()
        return ev

    async def _recently_emailed_canons(self) -> set[str]:
        cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
        # 只取需要的列:该查询在每次首触/重试时执行,避免整行 ORM 实例化。
        # 🔴 禁止加 LIMIT:"任一值见过"的判定只有覆盖完整 24h 窗口才正确。
        result = await self.db.execute(
            select(Evaluation.contact_info).where(
                Evaluation.email_sent.is_(True),
                Evaluation.evaluated_at >= cutoff,
            )
        )
        seen: set[str] = set()
        for (contact_info,) in result:
            for v in (contact_info or {}).values():
                if v:
                    seen.add(canon(str(v)))
        return seen

    async def _maybe_send_lead_email(
        self, conv: Conversation, profile: CustomerProfile, evaluation: dict,
        evaluated: bool = True,
    ) -> None:
        """D1: 联系方式是发邮件的唯一条件。confirmed 才触发;suspect 只进提醒。

        evaluated=False 表示本轮跳过了 GLM 评估(<3条客户消息)——此时邮件
        intent_level 必须传空串(→[未评估]),不得把客户端事件用的 "pending"
        透传成 [待确认](D2 两者语义不同)。
        """
        contacts_all = normalize_contacts((profile.raw_info or {}).get("联系方式"))
        confirmed, _suspect = split_validate(contacts_all)
        if not confirmed:
            return
        ev = await self._get_or_init_evaluation(conv.id)

        if not ev.email_sent:
            # S3: 任一值已见于24h内其他会话 → 轻提醒;否则全量
            seen = await self._recently_emailed_canons()
            if any(canon(str(v)) in seen for v in confirmed.values()):
                sent = await email_service.send_returning_lead_email(confirmed)
            else:
                messages = await self._load_history(conv.id)
                sent = await email_service.send_lead_email(
                    customer_profile=profile.raw_info or {},
                    contact_info=confirmed,
                    reasoning=evaluation.get("reasoning", ""),
                    conversation_messages=messages,
                    intent_level=evaluation.get("intent_level", "") if evaluated else "",
                )
            if sent:
                ev.email_sent = True
                ev.contact_info = confirmed
                self.db.add(ev)
        elif not ev.correction_sent:
            old_canons = {canon(str(v)) for v in (ev.contact_info or {}).values()}
            if any(canon(str(v)) not in old_canons for v in confirmed.values()):
                if await email_service.send_correction_email(confirmed):
                    ev.correction_sent = True
                    ev.contact_info = confirmed
                    self.db.add(ev)

顶部 import 追加：from datetime import datetime, timedelta, timezone、from app.services import email_service、contact_pipeline 导入补 canon。

(b) stream_process_message 在 evaluate 之后、should_end 之前插入：

        # 邮件触发(D1): 与会话结束彻底解耦,联系方式到手即发
        await self._maybe_send_lead_email(
            conv, profile, evaluation, evaluated=evaluation_ran
        )

        # 卡单标记(D8): 有意向但没联系方式 → 落库供 admin 人工兜底
        confirmed_now, _ = split_validate(
            normalize_contacts((profile.raw_info or {}).get("联系方式")))
        if (user_msg_count >= 3
                and evaluation.get("intent_level") == "interested"
                and not confirmed_now):
            ev = await self._get_or_init_evaluation(conv.id)
            ev.intent_level = IntentLevel.interested
            ev.reasoning = evaluation.get("reasoning", "")
            ev.contact_info = {}
            self.db.add(ev)

        evaluation_record = await self._get_or_init_evaluation(conv.id)
        evaluation_record.intent_level = intent_enum
        evaluation_record.reasoning = evaluation.get("reasoning", "")
        # 结束时的管道快照;保留 email_sent / correction_sent
        confirmed, _ = split_validate(
            normalize_contacts((profile.raw_info or {}).get("联系方式")))
        evaluation_record.contact_info = confirmed or evaluation_record.contact_info or {}
        self.db.add(evaluation_record)

Step 6.4: 运行确认通过

Run: pytest tests/test_lead_email_trigger.py -v Expected: 全部 PASS

Step 6.5: Commit

git add app/services/conversation.py tests/test_lead_email_trigger.py
git commit -m "feat(trigger): 联系方式驱动邮件触发器/卡单标记/finalize改UPDATE"

Task 7: 全来源结束守卫 + 拆除旧触发路径（D3/D9/D13）

Files:

Modify: app/services/conversation.py:226-236（should_end 区域）与 _has_real_contact（248-262，删除）
Modify: app/api/routes/chat.py:10,94-97,108-144
Modify: app/api/routes/chat_rest.py:138-142（及 docstring 第 12-13 行）
Modify: tests/test_chat_rest_message.py（删 3 个钉死旧路径的测试）
Test: tests/test_lead_email_trigger.py
Step 7.1: 写失败测试（追加）

def test_widget_never_ends_without_contact_all_sources(async_db, monkeypatch):
    # D3/P1: widget 来源 interested+sufficient 但无联系方式 → 永不结束
    (manager, conv), _ = _setup(async_db, monkeypatch,
                                ai=_mk_ai({}, evaluate=INTERESTED))
    for text in ("hi", "kenya", "20k", "tang", "ship how"):
        asyncio.run(_send(manager, conv.session_id, text))
    conv_now = asyncio.run(manager.get_by_session(conv.session_id))
    assert conv_now.status == ConversationStatus.active


def test_ends_with_contact_and_interested(async_db, monkeypatch):
    # D3: confirmed联系方式 + interested + ≥3轮 → 正常结束
    ai = _mk_ai({"联系方式": {"email": "a@b.com"}}, evaluate=INTERESTED)
    (manager, conv), _ = _setup(async_db, monkeypatch, ai=ai)
    for text in ("hi", "kenya", "20k", "tang"):
        asyncio.run(_send(manager, conv.session_id, text))
    conv_now = asyncio.run(manager.get_by_session(conv.session_id))
    assert conv_now.status == ConversationStatus.completed

Step 7.2: 运行确认失败

Run: pytest tests/test_lead_email_trigger.py -v Expected: widget 用例 FAIL（现状 widget 无守卫会结束）

Step 7.3: 修改 should_end（conversation.py:226-236 区域）

        # 结束守卫(D3): 全来源必须有 confirmed 联系方式 —— 无联系方式永不结束,
        # AI 保持争取机会。(推翻"不改widget"红线,2026-08-17 产品决策)
        should_end = (
            bool(confirmed_now)
            and evaluation.get("info_sufficient", False)
            and user_msg_count >= 3
            and evaluation.get("intent_level") == "interested"
        )

（confirmed_now 已在 Task 6 卡单标记处计算，紧邻复用。）随后：

删除 _has_real_contact 静态方法（248-262）及其上方的 app-only 注释（226-229）；
删除整个 process_message 方法（70-152）：grep 确认零调用方的死代码，且其对 _has_real_contact 的调用（137 行）在方法删除后会变成悬空引用——与其留一个必然 AttributeError 的死方法，不如随本提交一并清除（其 131-134 的红线注释同时消失）；
顺手删除旧区域遗留的 user_msg_count = sum(...) 重复赋值行（conversation.py:225，Step 5.3 已提前计算）。
Step 7.4: 拆除路由层旧触发（D9，同一提交）

chat.py：

删除 if event["type"] == "evaluated" and event.get("ended") 触发块（94-97 行区域，含 _try_send_email 调用）；
保留 ended = False 标志 + ended = True + 外层 if ended: break（HEAD 的连接生命周期语义：ended 后关闭连接。只拆邮件触发，不改连接行为——若去掉外层 break，连接会悬挂到代理超时 1h）；
删除 _try_send_email 整个函数（108-144）与 from app.services import email_service（第 10 行）；随之 from sqlalchemy.ext.asyncio import AsyncSession（第 5 行）不再被使用，一并删除。

chat_rest.py：

删除 138-142 的 lazy import 与调用块；
docstring 第 12-13 行的 "fires _try_send_email on ended && intent_level == "interested"" 改为 "lead email is triggered inside ConversationManager when contact info is detected (see contact-driven-lead-email plan)"。

tests/test_chat_rest_message.py：

删除 test_email_fires_on_interested_end、test_email_does_not_fire_when_not_interested 及模块 docstring 中的 🔴 段（其保护的行为已被 tests/test_lead_email_trigger.py 的 D1/D4 测试取代）；
保留 reply/404/429 用例。
Step 7.4b: 确认死代码删除无残留引用（须在 7.4 删除 docstring 中的旧引用之后执行）

Run: grep -rn "process_message\b" app/ tests/ | grep -v stream_process_message Expected: 无输出。（注意顺序：若在 7.4 之前跑，tests/test_chat_rest_message.py docstring 里的裸 process_message 字样会产生一条命中——那是即将被 7.4 删除的旧文案。）

Step 7.5: 全量测试

Run: pytest -v Expected: 全部 PASS（含未改动的 context_seeding/enum/rate_limit/smoke）

Step 7.6: Commit

git add app/services/conversation.py app/api/routes/chat.py app/api/routes/chat_rest.py tests/
git commit -m "feat(guard): 全来源结束守卫+拆除旧ended触发路径(邮件触发内聚到Manager)"

Task 8: 定向提醒（D6）— 不重复索要

Files:

Modify: app/services/conversation.py（_build_context_reminder 312-328、_enrich_short_message 279-310）
Test: tests/test_lead_email_trigger.py
Step 8.1: 写失败测试（追加）

def test_reminder_says_no_reasking_when_contact_present():
    raw = {"目标国家": "肯尼亚", "联系方式": {"whatsapp": "+254712345678"}}
    text = ConversationManager._build_context_reminder(raw, "yes", [])
    assert "禁止" in text and "+254712345678" in text


def test_reminder_suspect_variant_asks_to_confirm():
    raw = {"联系方式": {"whatsapp": "12345"}}  # 校验不过 → suspect
    text = ConversationManager._build_context_reminder(raw, "yes", [])
    assert "确认" in text and "禁止" in text


def test_reminder_hides_internal_keys():
    raw = {"目标国家": "肯尼亚", "_internal": "x"}
    text = ConversationManager._build_context_reminder(raw, "hi", [])
    assert "_internal" not in text

Step 8.2: 运行确认失败

Run: pytest tests/test_lead_email_trigger.py -v Expected: 3 用例 FAIL

Step 8.3: 修改 _build_context_reminder

    @staticmethod
    def _build_context_reminder(raw_info: dict, latest_msg: str, history: list[dict] | None = None) -> str:
        """Build a context reminder from extracted profile info."""
        if not raw_info:
            return ""

        parts = ["[上下文提醒 - 这是一条系统指令，绝对不要向客户提及这段内容]"]
        parts.append("客户已明确告知以下信息，你绝对不能再问这些：")

        for key, val in raw_info.items():
            if str(key).startswith("_") or key == "联系方式":
                continue  # 内部状态与联系方式单独渲染
            if val:
                parts.append(f"{key}: {val}")

        # 联系方式专项(D6): confirmed=禁止索要; suspect=礼貌确认。绝不重复索要已给的。
        contacts_all = normalize_contacts(raw_info.get("联系方式"))
        confirmed, suspect = split_validate(contacts_all)
        if confirmed:
            shown = ", ".join(
                f"{k}: {' '.join(str(v).split())[:60]}" for k, v in confirmed.items())
            parts.append(f"⚠️ 客户已提供联系方式（{shown}）。绝对禁止再索要任何联系方式！"
                         f"确认收到后继续解答客户问题即可，不必急着结束对话。")
        elif suspect:
            shown = ", ".join(
                f"{k}: {' '.join(str(v).split())[:60]}" for k, v in suspect.items())
            parts.append(f"⚠️ 客户疑似已提供联系方式（{shown}）。禁止重新索要；"
                         f"可礼貌地请客户确认一次。")

        if len(latest_msg.strip()) < 25:
            parts.append("⚠️ 客户发的是短消息！必须结合上面已知信息来理解，绝对不能当作独立问题！")

        return "\n".join(parts)

_enrich_short_message 的 facts 构造行加同样过滤，并跳过 联系方式（避免 dict-repr 进提示词，与 reminder 的专项渲染保持一致）：

            facts = [f"{k}: {v}" for k, v in raw_info.items()
                     if v and not str(k).startswith("_") and k != "联系方式"]

Step 8.4: 运行确认通过 + Commit

Run: pytest tests/test_lead_email_trigger.py -v Expected: 全部 PASS

git add app/services/conversation.py tests/test_lead_email_trigger.py
git commit -m "feat(reminder): 联系方式定向提醒(confirmed禁索要/suspect礼貌确认)"

Task 9: system.md 话术（D11）

Files:

Modify: app/prompts/system.md:60-82（对话流程 + 联系方式引导策略两节）
Step 9.1: 修改话术

"## 对话流程"第 3、4 条替换为：

3. 第3-4轮用"转接销售"或"发报价单"的方式引导联系方式（见下方策略）
4. 拿到联系方式后：确认一句（"Got it!"），然后继续正常解答客户问题——不必急着结束对话，也别再提联系方式

"## 联系方式引导策略"节追加（放在"### 注意"之前）：

### 价值交换优先（比单纯索要多给一个理由）
- "I'll send you our full price list — what's your WhatsApp?"
- "Want me to send a proper quote? Drop your WhatsApp or email."
- 话术固定为"完整报价单/报价"级别，禁止承诺具体价格、折扣或时限（仅指索要联系方式时的话术，正常聊车报价照旧）

### 引流兜底（先聊后引，顺序不能反）
- 聊天内按上述方式要联系方式，最多两次
- 两次都没要到，才指引用户点面板上的 WhatsApp 按钮直接找我们：
  "You can also just message us on WhatsApp (button at the top) — our guy replies fast."
- 禁止一上来就推 WhatsApp 按钮跳过聊天索要

### 系统提醒配合
- 系统会在上下文提醒里告诉你客户是否已留联系方式
- 提醒说"已提供"时：绝对禁止再要，确认收到即可
- 提醒说"疑似提供"时：礼貌请客户确认一次，不要当成没给过重新索要

"### 注意"节的"拿到后说一句…别再重复要"保持不变。

Step 9.2: 验证 prompt 加载

Run: python -c "from app.services.ai_service import SYSTEM_PROMPT; assert '价值交换优先' in SYSTEM_PROMPT; print('ok')" Expected: ok

Step 9.3: Commit

git add app/prompts/system.md
git commit -m "feat(prompt): 价值交换/先聊后引/系统提醒配合话术"

Task 10: 全量回归 + 人工验收

Step 10.1: 全量测试

Run: pytest -v Expected: 全部 PASS

Step 10.2: 本地起服务人工冒烟

Run: uvicorn app.main:app --port 8010（.env 指向测试 SMTP/_NOTIFY_EMAIL）操作：打开 http://127.0.0.1:8010/，第 1 条消息直接输入 "my whatsapp is +254712345678" → 收到 [新询盘·未评估] 邮件（含该号码、完整转义对话）；继续聊 3+ 轮不再收邮件；再说 "wrong number, use +254711111111" → 收到 [询盘·更正]；无联系方式的对话聊 5 轮 → 零邮件且输入框不锁死。 Expected: 全部符合

Step 10.3: 上线后一周观测（对照 2026-08-17-baseline.txt）

指标：①新邮件 100% 带联系方式 ②全量邮件量 vs 基线上界 ③[更正]/[回访]占比 ④admin 中卡单（interested+无联系方式）数量

Step 10.4: 最终 Commit（如有遗漏小修）

git add -A && git commit -m "chore: 批次1收尾"

Task 11: 移除自动结束（决策 A，2026-08-18 追加）

背景：邮件已与结束解耦，自动结束只剩收尾 UX/停止计费两个弱价值，却会踢出还没问完的客户（锁输入+刷新丢上下文+销售响应空窗）。用户拍板方案 A。

Files:

Modify: app/services/conversation.py（stream 结束区域 + _finalize 删除）
Test: tests/test_lead_email_trigger.py
Step 11.1: 写失败测试（先红）

def test_conversation_never_auto_ends_even_with_contact_and_interested(async_db, monkeypatch):
    # 决策A(2026-08-18): 自动结束移除——即使 contact+interested+≥3轮也保持 active
    ai = _mk_ai({"联系方式": {"email": "a@b.com"}}, evaluate=INTERESTED)
    (manager, conv), _ = _setup(async_db, monkeypatch, ai=ai)
    for text in ("hi", "kenya", "20k", "tang", "more questions"):
        asyncio.run(_send(manager, conv.session_id, text))
    conv_now = asyncio.run(manager.get_by_session(conv.session_id))
    assert conv_now.status == ConversationStatus.active
    asyncio.run(manager.db.close())


def test_evaluation_upsert_keeps_flags_and_intent(async_db, monkeypatch):
    # 评估落库泛化: intent 刷新为 interested,email_sent 不被触碰
    ai = _mk_ai({"联系方式": {"email": "a@b.com"}}, evaluate=INTERESTED)
    (manager, conv), sent = _setup(async_db, monkeypatch, ai=ai)
    for text in ("hi", "kenya", "20k"):
        asyncio.run(_send(manager, conv.session_id, text))

    async def check():
        ev = (await manager.db.execute(
            select(Evaluation).where(Evaluation.conversation_id == conv.id)
        )).scalar_one()
        return ev
    ev = asyncio.run(check())
    assert ev.intent_level == IntentLevel.interested
    assert ev.email_sent is True and ev.correction_sent is False
    asyncio.run(manager.db.close())

同时把 test_ends_with_contact_and_interested 删除（其断言的行为已被决策 A 废除，由上面的 never-auto-ends 用例取代）。

Step 11.2: 运行确认失败

Run: python -m pytest tests/test_lead_email_trigger.py -v -k "never_auto_ends or upsert_keeps" Expected: never_auto_ends FAIL（现状会结束）；upsert_keeps 视现状可能通过（先记录）

Step 11.3: 实现

(a) 删除结束区域（should_end 判定 + if should_end: await self._finalize(...)），evaluated 事件的 ended 恒为 False（事件本身保留，客户端协议不变）。

(b) 将 D8 卡单块泛化为每轮评估落库：

        # 评估落库(D8 泛化): 每轮评估后刷新意向;email_sent/correction_sent
        # 永不触碰。有意向无联系方式 = 卡单(admin 可筛);已发过邮件的
        # contact_info 由触发器更正分支管理,这里不覆盖。
        if user_msg_count >= 3:
            ev = await self._get_or_init_evaluation(conv.id)
            ev.intent_level = IntentLevel(evaluation.get("intent_level", "pending"))
            ev.reasoning = evaluation.get("reasoning", "")
            if not confirmed_now:
                ev.contact_info = {}
            elif not ev.email_sent:
                ev.contact_info = confirmed_now
            self.db.add(ev)

Step 11.4: 全量测试

Run: python -m pytest -q Expected: 57 passed（56 − 删1 + 增2）

Step 11.5: Commit — 🔴 SKIP（沿用不提交约定）

Task 12: 联系方式更正覆盖 + 更正上限 3 次（2026-08-18 用户实测反馈）

背景：用户实测 "wrong number, use X" 未触发更正——extractor 受"保留已有信息"倾向回传旧值，扫描器"首中+填空"结构性无法翻新（风险台账中该残余首次真实命中）。用户决策：①确定性"最新显式自述优先"覆盖；②extractor.md 教会覆盖语义；③更正上限放宽至 3 次；④锁邮件不锁快照。

Files: contact_pipeline.py（新增 latest_stated_contacts + CORRECTION_CTX_RE + PHONE_CTX 扩展 new/other/another）、conversation.py（Round 2.5 覆盖 + 触发器计数分支）、database.py + main.py（correction_sent→correction_count，迁移替换）、email_service.py（更正邮件带次数）、app/prompts/extractor.md（覆盖语义一行）、测试三文件。

核心规则：latest_stated_contacts(messages) 反向取每 key 全历史最后一次显式自述——上下文门控捕获（含 my new/other/another number is）与"更正语境 ∧ 裸国际号"有覆盖权；纯裸数字无覆盖权。Round 2.5 合并后：canon 与现值不同的 key 强制覆盖（唯一允许文本压过 extractor 的位置）。触发器更正分支：correction_count < 3 → 发邮件（主题 [询盘·更正N]）并计数；已达上限 → 仅刷新 ev.contact_info。

关键测试：test_correction_overrides_extractor_echo（实测场景复刻：extractor 恒回旧值 + wrong number 消息 → 管道翻新 + lead/fix 各一封）、test_bare_second_number_overrides（无更正语境的 my new number is）、test_correction_cap_three_then_snapshot_only（A→B→C→D→E：lead+3 fix，第5值仅进快照）、原 test_correction_email_on_change_once 期望更新为 ["lead","fix","fix"]、渲染测试断言 [询盘·更正1]、latest_stated_* 单测四例。

Task 12b: 同号同步（2026-08-18 用户实测反馈 II）

背景：用户实测更正邮件里联系方式新旧混排（phone=新号, whatsapp=旧号）。根因：extractor 习惯把同一号码镜像进 phone/whatsapp 两个 key，更正只翻新命中的 key，兄弟 key 滞留旧值进邮件/快照。

修复：contact_pipeline.sync_same_number_keys(before, after) 纯函数——after 中某 key 翻新时，与其旧值同号的兄弟 key 一并翻新（email 身份独立不同步；两 key 本就不同号时不动）。接线在 Round 2 深合并前捕获 before、Task 12 覆盖后执行同步（覆盖 extractor 与 override 两条更新路径）。

测试：3 个纯函数单测（镜像同步/真不同号不动/email 独立）+ 1 个集成复刻（extractor 镜像旧号 → wrong number → 断言邮件与快照两 key 均为最新号）。

Task 12c: 最近提及裁决（2026-08-18 用户实测反馈 III）

背景：用户实测更正配额被虚假更正烧光（仅 1 次真实变化却 correction_count=3，最后一封更正邮件显示旧号）。根因：extractor 在历史消息里的多个号码间振荡（GLM 每轮从全历史提取，回传值不稳定），深合并如实应用翻转，12b 同号同步进一步放大到镜像 key，每次翻转触发一封虚假更正。

修复：resolve_by_recency(before, after, messages) 取代 12b 的同号同步——镜像身份组（before 同 canon 的非 email key）内部分歧时，取"出现在最晚客户消息里"的值：旧号在更早消息里天然落败（杀振荡）、幻觉值 last_idx=-1 天然落败（压制幻觉）、客户最晚消息里的新号天然胜出（extractor 抓到的裸号正确翻新）。

验收测试：test_oscillation_does_not_burn_correction_quota——完整复刻用户序列（A→B→三轮振荡→裸号 C），断言仅 2 次更正且第 2 封内容就是 C；12b 的镜像同步场景由 recency 涵盖（原测试继续通过）。

4. 验收清单（对照已定决策）

决策	验收
D1 邮件唯一条件	`test_first_message_contact_sends_email_with_unevaluated_tag`
D3 永不自动结束（决策A）	`test_widget_never_ends_without_contact_all_sources` / `test_conversation_never_auto_ends_even_with_contact_and_interested` / `test_evaluation_upsert_keeps_flags_and_intent`
D4 防重发+更正	`test_email_sent_once_per_conversation` / `test_correction_email_on_change_once`
D5 24h 去重	`test_24h_dedup_sends_nudge`
D6 不重复索要	`test_reminder_*` 三例
D7 评估前置	`test_evaluation_skipped_below_3_user_messages`
D8 卡单标记	`test_no_contact_never_emails_and_marks_stuck`
D9/D10 旧路径拆除+UPDATE	Task 7 Step 7.4/7.5 全量回归绿
不丢失（T1/T2/T3）	`test_extractor_empty_round_preserves_contact` / 归一化用例 / `test_regex_catches_extractor_miss`
D12 转义	email_service 全字段 `_esc` + 人工冒烟

5. 残余风险台账（明示接受）

风险	缓解	残余
正则误报 → 错发一封 + 提醒压制再索要	高置信模式/校验分级/上线监控无效率	低概率，方向安全
号码变体击穿 24h 去重	canon 纯数字比对	best-effort
extractor 连续漏提取"原值+更正值"时，扫描器首个历史值滞留	邻接绑定+extractor优先合并压低概率	双重漏提取，极低
给号即沉默 + SMTP 首发失败	下条消息自然重试（email_sent 未置位）	双重失败丢线索（极低）
发送成功后落库失败（毫秒级窗口）→ 重试时重复发送	at-least-once 为有意选择：漏一条线索比重复一封邮件贵	接受
App 端 `ended` 恒为 false（决策A）	App 若有"对话结束"闭环 UI 需知晓并调整；widget/独立页零影响	已知会用户
第 1 轮 extractor 幻觉出从未提及的号码 → 首封全量邮件 phone/whatsapp 混排	需"幻觉+首轮无历史组"双条件叠加，概率极低；whatsapp 侧仍真实可联	观察期，若实测出现再做 12e（无消息依据的 canon 降级）
客户撤回已给联系方式	—	接受
邮件量上升（含无意向留号者）	标签分流 + 基线上界预估	预期内

6. 后续批次（另行成 plan，本计划不含）

批次 2：widget 会话持久化（localStorage 独立前缀 + 48h 过期）+ WS 限流/消息长度（消息级拒绝防重连风暴；session ≥60/h、IP 放宽；REST 长度复用 429）。
批次 3：admin"缺联系方式线索"筛选（基于 D8 卡单记录）+ UI 留资卡片（提交即写管道；端点限流+校验+只弹一次）。
健康轨：删 Redis、SQLite 备份+WAL、GLM 超时重试、admin 分页 total 修复、国家映射、ADMIN_PASSWORD 修改（强烈建议）、删除已无调用方的 AIService.chat（Task 7 后仅剩 provider 内部引用）。（process_message 死代码已并入本计划 Task 7 删除。）

70 KiB Raw Blame History Unescape Escape

联系方式驱动的询盘邮件（Contact-Driven Lead Email）实施计划 v1.1

1. 背景与问题（为什么做）

2. 已定决策（本计划的 spec，评审以此为准）

3. 文件结构

批次 0.5：基线统计

Task 0: 基线脚本

批次 1：核心实现

Task 1: contact_pipeline — 归一化 / 深合并 / 校验分级

Task 2: contact_pipeline — 正则扫描（高置信，S4）

Task 3: Evaluation 加 correction_sent 列 + 迁移（D4）

Task 4: email_service — 转义 / 标签 / 轻提醒 / 更正邮件（D2/D4/D12）

Task 5: conversation.py — 管道接线 + 深合并修复 + 评估前置（T1/T3/D7）

Task 6: 邮件触发器 + 卡单标记 + finalize UPDATE（D1/D4/D5/D8/D10）

Task 7: 全来源结束守卫 + 拆除旧触发路径（D3/D9/D13）

Task 8: 定向提醒（D6）— 不重复索要

Task 9: system.md 话术（D11）

Task 10: 全量回归 + 人工验收

Task 11: 移除自动结束（决策 A，2026-08-18 追加）

Task 12: 联系方式更正覆盖 + 更正上限 3 次（2026-08-18 用户实测反馈）

Task 12b: 同号同步（2026-08-18 用户实测反馈 II）

Task 12c: 最近提及裁决（2026-08-18 用户实测反馈 III）

4. 验收清单（对照已定决策）

5. 残余风险台账（明示接受）

6. 后续批次（另行成 plan，本计划不含）

70 KiB

Raw Blame History