from app.engine.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_normalize_coerces_bare_string_extractor_output(): # Task 11 真人验收实测: 真实 GLM extractor 偶发把联系方式压成裸字符串 # (mock 测不出)——唯一事实源处兜形,不崩溃不丢号: # 含 @ → email;裸号码 → phone;列表逐项;其他非 dict 按空 assert normalize_contacts("+966 50 123 4567") == {"phone": "+966 50 123 4567"} assert normalize_contacts("buyer@example.com") == {"email": "buyer@example.com"} assert normalize_contacts(["+966501234567", "a@b.com"]) == { "phone": "+966501234567", "email": "a@b.com"} assert normalize_contacts(" ") == {} assert normalize_contacts(12345) == {} # merge_contacts 两侧裸串同样不崩、正常深合并 assert merge_contacts("+966501234567", {"whatsapp": "+966509876543"}) == { "phone": "+966501234567", "whatsapp": "+966509876543"} 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"} def test_normalize_phone_alias_variants(): # 审查补充: 电话号码 / phone number 等带空格、中文长名变体必须命中 phone assert normalize_contacts({"电话号码": "13800138000"}) == {"phone": "13800138000"} assert normalize_contacts({"phone number": "+12345678901"}) == {"phone": "+12345678901"} def test_normalize_fuzzy_key_variants(): # extractor 常见即兴 key(精确别名表外)不得静默丢值 assert normalize_contacts({"联系电话": "13800138000"}) == {"phone": "13800138000"} assert normalize_contacts({"手机号": "13800138000"}) == {"phone": "13800138000"} assert normalize_contacts({"WhatsApp账号": "+254712345678"}) == { "whatsapp": "+254712345678"} def test_normalize_unknown_key_still_dropped(): # 完全不含渠道词的 key 仍然丢弃(如 "备注") assert normalize_contacts({"备注": "13800138000"}) == {} def test_normalize_slash_combined_key_deterministic(): # "电话/WhatsApp" 双渠道复合 key: 按 KEY_ALIASES 声明序取第一个命中(phone 先于 whatsapp) assert normalize_contacts({"电话/WhatsApp": "13800138000"}) == { "phone": "13800138000"} from app.engine.services.contact_pipeline import scan_user_messages def m(*texts): return [{"role": "user", "content": t} for t in texts] + [ {"role": "assistant", "content": "call me at +999111222333"}] 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 PE100 pipes", # 价格+年份 "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 pvc"}, ] 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"} def test_canon_normalizes_for_dedup(): from app.engine.services.contact_pipeline import canon assert canon("A@B.COM") == "a@b.com" assert canon("+254 712-345 678") == "254712345678" from app.engine.services.contact_pipeline import latest_stated_contacts def test_latest_stated_correction_context(): msgs = m("my phone is +8613800138000", "wrong number, use +8613999999999") assert latest_stated_contacts(msgs) == {"phone": "+8613999999999"} def test_latest_stated_new_number_no_correction_words(): msgs = m("my number is +254712345678", "my new number is +254700111222") assert latest_stated_contacts(msgs) == {"phone": "+254700111222"} def test_latest_stated_bare_number_has_no_authority(): # 更正语境缺席时,裸 +号 无覆盖权(价格/单号防误翻新) msgs = m("my number is +254712345678", "total is +254700111222 usd wait no") assert latest_stated_contacts(msgs) == {"phone": "+254712345678"} def test_latest_stated_same_value_no_conflict(): msgs = m("my number is +254712345678", "my number is +254712345678") assert latest_stated_contacts(msgs) == {"phone": "+254712345678"} def test_scan_my_phone_is_local_number(): # 评审恢复项: "my phone is <无+本地号>" 曾从门控模式掉落(extractor 漏时彻底失联) found = scan_user_messages(m("my phone is 07123456789")) assert found == {"phone": "07123456789"} def test_latest_stated_my_phone_is_form(): msgs = m("hi", "my phone is 07123456789") assert latest_stated_contacts(msgs) == {"phone": "07123456789"} from app.engine.services.contact_pipeline import resolve_by_recency def _u(*texts): return [{"role": "user", "content": t} for t in texts] def test_recency_mirror_lag_newest_wins(): # Task 12b 场景: extractor 只翻新 phone,镜像 whatsapp 滞后 → 最新提及统一 before = {"phone": "+8613800138000", "whatsapp": "+8613800138000"} after = {"phone": "+8613999999999", "whatsapp": "+8613800138000"} msgs = _u("my phone is +8613800138000", "wrong number, use +8613999999999") assert resolve_by_recency(before, after, msgs) == { "phone": "+8613999999999", "whatsapp": "+8613999999999"} def test_recency_kills_extractor_oscillation(): # 实测bug场景: extractor 翻回更早消息里的旧号 → 裁决回最新的 ...004 before = {"phone": "+8613800138004", "whatsapp": "+8613800138004"} after = {"phone": "+8613800138000", "whatsapp": "+8613800138004"} # extractor 振荡 msgs = _u("my whatsapp is +8613800138000", "my whatsapp is +8613800138004", "any pipe talk") assert resolve_by_recency(before, after, msgs) == before # 无变化→不触发更正 def test_recency_suppresses_hallucinated_value(): # 幻觉值从未出现在消息里(last_idx=-1) → 真实陈述胜 before = {"phone": "+8613800138004", "whatsapp": "+8613800138004"} after = {"phone": "+8613999999999", "whatsapp": "+8613800138004"} msgs = _u("my whatsapp is +8613800138004") assert resolve_by_recency(before, after, msgs) == before def test_recency_bare_newest_number_wins(): # 用户实测终局: 最晚消息里的裸新号(extractor 抓到)胜过更早的陈述 before = {"phone": "+8613800138004", "whatsapp": "+8613800138004"} after = {"phone": "18325698745", "whatsapp": "+8613800138004"} msgs = _u("my whatsapp is +8613800138004", "chat", "18325698745") assert resolve_by_recency(before, after, msgs) == { "phone": "18325698745", "whatsapp": "18325698745"} def test_recency_genuinely_different_untouched(): before = {"phone": "+8613800138000", "whatsapp": "+254712345678"} after = {"phone": "+8613999999999", "whatsapp": "+254712345678"} msgs = _u("my phone is +8613800138000", "my whatsapp is +254712345678", "wrong number, use +8613999999999") assert resolve_by_recency(before, after, msgs) == after # 不同组不动 from app.engine.services.contact_pipeline import standalone_number def test_standalone_number_captures_cn_and_ke_mobiles(): assert standalone_number("15236541128") == "15236541128" assert standalone_number("0712345678") == "0712345678" assert standalone_number("+86 138 0013 8000") == "+86 138 0013 8000" def test_standalone_number_rejects_prices_years_and_prose(): assert standalone_number("2024") is None # 年份 assert standalone_number("25000") is None # 美元价 assert standalone_number("1500000") is None # 先令价(7位) assert standalone_number("50,000 km") is None # 含字母逗号 assert standalone_number("I want 5 units") is None def test_latest_stated_standalone_overrides(): msgs = m("my number is +254712345678", "chat", "15236541128") assert latest_stated_contacts(msgs) == {"phone": "15236541128"} # ---- P0 捕获缺口扩充: 号码在前/无is/msg me/wa冒号 ---- # 全部经 scan_user_messages 断言,不直接 import 新正则符号 def test_scan_number_before_whatsapp(): # 倒装: "0712 345 678 is my whatsapp" assert scan_user_messages(m("0712 345 678 is my whatsapp")) == { "whatsapp": "0712 345 678"} def test_scan_number_before_wechat_cn(): # 中文倒装(我的后无空格——正则必须用 \s* 而非 \s+,实证 \s+ 永不命中) assert scan_user_messages(m("13800138000 我的微信")) == { "wechat": "13800138000"} assert scan_user_messages(m("13800138000 这是我的微信")) == { "wechat": "13800138000"} def test_scan_my_number_without_is(): # "my number 0712345678"(无 is/冒号) assert scan_user_messages(m("my number 0712345678")) == { "phone": "0712345678"} def test_scan_text_me_forms(): assert scan_user_messages(m("text me at 07123456789")) == { "phone": "07123456789"} assert scan_user_messages(m("msg me 07123456789")) == { "phone": "07123456789"} def test_scan_wa_colon_local_number(): assert scan_user_messages(m("wa: 07123456789")) == { "whatsapp": "07123456789"} def test_scan_new_forms_do_not_grab_prices(): # 误报防护: 价格/单号即使与 whatsapp 词同现也不得捕获(无 my 指称/无邻接) assert scan_user_messages(m("paid 2500000 via whatsapp transfer")) == {} assert scan_user_messages(m("my budget 25000 number one priority")) == {} assert scan_user_messages(m("order 1234567 is my order number")) == {} # 排位表述(实证: 若倒装正则收 number/phone,"2500000 is my number" 被误捕, # 且 latest_stated_contacts 覆盖权会把真实号码翻新成价格)——倒装只收显式渠道词 assert scan_user_messages(m("2500000 is my number one choice")) == {} # wa 词界守卫: "my watch/way" 的 wa 后无词边界,不得命中 assert scan_user_messages(m("0712345678 my watch")) == {} assert scan_user_messages(m("0712345678 my way home")) == {} def test_bare_my_number_has_no_latest_authority(): # 回归钉死: 裸 "my number X"(无 is/冒号)只进扫描辅源,不得进 # latest_stated_contacts 获覆盖权——否则首轮旧值长期压制 extractor # 后续改号(实证: test_correction_up_to_three 两封更正邮件全灭)。 # 预期 {} 依赖 "wrong,"(逗号后无 number/phone)不命中 CORRECTION_CTX_RE。 msgs = m("my number +8613800138000", "wrong, use +8613800139999", "again +8613800122222") assert latest_stated_contacts(msgs) == {} def test_number_first_whatsapp_whitespace_variants(): # 回归(评审): whats\s*app 捕获的任意空白变体(双空格/NBSP)不得让 # _NUMBER_FIRST_CHANNEL 查表 KeyError——scan 在每条用户消息的热路径上 # 运行(conversation.py 无 try/except),崩溃即断流整轮对话。 assert scan_user_messages(m("0712345678 is my whats app")) == { "whatsapp": "0712345678"} assert scan_user_messages(m("0712345678 is my whats\xa0app")) == { "whatsapp": "0712345678"} # 两个接线位(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.engine.services.contact_pipeline import canon assert canon("+٩٦٦ ٥٠١ ٢٣٤ ٥٦٧") == canon("+966 501 234 567")