#!/usr/bin/env python3 import argparse, fcntl, hashlib, json, os, re, signal, socket, sqlite3, sys, threading, time, urllib.error, urllib.request from datetime import datetime, timezone, timedelta from pathlib import Path BASE=Path('/opt/struktur/obsidian-graphiti-import'); CFG=json.loads((BASE/'config.json').read_text()); DB=Path(CFG['db']); NOW=lambda: datetime.now(timezone.utc).isoformat() GRAPHITI_REQUEST_DEADLINE_SECONDS=float(os.getenv('GRAPHITI_REQUEST_DEADLINE_SECONDS',CFG.get('graphiti_request_deadline_seconds',330))); IMPORTER_HTTP_TIMEOUT_SECONDS=float(os.getenv('IMPORTER_HTTP_TIMEOUT_SECONDS',CFG.get('importer_http_timeout_seconds',360))); OUTER_PROCESS_TIMEOUT_SECONDS=float(os.getenv('OUTER_PROCESS_TIMEOUT_SECONDS',CFG.get('outer_process_timeout_seconds',420))) LOCK_PATH=BASE/'import.lock'; LOCK_HEARTBEAT_SECONDS=15 PROVIDER_STATE_PATH=BASE/'provider_state.json'; RUN_BLOCK_PATH=BASE/'import_run_block.json'; CONSERVATIVE_PROVIDER_BLOCK_SECONDS=24*60*60 def process_start_metadata(pid): try: raw=Path(f'/proc/{pid}/stat').read_text(); tail=raw.rsplit(') ',1)[1].split(); ticks=int(tail[19]) btime=next(int(x.split()[1]) for x in Path('/proc/stat').read_text().splitlines() if x.startswith('btime ')); hz=os.sysconf(os.sysconf_names['SC_CLK_TCK']) return {'pid':pid,'process_start_ticks':ticks,'process_start_time':datetime.fromtimestamp(btime+ticks/hz,timezone.utc).isoformat()} except (FileNotFoundError,ValueError,StopIteration): return {'pid':pid,'process_start_ticks':None,'process_start_time':None} class ImportLock: def __init__(self, run_id): self.run_id=run_id; self.fd=None; self.inode=None; self.stop=threading.Event(); self.thread=None; self.old_handlers={} self.metadata={} def _write(self): self.metadata['heartbeat_at']=NOW(); data=(json.dumps(self.metadata,ensure_ascii=False,sort_keys=True)+'\n').encode() os.lseek(self.fd,0,os.SEEK_SET); os.ftruncate(self.fd,0); os.write(self.fd,data); os.fsync(self.fd) def _heartbeat(self): while not self.stop.wait(LOCK_HEARTBEAT_SECONDS): try: self._write() except OSError: return def acquire(self): fd=os.open(LOCK_PATH,os.O_RDWR|os.O_CREAT,0o644) try: fcntl.flock(fd,fcntl.LOCK_EX|fcntl.LOCK_NB) except BlockingIOError: try: os.lseek(fd,0,os.SEEK_SET); raw=os.read(fd,8192).decode(errors='replace').strip(); print(json.dumps({'already_running':True,'lock_path':str(LOCK_PATH),'lock_metadata':raw},ensure_ascii=False)) finally: os.close(fd) return False self.fd=fd; self.inode=os.fstat(fd).st_ino; start=process_start_metadata(os.getpid()) self.metadata={'process_id':os.getpid(),'process_start_time':start['process_start_time'],'process_start_ticks':start['process_start_ticks'],'run_id':self.run_id,'hostname':socket.gethostname(),'created_at':NOW(),'heartbeat_at':NOW(),'command':' '.join(sys.argv),'database_path':str(DB),'lock_inode':self.inode} self._write(); self.thread=threading.Thread(target=self._heartbeat,name='import-lock-heartbeat',daemon=True); self.thread.start() if threading.current_thread() is threading.main_thread(): for sig in (signal.SIGTERM,signal.SIGINT): self.old_handlers[sig]=signal.getsignal(sig); signal.signal(sig,self._signal_handler) return True def _signal_handler(self, signum, frame): self.release(); raise SystemExit(128+signum) def release(self): if self.fd is None: return self.stop.set() if self.thread: self.thread.join(timeout=2) try: if os.stat(LOCK_PATH).st_ino==self.inode: os.unlink(LOCK_PATH) except FileNotFoundError: pass finally: for sig,handler in self.old_handlers.items(): signal.signal(sig,handler) fcntl.flock(self.fd,fcntl.LOCK_UN); os.close(self.fd); self.fd=None def normalized_relative_path(rel): return Path(str(rel).replace('\\','/')).as_posix().lstrip('/') def import_key(rel,h): rel=normalized_relative_path(rel) if not h or len(h)!=64: raise ValueError('invalid content_hash') return hashlib.sha256(('obsidian|'+rel+'|'+h).encode()).hexdigest() def formal_import_identity(source_identity, version_hash): source_identity=(source_identity or '').strip() version_hash=(version_hash or '').strip().lower() if not source_identity or not re.fullmatch(r'[0-9a-f]{64}',version_hash): raise ValueError('invalid formal identity material') return hashlib.sha256(('v1\n'+source_identity+'\n'+version_hash).encode('utf-8')).hexdigest() YOUTUBE_URL_RE=re.compile(r'(?:youtube\.com/(?:watch\?v=|shorts/)|youtu\.be/)([A-Za-z0-9_-]{6,})',re.I) def source_identity_for(raw,rel): m=YOUTUBE_URL_RE.search(raw or '') return 'youtube:'+m.group(1) if m else 'obsidian:'+normalized_relative_path(rel) def canonical_content_bytes(raw): text=(raw or '').replace('\r\n','\n').replace('\r','\n') return '\n'.join(line.rstrip(' \t') for line in text.split('\n')).encode('utf-8') def content_version_hash(raw): return hashlib.sha256(canonical_content_bytes(raw)).hexdigest() def _metadata_from_source_description(sd): m=re.search(r'\|\s*(\{.*\})$',sd or '') if not m: return {} try: x=json.loads(m.group(1)); return x if isinstance(x,dict) else {} except Exception: return {} def _atomic_json(path, payload): tmp=Path(str(path)+'.tmp'); tmp.write_text(json.dumps(payload,ensure_ascii=False,sort_keys=True,indent=2)+'\n'); os.replace(tmp,path) def _read_json(path): try: return json.loads(Path(path).read_text()) except (FileNotFoundError,json.JSONDecodeError,OSError): return None def provider_reset_at(text): m=re.search(r'X-RateLimit-Reset[^0-9]*(\d{10,})',text or '',re.I) if not m: return None value=float(m.group(1)); value=value/1000 if value>10**11 else value try: return datetime.fromtimestamp(value,timezone.utc).isoformat() except (OverflowError,OSError,ValueError): return None def provider_rate_limit(text): t=(text or '').lower(); return 'provider_rate_limit' if ('429' in t or 'rate limit' in t or 'ratelimit' in t) else None def provider_block_active(): state=_read_json(PROVIDER_STATE_PATH) if not state or not state.get('blocked'): return False,state until=state.get('next_allowed_at') or state.get('reset_at') if until: try: return datetime.now(timezone.utc) < datetime.fromisoformat(until),state except ValueError: pass return True,state def set_provider_block(error_text,run_id,queue_id): now=datetime.now(timezone.utc); reset=provider_reset_at(error_text); reset_dt=datetime.fromisoformat(reset) if reset else None; next_allowed=reset if reset_dt and reset_dt>now else (now+timedelta(seconds=CONSERVATIVE_PROVIDER_BLOCK_SECONDS)).isoformat() state={'blocked':True,'reason':'provider_rate_limit','blocked_at':now.isoformat(),'reset_at':reset,'next_allowed_at':next_allowed,'run_id':run_id,'queue_id':str(queue_id),'provider':'OpenRouter','model':'nvidia/nemotron-3-super-120b-a12b:free'} _atomic_json(PROVIDER_STATE_PATH,state); return state def set_run_block(reason,run_id,queue_id): _atomic_json(RUN_BLOCK_PATH,{'blocked':True,'reason':reason,'blocked_at':NOW(),'run_id':run_id,'queue_id':str(queue_id)}) def run_block_active(): state=_read_json(RUN_BLOCK_PATH); return bool(state and state.get('blocked')),state def conn(): c=sqlite3.connect(DB,timeout=30); c.row_factory=sqlite3.Row; c.execute('pragma busy_timeout=30000'); return c def init(c): c.execute('''create table if not exists files(id integer primary key, relative_path text not null unique, absolute_path text not null, source_area text, document_type text, filename text, size integer, modified_at text, content_hash text, discovered_at text, status text not null, attempts integer default 0, last_attempt text, imported_at text, episode_ref text, last_error text, importer_version text, metadata_json text, supersedes_id integer, unique(relative_path,content_hash))''') c.execute('''create table if not exists runs(id integer primary key, started_at text, finished_at text, scanned integer, allowed integer, excluded integer, done integer, errors integer)''') for col,typ in [('import_key','text'),('source_identity','text'),('content_version_hash','text'),('import_identity','text'),('updated_at','text'),('graphiti_episode_name','text'),('graphiti_episode_uuid','text'),('graphiti_status','text'),('graphiti_created_at','text'),('last_verified_at','text'),('last_error_phase','text')]: try: c.execute(f'alter table files add column {col} {typ}') except sqlite3.OperationalError: pass c.execute('''create table if not exists status_history(id integer primary key,queue_id integer,old_status text,new_status text,timestamp text,phase text,run_id text,message text)''') c.execute('create unique index if not exists idx_files_import_key on files(import_key) where import_key is not null') for row in c.execute('select id,relative_path,absolute_path,content_hash,source_identity,content_version_hash,import_identity from files').fetchall(): sid=row['source_identity']; cvh=row['content_version_hash']; iid=row['import_identity'] try: p=Path(row['absolute_path']); raw=p.read_text(encoding='utf-8',errors='ignore') if p.exists() else '' sid=sid or source_identity_for(raw,row['relative_path']); cvh=cvh or content_version_hash(raw); iid=iid or (formal_import_identity(sid,cvh) if sid and cvh else None) except (OSError,UnicodeError,ValueError): pass c.execute('update files set import_key=coalesce(import_key,?),source_identity=coalesce(source_identity,?),content_version_hash=coalesce(content_version_hash,?),import_identity=coalesce(import_identity,?),graphiti_status=coalesce(graphiti_status,status),updated_at=coalesce(updated_at,?) where id=?',(import_key(row['relative_path'],row['content_hash']),sid,cvh,iid,NOW(),row['id'])) c.commit() def transition(c,qid,new,phase,message,run_id): old=c.execute('select status from files where id=?',(qid,)).fetchone()['status'] if old!=new: c.execute('insert into status_history(queue_id,old_status,new_status,timestamp,phase,run_id,message) values(?,?,?,?,?,?,?)',(qid,old,new,NOW(),phase,run_id,message)) c.execute('update files set status=?,graphiti_status=? where id=?',(new,new,qid)) def front(t): if not t.startswith('---'): return {} out={} for line in t.split('---',2)[1].splitlines(): m=re.match(r'^([\w-]+):\s*(.*)$',line) if m: out[m.group(1)]=m.group(2).strip().strip('"') return out def active_request(queue_id): p=Path('/opt/struktur/graphiti/request-data/requests.db') if not p.exists(): return False try: c=sqlite3.connect(p); n=c.execute("select count(*) from graphiti_requests where queue_id=? and status in ('started','processing')",(str(queue_id),)).fetchone()[0]; c.close(); return n>0 except Exception: return False def request_state(request_id): p=Path('/opt/struktur/graphiti/request-data/requests.db') if not p.exists(): return None try: c=sqlite3.connect(p); c.row_factory=sqlite3.Row; row=c.execute('select request_id,status,server_completed,episode_uuid,terminal_reason,recovery_result from graphiti_requests where request_id=?',(request_id,)).fetchone(); c.close(); return dict(row) if row else None except Exception: return None def allowed(p, cfg): rel=p.relative_to(Path(cfg['vault'])); s=str(rel).lower() if p.suffix.lower() not in cfg['extensions'] or any(x in s for x in [d.lower() for d in cfg['deny_patterns']]): return False,'denylist' if not cfg['allow_roots'] or rel.parts[0] not in cfg['allow_roots']: return False,'not-allowlisted' t=p.read_text(encoding='utf-8',errors='ignore') if len(t.strip()) {h}; import_key {old['import_key']} -> {key}; no episode present")) continue # Existing imported/active records are retained; versioning requires explicit reconciliation. continue meta=front(raw); status='queued' if not old else 'superseded' if old and old['status']=='done': status='queued' c.execute('insert into files(relative_path,absolute_path,source_area,document_type,filename,size,modified_at,content_hash,content_version_hash,source_identity,import_identity,import_key,discovered_at,updated_at,status,attempts,importer_version,metadata_json,supersedes_id) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)',(rel,str(p),p.relative_to(root).parts[0],meta.get('type','markdown'),p.name,st.st_size,datetime.fromtimestamp(st.st_mtime,timezone.utc).isoformat(),h,cvh,sid,iid,key,t,t,status,0,CFG['importer_version'],json.dumps(meta,ensure_ascii=False),old['id'] if old else None)) for row in c.execute('select id,relative_path from files where status not in ("deleted","superseded")').fetchall(): if row['relative_path'] not in seen and not (root/row['relative_path']).exists(): c.execute('update files set status="deleted" where id=?',(row['id'],)) c.commit(); return scanned,allowed_n,excluded def post(row): p=Path(row['absolute_path']); raw=p.read_text(encoding='utf-8',errors='ignore'); title=re.search(r'(?m)^#\s+(.+)$',raw); title=title.group(1).strip() if title else p.stem meta=json.loads(row['metadata_json'] or '{}'); run_id=os.getenv('OBSIDIAN_IMPORT_RUN_ID') or ('run-'+NOW()); request_id='req-'+hashlib.sha256((run_id+str(row['id'])+NOW()).encode()).hexdigest()[:24]; source_identity=row['source_identity'] or ('obsidian:'+row['relative_path']); version_hash=row['content_version_hash'] or row['content_hash']; formal_identity=row['import_identity'] or formal_import_identity(source_identity,version_hash); provenance={'import_identity':formal_identity,'identity_version':'v1','source_identity':source_identity,'content_version_hash':version_hash,'canonical_obsidian_path':row['relative_path'],'reconciliation_record_id':'legacy-queue-'+str(row['id']),'source_system':'obsidian-general','source_type':'markdown','request_id':request_id,'run_id':run_id,'import_key':row['import_key'],'obsidian_path':row['relative_path'],'document_title':title,'document_type':row['document_type'],'content_hash':row['content_hash'],'source_area':row['source_area'],'queue_id':row['id'],'importer_name':'obsidian-graphiti-single-writer','importer_version':CFG['importer_version'],'imported_at':NOW(),'metadata':meta}; payload={'content':raw[:8000],'source':'obsidian','actor':'obsidian-single-writer','name':'aa043_'+formal_identity,'context':json.dumps(provenance,ensure_ascii=False,sort_keys=True),'timestamp':NOW()} req=urllib.request.Request(CFG['graphiti_url']+'/episodes',json.dumps(payload).encode(),{'Content-Type':'application/json'}) try: with urllib.request.urlopen(req,timeout=IMPORTER_HTTP_TIMEOUT_SECONDS) as r: return json.loads(r.read()) except urllib.error.HTTPError as e: if e.code == 503: detail=e.read().decode(errors='replace')[:500] raise RuntimeError('maintenance_mode: '+detail) from e if e.code in (401,429,502,504): detail=e.read().decode(errors='replace')[:1500] phase='provider_rate_limit' if e.code==429 else 'provider_authentication' if e.code==401 else 'server_timeout' if e.code==504 else 'provider_response' raise RuntimeError(phase+': '+detail) from e raise except (socket.timeout, TimeoutError, urllib.error.URLError) as exc: reason=getattr(exc,'reason',exc) if not isinstance(exc,(socket.timeout,TimeoutError)) and not isinstance(reason,(socket.timeout,TimeoutError)): raise state=request_state(request_id) try: episodes=check(row) except Exception: episodes=[] if episodes: raise RuntimeError('request_timeout_episode_present: '+json.dumps({'request_id':request_id,'episode_count':len(episodes)})) from exc if state and state.get('status') in ('timed_out','cancelled','abandoned') and not state.get('episode_uuid'): raise RuntimeError('request_timeout_before_processing: '+json.dumps(state,ensure_ascii=False)) from exc raise RuntimeError('technical_review: request state unresolved after local HTTP timeout') from exc def check(row): with urllib.request.urlopen(CFG['graphiti_url']+'/episodes/inventory',timeout=30) as r: inventory=json.loads(r.read()).get('episodes',[]) matches=[]; partial=[]; expected_identity=row['import_identity'] or '' for e in inventory: sd=e.get('source_description') or ''; meta=_metadata_from_source_description(sd) exact=expected_identity and meta.get('import_identity')==expected_identity legacy=(e.get('name')=='obsidian_'+row['import_key'] or meta.get('import_key')==row['import_key'] or (meta.get('obsidian_path')==row['relative_path'] and meta.get('content_hash')==row['content_hash'])) if exact or legacy: matches.append(e); continue same_source=meta.get('source_identity')==row['source_identity'] same_path=meta.get('canonical_obsidian_path')==row['relative_path'] or meta.get('obsidian_path')==row['relative_path'] if same_source or same_path: partial.append(e) if partial and not matches: raise RuntimeError('identity_review: partial or historical Graphiti match without exact formal identity; candidates='+str(len(partial))) return matches def validate_force_selection(ids, force_ids): force_ids=set(force_ids or []) if not force_ids: return None if ids is None: return 'force_id_requires_explicit_ids' if len(force_ids) != 1: return 'exactly_one_force_id_allowed' if len(ids) != 1 or next(iter(force_ids)) not in set(ids): return 'force_id_must_match_exactly_one_explicit_id' return None def run(limit, ids=None, scan_enabled=True, force_ids=None, hold_after_success=False): validation_error=validate_force_selection(ids, force_ids) if validation_error: print(json.dumps({'error':validation_error,'ids':ids or [],'force_ids':sorted(set(force_ids or []))},ensure_ascii=False)); return 2 run_id=os.getenv('OBSIDIAN_IMPORT_RUN_ID') or ('run-'+NOW()); lock=ImportLock(run_id) if not lock.acquire(): return 2 force_ids=set(force_ids or []) blocked,state=provider_block_active() run_block,block_state=run_block_active() if blocked: print(json.dumps({'blocked':True,'reason':'provider_rate_limit','provider_state':state},ensure_ascii=False)); lock.release(); return 75 if run_block and not force_ids: print(json.dumps({'blocked':True,'reason':'import_run_block','run_state':block_state},ensure_ascii=False)); lock.release(); return 75 c=None try: c=conn(); init(c); started=NOW(); print(json.dumps({'scan_enabled':scan_enabled,'requested_ids':ids or [],'database_path':str(DB),'batch_limit':limit,'import_enabled':True,'graphiti_request_deadline_seconds':GRAPHITI_REQUEST_DEADLINE_SECONDS,'importer_http_timeout_seconds':IMPORTER_HTTP_TIMEOUT_SECONDS,'outer_process_timeout_seconds':OUTER_PROCESS_TIMEOUT_SECONDS,'provider_blocked':False,'run_blocked':bool(run_block)})); scanned,allowed_n,excluded=scan(c,started) if scan_enabled else (0,0,0); done=errors=0 if ids: marks=','.join('?' for _ in ids) if force_ids: rows=c.execute(f'select * from files where id in ({marks}) and status not in ("done","duplicate","verified") order by id',tuple(ids)).fetchall() else: rows=c.execute(f'select * from files where id in ({marks}) and status in ("queued","retry") and attempts< ? order by id',(*ids,CFG['max_attempts'])).fetchall() else: rows=c.execute('select * from files where status in ("queued","retry") and attempts< ? order by id limit ?', (CFG['max_attempts'],limit)).fetchall() for row in rows: if row['id'] in force_ids and row['attempts'] >= CFG['max_attempts']: c.execute('insert into status_history(queue_id,old_status,new_status,timestamp,phase,run_id,message) values(?,?,?,?,?,?,?)',(row['id'],row['status'],row['status'],NOW(),'manual_retry_override',started,'Controlled post-write failpoint test; previous attempts='+str(row['attempts']))); c.commit() transition(c,row['id'],'preflight','preflight','reconciliation',started); c.commit() try: existing=check(row) if len(existing)>1: transition(c,row['id'],'duplicate','preflight','multiple matching episodes',started); errors+=1; c.commit(); continue if len(existing)==1: e=existing[0]; transition(c,row['id'],'verified','reconcile','existing episode adopted',started); c.execute('update files set graphiti_episode_name=?,graphiti_episode_uuid=?,graphiti_created_at=?,last_verified_at=?,episode_ref=?,last_error=null where id=?',(e.get('name'),e.get('uuid'),str(e.get('created_at')),NOW(),e.get('name'),row['id'])); transition(c,row['id'],'done','verified','existing episode verified',started); done+=1; c.commit(); continue transition(c,row['id'],'processing','processing','POST /episodes',started); c.commit() result=post(row); c.execute('update files set attempts=attempts+1,last_attempt=? where id=?',(NOW(),row['id'])); transition(c,row['id'],'graphiti_created','graphiti_created','Graphiti POST completed',started); c.execute('update files set graphiti_episode_name=?,graphiti_episode_uuid=?,episode_ref=? where id=?',(result.get('name'),result.get('uuid'),result.get('name'),row['id'])); c.commit() verified=check(row) if len(verified)!=1: raise RuntimeError('verification did not find exactly one episode') if os.getenv('OBSIDIAN_IMPORT_FAILPOINT_ID') == str(row['id']): e=verified[0]; c.execute('update files set graphiti_episode_name=?,graphiti_episode_uuid=?,episode_ref=?,graphiti_created_at=? where id=?',(e.get('name'),e.get('uuid'),e.get('name'),str(e.get('created_at')),row['id'])) transition(c,row['id'],'graphiti_created','post_write_failpoint',f"test failpoint; episode_name={e.get('name')}; episode_uuid={e.get('uuid')}; run_id={started}",started); c.commit(); raise SystemExit(75) transition(c,row['id'],'verified','verified','episode verified',started); c.execute('update files set last_verified_at=?,last_error=null where id=?',(NOW(),row['id'])); transition(c,row['id'],'done','verified','complete',started); done+=1 except Exception as e: text_error=str(e) if text_error.startswith('identity_review:'): transition(c,row['id'],'review','identity_review',text_error[:1000],started); c.execute('update files set last_error=?,last_error_phase=? where id=?',(text_error[:1000],'identity_review',row['id'])); c.commit(); set_run_block('identity_review',started,row['id']); errors+=1; return 75 if str(e).startswith('maintenance_mode:'): transition(c,row['id'],row['status'],'maintenance_deferred',text_error[:1000],started); c.execute('update files set last_error=?,last_error_phase=? where id=?',(text_error[:1000],'maintenance_mode',row['id'])); c.commit(); errors+=1; set_run_block('maintenance_mode',started,row['id']); return 75 if provider_rate_limit(text_error): state=set_provider_block(text_error,started,row['id']); transition(c,row['id'],'provider_limited','provider_rate_limit',text_error[:1000],started); c.execute('update files set last_attempt=?,last_error=?,last_error_phase=? where id=?',(NOW(),text_error[:1000],'provider_rate_limit',row['id'])); c.commit(); set_run_block('provider_rate_limit',started,row['id']); errors+=1; print(json.dumps({'provider_blocked':True,'state':state},ensure_ascii=False)); return 75 attempts=row['attempts']+1; status='failed'; phase='server_timeout' if text_error.startswith('server_timeout:') else 'request_timeout_before_processing' if text_error.startswith('request_timeout_before_processing:') else 'technical_review' if text_error.startswith('technical_review:') else 'error'; transition(c,row['id'],status,phase,text_error[:1000],started); c.execute('update files set attempts=?,last_attempt=?,last_error=?,last_error_phase=? where id=?',(attempts,NOW(),text_error[:1000],phase,row['id'])); c.commit(); set_run_block('source_failure',started,row['id']); errors+=1; return 1 c.commit() if hold_after_success: set_run_block('pilot_completed',started,rows[0]['id'] if rows else 'none') c.execute('insert into runs(started_at,finished_at,scanned,allowed,excluded,done,errors) values(?,?,?,?,?,?,?)',(started,NOW(),scanned,allowed_n,excluded,done,errors)); c.commit() finally: if c is not None: c.close() lock.release() print(json.dumps({'scanned':scanned,'allowed':allowed_n,'excluded':excluded,'done':done,'errors':errors})); return 0 if errors==0 else 1 def status(): c=conn(); init(c); print('status:'); [print(r['status'],r['n']) for r in c.execute('select status,count(*) n from files group by status')]; print('areas:'); [print(r['source_area'],r['n']) for r in c.execute('select source_area,count(*) n from files group by source_area order by source_area')]; print('last_run:',c.execute('select finished_at from runs order by id desc limit 1').fetchone()); c.close() def audit(): c=conn(); init(c); out={'done_without_episode':0,'episode_without_queue_reference':0,'multiple_episodes':0,'retry_with_episode':0,'bad_hash':0,'ambiguous_episode_reference':0} inventory=[]; inventory_error=None try: with urllib.request.urlopen(CFG['graphiti_url']+'/episodes/inventory',timeout=30) as rr: inventory=json.loads(rr.read()).get('episodes',[]) except Exception as exc: inventory_error=str(exc) def matches(row): out=[] for e in inventory: sd=e.get('source_description') or ''; m=re.search(r'\|\s*(\{.*\})$',sd); meta={} try: meta=json.loads(m.group(1)) if m else {} except Exception: pass if e.get('name')=='obsidian_'+row['import_key'] or meta.get('import_key')==row['import_key'] or (meta.get('obsidian_path')==row['relative_path'] and meta.get('content_hash')==row['content_hash']): out.append(e) return out for row in c.execute('select * from files').fetchall(): if not row['content_hash'] or len(row['content_hash'])!=64: out['bad_hash']+=1 if row['episode_ref'] in ('obsidian','graphiti'): out['ambiguous_episode_reference']+=1 eps=matches(row) if inventory_error is None else [] if row['status']=='done' and len(eps)==0: out['done_without_episode']+=1 if row['status']=='retry' and len(eps)>0: out['retry_with_episode']+=1 if len(eps)>1: out['multiple_episodes']+=1 try: if inventory_error: raise RuntimeError(inventory_error) eps=inventory inv={'obsidian_general_without_queue':0,'youtube_without_source_id':0,'new_schema_without_request_id':0,'obsidian_general_without_import_key':0,'multiple_episodes_per_path_hash':0,'unknown_source_system':0,'episode_without_traceable_import_path':0,'legacy_youtube':0,'legacy_obsidian':0,'unknown_legacy':0} keys={x['import_key'] for x in c.execute('select import_key from files where import_key is not null')} for e in eps: sd=e.get('source_description') or ''; name=e.get('name') or ''; m=re.search(r'\|\s*(\{.*\})$',sd) try: meta=json.loads(m.group(1)) if m else {} except Exception: meta={} source=meta.get('source_system') if source=='obsidian-general': if meta.get('import_key') not in keys: inv['obsidian_general_without_queue']+=1 if not meta.get('request_id'): inv['new_schema_without_request_id']+=1 if not meta.get('import_key'): inv['obsidian_general_without_import_key']+=1 elif source=='youtube-research': if not meta.get('source_id'): inv['youtube_without_source_id']+=1 elif name.startswith('youtube_'): inv['legacy_youtube']+=1 elif name.startswith('obsidian_'): inv['legacy_obsidian']+=1 else: inv['unknown_legacy']+=1 out['episode_inventory']=inv; out['episode_total']=len(eps) except Exception as e: out['episode_inventory_error']=str(e) out['check_success']=len(inventory) if inventory_error is None else 0; out['check_not_found']=sum(1 for row in c.execute('select * from files').fetchall() if inventory_error is None and not matches(row)); out['check_multiple']=sum(1 for row in c.execute('select * from files').fetchall() if inventory_error is None and len(matches(row))>1); out['check_timeout']=1 if inventory_error and 'timed out' in inventory_error.lower() else 0; out['check_unavailable']=1 if inventory_error else 0; out['check_internal_error']=0; out['audit_duration_seconds']=None; out['exit_code']=2 if inventory_error else (1 if out.get('multiple_episodes',0) else 0) print(json.dumps(out)); c.close(); return out['exit_code'] if __name__=='__main__': ap=argparse.ArgumentParser(); ap.add_argument('--once',action='store_true'); ap.add_argument('--status',action='store_true'); ap.add_argument('--audit',action='store_true'); ap.add_argument('--scan',action='store_true'); ap.add_argument('--limit',type=int,default=5); ap.add_argument('--ids'); ap.add_argument('--force-id'); ap.add_argument('--hold-after-success',action='store_true'); a=ap.parse_args(); ids=[int(x) for x in a.ids.split(',')] if a.ids else None; force_ids=[int(x) for x in a.force_id.split(',')] if a.force_id else []; scan_enabled=a.scan or ids is None; sys.exit(audit() if a.audit else status() if a.status else run(a.limit,ids,scan_enabled,force_ids,a.hold_after_success))