|
|
@@ -0,0 +1,600 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+"""
|
|
|
+一次性修复脚本：重建 howard_dedup 版本的 capability / strategy / resource
|
|
|
+
|
|
|
+背景：同事的 agent 失控污染了 tao_dev_1 数据。
|
|
|
+这个脚本：
|
|
|
+ 1. 备份当前 DB 状态到 /tmp/knowhub_backup_<date>/
|
|
|
+ 2. 用 /tmp/capabilities_all.md（465 快照）+ MERGE_CLUSTERS 构建别名表
|
|
|
+ 3. Purge 当前 capability + strategy + resource + junctions（所有版本）
|
|
|
+ 4. 从 output 2/ 的 99 folder 重新 ingest，version='howard_dedup'
|
|
|
+ 5. 应用 RENAMES 保留改名工作
|
|
|
+
|
|
|
+不动的表：requirement、knowledge 及其 junction。
|
|
|
+
|
|
|
+用法：
|
|
|
+ python knowhub/scripts/rebuild_howard_dedup.py --backup-only
|
|
|
+ python knowhub/scripts/rebuild_howard_dedup.py --dry-run
|
|
|
+ python knowhub/scripts/rebuild_howard_dedup.py --execute
|
|
|
+"""
|
|
|
+import argparse
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import os
|
|
|
+import re
|
|
|
+import sys
|
|
|
+import time
|
|
|
+from datetime import date
|
|
|
+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.knowhub_db.pg_resource_store import PostgreSQLResourceStore
|
|
|
+from knowhub.knowhub_db.pg_requirement_store import PostgreSQLRequirementStore
|
|
|
+from knowhub.knowhub_db.pg_strategy_store import PostgreSQLStrategyStore
|
|
|
+from knowhub.scripts.merge_capabilities import MERGE_CLUSTERS
|
|
|
+from knowhub.scripts.rename_merged_capabilities import RENAMES
|
|
|
+
|
|
|
+OUTPUT_DIR = Path('/Users/sunlit/Downloads/output 2')
|
|
|
+SNAPSHOT_PATH = Path('/tmp/capabilities_all.md')
|
|
|
+BACKUP_DIR = Path(f'/tmp/knowhub_backup_{date.today().isoformat()}')
|
|
|
+DEDUP_VERSION = 'howard_dedup'
|
|
|
+
|
|
|
+# CAP-006 在同事操作后丢失，从 conversation 缓存中重建
|
|
|
+CAP_006 = {
|
|
|
+ 'id': 'CAP-006',
|
|
|
+ 'name': '图像细节增强与高清放大',
|
|
|
+ 'description': '对已生成的图像进行分辨率提升和细节增强，在放大的同时补充高频细节（后处理路径，区别于生成阶段直接高清输出的 CAP-016）',
|
|
|
+ 'criterion': '',
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+def norm(s):
|
|
|
+ return s.strip().lower() if s else ''
|
|
|
+
|
|
|
+
|
|
|
+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 parse_snapshot(path):
|
|
|
+ """Parse /tmp/capabilities_all.md -> {id: name}."""
|
|
|
+ if not path.exists():
|
|
|
+ print(f'⚠️ snapshot file missing: {path}', flush=True)
|
|
|
+ return {}
|
|
|
+ content = path.read_text(encoding='utf-8')
|
|
|
+ pat = re.compile(r'^## (.+?)\n\*\*(CAP-[\w\-]+)\*\*', re.MULTILINE)
|
|
|
+ return {cid: name.strip() for name, cid in pat.findall(content)}
|
|
|
+
|
|
|
+
|
|
|
+def build_alias_map(snapshot, current_caps):
|
|
|
+ """Build normalized_name -> final canonical_id (transitively resolved)."""
|
|
|
+ # Step A: build member→canonical map; resolve transitively (A→B→C → A,B→C)
|
|
|
+ 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
|
|
|
+
|
|
|
+ # Resolve: every member maps to final canonical
|
|
|
+ for m in list(member_to_canonical.keys()):
|
|
|
+ member_to_canonical[m] = final(m)
|
|
|
+
|
|
|
+ alias = {}
|
|
|
+
|
|
|
+ # 1. All snapshot names → final canonical (member) or self (non-member)
|
|
|
+ for cid, name in snapshot.items():
|
|
|
+ alias[norm(name)] = member_to_canonical.get(cid, cid)
|
|
|
+
|
|
|
+ # 2. Canonical names from current DB (post-rename) → final canonical
|
|
|
+ for canonical in MERGE_CLUSTERS.keys():
|
|
|
+ final_id = final(canonical)
|
|
|
+ if canonical in current_caps:
|
|
|
+ alias[norm(current_caps[canonical]['name'])] = final_id
|
|
|
+
|
|
|
+ # 3. RENAMES: new name → same canonical (final-resolved)
|
|
|
+ for cid, (new_name, _) in RENAMES.items():
|
|
|
+ alias[norm(new_name)] = final(cid)
|
|
|
+
|
|
|
+ # 4. v0 foundation CAP-001..021 + CAP-006 reconstruction → self
|
|
|
+ for cid, cap in current_caps.items():
|
|
|
+ if cid.startswith('CAP-') and len(cid) == 7 and cid[4:].isdigit():
|
|
|
+ alias[norm(cap['name'])] = cid
|
|
|
+ alias[norm(CAP_006['name'])] = CAP_006['id']
|
|
|
+
|
|
|
+ # 5. Any remaining current DB capability (non-VCAP) → self
|
|
|
+ for cid, cap in current_caps.items():
|
|
|
+ if cid.startswith('CAP-tao_dev_'):
|
|
|
+ continue
|
|
|
+ alias.setdefault(norm(cap['name']), cid)
|
|
|
+
|
|
|
+ return alias
|
|
|
+
|
|
|
+
|
|
|
+# ═══════════════════════════════════════════════════════════
|
|
|
+# PHASE 0: BACKUP
|
|
|
+def backup(stores):
|
|
|
+ BACKUP_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
+ print(f'📦 Backing up to {BACKUP_DIR}/', flush=True)
|
|
|
+ tables = [
|
|
|
+ 'capability', 'strategy', 'resource', 'requirement', 'knowledge',
|
|
|
+ 'requirement_capability', 'capability_tool', 'capability_knowledge',
|
|
|
+ 'capability_resource', 'strategy_capability', 'strategy_resource',
|
|
|
+ 'strategy_knowledge', 'requirement_strategy', 'requirement_resource',
|
|
|
+ ]
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ try:
|
|
|
+ for t in tables:
|
|
|
+ try:
|
|
|
+ cur.execute(f'SELECT * FROM {t}')
|
|
|
+ rows = [dict(r) for r in cur.fetchall()]
|
|
|
+ # strip embedding (too big, not essential for restore)
|
|
|
+ for r in rows:
|
|
|
+ r.pop('embedding', None)
|
|
|
+ (BACKUP_DIR / f'{t}.json').write_text(
|
|
|
+ json.dumps(rows, default=str, ensure_ascii=False))
|
|
|
+ print(f' ✓ {t}: {len(rows)} rows', flush=True)
|
|
|
+ except Exception as e:
|
|
|
+ print(f' ❌ {t}: {e}', flush=True)
|
|
|
+ finally:
|
|
|
+ cur.close()
|
|
|
+
|
|
|
+
|
|
|
+# PHASE 1: PURGE
|
|
|
+def purge(stores):
|
|
|
+ print('\n🧹 Purging capability / strategy / resource (all versions) + junctions...', flush=True)
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ try:
|
|
|
+ # junctions first (no FK but keep clean)
|
|
|
+ for t in ['requirement_capability', 'capability_tool', 'capability_knowledge',
|
|
|
+ 'capability_resource', 'strategy_capability', 'strategy_resource',
|
|
|
+ 'strategy_knowledge', 'requirement_strategy', 'requirement_resource']:
|
|
|
+ cur.execute(f'DELETE FROM {t}')
|
|
|
+ print(f' ✓ {t} cleared', flush=True)
|
|
|
+ for t in ['capability', 'strategy', 'resource']:
|
|
|
+ cur.execute(f'DELETE FROM {t}')
|
|
|
+ print(f' ✓ {t} cleared', flush=True)
|
|
|
+ finally:
|
|
|
+ cur.close()
|
|
|
+
|
|
|
+
|
|
|
+# PHASE 2: SEED
|
|
|
+def seed(stores, current_caps, snapshot):
|
|
|
+ """Insert:
|
|
|
+ - 21 v0 foundation caps (CAP-001..021) in howard_dedup
|
|
|
+ - CAP-006 reconstructed if missing
|
|
|
+ - All surviving non-VCAP tao_dev_1 canonicals (preserves R1/R2/C renames)
|
|
|
+ - Missing canonicals (in MERGE_CLUSTERS.keys but gone) recovered from snapshot
|
|
|
+ """
|
|
|
+ print('\n🌱 Seeding howard_dedup...', flush=True)
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ try:
|
|
|
+ inserted = 0
|
|
|
+ # 1. Insert from current_caps: everything non-VCAP gets version=howard_dedup
|
|
|
+ for cap_id, cap in current_caps.items():
|
|
|
+ if cap_id.startswith('CAP-tao_dev_'):
|
|
|
+ continue # skip VCAP
|
|
|
+ cur.execute(
|
|
|
+ """INSERT INTO capability (id, name, criterion, description, effects, version)
|
|
|
+ VALUES (%s, %s, %s, %s, %s, %s)""",
|
|
|
+ (cap_id, cap.get('name', ''), cap.get('criterion', '') or '',
|
|
|
+ cap.get('description', '') or '',
|
|
|
+ json.dumps(cap.get('effects', []) or [], ensure_ascii=False, default=str),
|
|
|
+ DEDUP_VERSION))
|
|
|
+ inserted += 1
|
|
|
+
|
|
|
+ # 2. CAP-006 reconstruct if missing
|
|
|
+ if 'CAP-006' not in current_caps:
|
|
|
+ cur.execute(
|
|
|
+ """INSERT INTO capability (id, name, criterion, description, effects, version)
|
|
|
+ VALUES (%s, %s, %s, %s, %s, %s)""",
|
|
|
+ (CAP_006['id'], CAP_006['name'], CAP_006['criterion'],
|
|
|
+ CAP_006['description'], '[]', DEDUP_VERSION))
|
|
|
+ inserted += 1
|
|
|
+ print(f' ✓ CAP-006 reconstructed', flush=True)
|
|
|
+
|
|
|
+ # 3. Missing canonicals (in MERGE_CLUSTERS but not in DB) recovered from snapshot
|
|
|
+ merged_members = set()
|
|
|
+ for members in MERGE_CLUSTERS.values():
|
|
|
+ merged_members.update(members)
|
|
|
+ recovered = 0
|
|
|
+ for cid in MERGE_CLUSTERS.keys():
|
|
|
+ if cid in current_caps or cid in merged_members:
|
|
|
+ continue
|
|
|
+ # Not in DB, not a member — need to recover
|
|
|
+ name = snapshot.get(cid)
|
|
|
+ if not name:
|
|
|
+ continue
|
|
|
+ # Use RENAMES version if available
|
|
|
+ if cid in RENAMES:
|
|
|
+ name, desc = RENAMES[cid]
|
|
|
+ else:
|
|
|
+ desc = ''
|
|
|
+ cur.execute(
|
|
|
+ """INSERT INTO capability (id, name, criterion, description, effects, version)
|
|
|
+ VALUES (%s, %s, %s, %s, %s, %s)""",
|
|
|
+ (cid, name, '', desc, '[]', DEDUP_VERSION))
|
|
|
+ recovered += 1
|
|
|
+ print(f' seeded: {inserted} existing + {recovered} recovered from snapshot', flush=True)
|
|
|
+ finally:
|
|
|
+ cur.close()
|
|
|
+
|
|
|
+
|
|
|
+# PHASE 3: INGEST from output 2/
|
|
|
+def ingest_folder(folder, stores, alias, cur, stats):
|
|
|
+ folder_key = folder.name
|
|
|
+ # blueprint → requirement match
|
|
|
+ bp_path = folder / 'blueprint.json'
|
|
|
+ if not bp_path.exists():
|
|
|
+ stats['bad_folders'].append(folder_key + ':no_blueprint')
|
|
|
+ return
|
|
|
+ try:
|
|
|
+ bp = json.loads(bp_path.read_text(encoding='utf-8'))
|
|
|
+ except Exception as e:
|
|
|
+ stats['bad_folders'].append(folder_key + f':bp_parse_err({e})')
|
|
|
+ return
|
|
|
+ req_text = bp.get('requirement', '')
|
|
|
+ if not req_text:
|
|
|
+ stats['bad_folders'].append(folder_key + ':empty_req')
|
|
|
+ return
|
|
|
+
|
|
|
+ cur.execute('SELECT id FROM requirement WHERE description = %s LIMIT 1', (req_text,))
|
|
|
+ row = cur.fetchone()
|
|
|
+ if not row:
|
|
|
+ # try fuzzier: first 80 chars prefix
|
|
|
+ cur.execute('SELECT id FROM requirement WHERE description LIKE %s LIMIT 1',
|
|
|
+ (req_text[:80].replace('%', '\\%') + '%',))
|
|
|
+ row = cur.fetchone()
|
|
|
+ if not row:
|
|
|
+ print(f' ⚠️ no matching requirement: {req_text[:60]}', flush=True)
|
|
|
+ stats['missing_req'].append(folder_key)
|
|
|
+ return
|
|
|
+ req_id = row['id']
|
|
|
+
|
|
|
+ # resources from raw_cases/
|
|
|
+ raw_dir = folder / 'raw_cases'
|
|
|
+ resource_ids = []
|
|
|
+ if raw_dir.exists():
|
|
|
+ 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_key},
|
|
|
+ ensure_ascii=False),
|
|
|
+ -int(likes), DEDUP_VERSION))
|
|
|
+ resource_ids.append(rid)
|
|
|
+ stats['resource'] += 1
|
|
|
+
|
|
|
+ # capabilities
|
|
|
+ caps_path = folder / 'capabilities_extracted.json'
|
|
|
+ cap_resolved = {} # source_key -> resolved_id
|
|
|
+ if caps_path.exists():
|
|
|
+ try:
|
|
|
+ caps_data = json.loads(caps_path.read_text(encoding='utf-8'))
|
|
|
+ except Exception as e:
|
|
|
+ stats['bad_folders'].append(folder_key + f':caps_parse_err({e})')
|
|
|
+ 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
|
|
|
+
|
|
|
+ # (1) source id exists in DB?
|
|
|
+ if src_id:
|
|
|
+ cur.execute('SELECT 1 FROM capability WHERE id = %s', (src_id,))
|
|
|
+ if cur.fetchone():
|
|
|
+ resolved = src_id
|
|
|
+ # (2) 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
|
|
|
+ # (3) create new with hash ID
|
|
|
+ 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['capability_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))
|
|
|
+ stats['criterion_backfilled'] += 1
|
|
|
+ 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['effects_backfilled'] += 1
|
|
|
+ stats['capability_linked'] += 1
|
|
|
+ cap_resolved[src_id or name] = resolved
|
|
|
+
|
|
|
+ # strategy (is_selected only)
|
|
|
+ strat_path = folder / 'strategy.json'
|
|
|
+ if not strat_path.exists():
|
|
|
+ return
|
|
|
+ try:
|
|
|
+ strat_data = json.loads(strat_path.read_text(encoding='utf-8'))
|
|
|
+ except Exception as e:
|
|
|
+ stats['bad_folders'].append(folder_key + f':strat_parse_err({e})')
|
|
|
+ return
|
|
|
+
|
|
|
+ selected = next((s for s in strat_data.get('strategies', []) if s.get('is_selected')), None)
|
|
|
+ if not selected:
|
|
|
+ sel_list = strat_data.get('strategies', [])
|
|
|
+ selected = sel_list[0] if sel_list else None
|
|
|
+ if not selected:
|
|
|
+ return
|
|
|
+
|
|
|
+ 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['strategy'] += 1
|
|
|
+
|
|
|
+ # 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))
|
|
|
+ cur.execute("""INSERT INTO strategy_resource (strategy_id, resource_id)
|
|
|
+ VALUES (%s, %s) ON CONFLICT DO NOTHING""", (strat_id, rid))
|
|
|
+ cur.execute("""INSERT INTO requirement_strategy (requirement_id, strategy_id)
|
|
|
+ VALUES (%s, %s) ON CONFLICT DO NOTHING""", (req_id, strat_id))
|
|
|
+
|
|
|
+ strat_cap_ids = set()
|
|
|
+ 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 c_ref in caps:
|
|
|
+ if not isinstance(c_ref, dict):
|
|
|
+ continue
|
|
|
+ key = c_ref.get('id') or c_ref.get('name', '')
|
|
|
+ resolved = cap_resolved.get(key) or alias.get(norm(c_ref.get('name', '')))
|
|
|
+ if resolved:
|
|
|
+ strat_cap_ids.add(resolved)
|
|
|
+ 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))
|
|
|
+ cur.execute("""INSERT INTO requirement_capability (requirement_id, capability_id)
|
|
|
+ VALUES (%s, %s) ON CONFLICT DO NOTHING""",
|
|
|
+ (req_id, cid))
|
|
|
+ stats['folder_cap_count'] = len(strat_cap_ids)
|
|
|
+ stats['folder_res_count'] = len(resource_ids)
|
|
|
+
|
|
|
+
|
|
|
+# PHASE 4: apply renames (defensive — some may have already been applied during seeding)
|
|
|
+def apply_renames(stores):
|
|
|
+ print('\n📝 Applying RENAMES...', flush=True)
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ try:
|
|
|
+ applied = 0
|
|
|
+ for cid, (name, desc) in RENAMES.items():
|
|
|
+ cur.execute('UPDATE capability SET name = %s, description = %s WHERE id = %s',
|
|
|
+ (name, desc, cid))
|
|
|
+ if (cur.rowcount or 0) > 0:
|
|
|
+ applied += 1
|
|
|
+ print(f' applied {applied}/{len(RENAMES)}', flush=True)
|
|
|
+ finally:
|
|
|
+ cur.close()
|
|
|
+
|
|
|
+
|
|
|
+# ═══════════════════════════════════════════════════════════
|
|
|
+def main():
|
|
|
+ ap = argparse.ArgumentParser()
|
|
|
+ g = ap.add_mutually_exclusive_group(required=True)
|
|
|
+ g.add_argument('--backup-only', action='store_true')
|
|
|
+ g.add_argument('--dry-run', action='store_true')
|
|
|
+ g.add_argument('--execute', action='store_true')
|
|
|
+ ap.add_argument('--skip-backup', action='store_true', help='Skip backup (if already done)')
|
|
|
+ args = ap.parse_args()
|
|
|
+
|
|
|
+ print(f'\n{"="*60}', flush=True)
|
|
|
+ print(f'{"BACKUP" if args.backup_only else "DRY RUN" if args.dry_run else "EXECUTE"} '
|
|
|
+ f'— rebuild howard_dedup', flush=True)
|
|
|
+ print(f'{"="*60}', flush=True)
|
|
|
+
|
|
|
+ stores = {
|
|
|
+ 'cap': PostgreSQLCapabilityStore(),
|
|
|
+ 'res': PostgreSQLResourceStore(),
|
|
|
+ 'req': PostgreSQLRequirementStore(),
|
|
|
+ 'strat': PostgreSQLStrategyStore(),
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ if not args.skip_backup:
|
|
|
+ backup(stores)
|
|
|
+ if args.backup_only:
|
|
|
+ print('\n✅ Backup only. Exit.', flush=True)
|
|
|
+ return
|
|
|
+
|
|
|
+ # dump current caps (used for seed)
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ cur.execute('SELECT id, name, criterion, description, effects, version FROM capability')
|
|
|
+ current_caps = {r['id']: dict(r) for r in cur.fetchall()}
|
|
|
+ cur.close()
|
|
|
+
|
|
|
+ snapshot = parse_snapshot(SNAPSHOT_PATH)
|
|
|
+ alias = build_alias_map(snapshot, current_caps)
|
|
|
+ (BACKUP_DIR / 'alias_map.json').write_text(
|
|
|
+ json.dumps(alias, ensure_ascii=False, indent=2))
|
|
|
+ print(f'\n🔗 Alias map: {len(alias)} entries (snapshot={len(snapshot)}, '
|
|
|
+ f'current_caps={len(current_caps)})', flush=True)
|
|
|
+
|
|
|
+ if args.dry_run:
|
|
|
+ print('\n[DRY-RUN] Simulating folder 001 ingest...', flush=True)
|
|
|
+ f1 = OUTPUT_DIR / '001'
|
|
|
+ if f1.exists():
|
|
|
+ caps = json.loads((f1 / 'capabilities_extracted.json').read_text())
|
|
|
+ new = linked = 0
|
|
|
+ for c in caps.get('extracted_capabilities', [])[:15]:
|
|
|
+ name = (c.get('name') or '').strip()
|
|
|
+ src = c.get('id')
|
|
|
+ cand = alias.get(norm(name))
|
|
|
+ status = 'LINK' if cand else 'NEW'
|
|
|
+ if cand: linked += 1
|
|
|
+ else: new += 1
|
|
|
+ print(f' [{status}] src_id={src!r} → {cand!r} | {name[:50]}', flush=True)
|
|
|
+ print(f'\n folder 001 sample (first 15): link={linked} new={new}', flush=True)
|
|
|
+ print('\n[DRY-RUN] Skipping purge + ingest. Use --execute to run.', flush=True)
|
|
|
+ return
|
|
|
+
|
|
|
+ # Skip purge+seed if howard_dedup already has canonical seeds
|
|
|
+ # (enables resume after connection drop)
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ cur.execute("SELECT COUNT(*) AS c FROM capability WHERE version = %s",
|
|
|
+ (DEDUP_VERSION,))
|
|
|
+ hd_count = cur.fetchone()['c']
|
|
|
+ cur.close()
|
|
|
+ if hd_count > 100:
|
|
|
+ print(f'\n⚠️ howard_dedup already has {hd_count} caps — skipping purge+seed (resume mode)',
|
|
|
+ flush=True)
|
|
|
+ else:
|
|
|
+ purge(stores)
|
|
|
+ seed(stores, current_caps, snapshot)
|
|
|
+
|
|
|
+ # Ingest
|
|
|
+ print('\n📂 Ingesting output 2/ ...', flush=True)
|
|
|
+ stats = {'folder_processed': 0, 'bad_folders': [], 'missing_req': [],
|
|
|
+ 'resource': 0, 'capability_new': 0, 'capability_linked': 0,
|
|
|
+ 'strategy': 0, 'criterion_backfilled': 0, 'effects_backfilled': 0,
|
|
|
+ 'folder_cap_count': 0, 'folder_res_count': 0}
|
|
|
+ folders = sorted([d for d in OUTPUT_DIR.iterdir() if d.is_dir()])
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ try:
|
|
|
+ for idx, folder in enumerate(folders, 1):
|
|
|
+ before_cap_new = stats['capability_new']
|
|
|
+ before_cap_link = stats['capability_linked']
|
|
|
+ before_res = stats['resource']
|
|
|
+ before_strat = stats['strategy']
|
|
|
+ stats['folder_cap_count'] = 0
|
|
|
+ stats['folder_res_count'] = 0
|
|
|
+ try:
|
|
|
+ ingest_folder(folder, stores, alias, cur, stats)
|
|
|
+ stats['folder_processed'] += 1
|
|
|
+ d_new = stats['capability_new'] - before_cap_new
|
|
|
+ d_link = stats['capability_linked'] - before_cap_link
|
|
|
+ d_res = stats['resource'] - before_res
|
|
|
+ d_strat = stats['strategy'] - before_strat
|
|
|
+ print(f"[{idx:3d}/{len(folders)}] 📁 {folder.name}/ "
|
|
|
+ f"cap:link={d_link} new={d_new} res={d_res} strat={d_strat} "
|
|
|
+ f"strat_caps={stats['folder_cap_count']}",
|
|
|
+ flush=True)
|
|
|
+ except Exception as e:
|
|
|
+ print(f'[{idx:3d}/{len(folders)}] 📁 {folder.name}/ ❌ {type(e).__name__}: {e}',
|
|
|
+ flush=True)
|
|
|
+ stats['bad_folders'].append(folder.name + f':{type(e).__name__}')
|
|
|
+ # reconnect cursor if connection dropped
|
|
|
+ try:
|
|
|
+ cur.close()
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ finally:
|
|
|
+ try:
|
|
|
+ cur.close()
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+ apply_renames(stores)
|
|
|
+
|
|
|
+ # Verify
|
|
|
+ print(f'\n{"="*60}\n验证:', flush=True)
|
|
|
+ cur = stores['cap']._get_cursor()
|
|
|
+ for tbl in ['capability', 'strategy', 'resource', 'requirement']:
|
|
|
+ cur.execute(f"SELECT version, COUNT(*) AS c FROM {tbl} GROUP BY version")
|
|
|
+ for r in cur.fetchall():
|
|
|
+ print(f" {tbl} / {r['version']}: {r['c']}", flush=True)
|
|
|
+ cur.close()
|
|
|
+ print(f'\n📊 Stats:', flush=True)
|
|
|
+ for k, v in stats.items():
|
|
|
+ if isinstance(v, list):
|
|
|
+ print(f' {k}: {len(v)} {v[:5]}{"..." if len(v)>5 else ""}', flush=True)
|
|
|
+ else:
|
|
|
+ print(f' {k}: {v}', flush=True)
|
|
|
+ print(f'{"="*60}', flush=True)
|
|
|
+ finally:
|
|
|
+ for s in stores.values():
|
|
|
+ s.close()
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ main()
|
Talegorithm