"""site_import 生成器：两段 AI + 校验器（Task 5，provider 全 fake）。""" import asyncio import json from pathlib import Path from unittest.mock import patch import pytest from app.platform.site_import import generator def _cjk_ratio(s: str) -> float: """中文字符占比（分母=非空白字符）——生成物中文断言的口径。""" total = sum(1 for c in s if not c.isspace()) cjk = sum(1 for c in s if "一" <= c <= "鿿") return cjk / max(1, total) def _placeholder_names() -> list[str]: """测试自读 placeholders.json（不依赖被测模块常量，RED 期可用）。""" path = (Path(__file__).resolve().parents[1] / "app" / "platform" / "site_import" / "template" / "placeholders.json") names: list[str] = [] for group in json.loads(path.read_text(encoding="utf-8")).values(): for n in group: if n not in names: names.append(n) return names class FakeProvider: """按预设脚本回放的假 provider——replies 依次返回。 判别两段调用：GEN1_SYSTEM 含「行业分析」锚词（实现侧约定 GEN2 不含）。 """ def __init__(self, replies: list[str]): self.replies = list(replies) self.calls: list[tuple[str, str]] = [] # (tag, user content 全文) async def chat(self, messages, system_prompt): tag = "1" if "行业分析" in system_prompt else "2" self.calls.append((tag, messages[-1]["content"])) if not self.replies: raise AssertionError("FakeProvider 回放耗尽") return self.replies.pop(0) ANALYSIS_JSON = { # 调用 1 的合法产物样例（schema 照 generator 实现） "company_name": "Acme Energy", "profile": "Acme Energy 是一家专注储能系统的出口制造商，成立于2015年，产品销往全球。", "product_lines": [ {"name": "户用储能系统", "points": ["一体机设计", "壁挂安装"]}, {"name": "工商业储能系统", "points": ["模块化扩容", "液冷散热"]}, ], "specs_published": True, "specs_evidence": "产品页含 950°C 与 75~90% 参数表", "certifications": ["CE", "IEC"], "applications": ["家庭备用电源", "工商业削峰"], "contacts": {"whatsapp": "+8618603858592", "email": "sales@acme.com"}, "customers": "海外安装商、经销商与企业主", "keywords": ["储能", "battery"], "es_product_word": "sistemas de almacenamiento", } ANALYSIS_RAW = json.dumps(ANALYSIS_JSON, ensure_ascii=False) GEN2_JSON = { # 调用 2 的合法产物样例：占位符全填 + 品牌四件 + 墙尾巴三语 "身份描述": "你是一位储能系统出口外贸销售（Acme Energy），户用储能和工商业储能都做，" "日常工作就是通过聊天和海外的安装商、经销商沟通。" "你说话像在 WhatsApp 上跟朋友聊天一样，简短、直接、有人情味。", "产品类别清单": "户用储能、工商业储能、逆变器等官网公开产品", "认证清单": "CE、IEC", "预算数字示例": "\"预算五万\" 这类数字形式", "固定追问句": "What's the scenario — home or business?", "零上下文首轮例句": "\"What's the scenario — home or business? And which country?\"", "产品统称EN": "storage systems", "推荐建议": "按用电场景直接推荐合适的品类", "产品统称": "储能系统", "已知信息重问示例": "客户说过场景后不要再问场景——重复问属于违规", "规格追问三要素": "场景、容量、并网要求——一次问一个，不要连环轰炸", "产品追问示例": "\"any specific capacity?\"", "品类错推示例": "客户指定户用后推荐工商业属于违规", "好差示例组": "- 差：\"What do you need?\"／好：\"Got it — home storage. Daily use or backup?\"", "关注信息": "目标国家、应用场景、规模、预算", "采购域": "储能设备", "示例字段清单": "目标国家、意向产品线、应用场景、规模、联系方式、其他备注", "字段说明组": "- 意向产品线：客户想采购的产品线\n- 应用场景：自由文本，保留客户原话", "外文对照例": "Kenya→肯尼亚", "原样保留字段": "应用场景、其他备注", "意向要素": "产品线与容量", "浏览对象": "产品资料", "缺失关键信息": "场景、规模或联系方式", "正例": "\"home battery for my house, 10kWh, in Kenya\"", "负例": "\"just looking\"", "浏览例句": "\"what products do you have\"", "brand_title": "Acme Support", "welcome": "Hi! What energy project are you working on?", "topics": ["Residential ESS", "C&I ESS", "Inverter"], "placeholder": "Type your message...", "wall_tail_en": ("Our manager will send you the full quotation for our " "storage systems — please leave your WhatsApp."), "wall_tail_zh": "客户经理会把储能系统的完整报价发您——请留下您的WhatsApp。", "wall_tail_es": ("Nuestro gerente le enviará la cotización completa de " "nuestros sistemas de almacenamiento — deje su WhatsApp."), } # 样例覆盖守卫：GEN2_JSON 必须含全部占位符键（缺键会让下游测试假绿/假红） assert set(_placeholder_names()) <= set(GEN2_JSON), ( set(_placeholder_names()) - set(GEN2_JSON)) GEN2_RAW = json.dumps(GEN2_JSON, ensure_ascii=False) SOURCE_TEXT = ( "Acme Energy 成立于2015年，专注储能系统制造与出口。" "产品页公开参数：工作温度 950°C，石墨化度 75~90%。" "户用储能 10kWh 一体机，壁挂安装；工商业储能模块化扩容。" "WhatsApp: +8618603858592 sales@acme.com 认证 CE、IEC。" ) CRAWL = { # Task 3/6 抓取产物形态（测试给假值） "color": "#0a7ea4", "whatsapp": "+8618603858592", "email": "sales@acme.com", "logo_data_url": "data:image/png;base64,aGVsbG8=", "pages_text": SOURCE_TEXT, } class TestCall1Parse: async def test_裸JSON直接解析(self): p = FakeProvider([ANALYSIS_RAW]) out = await generator.analyze(p, "页面文本") assert out["company_name"] == "Acme Energy" assert len(p.calls) == 1 and p.calls[0][0] == "1" async def test_markdown栅栏剥离(self): p = FakeProvider(["```json\n" + ANALYSIS_RAW + "\n```"]) out = await generator.analyze(p, "页面文本") assert out["company_name"] == "Acme Energy" async def test_杂文前后剥离(self): p = FakeProvider(["好的，以下是分析结果：\n" + ANALYSIS_RAW + "\n希望有帮助！"]) out = await generator.analyze(p, "页面文本") assert out["product_lines"][0]["name"] == "户用储能系统" async def test_解析失败自动重试一次成功(self): p = FakeProvider(["抱歉，我不知道怎么输出。", ANALYSIS_RAW]) out = await generator.analyze(p, "页面文本") assert out["company_name"] == "Acme Energy" assert len(p.calls) == 2 assert "严格只输出 JSON" in p.calls[1][1] async def test_重试再失败抛GenerateError(self): p = FakeProvider(["垃圾", "还是垃圾"]) with pytest.raises(generator.GenerateError): await generator.analyze(p, "页面文本") assert len(p.calls) == 2 async def test_schema缺键也触发重试(self): bad = json.dumps({"unrelated": 1}, ensure_ascii=False) p = FakeProvider([bad, ANALYSIS_RAW]) out = await generator.analyze(p, "页面文本") assert out["company_name"] == "Acme Energy" assert len(p.calls) == 2 async def test_入口截断24000(self): p = FakeProvider([ANALYSIS_RAW]) await generator.analyze(p, "x" * 30000) user = p.calls[0][1] assert "x" * 24000 in user assert "x" * 24001 not in user async def test_空文本抛GenerateError(self): p = FakeProvider([ANALYSIS_RAW]) with pytest.raises(generator.GenerateError): await generator.analyze(p, " ") class TestCall2Generate: async def test_生成返回占位符dict(self): p = FakeProvider([GEN2_RAW]) out = await generator.generate(p, ANALYSIS_JSON) assert out["身份描述"] == GEN2_JSON["身份描述"] assert out["wall_tail_es"] == GEN2_JSON["wall_tail_es"] assert p.calls[0][0] == "2" async def test_用户消息含fewshot与分析与键清单(self): p = FakeProvider([GEN2_RAW]) await generator.generate(p, ANALYSIS_JSON) user = p.calls[0][1] assert "语料摘录" in user # fewshot 文件标题锚 assert "brand_title" in user and "身份描述" in user assert "Acme Energy" in user # analysis JSON 注入 async def test_解析失败重试(self): p = FakeProvider(["不行", GEN2_RAW]) out = await generator.generate(p, ANALYSIS_JSON) assert out["brand_title"] == "Acme Support" assert "严格只输出 JSON" in p.calls[1][1] async def test_完全无可识别键重试再失败(self): p = FakeProvider(['{"foo": 1}', '{"bar": 2}']) with pytest.raises(generator.GenerateError): await generator.generate(p, ANALYSIS_JSON) assert len(p.calls) == 2 class TestNumbersValidator: def test_normalize_number全半角千分位(self): assert generator.normalize_number("１,０００") == "1000" assert generator.normalize_number("950") == "950" def test_全半角与单位归一视为同数字(self): # 源文 "950°C"，生成物 "950℃" → 合法（不剔除） # 源文 "75~90%"，生成物 "75～90%" → 合法 text, removed = generator.validate_numbers( "耐温 950℃，石墨化度 75～90%", "950°C 75~90%") assert removed == [] assert "950" in text and "75~90%" in text def test_千分位与全角数字归一(self): text, removed = generator.validate_numbers( "产能 1,000 台／功率１０００ kW", "capacity 1000") assert removed == [] assert "1000" in text def test_源文没有的数字剔除并注记(self): text, removed = generator.validate_numbers("质保 40 years 经验", "十年经验老厂") assert "40" not in text assert "40" in removed def test_剔除后文本不残留孤立单位(self): text, removed = generator.validate_numbers("强度 ≥40 MPa", "无参数") assert "40" not in text assert "MPa" not in text and "≥" not in text assert removed == ["40"] class TestIndustryBranch: def test_规格公开_人设允许聊参数(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert "可照事实卡答参数" in draft["prompts"]["system"] assert "可照卡作答" in draft["prompts"]["company_facts"] assert any(w.startswith("规格判定：公开") for w in draft["warnings"]) def test_规格不公开_参数转经理(self): analysis = dict(ANALYSIS_JSON, specs_published=False) draft = generator.build(GEN2_JSON, analysis, CRAWL, SOURCE_TEXT) facts = draft["prompts"]["company_facts"] assert "参数未公布" in facts and "兜底" in facts assert "可照事实卡答参数" not in draft["prompts"]["system"] assert "可照卡作答" not in facts # 不公开时无性能指标节 assert any(w.startswith("规格判定：不公开") for w in draft["warnings"]) class TestBuild: def test_占位符全填充_模板零残留(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) for name in ("system", "company_facts", "extractor", "evaluator"): assert "{{" not in draft["prompts"][name], name assert not any("已用默认" in w for w in draft["warnings"]) # 模板固定段原文保留（红线零漂移锚） assert "红线（任何情况下不可违反）" in draft["prompts"]["system"] assert "储能设备" in draft["prompts"]["extractor"] # 采购域填入 def test_品牌四件在输出(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) b = draft["brand"] assert b["title"] == "Acme Support" assert b["welcome"] == GEN2_JSON["welcome"] assert b["topics"] == GEN2_JSON["topics"] assert b["placeholder"] == GEN2_JSON["placeholder"] assert b["color"] == "#0a7ea4" assert b["whatsapp"] == "+8618603858592" assert b["email"] == "sales@acme.com" assert b["privacy_url"] == "" def test_头像透传与缺省null(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert draft["avatar"]["data_url"] == CRAWL["logo_data_url"] assert draft["avatar"]["mime"] == "image/png" draft2 = generator.build( GEN2_JSON, ANALYSIS_JSON, dict(CRAWL, logo_data_url=None), SOURCE_TEXT) assert draft2["avatar"] is None def test_wall尾巴三语(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert draft["wall_tails"] == { "en": GEN2_JSON["wall_tail_en"], "zh": GEN2_JSON["wall_tail_zh"], "es": GEN2_JSON["wall_tail_es"], } def test_es生成失败降级为null(self): gen = {k: v for k, v in GEN2_JSON.items() if k != "wall_tail_es"} draft = generator.build(gen, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert draft["wall_tails"]["es"] is None assert draft["wall_tails"]["en"] == GEN2_JSON["wall_tail_en"] assert not any("wall_tail_es" in w for w in draft["warnings"]) def test_wall尾巴en缺失出厂默认加warning(self): gen = {k: v for k, v in GEN2_JSON.items() if k != "wall_tail_en"} draft = generator.build(gen, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert draft["wall_tails"]["en"] assert any("wall_tail_en" in w and "默认" in w for w in draft["warnings"]) def test_残留命中在build层从联系方式与事实卡剔除(self): analysis = dict(ANALYSIS_JSON) analysis["profile"] = ANALYSIS_JSON["profile"] + "\n全球领先的 Lorem ipsum 宣传占位文本" crawl = dict(CRAWL, whatsapp="+15551234567") # 555 占位号段 draft = generator.build(GEN2_JSON, analysis, crawl, SOURCE_TEXT) assert draft["brand"]["whatsapp"] is None assert "Lorem ipsum" not in draft["prompts"]["company_facts"] assert any("已剔除疑似模板内容" in w for w in draft["warnings"]) def test_生成物中文断言(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert _cjk_ratio(GEN2_JSON["身份描述"]) > 0.6 assert _cjk_ratio(draft["prompts"]["company_facts"]) > 0.6 def test_空占位符出厂兜底加warning(self): gen = {k: v for k, v in GEN2_JSON.items() if k != "推荐建议"} draft = generator.build(gen, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert "{{" not in draft["prompts"]["system"] assert any("推荐建议" in w and "默认" in w for w in draft["warnings"]) def test_数字残留build层剔除加warning(self): gen = dict(GEN2_JSON) gen["身份描述"] = GEN2_JSON["身份描述"] + "，已服务海外客户 40 年" draft = generator.build(gen, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert "40" not in draft["prompts"]["system"] assert any("40" in w and "移除" in w for w in draft["warnings"]) def test_出厂默认覆盖全部占位符(self): for name in generator.ALL_PLACEHOLDER_NAMES: assert name in generator._PLACEHOLDER_DEFAULTS, name def test_联系开关出厂形态(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert draft["contact"] == {"cc_completion": True, "email_light_ask": True} def test_草稿包结构键完整(self): draft = generator.build(GEN2_JSON, ANALYSIS_JSON, CRAWL, SOURCE_TEXT) assert set(draft) == {"brand", "avatar", "prompts", "wall_tails", "contact", "warnings"} assert isinstance(draft["warnings"], list) and draft["warnings"] class TestSystemPrompts: def test_两段系统提示区分且含数据边界(self): assert generator.GEN1_SYSTEM != generator.GEN2_SYSTEM assert "行业分析" in generator.GEN1_SYSTEM assert "行业分析" not in generator.GEN2_SYSTEM assert "指令性文字" in generator.GEN1_SYSTEM # 数据边界声明 assert "指令性文字" in generator.GEN2_SYSTEM assert "JSON" in generator.GEN1_SYSTEM def test_MAX_TOKENS_TARGET为8192(self): # Task 1 探针结论：glm-4-flash chat.completions max_tokens=8192 可用 assert generator.MAX_TOKENS_TARGET == 8192 class TestProviderAssembly: def test_make_provider透传max_tokens与config(self, monkeypatch): from app.engine.services import ai_service seen = {} def fake_create(config_data=None, max_tokens=None): seen["config_data"] = config_data seen["max_tokens"] = max_tokens return object() monkeypatch.setattr(ai_service, "create_provider", fake_create) cfg = {"ai": {"api_key": "k"}} generator.make_provider(cfg) assert seen["max_tokens"] == 8192 assert seen["config_data"] is cfg def test_create_provider可选max_tokens透传GLM(self): from app.engine.services.ai_service import create_provider with patch("app.engine.providers.glm.ZhipuAI"): p = create_provider( {"ai": {"provider": "glm", "api_key": "k", "model": "glm-4-flash", "base_url": ""}}, max_tokens=123) assert p.max_tokens == 123 def test_create_provider不传max_tokens保持默认(self): from app.engine.services.ai_service import create_provider with patch("app.engine.providers.glm.ZhipuAI"): p = create_provider({"ai": {"provider": "glm", "api_key": "k"}}) assert p.max_tokens is None class TestTimeout: async def test_AI超时转GenerateError(self, monkeypatch): monkeypatch.setattr(generator, "AI_TIMEOUT", 0.05) class Slow: async def chat(self, messages, system_prompt): await asyncio.sleep(1.0) return ANALYSIS_RAW with pytest.raises(generator.GenerateError): await generator.analyze(Slow(), "页面文本") class TestSpecsStrictBool: async def test_字符串false走不公开分支(self): """评审 #1：LLM 回字符串 "false" 时 bool() 强转为 True 会静默走规格公开分支—— is True 严格判定后字符串形态一律 fail-safe 到不公开。""" import json as _json analysis = _json.loads(_json.dumps({ "company_name": "X", "product_lines": [{"name": "p", "points": []}], "specs_published": "false"})) gen = dict(_MIN_GEN) if "_MIN_GEN" in globals() else { "身份描述": "你是销售", "产品统称": "产品", "产品统称EN": "products", "产品类别清单": "产品", "规格追问三要素": "场景", "好差示例组": "示例", "关注信息": "信息", "固定追问句": "问句", "推荐建议": "建议"} draft = generator.build(gen, analysis, CRAWL, SOURCE_TEXT) facts = draft["prompts"]["company_facts"] assert ("参数未公布" in facts) or ("兜底" in facts) assert any(w.startswith("规格判定：不公开") for w in draft["warnings"]) class TestCrawlWarningsMerge: def test_crawl层注记并入草稿包warnings(self): """Task 8 评审：logo 跳过/薄子页跳过等 crawl 层注记必须进 warnings（spec §3.1）。""" gen = dict(_MIN_GEN) if "_MIN_GEN" in globals() else { "身份描述": "你是销售", "产品统称": "产品", "产品统称EN": "products", "产品类别清单": "产品", "规格追问三要素": "场景", "好差示例组": "示例", "关注信息": "信息", "固定追问句": "问句", "推荐建议": "建议"} crawl = dict(CRAWL, warnings=["logo 为不支持的格式，已跳过", "子页过薄已跳过：/thin"]) draft = generator.build(gen, ANALYSIS_JSON, crawl, SOURCE_TEXT) assert "logo 为不支持的格式，已跳过" in draft["warnings"] assert "子页过薄已跳过：/thin" in draft["warnings"] class TestNestedFlatten: def test_嵌套分组自动展平(self): """Task 10 实测：glm-4-flash 常把键按模板分组嵌套——展平容错。""" from app.platform.site_import.generator import _flatten_nested nested = {"system": {"身份描述": "x", "产品统称": "y"}, "extractor": {"采购域": "z"}, "brand_title": "T"} flat = _flatten_nested(nested) assert flat == {"身份描述": "x", "产品统称": "y", "采购域": "z", "brand_title": "T"} def test_本来就扁平原样返回(self): from app.platform.site_import.generator import _flatten_nested flat_in = {"身份描述": "x", "brand_title": "T"} assert _flatten_nested(dict(flat_in)) == flat_in def test_混合形态_dict值并入非dict值保留(self): """修订语义：dict 值并入顶层、非 dict 值原位保留（不要求全 dict）。""" from app.platform.site_import.generator import _flatten_nested weird = {"a": "字符串", "b": {"身份描述": "x"}} assert _flatten_nested(weird) == {"a": "字符串", "身份描述": "x"} async def test_generate接受嵌套输出(self): """端到端：fake 回嵌套 JSON 也能过 generate。""" import json as _json nested = {"system": {"身份描述": "x"}, "brand_title": "T"} class NestedP: async def chat(self, m, s): return "```json\n" + _json.dumps(nested, ensure_ascii=False) + "\n```" out = await generator.generate(NestedP(), ANALYSIS_JSON) assert out.get("身份描述") == "x" and out.get("brand_title") == "T"