import json,sqlite3,subprocess,time,urllib.request,datetime,os,sys
from pathlib import Path
BASE='/opt/struktur/obsidian-graphiti-import'; Q=BASE+'/obsidian_graphiti_import.db'; R='/opt/struktur/graphiti/request-data/requests.db'; EV='/opt/struktur/graphiti/request-data/processing-events.jsonl'; OUT='/opt/struktur/aa037'; Path(OUT).mkdir(exist_ok=True)
EXCLUDE={11,13,31,69,166,182,184}
def conn(path): return sqlite3.connect('file:'+path+'?mode=ro',uri=True)
def row(qid):
c=conn(Q); c.row_factory=sqlite3.Row; r=c.execute('select id,relative_path,status,attempts,size,last_error,last_error_phase,graphiti_episode_uuid from files where id=?',(qid,)).fetchone(); c.close(); return dict(r) if r else None
def health(): return json.loads(urllib.request.urlopen('http://127.0.0.1:8644/health',timeout=30).read())
def active():
c=conn(R); n=c.execute("select count(*) from graphiti_requests where status in ('started','processing')").fetchone()[0]; c.close(); return n
def latest_request(qid, before):
c=conn(R); c.row_factory=sqlite3.Row; rs=c.execute('select rowid,* from graphiti_requests where queue_id=? and started_at>=? order by rowid desc limit 1',(str(qid),before)).fetchall(); c.close(); return dict(rs[0]) if rs else None
def telemetry(rid):
out=[]
for line in Path(EV).read_text(errors='replace').splitlines():
try:
x=json.loads(line)
if x.get('request_id')==rid: out.append(x)
except Exception: pass
llm=[x for x in out if x.get('phase','').startswith('llm_call') and x.get('outcome')=='completed']
edge=[x for x in llm if x.get('graphiti_phase')=='edge_extraction']
return {'events':len(out),'max_tokens':sorted(set(x.get('max_tokens') for x in llm)),'max_output':max([x.get('output_tokens') or 0 for x in llm],default=0),'edge_output':[x.get('output_tokens') for x in edge],'edge_total':max([x.get('edge_total') or 0 for x in out],default=0),'edge_completed':len([x for x in out if x.get('phase')=='edge_completed' and x.get('outcome')=='completed']),'truncation':any(x.get('error_class')=='LLM_OUTPUT_TRUNCATED' for x in out),'json_error':any('JSONDecodeError' in json.dumps(x) for x in out),'neo4j_commit':[(x.get('outcome'),x.get('error_message')) for x in out if x.get('phase')=='neo4j_write_commit'],'errors':[x for x in out if x.get('outcome') in ('failed','error') or x.get('error_class') in ('LLM_OUTPUT_TRUNCATED','JSONDecodeError')]}
def candidates():
c=conn(Q); c.row_factory=sqlite3.Row; rs=[]
for r in c.execute("select id,relative_path,status,attempts,size,last_error,last_error_phase,graphiti_episode_uuid from files where status in ('retry','failed') and graphiti_episode_uuid is null order by id"):
d=dict(r)
if d['id'] not in EXCLUDE: rs.append(d)
c.close(); return rs
cands=candidates(); Path(OUT+'/candidates.json').write_text(json.dumps(cands,ensure_ascii=False,indent=2))
results=[]
for idx,cand in enumerate(cands,1):
qid=cand['id']; before=datetime.datetime.now(datetime.timezone.utc).isoformat(); pre=row(qid)
if not pre or pre['status'] not in ('retry','failed') or pre['graphiti_episode_uuid'] is not None or active()!=0:
results.append({'queue_id':qid,'relative_path':cand['relative_path'],'outcome':'PRECHECK_FAIL','pre':pre,'active':active()}); Path(OUT+'/results.json').write_text(json.dumps(results,ensure_ascii=False,indent=2)); break
cmd=['python3',BASE+'/importer.py','--once','--ids',str(qid),'--force-id',str(qid),'--limit','1','--hold-after-success']
started=time.time();
try: p=subprocess.run(cmd,capture_output=True,text=True,timeout=1100)
except subprocess.TimeoutExpired as e: p=type('P',(),{'returncode':124,'stdout':e.stdout or '','stderr':str(e)})()
post=row(qid); req=latest_request(qid,before); tele=telemetry(req['request_id']) if req else {}
try: h=health()
except Exception as e: h={'error':str(e)}
ok=bool(req and req.get('status')=='completed' and req.get('http_status')==200 and post and post.get('status')=='done' and post.get('graphiti_episode_uuid') and tele.get('neo4j_commit') and tele.get('neo4j_commit')[-1][0]=='completed' and not tele.get('truncation') and not tele.get('json_error') and h.get('ready') is True and h.get('neo4j_ready') is True and h.get('provider_ready') is True and h.get('provider_rate_limited') is False and active()==0)
result={'queue_id':qid,'relative_path':cand['relative_path'],'pre':pre,'post':post,'request':req,'telemetry':tele,'health':h,'returncode':p.returncode,'stdout':p.stdout[-2000:],'stderr':p.stderr[-2000:],'duration_s':round(time.time()-started,2),'outcome':'success' if ok else 'FAIL_STOP'}
results.append(result); Path(OUT+'/results.json').write_text(json.dumps(results,ensure_ascii=False,indent=2))
if not ok: break
print(json.dumps({'candidate_count':len(cands),'processed_results':len(results),'last_outcome':results[-1]['outcome'] if results else None,'output':OUT},ensure_ascii=False))