feat: 会话编排建材化+阿语号码转ASCII+12小时话术跨层同步

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

app/services/boundary_filter.py

@@ -204,8 +204,10 @@ def deflection_for(user_msg: str, rotate: int = 0,
 LONG_NUMBER_RE = re.compile(r"\+?\d[\d\s\-().]{9,17}\d")   # ≥10位连续数字(允许分隔形态)
 ANY_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
 
+# 12 小时口径与 system.md 对齐(Task 3 质量审查 Minor 4: 原句 "replies fast"
+# 与 system.md 的 "reply within 12 hours" 承诺不一致)——hours 不触时间承诺墙
 NO_CONTACT_WEB = ("Just tap the WhatsApp button at the top of this chat — "
-                  "our manager replies fast!")
+                  "our manager will reply within 12 hours!")
 NO_CONTACT_APP = ("Drop your WhatsApp and our manager will reach out to you directly "
                   "with the full quote!")
 # F6 同款(15.3 补): 已留资时零号码墙替换句不得索要

app/services/contact_pipeline.py

@@ -10,6 +10,19 @@
 """
 import re
 
+# Unicode 十进制数字 → ASCII(阿语٠-٩/波斯语۰-۹), 中东买家号码 CRM 可直接拨。
+# 挂在归一化链最前端(normalize_contacts 的值 / digits_only / scan 与
+# latest 的捕获赋值): 存库、canon 去重比对、boundary_filter 客户回显豁免
+# 两侧全部经此转成 ASCII——阿语消息与 ASCII 提取值比对口径一致,
+# 同一号码不会被误判成"改号"触发更正邮件。
+_DIGIT_TRANSLATE = str.maketrans(
+    "٠١٢٣٤٥٦٧٨٩۰۱۲۳۴۵۶۷۸۹", "01234567890123456789")
+
+
+def _to_ascii_digits(value: str) -> str:
+    """捕获源头归一: 阿语/波斯语数字转 ASCII(translate 对 ASCII 恒等)。"""
+    return value.translate(_DIGIT_TRANSLATE)
+
 KEY_ALIASES = {
     "phone": "phone", "手机": "phone", "电话": "phone", "电话号码": "phone",
     "联系电话": "phone", "手机号": "phone", "手机号码": "phone",
@@ -38,7 +51,7 @@ def normalize_contacts(raw: dict | None) -> dict:
                 if name in key:
                     alias = target
                     break
-        val = str(v or "").strip()
+        val = _to_ascii_digits(str(v or "").strip())
         if alias and val:
             out[alias] = val
     return out
@@ -50,8 +63,11 @@ def merge_contacts(old: dict | None, new: dict | None) -> dict:
 
 
 def digits_only(s: str) -> str:
-    """提取纯数字（split_validate 位数判断与后续 canon 复用）。"""
-    return re.sub(r"\D", "", s)
+    """提取纯数字（split_validate 位数判断与后续 canon 复用）。
+    阿语/波斯语数字先转 ASCII——canon 与 boundary_filter 回显豁免两侧
+    都走本函数,天然保证同形比较(Unicode 正则的数字类本就匹配阿语数字,
+    位数判定不变;此处只统一"比对与存储的形态")。"""
+    return re.sub(r"\D", "", s.translate(_DIGIT_TRANSLATE))
 
 
 def split_validate(contacts: dict) -> tuple[dict, dict]:
@@ -72,7 +88,7 @@ def split_validate(contacts: dict) -> tuple[dict, dict]:
 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：车架号/订单号/先令价格）
+# 有上下文词 + 任意位置有 7 位数字"的消息级误绑（S1：单号/订单号/先令价格）
 WHATSAPP_CTX_RE = re.compile(
     r"(?:whats\s*app|wa\.?me|(?:^|\b)wa\s*[:：])\D{0,12}(\+?\d[\d\s\-().]{5,17}\d)",
     re.IGNORECASE,
@@ -129,7 +145,7 @@ CORRECTION_CTX_RE = re.compile(
 
 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":
@@ -142,30 +158,30 @@ def scan_user_messages(messages: list[dict]) -> dict:
         if "phone" not in found:
             sn = standalone_number(text)
             if sn:
-                found["phone"] = sn
+                found["phone"] = _to_ascii_digits(sn)
         if "phone" not in found:
             hit = INTL_PHONE_RE.search(text)
             if hit:
-                found["phone"] = hit.group(0)
+                found["phone"] = _to_ascii_digits(hit.group(0))
         if "whatsapp" not in found:
             hit = WHATSAPP_CTX_RE.search(text)
             if hit:
-                found["whatsapp"] = hit.group(1)
+                found["whatsapp"] = _to_ascii_digits(hit.group(1))
         if "phone" not in found:
             hit = PHONE_CTX_RE.search(text)
             if hit:
-                found["phone"] = hit.group(1)
+                found["phone"] = _to_ascii_digits(hit.group(1))
         if "phone" not in found:
             hit = PHONE_ADJACENCY_SCAN_RE.search(text)
             if hit:
-                found["phone"] = hit.group(1)
+                found["phone"] = _to_ascii_digits(hit.group(1))
         # 倒装属最低置信层: whatsapp/phone 任一已被更高置信源捕获即跳过,
         # 不与之竞争(跨 key 门控故含 phone)。
         if "whatsapp" not in found and "phone" not in found:
             hit = NUMBER_FIRST_RE.search(text)
             if hit:
                 ch = _number_first_channel(hit)
-                found[ch] = hit.group(1)
+                found[ch] = _to_ascii_digits(hit.group(1))
     return found
 
 
@@ -211,21 +227,21 @@ def latest_stated_contacts(messages: list[dict]) -> dict:
             latest["email"] = hit.group(0)
         hit = WHATSAPP_CTX_RE.search(text)
         if hit:
-            latest["whatsapp"] = hit.group(1)
+            latest["whatsapp"] = _to_ascii_digits(hit.group(1))
         sn = standalone_number(text)
         if sn:
-            latest["phone"] = sn
+            latest["phone"] = _to_ascii_digits(sn)
         hit = PHONE_CTX_RE.search(text)
         if hit:
-            latest["phone"] = hit.group(1)
+            latest["phone"] = _to_ascii_digits(hit.group(1))
         elif CORRECTION_CTX_RE.search(text):
             hit = INTL_PHONE_RE.search(text)
             if hit:
-                latest["phone"] = hit.group(0)
+                latest["phone"] = _to_ascii_digits(hit.group(0))
         hit = NUMBER_FIRST_RE.search(text)
         if hit:
             ch = _number_first_channel(hit)
-            latest[ch] = hit.group(1)
+            latest[ch] = _to_ascii_digits(hit.group(1))
     return latest
 
 

app/services/conversation.py

@@ -27,7 +27,7 @@ from app.services.contact_pipeline import (
 logger = logging.getLogger(__name__)
 
 # 职能边界触发词(Task 13.2, 评审#1/#4 修订): 命中即注入红线提醒。精修说明:
-# - 裸 "how long"/"多久" 不收(续航类问题常用),英文要求 take/arrive/deliver/ship 邻接,
+# - 裸 "how long"/"多久" 不收(规格参数类问题常用),英文要求 take/arrive/deliver/ship 邻接,
 #   中文只收 到货/多久能到/交货 等明确交付语境
 # - 【评审#1】施压话术本身也触发: 磨价轮("just roughly"/"说个大概")没有价格词,
 #   却是最易破防的时刻——压力词单列一组
@@ -38,13 +38,15 @@ BOUNDARY_TRIGGER_RE = re.compile(
     r"price|pricing|how\s+much|\bcost\b|discount|cheaper|expensive|\bfee\b|"
     r"quote|quotation|shipping\s+(fee|cost|time)|freight|tariff|customs\s+fee|"
     r"best\s+offer|any\s+offer|free\s+shipping|"
-    r"test\s+drive|book(ing)?\s|reserv|arrange|"
+    r"sample|certificat|\bmoq\b|container\s+loading|"
+    r"book(ing)?\s|reserv|arrange|"
     r"delivery|lead\s+time|arriv|"
     r"how\s+long\s+(does|will|would|to).*(take|arriv|deliver|ship)|"
     r"roughly|ballpark|just\s+tell|come\s+on|at\s+least\s+(a|some|give)|"
     r"budget|预算|价格|价钱|多少钱|报价|优惠|折扣|便宜|贵|划算|"
     r"总价|落地|运费|关税|免费|"
-    r"到货|多久能到|交货|时效|发货|试驾|预约|下单|定金|提车|安排|"
+    r"到货|多久能到|交货|时效|发货|样品|证书|装柜|起订|"
+    r"预约|下单|定金|安排|"
     r"大概|估个|估一|随便说|说个|求你|求了",
     re.IGNORECASE,
 )
@@ -101,14 +103,14 @@ class ConversationManager:
         await self.db.refresh(conv)
 
         # Conditional whitelist seed of app context into raw_info (red line):
-        # only when `context` is provided, and only vehicle_name/vehicle_id
+        # only when `context` is provided, and only product_name/product_id
         # mapped to app_* keys. WS path passes no context -> raw_info stays {}.
         raw_info: dict = {}
         if context:
-            if context.get("vehicle_name"):
-                raw_info["app_vehicle_name"] = context["vehicle_name"]
-            if context.get("vehicle_id"):
-                raw_info["app_vehicle_id"] = context["vehicle_id"]
+            if context.get("product_name"):
+                raw_info["app_product_name"] = context["product_name"]
+            if context.get("product_id"):
+                raw_info["app_product_id"] = context["product_id"]
         profile = CustomerProfile(conversation_id=conv.id, raw_info=raw_info)
         self.db.add(profile)
         await self.db.commit()
@@ -554,8 +556,8 @@ class ConversationManager:
             if BUDGET_TRIGGER_RE.search(latest_msg or ""):
                 # 评审#9: 预算求推荐 → 预算路径(推荐+反问+引导),不一刀切要号
                 boundary_note = (
-                    "⚠️ 职能边界提醒：客户报预算求推荐。推荐车型但绝不复述预算数字、"
-                    "不报车价 → 反问意向车型 → 顺势引导留联系方式。")
+                    "⚠️ 职能边界提醒：客户报预算求推荐。推荐产品方案但绝不复述预算数字、"
+                    "不报价格 → 反问意向产品与需求规格 → 顺势引导留联系方式。")
             else:
                 boundary_note = (
                     "⚠️ 职能边界提醒：客户正在询问价格/优惠/交付时效/实质安排。绝对禁止"
@@ -590,7 +592,7 @@ class ConversationManager:
             else:
                 contact_note = ("⚠️ 客户正在索要我方联系方式（最高意向信号）。请逐字使用这句指路："
                                 "\"Just tap the WhatsApp button at the top of this chat — "
-                                "our manager replies fast!\"（邮箱同理指 Email 按钮）。"
+                                "our manager will reply within 12 hours!\"（邮箱同理指 Email 按钮）。"
                                 "绝对禁止口头报出任何号码或邮箱。")
 
         if not raw_info:

docs/superpowers/plans/2026-09-29-building-materials-ai-kefu.md

@@ -220,7 +220,7 @@ python -m pytest -q && git add -A && git commit -m "feat: 提取/评估提示词
 - Modify: `app/services/boundary_filter.py`
 - Test: `tests/test_boundary_filter.py`
 
-- [ ] **Step 1: 先写失败的多语言墙测试**
+- [x] **Step 1: 先写失败的多语言墙测试**
 
 在 `tests/test_boundary_filter.py` 追加（沿用该文件现有参数化风格）：
 
@@ -260,14 +260,14 @@ SPEC_NON_HITS = [
 
 断言：前者每个 `violates_boundary(s)` 为 True；后者每个为 False。
 
-- [ ] **Step 2: 跑测试确认失败**
+- [x] **Step 2: 跑测试确认失败**
 
 ```bash
 python -m pytest tests/test_boundary_filter.py -q
 ```
 预期：多语言 HITS 用例 FAIL（现有正则拦不住）
 
-- [ ] **Step 3: 扩充 PRICE_PATTERNS**
+- [x] **Step 3: 扩充 PRICE_PATTERNS**
 
 在 `PRICE_PATTERNS` 列表追加（保持参考项目"注释说明设计意图"的风格）：
 
@@ -291,14 +291,14 @@ re.compile(r"\d+\s*[- ]?(?:d[ií]as?|semanas?|дн(?:ей|я)|недел[ьи]|
 - `(两|三|四|五|六|七|八|九|十)\s*万(?!\s*(公里|km))` 保留（中文预算）
 - 时间承诺 `\d+\s*[- ]?(day|week|month)s?\b` 保留（英文交期承诺）
 
-- [ ] **Step 4: 跑测试至全绿**
+- [x] **Step 4: 跑测试至全绿**
 
 ```bash
 python -m pytest tests/test_boundary_filter.py -q
 ```
 预期：PASS（含原有全部用例不回归）
 
-- [ ] **Step 5: 迂回话术建材化 + ES 话术池（spec §6 第3层）**
+- [x] **Step 5: 迂回话术建材化 + ES 话术池（spec §6 第3层）**
 
 1. **行业残留清零**（用户可见输出，实测在参考项目 L54/L57）：
    - `DEFLECTIONS_EN_CONTACT` 的 "Anything else about the cars?" → "...about pipes?"
@@ -313,7 +313,7 @@ python -m pytest tests/test_boundary_filter.py -q
 3. **测试**：`tests/test_boundary_filter.py` 追加——西语消息选 ES 池、英文选 EN、
    中文选 ZH；"cars"/"什么车" 不再出现在任何话术池
 
-- [ ] **Step 6: 同步 scripts/breach_rate.py**
+- [x] **Step 6: 同步 scripts/breach_rate.py**
 
 把上述新词表同步进 breach_rate.py 的统计模式（保持与墙一致，供上线观察期用）。
 
@@ -326,7 +326,7 @@ python -m pytest tests/test_boundary_filter.py -q
 6. system.md line 18 与语言规则段确认无残留字面张力（Task 3 复审 nano-nit 已修 line 19，复核即可）
 修完跑 test_extractor_evaluator_prompts.py 相关断言（必要时同步微调断言）。
 
-- [ ] **Step 7: 全量测试 + Commit**
+- [x] **Step 7: 全量测试 + Commit**
 
 ```bash
 python -m pytest -q && git add -A && git commit -m "feat: 多语言输出墙(€/£/西葡俄阿价格词+时间承诺+ES话术池)+话术建材化+防误杀"

tests/test_boundary_filter.py

@@ -158,12 +158,12 @@ def test_asks_contact_detects_production_log_phrasings():
     assert asks_contact("Drop yours too and our manager will send you the full quote!")
     assert asks_contact("留个WhatsApp或者邮箱吧?")
     assert asks_contact("你WhatsApp多少?")
-    assert asks_contact("你给我个联系方式,我让客户经理给你推几个车型。")
+    assert asks_contact("你给我个联系方式,我让客户经理给你推几个产品方案。")
 
 
 def test_asks_contact_clean_phrasings_pass():
     assert not asks_contact("Got your number, thanks!")               # 确认,非问句
-    assert not asks_contact("Just tap the WhatsApp button at the top of this chat — our manager replies fast!")
+    assert not asks_contact("Just tap the WhatsApp button at the top of this chat — our manager will reply within 12 hours!")
     assert not asks_contact("Which color do you prefer?")
     assert not asks_contact("What's your budget range for this one?")
     assert not asks_contact("Can you send me more details about the Tang?")  # 泛请求无渠道词

tests/test_contact_pipeline.py

@@ -342,3 +342,25 @@ def test_number_first_whatsapp_whitespace_variants():
     # 两个接线位(scan + latest)都要走空白归一,任一崩都算回归
     assert latest_stated_contacts(m("0712345678 is my whats  app")) == {
         "whatsapp": "0712345678"}
+
+
+# ---- 阿语/波斯语数字归一化(中东买家): 存库与比对全链 ASCII ----
+
+
+def test_arabic_digits_normalized():
+    # 阿语数字(中东买家)提取后应转 ASCII,CRM 可直接拨
+    confirmed, _ = split_validate({"whatsapp": "+٩٦٦٥٠١٢٣٤٥٦٧"})
+    assert confirmed["whatsapp"] == "+966501234567"
+
+
+def test_arabic_digits_scan_latest_translated():
+    # 捕获源头即转 ASCII: 用户消息里的阿语号码,scan/latest 的落库形态
+    # 直接可拨;canon 比对也与 ASCII 形态同口径(不触发假性"改号"更正邮件)
+    msgs = [{"role": "user", "content": "whatsapp: ٩٦٦٥٠١٢٣٤٥٦٧"}]
+    assert scan_user_messages(msgs) == {"whatsapp": "966501234567"}
+    assert latest_stated_contacts(msgs) == {"whatsapp": "966501234567"}
+
+
+def test_arabic_digits_canon_equals_ascii_form():
+    from app.services.contact_pipeline import canon
+    assert canon("+٩٦٦ ٥٠١ ٢٣٤ ٥٦٧") == canon("+966 501 234 567")

tests/test_context_seeding.py

@@ -1,7 +1,7 @@
 """Red-line: `context` injection must be CONDITIONAL + WHITELIST only.
 
 - No `context` passed  -> `raw_info` stays `{}` (widget WS path unchanged).
-- `context` passed     -> only `vehicle_name`/`vehicle_id` map to `app_*`.
+- `context` passed     -> only `product_name`/`product_id` map to `app_*`.
 """
 from sqlalchemy import select
 
@@ -33,13 +33,13 @@ async def test_context_seeds_whitelist_only(async_db):
         mgr = ConversationManager(db)
         conv = await mgr.create_conversation(
             context={
-                "vehicle_name": "Tesla Model Y",
-                "vehicle_id": "v123",
+                "product_name": "HDPE PE100 Pipe DN110",
+                "product_id": "p123",
                 "evil_key": "should_be_ignored",
             }
         )
     raw = await _raw_info_for(async_db, conv.id)
     assert raw == {
-        "app_vehicle_name": "Tesla Model Y",
-        "app_vehicle_id": "v123",
+        "app_product_name": "HDPE PE100 Pipe DN110",
+        "app_product_id": "p123",
     }

tests/test_lead_email_trigger.py

@@ -482,17 +482,17 @@ def test_reminder_boundary_covers_delivery_and_arrangement():
     assert "职能边界" in ConversationManager._build_context_reminder(
         {}, "how long does shipping take?", [])
     assert "职能边界" in ConversationManager._build_context_reminder(
-        {}, "可以约试驾吗", [])
+        {}, "可以安排验厂吗", [])
     assert "职能边界" in ConversationManager._build_context_reminder(
         {}, "my budget is 20000, any recommendation?", [])
 
 
 def test_reminder_no_boundary_on_normal_car_questions():
-    # 续航/配置/库存类正常问题不得误触发
+    # 规格/配置/库存类正常问题不得误触发
     assert "职能边界" not in ConversationManager._build_context_reminder(
-        {}, "what is the range of the Tang?", [])
+        {}, "what is the pressure rating of the PE100 pipe?", [])
     assert "职能边界" not in ConversationManager._build_context_reminder(
-        {}, "这款车续航多久", [])
+        {}, "这根管耐压多少", [])
 
 
 def test_reminder_boundary_on_pressure_phrases():