#!/usr/bin/env python3 """ 处理 5 个重跑数据 folder：032/046/069/085/097。 策略（对每个 folder）： 1. 定位 req_id（blueprint.json 解析失败时 fallback strategy.json） 2. 清理旧数据： - 删除 folder 标签为 F 的 resource 及其所有 junction - 删除 req 关联的 strategy + 其 junction - 删除 requirement_capability (req, *) 条目（后续重建全集） 3. 重新 ingest： - resources from raw_cases/（case_bili.json parse fail → 正则 fallback） - capabilities via alias（不存在则新建 howard_dedup） - strategy（is_selected 或第一条） - junctions: · req_res / strat_res: 按 resource 逐条写 · strat_cap: workflow_outline 的 caps（relation_type='compose'） · req_strat: 1 条 · req_cap: capabilities_extracted.json 的所有 caps（A 方案：研究全集） 所有操作 autocommit=True；脚本幂等、允许断点重跑。 """ import hashlib import json import re 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.merge_capabilities import MERGE_CLUSTERS from knowhub.scripts.rename_merged_capabilities import RENAMES RERUN_DIR = Path('/Users/sunlit/Downloads/5') FOLDERS = ['032', '046', '069', '085', '097'] DEDUP_VERSION = 'howard_dedup' def norm(s): return (s or '').strip().lower() def hash8(text): return hashlib.sha256(text.encode('utf-8')).hexdigest()[:8] def hash12(text): return hashlib.sha256(text.encode('utf-8')).hexdigest()[:12] def gen_cap_id(name): return f'CAP-{hash8(norm(name))}' def gen_resource_id(platform, url): p = (platform or 'unknown').lower().strip() return f'resource/research/{p}/{hash12(url)}' def gen_strategy_id(req_text, strategy_name): return f'strategy-{hash8((req_text or "") + "|" + (strategy_name or ""))}' # ═══════════════════════════════════════════════════════════ def build_alias_map(cur): """Build norm(name) -> canonical_id alias from current DB + MERGE_CLUSTERS + RENAMES.""" # Step A: member→canonical with transitive closure member_to_canonical = {} for canonical, members in MERGE_CLUSTERS.items(): for m in members: member_to_canonical[m] = canonical def final(cid, limit=10): seen = set() while cid in member_to_canonical and cid not in seen and limit > 0: seen.add(cid) cid = member_to_canonical[cid] limit -= 1 return cid for m in list(member_to_canonical.keys()): member_to_canonical[m] = final(m) alias = {} # Current DB names cur.execute('SELECT id, name FROM capability') for r in cur.fetchall(): alias[norm(r['name'])] = r['id'] # RENAMES new names for cid, (new_name, _) in RENAMES.items(): alias[norm(new_name)] = final(cid) return alias # ═══════════════════════════════════════════════════════════ def load_raw_cases(folder_path): """Return list of case dicts; fallback to regex when json.load fails.""" raw_dir = folder_path / 'raw_cases' all_cases = [] if not raw_dir.exists(): return all_cases for cf in sorted(raw_dir.glob('*.json')): platform = cf.stem.replace('case_', '') try: data = json.loads(cf.read_text(encoding='utf-8')) cases = data.get('cases', []) if isinstance(data, dict) else data if isinstance(cases, list): for c in cases: if isinstance(c, dict): c.setdefault('platform', platform) all_cases.append(c) continue except Exception as e: print(f' ⚠️ {cf.name} parse fail ({e}); trying regex fallback', flush=True) # regex fallback: anchor by source_url (titles too unreliable) text = cf.read_text(encoding='utf-8') urls = re.findall(r'"source_url"\s*:\s*"([^"]+)"', text) ids = re.findall(r'"id"\s*:\s*"(case_[^"]+)"', text) recovered_n = 0 for i, url in enumerate(urls): case = { 'id': ids[i] if i < len(ids) else f'{platform}_fallback_{i}', 'title': '', # unreliable without proper JSON parse 'platform': platform, 'source_url': url, } all_cases.append(case) recovered_n += 1 print(f' ⇒ recovered {recovered_n} {platform} cases via regex (titles skipped)', flush=True) return all_cases # ═══════════════════════════════════════════════════════════ def cleanup_folder_data(cur, req_id, folder_key, stats): """Remove old resources with folder tag + related junctions + old strategy for this req + req_cap.""" # resources tagged with this folder cur.execute("SELECT id FROM resource WHERE metadata::jsonb->>'folder' = %s", (folder_key,)) old_res = [r['id'] for r in cur.fetchall()] for rid in old_res: cur.execute('DELETE FROM requirement_resource WHERE resource_id = %s', (rid,)) cur.execute('DELETE FROM strategy_resource WHERE resource_id = %s', (rid,)) cur.execute('DELETE FROM capability_resource WHERE resource_id = %s', (rid,)) cur.execute('DELETE FROM resource WHERE id = %s', (rid,)) stats['deleted_resources'] = len(old_res) # any remaining req_res junctions for this req (untagged orphans) cur.execute('DELETE FROM requirement_resource WHERE requirement_id = %s', (req_id,)) # strategies linked to this req cur.execute('SELECT strategy_id FROM requirement_strategy WHERE requirement_id = %s', (req_id,)) old_strats = [r['strategy_id'] for r in cur.fetchall()] for sid in old_strats: cur.execute('DELETE FROM requirement_strategy WHERE strategy_id = %s', (sid,)) cur.execute('DELETE FROM strategy_capability WHERE strategy_id = %s', (sid,)) cur.execute('DELETE FROM strategy_resource WHERE strategy_id = %s', (sid,)) cur.execute('DELETE FROM strategy_knowledge WHERE strategy_id = %s', (sid,)) cur.execute('DELETE FROM strategy WHERE id = %s', (sid,)) stats['deleted_strategies'] = len(old_strats) # req_cap for this req (will be rebuilt) cur.execute('DELETE FROM requirement_capability WHERE requirement_id = %s', (req_id,)) # ═══════════════════════════════════════════════════════════ def ingest_folder(folder_path, cur, alias, stats): folder_key = folder_path.name # (1) requirement text — try blueprint, fallback to strategy req_text = '' try: bp = json.loads((folder_path / 'blueprint.json').read_text(encoding='utf-8')) req_text = bp.get('requirement', '') except Exception as e: print(f' ⚠️ blueprint parse fail ({e}); trying strategy.json', flush=True) if not req_text: try: sd = json.loads((folder_path / 'strategy.json').read_text(encoding='utf-8')) req_text = sd.get('requirement', '') except Exception as e: print(f' ❌ no requirement text available ({e})', flush=True) return cur.execute('SELECT id FROM requirement WHERE description = %s LIMIT 1', (req_text,)) row = cur.fetchone() if not row: print(f' ❌ no matching requirement for {folder_key}', flush=True) return req_id = row['id'] print(f' → req_id={req_id}', flush=True) # (2) cleanup cleanup_folder_data(cur, req_id, folder_key, stats) del_r, del_s = stats['deleted_resources'], stats['deleted_strategies'] print(f' cleaned: del_res={del_r}, del_strat={del_s}', flush=True) # (3) resources cases = load_raw_cases(folder_path) resource_ids = [] for case in cases: url = case.get('source_url') or case.get('url') if not url: continue platform = case.get('platform') or 'unknown' 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_key}, ensure_ascii=False), -int(likes), DEDUP_VERSION)) resource_ids.append(rid) stats['resources'] = len(resource_ids) # (4) capabilities (from capabilities_extracted.json) — track ALL of them for req_cap superset caps_path = folder_path / 'capabilities_extracted.json' all_cap_ids_research = set() # for req_cap (A-plan research superset) cap_key_to_id = {} # source_key -> resolved_id (for strat_cap resolution) if caps_path.exists(): try: caps_data = json.loads(caps_path.read_text(encoding='utf-8')) except Exception as e: print(f' ⚠️ capabilities_extracted parse fail: {e}', flush=True) caps_data = {'extracted_capabilities': []} for cap in caps_data.get('extracted_capabilities', []): name = (cap.get('name') or '').strip() if not name: continue src_id = cap.get('id') resolved = None # (a) source id exists? if src_id: cur.execute('SELECT 1 FROM capability WHERE id = %s', (src_id,)) if cur.fetchone(): resolved = src_id # (b) alias by name if not resolved: cand = alias.get(norm(name)) if cand: cur.execute('SELECT 1 FROM capability WHERE id = %s', (cand,)) if cur.fetchone(): resolved = cand # (c) create new if not resolved: new_id = gen_cap_id(name) 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, name, cap.get('criterion', '') or '', cap.get('description', '') or '', json.dumps(cap.get('effects', []) or [], ensure_ascii=False, default=str), DEDUP_VERSION)) alias[norm(name)] = new_id stats['cap_new'] += 1 resolved = new_id else: # backfill criterion/effects if missing cur.execute('SELECT criterion, effects FROM capability WHERE id = %s', (resolved,)) ex = cur.fetchone() if ex: if (not (ex.get('criterion') or '').strip()) and cap.get('criterion'): cur.execute('UPDATE capability SET criterion = %s WHERE id = %s', (cap['criterion'], resolved)) cur_eff = ex.get('effects') if (not cur_eff or cur_eff in ([], '[]')) and cap.get('effects'): cur.execute('UPDATE capability SET effects = %s WHERE id = %s', (json.dumps(cap['effects'], ensure_ascii=False, default=str), resolved)) stats['cap_linked'] += 1 all_cap_ids_research.add(resolved) cap_key_to_id[src_id or name] = resolved # (5) strategy strat_path = folder_path / 'strategy.json' strat_id = None strat_cap_ids = set() if strat_path.exists(): try: strat_data = json.loads(strat_path.read_text(encoding='utf-8')) except Exception as e: print(f' ⚠️ strategy parse fail: {e}', flush=True) strat_data = {'strategies': []} selected = next((s for s in strat_data.get('strategies', []) if s.get('is_selected')), None) if not selected and strat_data.get('strategies'): selected = strat_data['strategies'][0] if selected: strategy_name = selected.get('name') or f'Strategy-{folder_key}' 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, (selected.get('reasoning') or '')[:2000], json.dumps(selected, ensure_ascii=False, indent=2), 'draft', now, now, DEDUP_VERSION)) stats['strategies'] = 1 # workflow_outline cap resolution wo = selected.get('workflow_outline') or [] if isinstance(wo, list): for phase in wo: if not isinstance(phase, dict): continue caps = phase.get('capabilities') or [] if not isinstance(caps, list): continue for cref in caps: if not isinstance(cref, dict): continue key = cref.get('id') or cref.get('name', '') resolved = cap_key_to_id.get(key) or alias.get(norm(cref.get('name', ''))) if resolved: strat_cap_ids.add(resolved) # (6) wire junctions for rid in resource_ids: cur.execute("""INSERT INTO requirement_resource (requirement_id, resource_id) VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, rid)) if strat_id: cur.execute("""INSERT INTO strategy_resource (strategy_id, resource_id) VALUES (%s, %s) ON CONFLICT DO NOTHING""", (strat_id, rid)) if strat_id: cur.execute("""INSERT INTO requirement_strategy (requirement_id, strategy_id) VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, strat_id)) for cid in strat_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)) # req_cap: research superset (A plan) for cid in all_cap_ids_research: cur.execute("""INSERT INTO requirement_capability (requirement_id, capability_id) VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, cid)) # also include strat-only caps (in case some are in workflow_outline but not in extracted list) for cid in strat_cap_ids: cur.execute("""INSERT INTO requirement_capability (requirement_id, capability_id) VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, cid)) stats['req_cap_wired'] = len(all_cap_ids_research | strat_cap_ids) stats['strat_cap_wired'] = len(strat_cap_ids) rc_n, sc_n = stats['req_cap_wired'], stats['strat_cap_wired'] s_n = 1 if strat_id else 0 print(f' ingested: res={len(resource_ids)}, strat={s_n}, req_cap={rc_n}, strat_cap={sc_n}', flush=True) # ═══════════════════════════════════════════════════════════ def main(): s = PostgreSQLCapabilityStore() cur = s._get_cursor() try: print('Building alias map...', flush=True) alias = build_alias_map(cur) print(f' alias entries: {len(alias)}', flush=True) totals = {'deleted_resources': 0, 'deleted_strategies': 0, 'resources': 0, 'cap_new': 0, 'cap_linked': 0, 'strategies': 0, 'req_cap_wired': 0, 'strat_cap_wired': 0} for f in FOLDERS: print(f'\n=== {f} ===', flush=True) stats = {'deleted_resources': 0, 'deleted_strategies': 0, 'resources': 0, 'cap_new': 0, 'cap_linked': 0, 'strategies': 0, 'req_cap_wired': 0, 'strat_cap_wired': 0} try: ingest_folder(RERUN_DIR / f, cur, alias, stats) for k in totals: totals[k] += stats.get(k, 0) except Exception as e: print(f' ❌ {type(e).__name__}: {e}', flush=True) try: cur.close() except Exception: pass cur = s._get_cursor() print(f'\n{"="*50}\nTOTALS: {totals}', flush=True) finally: cur.close() s.close() if __name__ == '__main__': main()