group_fqcd_jr/tools/drop_table_header_vectors.py

"""删除「表格表头被误当数据行」产生的零信息量知识向量（一次性清理；**默认 dry-run**）。

    python tools/drop_table_header_vectors.py            # 只看要删什么
    python tools/drop_table_header_vectors.py --apply    # 真删（并导出证据）

## 这是什么

`tools/build_knowledge_chunks.py` 的 `expand_table_rows()` 曾有一个 bug：它用「第一个非分隔行」
当表格表头，且 `header` 从不重置 → **同一节里第 2 张表格起，表头行被当成数据行**，
产出形如「第九条 问卷内容及评分标准：选项 分值」的 19 字零信息量块。
《个人投资者适当性管理指南》第九条下有 16 张问卷表格 → 15 条**正文逐字相同**的碎片。

后果不是"多占点存储"，而是**检索质量**：这些碎片彼此分数贴得极近（实测 top1/次优差
**0.002**），而客服的判定是"中置信必须领先次优 ≥0.07"（`customer_service.py` MIN_GAP），
于是「风险评估问卷怎么评分」这类问题被**永远判成并列 → 转人工**，
真正有内容的块被挤到第 5 名。

脚本侧的 bug 已修（`expand_table_rows` + 自带守卫 `assert_no_duplicate_contents`），
本工具负责**清掉已经灌进 Milvus 的那批历史碎片**。

## 判定规则（为什么这样判，而不是"删所有短块"）

待删 = **语义 id**（非纯数字）的向量，且它的正文**不在**修正后产物的正文集合里
（`knowledge/_chunks.jsonl`，由修好的脚本生成）。

两条限定都是有意的：

1. **只动语义 id**：纯数字 id 是**上传路径**产生的（`POST /api/v1/knowledge/upload`），
   它们本来就不在 jsonl 里，是**合法内容**（实测「场内基金的管理费率是多少」的 top1 就是
   数字 id 179）——"不在 jsonl 里"对它们不构成删除理由。
2. **不按长度判**：短块本身是设计的一部分（「评审标准：管理人资质 15%」13 字，是有效答案）。
   缺陷特征是"**正文在正确产物里不存在**"，不是"短"。

## 为什么不走 `knowledge.vector_delete_requested` 事件

事件消费侧（`knowledge_vector_worker.remove`）要按 `knowledge_id` **回读 MySQL 行**
（`_load()`），而这批是**无元数据种子向量**（MySQL 里一行都没有）→ 事件会一路
`RecoverableAgentError` 变成 failed/dead。没有可挂事件的载体，只能直连向量库删。
**这是本条路径唯一一次绕过 Outbox**，所以本工具：默认只读、删除前导出证据、
且删除范围由上面那条可复现的规则限定（不是"看着像垃圾就删"）。
"""

from __future__ import annotations

import argparse
import json
import sys
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from app.core.config import get_settings  # noqa: E402
from app.core.knowledge_contracts import ALLOWED_COLLECTIONS  # noqa: E402
from app.core.knowledge_schema import detect_schema  # noqa: E402

CHUNKS = Path("knowledge") / "_chunks.jsonl"
DEFAULT_OUT = Path("docs/evidence/table-header-vectors-dropped.json")

#: 复核前的等待与轮询次数。**实测依据**（隔离临时集合做的实验）：`client.delete()` 是
#: **标记删除**，删完立刻 query **仍能看到**该行，约 **1 秒**后才不可见
#: （`+1s` 起精确 filter 与空 filter 全量都查不到）。
#: 所以"删完立刻复核"必然谎报"没删掉"——本工具第一版就是这么错的（报 19 条残留，
#: 换个进程一跑却是 0 条）。复核必须轮询，不能删完立即断言。
VERIFY_WAIT_SECONDS = 1.5
VERIFY_ATTEMPTS = 4


def _is_semantic_id(doc_id: str) -> bool:
    """语义 id（`POL-AST-009-07`）→ True；上传路径的数字 id（`179`）→ False。"""
    return not doc_id.isdigit()


def valid_contents() -> set[str]:
    """修正后产物的正文集合（白名单）。文件不存在时明确失败，不"当成空集"全删。"""
    if not CHUNKS.exists():
        raise SystemExit(
            f"找不到 {CHUNKS}：请先跑 python tools/build_knowledge_chunks.py 生成正确产物。\n"
            "（本工具靠它区分「该有的」与「表头碎片」；缺了它会退化成「删掉所有种子向量」。）"
        )
    return {
        str(json.loads(line)["content"])
        for line in CHUNKS.read_text(encoding="utf-8").splitlines()
        if line.strip()
    }


def collect() -> tuple[dict[str, list[str]], dict[str, Any]]:
    """返回 `({collection: [待删 doc_id]}, 明细)`；**只读**，不删任何东西。"""
    from pymilvus import MilvusClient  # type: ignore[import-untyped]

    allowed = valid_contents()
    settings = get_settings()
    client = MilvusClient(uri=settings.milvus_uri, token=settings.milvus_token or None)

    targets: dict[str, list[str]] = {}
    detail: dict[str, Any] = {"collections": {}, "checked_at": datetime.now(UTC).isoformat()}
    for name in sorted(ALLOWED_COLLECTIONS):
        schema = detect_schema(client, name)
        if not schema.usable:
            detail["collections"][name] = {"error": schema.error or "集合不可用"}
            continue
        id_field = schema.resolve("doc_id")
        content_field = schema.resolve("content")
        rows = client.query(
            collection_name=name, filter="",
            output_fields=[id_field, content_field], limit=16384,
        )
        semantic = [
            {"doc_id": str(row[id_field]), "content": str(row.get(content_field) or "")}
            for row in rows
            if _is_semantic_id(str(row[id_field]))
        ]
        doomed = [item for item in semantic if item["content"] not in allowed]
        if doomed:
            targets[name] = [item["doc_id"] for item in doomed]
        by_content: dict[str, list[str]] = {}
        for item in doomed:
            by_content.setdefault(item["content"], []).append(item["doc_id"])
        detail["collections"][name] = {
            "semantic_id_vectors": len(semantic),
            "doomed": len(doomed),
            "groups": [
                {"content": content, "count": len(ids), "doc_ids": ids}
                for content, ids in sorted(by_content.items(), key=lambda kv: -len(kv[1]))
            ],
        }
    return targets, detail


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="删除表格表头碎片向量（默认 dry-run）")
    parser.add_argument("--apply", action="store_true", help="真正删除（默认只列出）")
    parser.add_argument("--out", type=Path, default=DEFAULT_OUT, help="证据落盘路径")
    args = parser.parse_args(argv)

    targets, detail = collect()
    total = sum(len(ids) for ids in targets.values())

    print("表格表头碎片向量清理（判定：语义 id 且正文不在修正后产物里）")
    print(f"修正后产物：{CHUNKS}（{len(valid_contents())} 条正文白名单）\n")
    if total == 0:
        print("没有需要清理的向量（数据库已经是干净状态）。")
        return 0
    for name, item in detail["collections"].items():
        if not isinstance(item, dict) or "error" in item:
            print(f"⚠️  {name}: {item}")
            continue
        print(f"{name}: 语义 id 向量 {item['semantic_id_vectors']} 条，判定待删 {item['doomed']} 条")
        for group in item["groups"]:
            print(f"    {group['count']:>3} 份  {group['content'][:56]!r}")
            print(f"           {group['doc_ids'][:8]}{' …' if group['count'] > 8 else ''}")
    print(f"\n合计待删 {total} 条。")

    if not args.apply:
        print("\n[dry-run] 未删除。加 --apply 真删（并导出证据到 "
              f"{args.out}）。")
        return 0

    from pymilvus import MilvusClient  # type: ignore[import-untyped]

    settings = get_settings()
    client = MilvusClient(uri=settings.milvus_uri, token=settings.milvus_token or None)
    deleted = 0
    for name, ids in targets.items():
        # 与写路径同一个调用（`MilvusKnowledgeWriter.delete` 里也是 client.delete(ids=[...])）
        client.delete(collection_name=name, ids=ids)
        deleted += len(ids)
        print(f"  已删 {name}: {len(ids)} 条")
    detail["deleted"] = deleted
    detail["applied"] = True

    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(detail, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"\n共删除 {deleted} 条；证据（含全部 doc_id 与正文，可据此重建）已写入 {args.out}")

    after_left = total
    for attempt in range(VERIFY_ATTEMPTS):
        time.sleep(VERIFY_WAIT_SECONDS)
        after_targets, _ = collect()
        after_left = sum(len(ids) for ids in after_targets.values())
        if after_left == 0:
            break
    if after_left == 0:
        print(f"复核：待删向量已清零 ✅（第 {attempt + 1} 次轮询，"
              f"每次间隔 {VERIFY_WAIT_SECONDS}s —— Milvus 标记删除约 1 秒后才不可见）")
    else:
        print(f"⚠️ 复核：仍有 {after_left} 条待删向量，轮询 {VERIFY_ATTEMPTS} 次仍未清零。"
              "请确认 Milvus 可写、以及删除语句没有报错（本工具未吞异常）。")
    return 1 if after_left else 0


if __name__ == "__main__":
    raise SystemExit(main())