#!/usr/bin/env python3
import sqlite3, json, os, re, hashlib
from datetime import datetime, timezone
K='/opt/struktur/youtube-research/knowledge.db'
CE='/opt/struktur/content-extraction/data/content_extraction.db'
BACK='/opt/struktur/reports/aa043-p34/20260827T223847Z/knowledge.db.pre-migration.bak'
OUT='/tmp/aa043_p34r1_audit.json'
def ro(path):
return sqlite3.connect('file:'+path+'?mode=ro',uri=True)
def rows(c,sql,args=()):
c.row_factory=sqlite3.Row
return [dict(x) for x in c.execute(sql,args)]
def cols(c,t):
c.row_factory=sqlite3.Row
return [dict(x) for x in c.execute('pragma table_info('+t+')')]
def tables(c): return [x[0] for x in c.execute("select name from sqlite_master where type='table' and name not like 'sqlite_%' order by name")]
def safe_sample(v):
if isinstance(v,str): return v[:240]
return v
def db_schema(path, target_tables=None):
c=ro(path); ts=tables(c); target_tables=target_tables or ts
d={'path':path,'tables':ts,'tables_detail':{},'foreign_keys':[]}
for t in target_tables:
if t not in ts: continue
d['tables_detail'][t]={'columns':cols(c,t),'fks':rows(c,'pragma foreign_key_list('+t+')'),'sql':(c.execute('select sql from sqlite_master where name=?',(t,)).fetchone() or [None])[0]}
for fk in d['tables_detail'][t]['fks']:
d['foreign_keys'].append({'table':t,**fk})
c.close(); return d
def fk_rows(path):
c=ro(path); c.row_factory=sqlite3.Row
violations=[dict(x) for x in c.execute('pragma foreign_key_check')]
result=[]
for v in violations:
t=v['table']; rid=v['rowid']; fkid=v['fkid']; fks=rows(c,'pragma foreign_key_list('+t+')')
fk=next((x for x in fks if x['id']==fkid),None)
child={}
if rid is not None:
try:
rr=c.execute('select * from "'+t.replace('"','""')+'" where rowid=?',(rid,)).fetchone()
if rr: child={k:safe_sample(rr[k]) for k in rr.keys()}
except Exception as e: child={'_read_error':str(e)}
result.append({'table':t,'rowid':rid,'parent':v['parent'],'fkid':fkid,'fk':fk,'child':child})
c.close(); return result
def logical_refs(path):
c=ro(path); out=[]
for t in tables(c):
for col in cols(c,t):
n=col['name'].lower()
if 'video_id' in n or 'youtube_id' in n or n in ('source_id','source_identity','canonical_url'):
try:
nonnull=c.execute('select count(*) from "'+t+'" where "'+col['name']+'" is not null').fetchone()[0]
distinct=c.execute('select count(distinct "'+col['name']+'") from "'+t+'" where "'+col['name']+'" is not null').fetchone()[0]
except Exception as e: nonnull=distinct=None
out.append({'table':t,'column':col['name'],'type':col['type'],'nonnull':nonnull,'distinct':distinct})
c.close(); return out
def backup_inventory():
root='/opt/struktur/reports'
found=[]
for base,dirs,files in os.walk(root):
dirs[:]=[d for d in dirs if not any(x in d.lower() for x in ['logs','node_modules'])]
for f in files:
if f.endswith(('.db','.bak','.sqlite')) and ('knowledge' in f.lower() or 'youtube' in f.lower()):
p=os.path.join(base,f)
try:
found.append({'path':p,'size':os.path.getsize(p),'mtime':datetime.fromtimestamp(os.path.getmtime(p),timezone.utc).isoformat()})
except OSError: pass
return sorted(found,key=lambda x:x['mtime'])
def code_search():
roots=['/opt/struktur','/etc/systemd','/home/hermes']
terms=['INSERT OR REPLACE INTO videos','REPLACE INTO videos','DELETE FROM videos','PRAGMA foreign_keys','foreign_keys=','DROP TABLE videos','ALTER TABLE videos']
hits=[]
for root in roots:
for base,dirs,files in os.walk(root):
dirs[:]=[d for d in dirs if d not in ['.git','__pycache__','logs','reports','backups','.venv','venv'] and not d.endswith('.bak')]
for f in files:
if not f.endswith(('.py','.sql','.service','.timer','.sh','.toml','.yaml','.yml')): continue
p=os.path.join(base,f)
try: lines=open(p,encoding='utf-8',errors='ignore').read().splitlines()
except Exception: continue
for i,line in enumerate(lines,1):
if any(term.lower() in line.lower() for term in terms):
hits.append({'path':p,'line':i,'text':line.strip()[:500]})
return hits
def ce_audit():
if not os.path.exists(CE): return {'exists':False}
c=ro(CE); ts=tables(c); d={'exists':True,'tables':ts,'details':{}}
for t in ts:
d['details'][t]={'columns':cols(c,t)}
# bounded, text-free inventory counts for likely identifiers
for t in ts:
names=[x['name'] for x in d['details'][t]['columns']]
idcols=[n for n in names if any(q in n.lower() for q in ['video_id','youtube_id','source_id','url','title'])]
if idcols:
d['details'][t]['identifier_columns']=idcols
for n in idcols:
try: d['details'][t][n]={'nonnull':c.execute('select count(*) from "'+n+'" where "'+n+'" is not null').fetchone()[0],'distinct':c.execute('select count(distinct "'+n+'") from "'+n+'" where "'+n+'" is not null').fetchone()[0]}
except Exception: pass
c.close(); return d
now=datetime.now(timezone.utc).isoformat()
prod_schema=db_schema(K,['videos','chapters','video_links','knowledge_units','source_versions','source_processing_registry','reconciliation_decisions','graphiti_import_queue'])
viol=fk_rows(K)
# compact distribution and missing parent ids for each violated FK
by={}
for x in viol:
key=(x['table'],x['fkid'],x['parent'],(x['fk'] or {}).get('from'),(x['fk'] or {}).get('to'))
z=by.setdefault(key,{'table':key[0],'fkid':key[1],'parent':key[2],'child_column':key[3],'parent_column':key[4],'on_update':(x['fk'] or {}).get('on_update'),'on_delete':(x['fk'] or {}).get('on_delete'),'rows':0,'rowids':[],'missing_parent_values':{}})
z['rows']+=1; z['rowids'].append(x['rowid'])
val=x['child'].get(key[3]) if x['child'] else None
z['missing_parent_values'][str(val)]=z['missing_parent_values'].get(str(val),0)+1
for z in by.values():
z['rowids']=z['rowids'][:20]; z['missing_parent_values']=dict(sorted(z['missing_parent_values'].items(),key=lambda kv:-kv[1]))
# source-side data samples/counts for affected video ids
c=ro(K); c.row_factory=sqlite3.Row
missing_ids=sorted({int(v) for x in viol for v in [x['child'].get((x['fk'] or {}).get('from'))] if isinstance(v,(int,str)) and str(v).isdigit()})
video_info=[]
for vid in missing_ids:
refs=[]
for t in ['chapters','video_links']:
if t in tables(c):
cs=[x['name'] for x in cols(c,t)]
if 'video_id' in cs: refs.append({'table':t,'count':c.execute('select count(*) from '+t+' where video_id=?',(vid,)).fetchone()[0]})
video_info.append({'video_id':vid,'child_refs':refs})
# source reconstruction by scanning all textual columns for exact video ids/youtube ids is bounded to table/column counts
c.close()
prod={'checked_at':now,'schema':prod_schema,'foreign_key_check':{'count':len(viol),'rows':viol},'distribution':list(by.values()),'missing_video_ids':missing_ids,'video_child_info':video_info,'logical_references':logical_refs(K),'backup_inventory':backup_inventory(),'code_search':code_search(),'ce':ce_audit()}
# compare exact violations with backup
for name,path in [('production',K),('backup',BACK)]:
try:
vv=fk_rows(path); prod['comparison_'+name]={'fk_count':len(vv),'fingerprint':hashlib.sha256(json.dumps([(x['table'],x['rowid'],x['parent'],x['fkid']) for x in vv],sort_keys=True).encode()).hexdigest()}
except Exception as e: prod['comparison_'+name]={'error':str(e)}
open(OUT,'w',encoding='utf-8').write(json.dumps(prod,ensure_ascii=False,indent=2,sort_keys=True))
print(json.dumps({'out':OUT,'checked_at':now,'fk_count':len(viol),'distribution':list(by.values()),'missing_video_ids':missing_ids,'prod_vs_backup':prod.get('comparison_production'), 'backup':prod.get('comparison_backup')},ensure_ascii=False))