group_fqcd_jr/tools/purge_expired_knowledge_vectors.py

"""批量补投「向量残留清理」：把**已过期但向量还在**的知识逐条送去清理（走管理接口）。

    python tools/purge_expired_knowledge_vectors.py                # 只读：列出要清理哪些
    python tools/purge_expired_knowledge_vectors.py --apply        # 真投（每条一个 HTTP 调用）
    python tools/purge_expired_knowledge_vectors.py --apply --wait 30   # 投完等 Worker 清完再复核

## 它解决什么

`fin_knowledge_meta.status = 'expired'` 只说明"这行下架了"，**不代表 Milvus 里的向量没了**。
向量删除靠一条 `knowledge.vector_delete_requested` 事件，历史上有一批行是**别的途径**变成
expired 的（早期脚本直接改库、导入侧幂等上线前的重复副本），它们**从来没投过**这条事件：

- 检索侧只过滤 `visibility`、**不看 `status`**（`app/service/knowledge_search_service.py`），
  所以这些死向量**继续参与排序**，和同题的活块抢答；
- 而 `DELETE /api/v1/knowledge/{id}` 对已经 expired 的行**刻意返回 404**（"不静默成功"），
  于是这些行在管理端口上**无路可走** —— 本脚本 + 新端点 `POST /{id}/vector-cleanups`
  就是补这个缺口。

## 与对账工具的关系

先用 `python tools/reconcile_knowledge_vectors.py` 看全局，再用本脚本动手。本脚本**只清理
对账认定"确实还有向量"的那些 id**（`stale_vectors_ids`）—— 对一堆早就干净的 expired 行
盲投事件，只会往 `domain_event_outbox` 里灌没有意义的删除事件，让"事件堆积"变成常态噪音。

## 为什么不直接连 Milvus 删

那样会**同时绕过**权限、审计与 Outbox 幂等语义（删除是不可逆动作，必须留痕、必须有唯一入口）。
孤儿向量（MySQL 里连行都没有的那种）没有可挂事件的载体，本脚本**默认不处理**，
只在对账报告里点出来 —— 真要删属于"直接操作向量库"的运维动作，须单独授权。
"""

from __future__ import annotations

import argparse
import importlib.util
import sys
import time
import uuid
from pathlib import Path
from typing import Any

import httpx

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(errors="replace")

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

BASE_DEFAULT = "http://127.0.0.1:8000"


def _load_reconcile_module() -> Any:
    """按文件路径加载对账工具（`tools/` 不是包，没有 `__init__.py`，不能 `import tools.x`）。"""
    path = ROOT / "tools" / "reconcile_knowledge_vectors.py"
    spec = importlib.util.spec_from_file_location("reconcile_knowledge_vectors", path)
    if spec is None or spec.loader is None:  # pragma: no cover - 只在文件缺失时发生
        raise SystemExit(f"找不到对账工具：{path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}", "Idempotency-Key": uuid.uuid4().hex}


def _login(client: httpx.Client, base: str, username: str, password: str) -> str:
    response = client.post(
        f"{base}/api/v1/auth/tokens", json={"username": username, "password": password}
    )
    if response.status_code != 200:
        raise SystemExit(f"登录失败：HTTP {response.status_code} {response.text[:160]}")
    return str(response.json()["data"]["access_token"])


def stale_ids_by_collection() -> dict[str, list[str]]:
    """本机对账一次，返回「还有向量的已过期行」：`{集合: [knowledge_id, ...]}`。"""
    reconcile = _load_reconcile_module()
    report = reconcile.reconcile(
        reconcile.mysql_rows(), reconcile.milvus_vectors(sorted(_collections()))
    )
    return {
        name: list(item["detail"]["stale_vectors_ids"])
        for name, item in report.get("collections", {}).items()
        if item["detail"]["stale_vectors_ids"]
    }


def _collections() -> list[str]:
    from app.core.knowledge_contracts import ALLOWED_COLLECTIONS

    return sorted(ALLOWED_COLLECTIONS)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="补投已过期知识的向量清理（走管理接口）")
    parser.add_argument("--base", default=BASE_DEFAULT, help="平台地址")
    parser.add_argument("--username", default="admin_t")
    parser.add_argument("--password", default="88888888")
    parser.add_argument("--apply", action="store_true", help="真正投递（默认只列出要清理的）")
    parser.add_argument("--wait", type=int, default=0, metavar="秒",
                        help="投完等 Worker 消费，再复核一次残留数")
    parser.add_argument("--limit", type=int, default=0, help="最多处理多少条（0=不限）")
    args = parser.parse_args(argv)

    stale = stale_ids_by_collection()
    total = sum(len(ids) for ids in stale.values())
    if total == 0:
        print("对账结果：没有任何「已过期但向量还在」的知识行，无需清理。")
        print("（要看全局请跑 python tools/reconcile_knowledge_vectors.py）")
        return 0

    print(f"对账发现 {total} 条已过期知识仍有向量（检索侧不看 status，它们仍在参与排序）：")
    targets: list[str] = []
    for name, ids in stale.items():
        print(f"  · {name}: {len(ids)} 条 → {ids[:12]}{' …' if len(ids) > 12 else ''}")
        targets.extend(ids)
    if args.limit > 0:
        targets = targets[: args.limit]
        print(f"（--limit {args.limit}：本次只处理前 {len(targets)} 条）")

    if not args.apply:
        print("\n[dry-run] 未投递。加 --apply 真投（每条一次 POST /api/v1/knowledge/{id}/vector-cleanups）。")
        return 0

    with httpx.Client(base_url=args.base, timeout=60) as client:
        token = _login(client, args.base, args.username, args.password)
        ok = 0
        failed: list[tuple[str, str]] = []
        for knowledge_id in targets:
            response = client.post(
                f"{args.base}/api/v1/knowledge/{knowledge_id}/vector-cleanups",
                headers=_headers(token),
            )
            if response.status_code == 200:
                ok += 1
            else:
                failed.append((knowledge_id, f"HTTP {response.status_code} {response.text[:120]}"))
        print(f"\n投递完成：成功 {ok} / 失败 {len(failed)}")
        for knowledge_id, reason in failed[:10]:
            print(f"  ✗ {knowledge_id}: {reason}")

    if args.wait > 0:
        print(f"等 {args.wait} 秒让 Worker 消费删除事件 …")
        time.sleep(args.wait)
        remaining = stale_ids_by_collection()
        left = sum(len(ids) for ids in remaining.values())
        print(f"复核：仍有向量的已过期行 {left} 条"
              + ("" if left == 0 else f" → { {k: v[:8] for k, v in remaining.items()} }"))
        if left > 0:
            print("  提示：确认 Worker 在跑（python -m app.worker），并看 domain_event_outbox 里"
                  " knowledge.vector_delete_requested 是否卡在 failed/dead。")
    return 0


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