浅念
e924b3a6a4
Task 5：app/platform/site_import/generator.py——AI 两段（行业分析/ 内容填槽，GEN1/GEN2 系统提示含数据边界声明与 JSON 严格指令）+ 数字校验器（全半角/单位/千分位/波浪号归一比对，违例剔除+孤立单位 清理）+ spec §4 草稿组装（占位符填槽/出厂默认兜底/规格公开分支/ 模板残留剔除/墙尾巴三语 es 降级 null）。create_provider 加可选 max_tokens 透传（默认 None 既有调用零影响）；MAX_TOKENS_TARGET=8192 （Task 1 探针结论）。AI 调用 wait_for 120s 包裹，超时不吃重试预算。 测试 38 条（provider 全 fake），全量 774 passed + 2 skipped。 Co-Authored-By: Claude Code <noreply@anthropic.com>
404 lines
18 KiB
Python
404 lines
18 KiB
Python
"""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(), "页面文本")
|