"""语料入库门禁（B-01 四查，`DEC-16` / `乙-18`）。

**为什么单独成模块**：原先这段写在 `tools/build_knowledge_chunks.py` 里，而该脚本
一 import 就会重写 `knowledge/_chunks.jsonl` —— 单测没法安全地 import 它来验证门禁，
于是门禁本身成了"只有跑构建时才顺带执行、谁也没测过"的代码。抽出来之后两侧都可测：
构建脚本 import 它，单测也 import 它。

**为什么放在写入前**：这四个条件一旦漏掉，错误会以「检索出来的答案」的形式出现在客户面前，
而不是以报错的形式出现在构建日志里。所以宁可中止构建、并报出块号。

四查：

1. **占位符**：`400-XXX-XXXX` / `XX科技` / `TBD` 这类没填完的值 —— 一眼假。
2. **品牌位**：以「南方」开头的机构名必须落在白名单里，域名必须是 `nffund.com`；
   另显式列出已知**错值**（`南方财富`、`南方科技`、`nanfangwm.com`、
   少一个「股份」的 `南方基金管理有限公司`）。白名单值**只用 `南方基金` + `nffund.com`**：
   旧的 `南方财富` + `nanfangwm.com` 会把**错误品牌**放行，等于门禁白做。
   第三方机构（华泰证券、厦门国际信托等）不以「南方」开头，不参与品牌位判定 ——
   它们是**股东/交易对手**，不是本平台品牌；`南方东英` / `南方资本` 是真实存在的
   境内外子公司，用负向断言显式排除。
3. **禁止类目**：内部管理类源文件（新人指南、反洗钱、薪酬考勤）不进客户侧知识库。
4. **档位显式声明**：每块必须有非空 `visibility`，缺省即拒绝写入 —— 堵住
   `milvus_knowledge_writer` 的 `FIELD_DEFAULTS` 把缺省当 `public` 的 fail-open。

另有一条**不在「四查」里但必须一起守**的检查：`doc_id` 唯一性。2026-09-19 实测踩到过 ——
两个新源文件若共用同一个 `prefix`，生成的 doc_id 会**整体重号**，灌库时后一份按 `upsert`
直接**覆盖**前一份，而构建日志一切正常（块数、档位都对），只有在下游检索里才发现
"有一半内容不存在"。
"""

import re
from collections.abc import Mapping, Sequence

PLACEHOLDER_PATTERN = re.compile(r"400-XXX-XXXX|XXX-XXXX-XXXX|XX科技|xxtech|待填|TBD|TODO")
BRAND_SLOT_PATTERN = re.compile(
    r"南方(?!基金|东英|资本)[\u4e00-\u9fa5]{0,8}(?:基金|财富|科技|证券|资产管理|银行|信托|保险)"
    r"(?:管理)?(?:股份)?(?:有限)?公司"
)
DOMAIN_PATTERN = re.compile(r"(?:[A-Za-z0-9][A-Za-z0-9-]*\.)+(?:com|cn|net|org)")
ALLOWED_BRAND_DOMAINS = ("nffund.com",)
BANNED_BRAND_LITERALS = ("南方财富", "南方科技", "nanfangwm.com", "南方基金管理有限公司")
FORBIDDEN_SOURCE_KEYWORDS = ("新人指南", "反洗钱", "薪酬", "考勤", "晋升", "绩效")


def corpus_gate_problems(records: Sequence[Mapping[str, object]]) -> list[str]:
    """返回违规清单（空列表 = 通过）。分离出来是为了让单测能逐条断言命中原因。"""
    problems: list[str] = []
    seen_doc_ids: set[str] = set()
    for record in records:
        doc_id = str(record.get("doc_id", ""))
        content = str(record.get("content", ""))
        if doc_id in seen_doc_ids:
            problems.append(f"{doc_id}: doc_id 重复（两个源的 prefix 撞了，后写的会覆盖前一份）")
        seen_doc_ids.add(doc_id)
        if not str(record.get("visibility") or "").strip():
            problems.append(f"{doc_id}: 未显式声明 visibility")
        placeholder = PLACEHOLDER_PATTERN.search(content)
        if placeholder:
            problems.append(f"{doc_id}: 含占位符 {placeholder.group(0)!r}")
        for banned in BANNED_BRAND_LITERALS:
            if banned in content:
                problems.append(f"{doc_id}: 含错误品牌值 {banned!r}")
        slot = BRAND_SLOT_PATTERN.search(content)
        if slot:
            problems.append(f"{doc_id}: 品牌位出现白名单外的机构名 {slot.group(0)!r}")
        for domain in DOMAIN_PATTERN.findall(content):
            if not any(
                domain == allowed or domain.endswith("." + allowed)
                for allowed in ALLOWED_BRAND_DOMAINS
            ):
                problems.append(f"{doc_id}: 域名 {domain!r} 不在白名单内")
        source_file = str(record.get("source_file") or "")
        for keyword in FORBIDDEN_SOURCE_KEYWORDS:
            if keyword in source_file:
                problems.append(f"{doc_id}: 源文件 {source_file!r} 落在禁止类目（{keyword}）")
    return problems


def assert_corpus_gate(records: Sequence[Mapping[str, object]]) -> None:
    """B-01 四查；任一不满足即中止构建，绝不写出半成品 jsonl。"""
    problems = corpus_gate_problems(records)
    if problems:
        detail = "\n".join(f"  - {line}" for line in problems[:20])
        raise SystemExit(
            f"B-01 语料入库门禁未通过（{len(problems)} 处）：\n{detail}\n已中止，未写入 jsonl。"
        )