Agent/knowhub/scripts/ingest_research_output.py

#!/usr/bin/env python3
"""
入库研究调研结果（case.json + strategy.json）到 knowhub。

输入目录结构：
    <root>/
    ├── 00/
    │   ├── case.json      ({requirement, cases:[{...}]})
    │   └── strategy.json  ({selected_strategy:{...}, vs_alternatives, uncovered_requirements})
    ├── 01/
    │   ...

入库映射：
    case.json.requirement      → 按完整描述文本精确匹配现有 requirement（跨版本）
                                → 找不到则报错（不创建新 requirement——避免语义重复）
    case.json.cases[]          → resource 实体（多条，URL hash 去重）
    strategy.json              → strategy 实体（1 条 per folder）
    strategy.workflow_outline[].capabilities[].is_new=true
                                → capability 实体（新建，version 隔离）
                                → capability_id 已存在则直接引用

关联：
    requirement_resource  (requirement → all cases in its folder)
    requirement_strategy  (requirement → strategy)
    strategy_resource     (strategy → all cases in its folder — provenance)
    strategy_capability   (strategy → all mentioned capabilities, compose)

去重策略：
    resource id = "resource/research/{platform}/{hash12(source_url)}"
    ON CONFLICT (id) DO UPDATE → 多次调研遇到同一 URL 自动覆盖

使用：
    python knowhub/scripts/ingest_research_output.py <output_root> <version>
    e.g. python knowhub/scripts/ingest_research_output.py /Users/sunlit/Downloads/output tao_dev_1

幂等：反复执行不破坏数据。可选 --purge-first 会先清掉该版本的已有数据。
"""

import argparse
import hashlib
import json
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from knowhub.knowhub_db.pg_store import PostgreSQLStore
from knowhub.knowhub_db.pg_resource_store import PostgreSQLResourceStore
from knowhub.knowhub_db.pg_requirement_store import PostgreSQLRequirementStore
from knowhub.knowhub_db.pg_capability_store import PostgreSQLCapabilityStore
from knowhub.knowhub_db.pg_strategy_store import PostgreSQLStrategyStore
from knowhub.knowhub_db.cascade import purge_version


def _hash12(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()[:12]


def _resource_id(platform: str, source_url: str) -> str:
    p = (platform or "unknown").lower().strip()
    return f"resource/research/{p}/{_hash12(source_url)}"


def _format_case_body(case: dict) -> str:
    parts = []
    if case.get("user_feedback"):
        parts.append(f"【用户反馈】{case['user_feedback']}")
    if case.get("input_details"):
        parts.append(f"【输入详情】{case['input_details']}")
    if case.get("output_details"):
        parts.append(f"【输出详情】{case['output_details']}")
    workflow = case.get("workflow_process", [])
    if workflow:
        parts.append("【工作流】\n" + "\n".join(f"- {s}" for s in workflow))
    return "\n\n".join(parts)


def _format_strategy_body(selected: dict) -> str:
    """strategy.body 存完整 workflow_outline JSON（前端可解析展示）"""
    return json.dumps(selected, ensure_ascii=False, indent=2)


def _find_existing_requirement(cursor, description: str):
    """按完整描述文本精确匹配现有 requirement（任何版本）。返回 id 或 None。"""
    cursor.execute("SELECT id FROM requirement WHERE description = %s LIMIT 1", (description,))
    row = cursor.fetchone()
    if row:
        # Dict-like (RealDictRow) or tuple
        return row['id'] if isinstance(row, dict) or hasattr(row, 'keys') else row[0]
    return None


def ingest_folder(folder: Path, version: str, stores: dict, stats: dict):
    """入库单个 output/{NN}/ 目录。所有 ID 都在 version namespace 内（除 requirement 复用已有的）。"""
    folder_key = folder.name  # "00", "01", ...
    case_path = folder / "case.json"
    strategy_path = folder / "strategy.json"

    case_doc = json.loads(case_path.read_text(encoding="utf-8"))
    strategy_doc = json.loads(strategy_path.read_text(encoding="utf-8"))

    requirement_text = case_doc.get("requirement", "")
    searched_at = case_doc.get("searched_at", "")
    cases = case_doc.get("cases", [])

    # ─── 1. requirement：按完整描述精确匹配现有 REQ（不创建新 REQ）─────
    cursor = stores["req"]._get_cursor()
    try:
        req_id = _find_existing_requirement(cursor, requirement_text)
    finally:
        cursor.close()
    if not req_id:
        raise RuntimeError(
            f"[{folder_key}] 找不到匹配的 requirement。case.json.requirement 应与现有 "
            f"requirement.description 精确相等。首 80 字：{requirement_text[:80]!r}"
        )
    stats["requirement_matched"] += 1
    print(f"  ✓ requirement: {req_id} (复用已有)", flush=True)

    # ─── 2. resource 实体（每个 case 一条，URL hash 去重）─────────────
    resource_ids = []
    for case in cases:
        src_url = case.get("source_url", "")
        if not src_url:
            continue
        platform = case.get("platform", "unknown")
        rid = _resource_id(platform, src_url)
        metrics = case.get("metrics") or {}
        likes = metrics.get("likes") or 0
        stores["res"].insert_or_update({
            "id": rid,
            "title": case.get("title", ""),
            "body": _format_case_body(case),
            "content_type": "research_case",
            "images": case.get("images", []),
            "metadata": {
                "platform": platform,
                "source_url": src_url,
                "metrics": metrics,
                "user_feedback": case.get("user_feedback") or "",
                "input_details": case.get("input_details") or "",
                "output_details": case.get("output_details") or "",
                "workflow_process": case.get("workflow_process") or [],
                "last_seen": searched_at,
                "local_case_id": case.get("id") or "",  # 保留原 case_001 便于 strategy 里的文字交叉引用追溯
            },
            "sort_order": -likes,
            "version": version,
        })
        resource_ids.append(rid)
        stats["resource"] += 1
    print(f"  ✓ resource: {len(resource_ids)} 条", flush=True)

    # ─── 3. strategy 实体 ────────────────────────────────────────────
    selected = strategy_doc.get("selected_strategy", {})
    strategy_id = f"strategy-{version}-{folder_key}"
    stores["strat"].insert_or_update({
        "id": strategy_id,
        "name": selected.get("name", ""),
        "description": selected.get("reasoning", "")[:2000],  # reasoning 太长时截断
        "body": _format_strategy_body(strategy_doc),  # 完整 JSON 原样入库
        "status": "draft",
        "created_at": int(time.time()),
        "updated_at": int(time.time()),
        "version": version,
    })
    stats["strategy"] += 1
    print(f"  ✓ strategy: {strategy_id}", flush=True)

    # ─── 4. 处理 strategy 提到的 capability ──────────────────────────
    # 收集所有 (name, is_new, existing_id) 元组
    workflow_outline = selected.get("workflow_outline", [])
    capability_ids_linked = []
    new_cap_counter = 0
    for phase_idx, phase in enumerate(workflow_outline):
        for cap_ref in phase.get("capabilities", []):
            if cap_ref.get("is_new"):
                new_cap_counter += 1
                new_cap_id = f"CAP-{version}-{folder_key}-{new_cap_counter:02d}"
                stores["cap"].insert_or_update({
                    "id": new_cap_id,
                    "name": cap_ref.get("capability_name", ""),
                    "description": (
                        f"[测试生成] 来自调研策略 {strategy_id} 阶段 {phase_idx+1}。"
                        f" 建议工具：{', '.join(cap_ref.get('suggested_tools', [])[:3])}。"
                    ),
                    "criterion": "",
                    "version": version,
                })
                capability_ids_linked.append(new_cap_id)
                stats["capability_new"] += 1
            elif cap_ref.get("capability_id"):
                # 引用已有 capability（在 v0 或其他版本）
                capability_ids_linked.append(cap_ref["capability_id"])
    print(f"  ✓ capability: new={new_cap_counter}, 总引用={len(capability_ids_linked)}", flush=True)

    # ─── 5. 关联关系 ──────────────────────────────────────────────────
    # requirement → resources (provenance)
    for rid in resource_ids:
        stores["req"].add_resource(req_id, rid)
    # requirement → strategy (satisfies)
    stores["req"].add_strategy(req_id, strategy_id)
    # strategy → resources (build-from provenance)
    for rid in resource_ids:
        stores["strat"].add_resource(strategy_id, rid)
    # strategy → capabilities (compose)
    for cap_id in capability_ids_linked:
        stores["strat"].add_capability(strategy_id, cap_id, relation_type="compose")
    print(f"  ✓ junctions: req_res={len(resource_ids)}, req_strat=1, "
          f"strat_res={len(resource_ids)}, strat_cap={len(capability_ids_linked)}",
          flush=True)


def main():
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("root", help="包含 00/ 01/ 等子目录的 output 根")
    p.add_argument("version", help="多租户版本标签，如 tao_dev_1")
    p.add_argument("--purge-first", action="store_true",
                   help="先清除该 version 的所有已有数据再入库（迭代测试用）")
    args = p.parse_args()

    root = Path(args.root).expanduser().resolve()
    if not root.is_dir():
        print(f"❌ root 不存在或非目录: {root}", file=sys.stderr)
        sys.exit(1)

    folders = sorted([d for d in root.iterdir() if d.is_dir() and (d / "case.json").exists()])
    if not folders:
        print(f"❌ 在 {root} 下未找到任何含 case.json 的子目录", file=sys.stderr)
        sys.exit(1)

    print(f"📂 检测到 {len(folders)} 个调研目录：{[f.name for f in folders]}")
    print(f"🏷  version: {args.version}")

    # 初始化 store（autocommit 模式，不持锁）
    stores = {
        "k": PostgreSQLStore(),
        "res": PostgreSQLResourceStore(),
        "req": PostgreSQLRequirementStore(),
        "cap": PostgreSQLCapabilityStore(),
        "strat": PostgreSQLStrategyStore(),
    }

    try:
        # Optional: 先 purge 该版本的已有数据
        if args.purge_first:
            print(f"\n🧹 purge version={args.version!r}...")
            conn = stores["res"].conn
            cur = conn.cursor()
            try:
                purge_stats = purge_version(cur, args.version)
                print(f"  deleted: {purge_stats}")
            finally:
                cur.close()

        stats = {"requirement_matched": 0, "resource": 0, "strategy": 0, "capability_new": 0}
        for folder in folders:
            print(f"\n📁 {folder.name}/ ...")
            ingest_folder(folder, args.version, stores, stats)

        print("\n" + "=" * 60)
        print("入库完成")
        print("=" * 60)
        print(f"  requirement (matched): {stats['requirement_matched']}  (不创建新的，复用已有 REQ)")
        print(f"  resource:              {stats['resource']}")
        print(f"  strategy:              {stats['strategy']}")
        print(f"  new capability:        {stats['capability_new']}")
    finally:
        for s in stores.values():
            s.close()


if __name__ == "__main__":
    main()