business-assistant/src/step1_knowledge_prepare/prepare.py

"""Deterministic local ETL. Source text is data; no network, eval or model calls."""

import argparse
import ast
import csv
import hashlib
import io
import json
import re
import sys
import tempfile
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
KINDS = {"general": "通用知识", "policy": "惠企政策", "qa": "问答知识", "finance": "金融产品"}
POLICY_FIELDS = {
    "policy_name": "政策名称", "department": "政策所属部门", "document_number": "政策文号",
    "reported_item": "部门上报事项", "model_suggested_item": "大模型拆解申报事项",
    "target": "支持对象", "support": "支持方式和标准", "conditions": "申报条件",
    "benefit": "企业收益", "start_text": "开始时间", "end_text": "结束时间",
    "time_note": "时间备注", "source_annotations": "metadata",
}
FINANCE_FIELDS = dict(zip(
    ("source_number", "product_name", "product_type", "institution", "institution_type",
     "approval_mode", "policy_backed", "approval_document", "interest_rate_text",
     "loan_amount_text", "loan_term_text", "guarantee", "approval_time", "repayment",
     "fees", "service_region", "target", "special_zone", "entry_conditions",
     "features", "target_description", "description", "detail_url"),
    ("序号", "产品名称", "产品类型", "金融机构", "机构类型", "审批模式", "是否政策性",
     "审批备案文号", "利率范围", "贷款额度(万元)", "贷款期限(月)", "担保方式", "审批时间",
     "还款方式", "相关费用", "服务区域", "适用对象", "特色专区", "准入条件", "产品特色",
     "适用群体说明", "产品简介", "详情页网址"), strict=True))


def dump(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)


def digest(value):
    return hashlib.sha256(value if isinstance(value, bytes) else value.encode("utf-8")).hexdigest()


def clean(value):
    if isinstance(value, str):
        return value.replace("\r\n", "\n").replace("\r", "\n").strip()
    if isinstance(value, list):
        return [clean(x) for x in value]
    if isinstance(value, dict):
        return {k: clean(v) for k, v in value.items()}
    return value


def pointer(key):
    return str(key).replace("~", "~0").replace("/", "~1")


def leaves(value, path=""):
    """Preserve field and array boundaries, including empty values, in readable text."""
    if isinstance(value, dict) and value:
        for key, item in value.items():
            yield from leaves(item, path + "/" + pointer(key))
    elif isinstance(value, list) and value:
        for i, item in enumerate(value):
            yield from leaves(item, path + "/" + str(i))
    else:
        yield {"field_path": path or "/", "text": value if isinstance(value, str) else dump(value)}


def strict_json(text):
    def pairs(items):
        result = {}
        for k, v in items:
            if k in result:
                raise ValueError("duplicate JSON key")
            result[k] = v
        return result
    def bad_constant(_):
        raise ValueError("non-finite JSON number")
    return json.loads(text, object_pairs_hook=pairs, parse_constant=bad_constant)


def read_units(path):
    raw_bytes = path.read_bytes()
    if path.suffix.lower() == '.xlsx':
        from .police_workbook import read_police_workbook
        return raw_bytes, read_police_workbook(raw_bytes)
    text = raw_bytes.decode("utf-8-sig")
    suffix = path.suffix.lower()
    if suffix == ".json":
        values = strict_json(text)
        if not isinstance(values, list) or not all(isinstance(x, dict) for x in values):
            raise ValueError("expected JSON array of objects")
        return raw_bytes, [(f"/{i}", value) for i, value in enumerate(values)]
    if suffix == ".csv":
        reader = csv.reader(io.StringIO(text, newline=""), strict=True)
        headers = next(reader)
        if not headers or len(headers) != len(set(headers)):
            raise ValueError("missing or duplicate CSV columns")
        rows = []
        previous = reader.line_num
        for i, row in enumerate(reader, 1):
            if len(row) != len(headers):
                raise ValueError(f"CSV column count mismatch at record {i}")
            rows.append((f"record:{i};lines:{previous + 1}-{reader.line_num}", dict(zip(headers, row))))
            previous = reader.line_num
        return raw_bytes, rows
    if suffix in {".txt", ".md"}:
        return raw_bytes, [(f"lines:1-{len(text.splitlines())}", {"text": text})]
    raise ValueError("unsupported file extension")


def first(value, keys):
    return next((value[k] for k in keys if value.get(k)), None)


def normalize(rel, locator, raw, sha):
    value = clean(raw)
    path = Path(rel)
    flags = ["validity_not_verified"]
    kind = {"惠企政策.json": "policy", "qa.json": "qa", "E企融金融产品列表.csv": "finance"}.get(path.name, "general")
    if kind == "policy" and "content" in value:
        # New exports wrap the original policy object in content; retain the
        # untouched wrapper in raw_record and reject malformed bodies explicitly.
        detail = strict_json(value["content"]) if isinstance(value["content"], str) else value["content"]
        if not isinstance(detail, dict):
            raise ValueError("expected policy content object")
        value = {**clean(detail), **value}
    title = first(value, ["title", "标题", "政策名称", "question", "产品名称"])
    url = first(value, ["link", "原文地址", "政策原文地址", "详情页网址"])
    if kind == "policy":
        data = {key: value.get(source) for key, source in POLICY_FIELDS.items()}
        # The 170-item export renamed the item field and flattened the
        # model-assisted annotations. Keep those annotations separate from
        # official conditions and expose the cited policy text as evidence.
        data["reported_item"] = data["reported_item"] or value.get("申报事项")
        data["source_text"] = value.get("涉及原文")
        if data["source_annotations"] is None:
            annotations = {key: value[key] for key in ("梳理后建议申报条件", "是否“免申即享”") if key in value}
            data["source_annotations"] = annotations or None
        conditions = data["conditions"]
        if isinstance(conditions, str) and conditions.startswith("["):
            try:
                parsed = ast.literal_eval(conditions)
                if isinstance(parsed, list) and all(isinstance(x, str) for x in parsed):
                    data["conditions"] = clean(parsed)
                else:
                    flags.append("conditions_unparsed")
            except (ValueError, SyntaxError):
                flags.append("conditions_unparsed")
        flags.append("source_model_annotations_unverified")
    elif kind == "qa":
        data = {"question": value.get("question"), "answer": value.get("answer")}
    elif kind == "finance":
        data = {key: value.get(source) for key, source in FINANCE_FIELDS.items()}
        flags.append("financial_terms_not_verified")
    else:
        structured = None
        body = value.get("content")
        if isinstance(body, str) and body.startswith(("{", "[")):
            try:
                structured = strict_json(body)
                body = None
            except ValueError:
                flags.append("embedded_json_unparsed")
        if "text" in value:
            body = value["text"]
            header, separator, remaining = body.partition("=" * 20)
            if separator:
                title_match = re.search(r"^标题：(.*)$", header, re.M)
                url_match = re.search(r"^来源：(.*)$", header, re.M)
                title = title_match[1].strip() if title_match else title
                url = url_match[1].strip() if url_match else url
                body = remaining.lstrip("=").strip()
            title = title or path.stem
        if body is None and structured is None:
            structured = {k: v for k, v in value.items() if k not in {"标题", "原文地址"}}
        data = {"body": body, "structured_content": structured,
                "document_role": "catalog" if path.suffix == ".md" else "knowledge"}
        if data["document_role"] == "catalog":
            flags.append("catalog_not_policy_evidence")
    if not title:
        title = path.stem
        flags.append("title_from_filename")
    if not url:
        flags.append("source_url_missing")
    elif not isinstance(url, str) or not re.match(r"^https?://[^\s]+$", url):
        flags.append("source_url_invalid")
    if kind == "qa" and not data.get("answer"):
        flags.append("empty_answer")
    if kind == "general" and not (data["body"] or data["structured_content"]):
        flags.append("empty_content")
    # Only explicit source fields become scope. Directory classification is not applicability.
    scope = {"region": value.get("服务区域"), "target": first(value, ["支持对象", "适用对象"])}
    if kind == "general" and isinstance(data["structured_content"], dict):
        detail = data["structured_content"]
        scope["region"] = first(detail, ["区划名称", "所属区划"])
        basic = detail.get("基本信息", {})
        if isinstance(basic, dict):
            scope["target"] = basic.get("服务对象")
    if not scope["region"]:
        flags.append("region_not_structured")
    if not scope["target"]:
        flags.append("target_not_structured")
    record_id = "k_" + digest(rel + "#" + locator)[:24]
    sections = list(leaves(data))
    for i, section in enumerate(sections, 1):
        section["id"] = f"{record_id}:s{i}"
    return {
        "schema_version": "1.0", "id": record_id, "kind": kind, "title": title,
        "collection": path.parts[0] if len(path.parts) > 1 else path.stem,
        "classification_path": list(path.parts[:-1]), "applicability": scope,
        "source": {"file": "reference/" + rel, "locator": locator, "sha256": sha,
                   "url": url, "publisher": first(value, ["source", "部门", "政策所属部门"]),
                   "published_text": value.get("date"), "updated_text": value.get("update_time"),
                   "acquired_at": None},
        "validity": {"status": "unverified", "effective_from": None, "effective_to": None},
        "data": data, "sections": sections, "raw_record": raw,
        "content_sha256": digest(dump({"kind": kind, "title": title, "data": data})),
        "quality_flags": flags, "duplicate_of": None,
    }


def collect(source):
    records, files = [], []
    if not source.is_dir():
        raise ValueError("source directory does not exist")
    paths = sorted((p for p in source.rglob("*") if p.is_file()), key=lambda p: p.relative_to(source).as_posix())
    if not paths:
        raise ValueError("source directory is empty")
    for path in paths:
        rel = path.relative_to(source).as_posix()
        if path.is_symlink() or not path.resolve().is_relative_to(source.resolve()):
            raise ValueError(f"source escapes input directory: {rel}")
        try:
            raw_bytes, units = read_units(path)
            sha = digest(raw_bytes)
            records.extend(normalize(rel, locator, raw, sha) for locator, raw in units)
        except (ValueError, UnicodeError, csv.Error, StopIteration) as exc:
            raise ValueError(f"cannot parse {rel}: {type(exc).__name__}") from exc
        files.append({"file": "reference/" + rel, "sha256": sha, "bytes": len(raw_bytes), "records": len(units)})
    seen = {}
    for record in records:
        fingerprint = record["content_sha256"]
        if fingerprint in seen:
            record["duplicate_of"] = seen[fingerprint]
            record["quality_flags"].append("exact_content_duplicate")
        else:
            seen[fingerprint] = record["id"]
    return records, files


def readable(records, label):
    parts = [f"# {label}\n", "此文件由程序生成。原始事实及其有效性未经联网核验；定位信息见每条来源。\n"]
    for r in records:
        parts.extend([f"## {r['title']}\n", f"- 标识：{r['id']}\n- 分类：{' / '.join(r['classification_path']) or r['collection']}\n",
                      f"- 来源文件：{r['source']['file']}\n- 来源定位：{r['source']['locator']}\n",
                      f"- 原文地址：{r['source']['url'] or '未提供'}\n- 待核验：{', '.join(r['quality_flags'])}\n",
                      f"- 重复指向：{r['duplicate_of'] or '无'}\n"])
        for section in r["sections"]:
            # Text is quoted, so embedded Markdown/HTML is never presented as executable HTML.
            text = section["text"].replace("<", "&lt;").replace(">", "&gt;")
            parts.append(f"### {section['field_path']}\n\n" + "\n".join("> " + line for line in text.split("\n")) + "\n")
    return "\n".join(parts)


def build_payloads(records, files):
    payloads = {}
    for kind, label in KINDS.items():
        selected = [r for r in records if r["kind"] == kind]
        payloads[f"{kind}.jsonl"] = "".join(dump(r) + "\n" for r in selected)
        payloads[f"{kind}.md"] = readable(selected, label)
    report = {"schema_version": "1.0", "source_files": len(files), "records": len(records),
              "counts": dict(Counter(r["kind"] for r in records)), "files": files,
              "flags": dict(Counter(f for r in records for f in r["quality_flags"])),
              "duplicates": [{"id": r["id"], "duplicate_of": r["duplicate_of"]} for r in records if r["duplicate_of"]],
              "output_sha256": {name: digest(text) for name, text in payloads.items()}}
    payloads["report.json"] = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    return payloads, report


def verify_output(source, output, *, collector=collect):
    records, files = collector(source)
    expected, report = build_payloads(records, files)
    for name, text in expected.items():
        if (output / name).read_bytes() != text.encode("utf-8"):
            raise ValueError(f"output mismatch: {name}")
    if len({r["id"] for r in records}) != len(records):
        raise ValueError("duplicate record IDs")
    if sum(f["records"] for f in files) != len(records):
        raise ValueError("source coverage mismatch")
    return report


def run(source, output, *, collector=collect):
    source, output = source.resolve(), output.resolve()
    if output.is_relative_to(source) or source.is_relative_to(output):
        raise ValueError("input and output directories must not overlap")
    records, files = collector(source)  # Parse all inputs before touching existing output.
    payloads, report = build_payloads(records, files)
    output.mkdir(parents=True, exist_ok=True)
    for name, text in payloads.items():
        target = output / name
        if target.is_symlink():
            raise ValueError("output must not be a symbolic link")
        if target.is_file() and target.read_bytes() == text.encode("utf-8"):
            continue
        # Unique, exclusively created temporary files cannot follow an existing link.
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", newline="\n",
                                         dir=output, prefix=name + ".", suffix=".tmp", delete=False) as stream:
            tmp = Path(stream.name)
            try:
                stream.write(text)
            except BaseException:
                stream.close()
                tmp.unlink(missing_ok=True)
                raise
        try:
            tmp.replace(target)
        finally:
            tmp.unlink(missing_ok=True)
    return report


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source", type=Path, default=ROOT / "reference" / "营商助手",
                        help="raw source directory (default: reference/营商助手)")
    parser.add_argument("--output", type=Path, default=ROOT / "knowledge" / "prepared")
    parser.add_argument("--verify", action="store_true", help="verify existing output without rewriting")
    args = parser.parse_args(argv)
    try:
        report = verify_output(args.source, args.output) if args.verify else run(args.source, args.output)
    except (ValueError, OSError) as exc:
        print(f"Preparation failed: {exc}", file=sys.stderr)
        return 1
    print(dump({"files": report["source_files"], "records": report["records"], "counts": report["counts"],
                "duplicates": len(report["duplicates"]), "verified": args.verify}))
    return 0