Agent/knowhub/scripts/salvage_malformed_folders.py

#!/usr/bin/env python3
"""
Salvage folders with non-standard JSON schemas that standard ingest missed.

Target folders (from data problem audit):
  004, 031, 053, 066, 070 — non-standard strategy.json structure (LLM 自由发挥)
  044 — capabilities_extracted uses old schema (capability_id/capability_name)

Strategy: for each folder, manually locate the relevant fields and wire the junctions.
All insertions use version='howard_dedup'.

用法：
  python knowhub/scripts/salvage_malformed_folders.py --execute
"""
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_capability_store import PostgreSQLCapabilityStore
from knowhub.scripts.rebuild_howard_dedup import (
    parse_snapshot, build_alias_map, norm, gen_cap_id, gen_strategy_id,
    gen_resource_id, SNAPSHOT_PATH, OUTPUT_DIR, DEDUP_VERSION,
)

DEDUP = DEDUP_VERSION


def resolve_cap(cur, cap_name_or_id, alias, current_caps, create_if_missing=False,
                cap_desc='', cap_criterion=''):
    """Resolve a capability name/id → DB id. Optionally create if missing."""
    if not cap_name_or_id:
        return None
    # If it looks like a cap ID and exists in DB, use it
    if cap_name_or_id.startswith('CAP-') and cap_name_or_id in current_caps:
        return cap_name_or_id
    # Try alias
    cand = alias.get(norm(cap_name_or_id))
    if cand and cand in current_caps:
        return cand
    # Try checking if name matches DB
    cur.execute('SELECT id FROM capability WHERE name = %s LIMIT 1', (cap_name_or_id,))
    row = cur.fetchone()
    if row:
        return row['id']
    if create_if_missing:
        new_id = gen_cap_id(cap_name_or_id)
        cur.execute('SELECT 1 FROM capability WHERE id = %s', (new_id,))
        if not cur.fetchone():
            cur.execute(
                """INSERT INTO capability (id, name, criterion, description, effects, version)
                   VALUES (%s, %s, %s, %s, %s, %s)""",
                (new_id, cap_name_or_id, cap_criterion, cap_desc, '[]', DEDUP))
            current_caps[new_id] = {'name': cap_name_or_id}
            alias[norm(cap_name_or_id)] = new_id
        return new_id
    return None


def insert_strategy(cur, folder_key, req_id, req_text, strategy_name, strategy_body,
                    reasoning, resource_ids, cap_ids):
    strat_id = gen_strategy_id(req_text, strategy_name)
    now = int(time.time())
    cur.execute('DELETE FROM strategy WHERE id = %s', (strat_id,))
    cur.execute(
        """INSERT INTO strategy (id, name, description, body, status, created_at, updated_at, version)
           VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
        (strat_id, strategy_name, (reasoning or '')[:2000],
         json.dumps(strategy_body, ensure_ascii=False, indent=2),
         'draft', now, now, DEDUP))
    # requirement → strategy
    cur.execute("""INSERT INTO requirement_strategy (requirement_id, strategy_id)
                   VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, strat_id))
    # strategy → resources
    for rid in resource_ids:
        cur.execute("""INSERT INTO strategy_resource (strategy_id, resource_id)
                       VALUES (%s, %s) ON CONFLICT DO NOTHING""", (strat_id, rid))
    # strategy → capabilities (+ req → cap)
    for cid in cap_ids:
        cur.execute("""INSERT INTO strategy_capability (strategy_id, capability_id, relation_type)
                       VALUES (%s, %s, 'compose') ON CONFLICT DO NOTHING""", (strat_id, cid))
        cur.execute("""INSERT INTO requirement_capability (requirement_id, capability_id)
                       VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, cid))
    return strat_id


def ensure_resources(cur, folder, req_id):
    """Read raw_cases/, insert resources, return list of resource IDs."""
    raw_dir = folder / 'raw_cases'
    resource_ids = []
    if not raw_dir.exists():
        return resource_ids
    for cf in sorted(raw_dir.glob('*.json')):
        try:
            data = json.loads(cf.read_text(encoding='utf-8'))
        except Exception:
            continue
        cases = data.get('cases', []) if isinstance(data, dict) else data
        if not isinstance(cases, list):
            continue
        for case in cases:
            if not isinstance(case, dict):
                continue
            url = case.get('source_url') or case.get('url')
            if not url:
                continue
            platform = case.get('platform') or cf.stem.replace('case_', '')
            rid = gen_resource_id(platform, url)
            title = (case.get('title') or '')[:200]
            metrics = case.get('metrics') if isinstance(case.get('metrics'), dict) else {}
            likes = (metrics.get('likes') or 0) if metrics else 0
            cur.execute('DELETE FROM resource WHERE id = %s', (rid,))
            cur.execute(
                """INSERT INTO resource (id, title, body, content_type, images, metadata, sort_order, version)
                   VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
                (rid, title, json.dumps(case, ensure_ascii=False)[:8000],
                 'research_case',
                 json.dumps(case.get('images', []) or [], ensure_ascii=False),
                 json.dumps({'platform': platform, 'source_url': url,
                             'metrics': metrics, 'folder': folder.name},
                            ensure_ascii=False),
                 -int(likes), DEDUP))
            resource_ids.append(rid)
            cur.execute("""INSERT INTO requirement_resource (requirement_id, resource_id)
                           VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, rid))
    return resource_ids


# ═══════════════════════════════════════════════════════════
# Per-folder handlers

def salvage_004(cur, alias, current_caps):
    """004: strategy.strategy.phases + capability_mapping (capability=name)."""
    folder = OUTPUT_DIR / '004'
    req_id = 'REQ_004'
    d = json.loads((folder / 'strategy.json').read_text())
    req_text = d.get('requirement', '')
    strat_dict = d.get('strategy', {})
    overview = strat_dict.get('overview', '')
    name = f'Strategy-004'
    # capability_mapping: [{capability: name, role, used_in_phases, key_tools}]
    cap_ids = set()
    for entry in d.get('capability_mapping', []) or []:
        if not isinstance(entry, dict):
            continue
        cap_name = entry.get('capability') or entry.get('capability_name')
        cid = resolve_cap(cur, cap_name, alias, current_caps, create_if_missing=True)
        if cid: cap_ids.add(cid)
    resource_ids = ensure_resources(cur, folder, req_id)
    body = {'strategy': strat_dict, 'capability_mapping': d.get('capability_mapping', [])}
    insert_strategy(cur, '004', req_id, req_text, name, body, overview,
                    resource_ids, cap_ids)
    return len(cap_ids), len(resource_ids)


def salvage_031(cur, alias, current_caps):
    """031: strategy.strategy.phases + capability_mapping (capability_id/name)."""
    folder = OUTPUT_DIR / '031'
    req_id = 'REQ_031'
    d = json.loads((folder / 'strategy.json').read_text())
    req_text = d.get('requirement', '')
    strat_dict = d.get('strategy', {})
    overview = strat_dict.get('overview', '')
    name = f'Strategy-031'
    cap_ids = set()
    for entry in d.get('capability_mapping', []) or []:
        if not isinstance(entry, dict):
            continue
        cid_src = entry.get('capability_id')
        cname = entry.get('capability_name')
        cid = None
        if cid_src:
            cid = resolve_cap(cur, cid_src, alias, current_caps)
        if not cid and cname:
            cid = resolve_cap(cur, cname, alias, current_caps, create_if_missing=True)
        if cid: cap_ids.add(cid)
    resource_ids = ensure_resources(cur, folder, req_id)
    body = {'strategy': strat_dict, 'capability_mapping': d.get('capability_mapping', [])}
    insert_strategy(cur, '031', req_id, req_text, name, body, overview,
                    resource_ids, cap_ids)
    return len(cap_ids), len(resource_ids)


def salvage_053(cur, alias, current_caps):
    """053: phases[] (phase/description/capabilities) + core_workflow (text)."""
    folder = OUTPUT_DIR / '053'
    req_id = 'REQ_053'
    d = json.loads((folder / 'strategy.json').read_text())
    req_text = d.get('requirement', '')
    name = f'Strategy-053'
    cap_ids = set()
    for phase in d.get('phases', []) or []:
        if not isinstance(phase, dict):
            continue
        for cap_ref in phase.get('capabilities', []) or []:
            if isinstance(cap_ref, str):
                # string name or id
                cid = resolve_cap(cur, cap_ref, alias, current_caps, create_if_missing=True)
                if cid: cap_ids.add(cid)
            elif isinstance(cap_ref, dict):
                src = cap_ref.get('id') or cap_ref.get('capability_id')
                nm = cap_ref.get('name') or cap_ref.get('capability_name')
                cid = None
                if src: cid = resolve_cap(cur, src, alias, current_caps)
                if not cid and nm:
                    cid = resolve_cap(cur, nm, alias, current_caps, create_if_missing=True)
                if cid: cap_ids.add(cid)
    resource_ids = ensure_resources(cur, folder, req_id)
    body = {'phases': d.get('phases', []), 'core_workflow': d.get('core_workflow', '')}
    insert_strategy(cur, '053', req_id, req_text, name, body,
                    d.get('core_workflow', '')[:500], resource_ids, cap_ids)
    return len(cap_ids), len(resource_ids)


def salvage_066(cur, alias, current_caps):
    """066: execution_phases + key_capabilities (id/name/role)."""
    folder = OUTPUT_DIR / '066'
    req_id = 'REQ_066'
    d = json.loads((folder / 'strategy.json').read_text())
    req_text = d.get('requirement', '')
    name = f'Strategy-066'
    cap_ids = set()
    # key_capabilities is primary
    for cap in d.get('key_capabilities', []) or []:
        if not isinstance(cap, dict): continue
        cid = None
        if cap.get('id'):
            cid = resolve_cap(cur, cap['id'], alias, current_caps)
        if not cid and cap.get('name'):
            cid = resolve_cap(cur, cap['name'], alias, current_caps, create_if_missing=True)
        if cid: cap_ids.add(cid)
    # execution_phases may reference more
    for phase in d.get('execution_phases', []) or []:
        if not isinstance(phase, dict): continue
        for c in phase.get('capabilities_used', []) or []:
            if isinstance(c, str):
                cid = resolve_cap(cur, c, alias, current_caps, create_if_missing=True)
                if cid: cap_ids.add(cid)
            elif isinstance(c, dict):
                cid = None
                if c.get('id'): cid = resolve_cap(cur, c['id'], alias, current_caps)
                if not cid and c.get('name'):
                    cid = resolve_cap(cur, c['name'], alias, current_caps, create_if_missing=True)
                if cid: cap_ids.add(cid)
    resource_ids = ensure_resources(cur, folder, req_id)
    body = {'execution_phases': d.get('execution_phases', []),
            'key_capabilities': d.get('key_capabilities', []),
            'recommended_tools': d.get('recommended_tools', [])}
    insert_strategy(cur, '066', req_id, req_text, name, body, '', resource_ids, cap_ids)
    return len(cap_ids), len(resource_ids)


def salvage_070(cur, alias, current_caps):
    """070: capabilities_mapping (id/name/role_in_strategy/is_new) + tool_chain."""
    folder = OUTPUT_DIR / '070'
    req_id = 'REQ_070'
    d = json.loads((folder / 'strategy.json').read_text())
    req_text = d.get('requirement', '')
    name = f'Strategy-070'
    cap_ids = set()
    for cap in d.get('capabilities_mapping', []) or []:
        if not isinstance(cap, dict): continue
        cid = None
        if cap.get('id'): cid = resolve_cap(cur, cap['id'], alias, current_caps)
        if not cid and cap.get('name'):
            cid = resolve_cap(cur, cap['name'], alias, current_caps, create_if_missing=True)
        if cid: cap_ids.add(cid)
    resource_ids = ensure_resources(cur, folder, req_id)
    body = {'capabilities_mapping': d.get('capabilities_mapping', []),
            'tool_chain': d.get('tool_chain', []),
            'selected_blueprint': d.get('selected_blueprint', '')}
    insert_strategy(cur, '070', req_id, req_text, name, body, '', resource_ids, cap_ids)
    return len(cap_ids), len(resource_ids)


def salvage_044(cur, alias, current_caps):
    """044: capabilities_extracted uses capability_id/capability_name schema.
    Just link these caps to existing REQ_044's strategy (already created)."""
    folder = OUTPUT_DIR / '044'
    req_id = 'REQ_044'
    # find existing strategy
    cur.execute('SELECT strategy_id FROM requirement_strategy WHERE requirement_id=%s LIMIT 1', (req_id,))
    row = cur.fetchone()
    if not row:
        print('  ⚠️  044 has no strategy yet', flush=True)
        return 0, 0
    strat_id = row['strategy_id']

    d = json.loads((folder / 'capabilities_extracted.json').read_text())
    cap_ids = set()
    for cap in d.get('extracted_capabilities', []) or []:
        if not isinstance(cap, dict): continue
        cid_src = cap.get('capability_id')
        cname = cap.get('capability_name')
        cid = None
        if cid_src:
            cid = resolve_cap(cur, cid_src, alias, current_caps)
        if not cid and cname:
            cid = resolve_cap(cur, cname, alias, current_caps, create_if_missing=True)
        if cid: cap_ids.add(cid)
    for cid in cap_ids:
        cur.execute("""INSERT INTO strategy_capability (strategy_id, capability_id, relation_type)
                       VALUES (%s, %s, 'compose') ON CONFLICT DO NOTHING""", (strat_id, cid))
        cur.execute("""INSERT INTO requirement_capability (requirement_id, capability_id)
                       VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, cid))
    return len(cap_ids), 0


# ═══════════════════════════════════════════════════════════
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--execute', action='store_true', required=True)
    args = ap.parse_args()

    s = PostgreSQLCapabilityStore()
    cur = s._get_cursor()
    cur.execute("SELECT id, name FROM capability WHERE version = %s", (DEDUP,))
    current_caps = {r['id']: {'name': r['name']} for r in cur.fetchall()}
    snapshot = parse_snapshot(SNAPSHOT_PATH)
    alias = build_alias_map(snapshot, current_caps)
    print(f'alias: {len(alias)}, caps: {len(current_caps)}', flush=True)

    handlers = [
        ('004', salvage_004), ('031', salvage_031),
        ('044', salvage_044), ('053', salvage_053),
        ('066', salvage_066), ('070', salvage_070),
    ]
    for key, fn in handlers:
        print(f'\n📁 salvaging {key}/ ...', flush=True)
        try:
            n_cap, n_res = fn(cur, alias, current_caps)
            print(f'  ✓ {key}: {n_cap} caps, {n_res} resources', flush=True)
        except Exception as e:
            print(f'  ❌ {key}: {type(e).__name__}: {e}', flush=True)
            import traceback; traceback.print_exc()

    # Final check
    print(f'\n{"="*50}\nFinal requirement coverage:', flush=True)
    for req in ['REQ_004','REQ_031','REQ_044','REQ_053','REQ_066','REQ_070']:
        cur.execute('SELECT COUNT(*) AS c FROM requirement_strategy WHERE requirement_id=%s', (req,))
        strat = cur.fetchone()['c']
        cur.execute('SELECT COUNT(*) AS c FROM requirement_capability WHERE requirement_id=%s', (req,))
        cap = cur.fetchone()['c']
        cur.execute('SELECT COUNT(*) AS c FROM requirement_resource WHERE requirement_id=%s', (req,))
        res = cur.fetchone()['c']
        print(f'  {req}: strat={strat} cap={cap} res={res}', flush=True)
    cur.close()
    s.close()


if __name__ == '__main__':
    main()