import sqlite3,json,os,hashlib,datetime,glob,subprocess,collections,re,itertools,statistics
DB='/opt/struktur/youtube-research/knowledge.db'; D='/opt/struktur/reports/aa043-p39/20260828T084500Z'; now=datetime.datetime.now(datetime.timezone.utc).isoformat()
def ro():
 c=sqlite3.connect('file:'+DB+'?mode=ro&immutable=1',uri=True); c.row_factory=sqlite3.Row; return c
c=ro(); rows=[dict(x) for x in c.execute('select * from knowledge_units')]; fk=[tuple(x) for x in c.execute('pragma foreign_key_check')]; integrity=c.execute('pragma integrity_check').fetchone()[0]; q=dict(c.execute('select graphiti_status,count(*) from graphiti_import_queue group by graphiti_status')); rel={str(x[0]):x[1] for x in c.execute('select graphiti_release,count(*) from graphiti_import_queue group by graphiti_release')}; c.close()
batches=[json.load(open(f'{D}/p39-batch-{i:02d}.json')) for i in range(1,6)]
new_ids=set();
for b in batches:
 new_ids.update(x['source_id'] for x in b['sources'])
# exact source-based new KU count and status distributions
newrows=[x for x in rows if x.get('promotion_reason','').startswith('P39 controlled historical batch')]
agg={'candidates':sum(b['candidate_count'] for b in batches),'valid':sum(b['valid_count'] for b in batches),'rejects':sum(b['reject_count'] for b in batches),'exact_duplicates':sum(b['duplicate_count'] for b in batches),'semantic_duplicates':sum(b['semantic_duplicate_count'] for b in batches),'new_unique_kus':sum(b['new_unique_kus'] for b in batches),'new_provenance':sum(b['new_provenance'] for b in batches),'new_review_audit':sum(b['new_review_audit'] for b in batches),'review_verdicts':dict(collections.Counter(k for b in batches for k,v in b['review_verdicts'].items() for _ in range(v))),'placeholder_count':sum(b['placeholder_count'] for b in batches),'hallucination_count':sum(b['hallucination_count'] for b in batches),'provenance_errors':sum(b['provenance_error_count'] for b in batches),'canonical_dedup_errors':sum(b['canonical_dedup_error_count'] for b in batches),'zero_ku_sources':sum(1 for b in batches for s in b['sources'] if s['new_kus']==0)}
# density by source and batch over actual KUs; statement KUs use source text denominator from result source metadata
per=[]
for b in batches:
 per.extend([s['new_kus']*1000/s['word_count'] for s in b['sources'] if s['word_count']])
density={'median':statistics.median(per),'mean':statistics.mean(per),'min':min(per),'max':max(per)} if per else {}
# Fresh baseline fields and all requested final statuses.
final={'checked_at':now,'db':DB,'db_sha256':hashlib.sha256(open(DB,'rb').read()).hexdigest(),'db_size':os.path.getsize(DB),'integrity_check':integrity,'foreign_key_check':len(fk),'review':dict(collections.Counter(x['review_state'] for x in rows)),'evidence':dict(collections.Counter(x['evidence_class'] for x in rows)),'gold':dict(collections.Counter(x['gold_state'] for x in rows)),'ku_total':len(rows),'real_kus':sum(not x['legacy_placeholder'] for x in rows),'legacy':sum(bool(x['legacy_placeholder']) for x in rows),'new_p39_kus':len(newrows),'queue':q,'queue_release':rel,'writer':{'active':subprocess.run(['systemctl','is-active','aa043-single-writer-loop.service'],capture_output=True,text=True).stdout.strip(),'substate':subprocess.run(['systemctl','show','-p','SubState','--value','aa043-single-writer-loop.service'],capture_output=True,text=True).stdout.strip(),'mainpid':subprocess.run(['systemctl','show','-p','MainPID','--value','aa043-single-writer-loop.service'],capture_output=True,text=True).stdout.strip()},'fk_fingerprint':hashlib.sha256(json.dumps(sorted(fk),sort_keys=True).encode()).hexdigest(),'aggregate':agg,'density':density,'batch_gates':[(b['batch'],b['gate'],b['idempotency']) for b in batches]}
json.dump(final,open(D+'/p39-final-metrics.json','w'),ensure_ascii=False,indent=2,sort_keys=True)
print(json.dumps(final,ensure_ascii=False))