ai-kefu-admin/tests/test_site_import_generator.py

"""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"]