#!/usr/bin/env python3
import csv,hashlib,json,os,re,sqlite3,subprocess,urllib.request,shlex
from pathlib import Path
from datetime import datetime,timezone
BASE=Path('/opt/struktur/knowledge-pipeline-aggregator-staging'); VAULT=Path('/opt/obsidian-vault'); IMP=Path('/opt/struktur/obsidian-graphiti-import'); CFG=json.loads((IMP/'config.json').read_text()); DB=Path(CFG['db']); OUT=BASE/'obsidian-graphiti-reconciliation-AA040.csv'; JOUT=BASE/'obsidian-graphiti-reconciliation-AA040.json'; REPORT=BASE/'AA-040.md'
PROBE=datetime.now(timezone.utc).isoformat()
def cypher(q):
# Password is used only inside the container command and is never printed.
inner='p="${NEO4J_AUTH#*/}"; /var/lib/neo4j/bin/cypher-shell -u neo4j -p "$p" --format plain --wrap false '+shlex.quote(q)
p=subprocess.run(['docker','exec','graphiti-neo4j','sh','-lc',inner],text=True,capture_output=True,timeout=180)
if p.returncode: raise RuntimeError(p.stderr.strip() or p.stdout.strip())
return list(csv.reader(p.stdout.splitlines(),skipinitialspace=True))[1:]
def parse_bool(s): return s.strip().lower() in ('true','1','yes')
def safe(v): return '' if v is None else str(v)
def graph_source(e):
s=((e.get('name') or '')+' '+(e.get('source_description') or '')).lower()
if 'youtube' in s: return 'YouTube'
if 'obsidian' in s: return 'Obsidian'
if 'hermes' in s: return 'sonstige'
return 'sonstige' if s.strip() else 'unbekannt'
def first_reason(p,text):
rel=p.relative_to(VAULT); s=rel.as_posix().lower(); parts=rel.parts
if p.suffix.lower() not in [x.lower() for x in CFG['extensions']] or any(x.lower() in s for x in CFG['deny_patterns']): return 'denylist_or_extension'
if not CFG['allow_roots'] or not parts or parts[0] not in CFG['allow_roots']: return 'not-allowlisted'
if len(text.strip())<CFG['min_chars'] or re.search(r'(?im)^\s*(draft|private|no-graphiti|exclude-from-graphiti)\s*:\s*(true|yes)\b',text): return 'content'
return ''
# Source inventory, exact current content hashes.
files=[]
for p in sorted(VAULT.rglob('*.md'),key=lambda x:x.relative_to(VAULT).as_posix()):
rel=p.relative_to(VAULT).as_posix(); raw=p.read_text(encoding='utf-8',errors='ignore'); st=p.stat()
files.append({'relative_path':rel,'filename':p.name,'directory':str(Path(rel).parent) if Path(rel).parent!=Path('.') else '', 'size_bytes':st.st_size,'mtime':datetime.fromtimestamp(st.st_mtime,timezone.utc).isoformat(),'sha256':hashlib.sha256(raw.encode()).hexdigest(),'exclude_reason':first_reason(p,raw)})
# Importer DB read-only.
c=sqlite3.connect('file:'+str(DB)+'?mode=ro',uri=True); c.row_factory=sqlite3.Row
qrows=c.execute('select * from files').fetchall(); bypath={r['relative_path']:dict(r) for r in qrows}
last_hist={}
for r in c.execute('select * from status_history order by id').fetchall(): last_hist[r['queue_id']]=dict(r)
status_counts={r['status']:r['n'] for r in c.execute('select status,count(*) n from files group by status')}
# importer-level run counts and latest run
latest_run=dict(c.execute('select * from runs order by id desc limit 1').fetchone()) if c.execute('select count(*) from runs').fetchone()[0] else None
c.close()
# Neo4j document and chunk inventories.
doc_rows=cypher('MATCH (n:ObsidianDocument) RETURN elementId(n),n.relative_path,n.content_hash,n.import_key,n.title,n.imported_at,n.modified_at,n.file_size ORDER BY n.relative_path, n.imported_at')
docs=[]
for r in doc_rows:
vals=(r+['']*8)[:8]; docs.append({'node_id':vals[0],'relative_path':vals[1],'content_hash':vals[2],'import_key':vals[3],'title':vals[4],'imported_at':vals[5],'modified_at':vals[6],'file_size':vals[7]})
chunk_rows=cypher('MATCH (n:ObsidianChunk) RETURN n.import_key,count(*) ORDER BY n.import_key')
chunks={r[0]:int(r[1]) for r in chunk_rows if r and r[0]}
label_rows=cypher('MATCH (n) UNWIND labels(n) AS label RETURN label,count(*) ORDER BY label')
label_counts={r[0]:int(r[1]) for r in label_rows}
rel_rows=cypher('MATCH ()-[r]->() RETURN type(r),count(*) ORDER BY type(r)')
rel_counts={r[0]:int(r[1]) for r in rel_rows}; rel_total=sum(rel_counts.values())
# provenance-adjacent relation counts explicitly available above.
# Graphiti endpoint inventory.
episodes=json.load(urllib.request.urlopen('http://127.0.0.1:8644/episodes/inventory',timeout=60)).get('episodes',[])
# Normalize episode identifiers into searchable safe strings.
ep_index=[]
for e in episodes:
name=e.get('name') or ''; sd=e.get('source_description') or ''; blob=name+' '+sd
ep_index.append({'uuid':e.get('uuid'),'name':name,'source_description':sd,'created_at':e.get('created_at'),'valid_at':e.get('valid_at'),'source_class':graph_source(e),'blob':blob})
# Create current-file matrix.
rows=[]
for f in files:
rel=f['relative_path']; h=f['sha256']; imp=bypath.get(rel); reason=f['exclude_reason']
relevant=not bool(reason)
# Documents by exact canonical path; same path can have historical versions or duplicates.
dpath=[d for d in docs if d['relative_path']==rel]
dcurrent=[d for d in dpath if d['content_hash']==h]
older=[d for d in dpath if d['content_hash'] and d['content_hash']!=h]
# Same current hash under another path is only a probable historical match, never exact.
dhash=[d for d in docs if d['content_hash']==h]
# Episode matching by explicit import key/path/hash/name, not filename-only.
ik=imp.get('import_key') if imp else None
eps=[]
for e in ep_index:
blob=e['blob']
if (ik and ik in blob) or (rel in blob and h in blob) or (e['name']=='obsidian_'+ik if ik else False): eps.append(e)
# If no importer key, exact path+hash metadata is still a strong match.
if not eps:
eps=[e for e in ep_index if rel in e['blob'] and h in e['blob']]
ep_uuids=sorted(set(e['uuid'] for e in eps if e['uuid']))
ep_names=sorted(set(e['name'] for e in eps if e['name']))
doc_keys=set(d['import_key'] for d in dcurrent if d['import_key'])
chunk_count=sum(chunks.get(k,0) for k in doc_keys)
queue_status=imp.get('status') if imp else ''
attempts=imp.get('attempts','') if imp else ''
last_error=imp.get('last_error') if imp else ''
last_attempt=imp.get('last_attempt') if imp else ''
# Strong classification hierarchy.
if len(dpath)>1 and (dcurrent or older): final='DUPLICATE_OR_INCONSISTENT'; match_type='multiple_documents_same_path'; confidence='high'; match_reason=f'{len(dpath)} ObsidianDocument nodes for canonical path'
elif dcurrent:
final='EXACT_CURRENT'; match_type='canonical_path_and_hash'; confidence='high'; match_reason='ObsidianDocument canonical relative_path and current SHA256 match'
elif older:
final='EXACT_OLDER_VERSION'; match_type='canonical_path_different_hash'; confidence='high'; match_reason='ObsidianDocument canonical relative_path matches, content_hash differs'
elif dhash:
final='PROBABLE_HISTORICAL'; match_type='hash_other_path'; confidence='medium'; match_reason='current hash exists in Graphiti under another relative_path'
else:
final='NOT_IN_GRAPHITI'; match_type='none'; confidence='high'; match_reason='no canonical ObsidianDocument, episode, or hash match'
if final=='NOT_IN_GRAPHITI':
if reason: missing_reason=reason
elif not imp: missing_reason='never_queued_no_importer_row'
elif queue_status in ('retry','failed','processing','provider_limited'): missing_reason=f'queue_status_{queue_status}'
elif queue_status in ('duplicate',): missing_reason='duplicate_without_unique_graphiti_target'
elif queue_status=='done': missing_reason='importer_done_but_no_graphiti_document_match'
else: missing_reason=f'queue_status_{queue_status or "unknown"}'
else: missing_reason=''
graph_hash_match='YES' if dcurrent else ('NO' if dpath else 'NOT_APPLICABLE')
rows.append({**f,'importer_relevant':'JA' if relevant else 'NEIN','queue_id':imp.get('id','') if imp else '','queue_status':queue_status,'attempts':attempts,'last_error':last_error or missing_reason,'last_attempt':last_attempt,'graphiti_status':'present' if (dpath or eps) else 'absent','graphiti_match_type':match_type,'episode_count':len(ep_uuids),'episode_uuid':'|'.join(ep_uuids),'episode_name':'|'.join(ep_names),'obsidian_document_count':len(dpath),'obsidian_document_id':'|'.join(d['node_id'] for d in dpath),'chunk_count':chunk_count,'graphiti_version_match':'YES' if dcurrent else ('OLDER' if older else 'NO'),'graphiti_current_hash_match':graph_hash_match,'match_confidence':confidence,'match_reason':match_reason,'final_classification':final,'missing_reason':missing_reason,'document_import_keys':'|'.join(sorted(doc_keys))})
fields=['relative_path','filename','directory','size_bytes','mtime','sha256','importer_relevant','exclude_reason','queue_id','queue_status','attempts','last_error','last_attempt','graphiti_status','graphiti_match_type','episode_count','episode_uuid','episode_name','obsidian_document_count','obsidian_document_id','chunk_count','graphiti_version_match','graphiti_current_hash_match','match_confidence','match_reason','missing_reason','final_classification','document_import_keys']
with OUT.open('w',encoding='utf-8',newline='') as fh:
w=csv.DictWriter(fh,fieldnames=fields,delimiter='\t'); w.writeheader(); w.writerows(rows)
# Separate full graphiti Obsidian episode list and all Neo4j documents for JSON audit.
current=set(f['relative_path'] for f in files)
obs_eps=[e for e in ep_index if e['source_class']=='Obsidian']
obs_eps_no_current=[e for e in obs_eps if not any(rel in e['blob'] for rel in current)]
# conservative: document nodes without current path
obs_docs_no_current=[d for d in docs if d['relative_path'] not in current]
# duplicated canonical paths in Neo4j
from collections import Counter
path_counts=Counter(d['relative_path'] for d in docs); duplicate_docs=[{'relative_path':p,'count':n} for p,n in sorted(path_counts.items()) if n>1]
cls=Counter(r['final_classification'] for r in rows)
# queue status only among current-file rows
current_queue=Counter(r['queue_status'] for r in rows if r['queue_status'])
for _s in ('done','retry','failed','processing','duplicate'): current_queue.setdefault(_s,0)
duplicate_episode_names=[{'name':k,'count':n} for k,n in sorted(Counter(e['name'] for e in obs_eps).items()) if n>1]
summary={'probe_time_utc':PROBE,'sources':{'vault':str(VAULT),'importer_db':str(DB),'graphiti_inventory':'http://127.0.0.1:8644/episodes/inventory','neo4j_container':'graphiti-neo4j','neo4j_labels':'MATCH (n) UNWIND labels(n)','neo4j_relationships':'MATCH ()-[r]->()'},'obsidian':{'markdown_total':len(rows),'classes':dict(cls),'importer_relevant':sum(r['importer_relevant']=='JA' for r in rows),'importer_excluded':sum(r['importer_relevant']=='NEIN' for r in rows),'without_graphiti':[r for r in rows if r['final_classification']=='NOT_IN_GRAPHITI'],'older':[r for r in rows if r['final_classification']=='EXACT_OLDER_VERSION'],'probable_historical':[r for r in rows if r['final_classification']=='PROBABLE_HISTORICAL'],'duplicates':[r for r in rows if r['final_classification']=='DUPLICATE_OR_INCONSISTENT']},'importer':{'all_rows':len(qrows),'status_counts':status_counts,'current_markdown_queue_status_counts':dict(current_queue),'latest_run':latest_run},'graphiti':{'episodes_total':len(episodes),'episode_source_counts':dict(__import__('collections').Counter(e['source_class'] for e in ep_index)),'obsidian_episodes':obs_eps,'obsidian_episodes_without_current_vault_file':obs_eps_no_current,'neo4j_label_counts':label_counts,'neo4j_relationship_counts':rel_counts,'neo4j_relationship_total':rel_total,'obsidian_documents_total':len(docs),'obsidian_documents_without_current_vault_file':obs_docs_no_current,'duplicate_obsi_document_paths':duplicate_docs,'duplicate_obsidian_episode_names':duplicate_episode_names,'obsidian_chunks_total':label_counts.get('ObsidianChunk',0),'chunks_by_import_key':chunks},'integrity':{'sum_classes':sum(cls.values()),'markdown_total':len(rows),'sum_classes_ok':sum(cls.values())==len(rows),'matrix_rows':len(rows)}}
JOUT.write_text(json.dumps(summary,ensure_ascii=False,indent=2,default=str)+'\n')
# Generate full report with embedded complete lists and compact causes; exact matrix is TSV/JSON.
def mdrow(r): return '| '+ ' | '.join(str(r.get(k,'' )).replace('|','\\|').replace('\n',' ') for k in ['relative_path','queue_status','attempts','missing_reason','final_classification'])+' |'
report=[]
report += [f'# AA-040 – Vollständiger Obsidian↔Graphiti/Neo4j-Bestandsabgleich', '', f'Prüfzeitpunkt: `{PROBE}` (UTC).', '', 'AUSSCHLIESSLICH READ-ONLY. Kein Import, keine Queue-, Episode-, Neo4j-, Code-, Konfigurations- oder Run-Block-Änderung.', '', '## Kompakte Klassifikation', '', '| Kategorie | Anzahl |','|---|---:|',f'| Obsidian gesamt | {len(rows)} |',f'| exakt aktuell in Graphiti | {cls.get("EXACT_CURRENT",0)} |',f'| ältere Version in Graphiti | {cls.get("EXACT_OLDER_VERSION",0)} |',f'| wahrscheinlich historisch | {cls.get("PROBABLE_HISTORICAL",0)} |',f'| nicht in Graphiti | {cls.get("NOT_IN_GRAPHITI",0)} |',f'| Dublette/inkonsistent | {cls.get("DUPLICATE_OR_INCONSISTENT",0)} |', '', '## Autoritative Quellen', '', f'- Vault: `{VAULT}`; direkte Dateisysteminventur aller `*.md`.', f'- Importer: `{IMP/"importer.py"}` und `{IMP/"config.json"}`; SQLite `{DB}` im `mode=ro`.', '- Graphiti-Episoden: `http://127.0.0.1:8644/episodes/inventory` (GET).', '- Neo4j: Container `graphiti-neo4j`, ausschließlich Cypher-`MATCH`/`RETURN`-Abfragen.', f'- Matrix: `{OUT}` (TSV-Inhalt trotz `.csv`-Dateiname, delimiter=TAB).', '', '## OBSIDIAN / IMPORTER', '', f'- Markdown-Dateien gesamt: **{len(rows)}**', f'- importer-relevant: **{sum(r["importer_relevant"]=="JA" for r in rows)}**', f'- importer-excluded: **{sum(r["importer_relevant"]=="NEIN" for r in rows)}**', f'- aktuelle Dateien mit Queue-Eintrag: **{sum(bool(r["queue_id"]) for r in rows)}**', f'- aktuelle Dateien ohne Queue-Eintrag: **{sum(not bool(r["queue_id"]) for r in rows)}**', f'- Queue-Status in den aktuellen Markdown-Zeilen: `{dict(current_queue)}`', f'- gesamte Importer-Statusverteilung aller `{len(qrows)}` DB-Zeilen: `{status_counts}`', '', 'Die konkrete Ausschlussursache ist in Matrixspalten `exclude_reason` und `missing_reason` enthalten. `excluded` ist nicht mit `failed`, `retry`, `duplicate` oder fehlenden Graphiti-Nachweisen gleichzusetzen.', '', '## GRAPHITI / NEO4J', '', f'- Episoden gesamt: **{len(episodes)}**', f'- Episoden nach Quelle: `{dict(__import__("collections").Counter(e["source_class"] for e in ep_index))}`', f'- ObsidianDocument gesamt: **{len(docs)}**', f'- ObsidianChunk gesamt: **{label_counts.get("ObsidianChunk",0)}**', f'- Entity: **{label_counts.get("Entity",0)}**', f'- Relationships gesamt: **{rel_total}**', f'- Relationship-Typen: `{rel_counts}`', f'- DocumentVersion: **{label_counts.get("DocumentVersion",0)}**; DocumentMetadata: **{label_counts.get("DocumentMetadata",0)}**; ProvenanceRecord: **{label_counts.get("ProvenanceRecord",0)}**; KnowledgeRule: **{label_counts.get("KnowledgeRule",0)}**; KnowledgeEvent: **{label_counts.get("KnowledgeEvent",0)}**', '', 'Die Episodeninventur enthält nur die vom produktiven GET-Endpunkt gelieferten Felder `uuid`, `name`, `source_description`, `created_at`; `valid_at` wurde dort nicht geliefert. Neo4j-Dokumentknoten lieferten `elementId`, `relative_path`, `content_hash`, `import_key`, `title`, `imported_at`, `modified_at`, `file_size`.', '', '## MATCH-REGELN', '', '1. `EXACT_CURRENT`: genau ein canonical `ObsidianDocument.relative_path` und aktueller SHA-256 identisch.', '2. `EXACT_OLDER_VERSION`: canonical Pfad vorhanden, aber gespeicherter Hash abweichend.', '3. `PROBABLE_HISTORICAL`: aktueller Hash in Graphiti unter anderem Pfad; bewusst nicht als exakt gewertet.', '4. `DUPLICATE_OR_INCONSISTENT`: mehr als ein `ObsidianDocument`-Knoten beansprucht denselben canonical Pfad.', '5. `NOT_IN_GRAPHITI`: kein canonical Dokument, kein expliziter Episode-/Import-Key-/Pfad+Hash-Match und kein Hash-Match unter anderem Pfad.', '', 'Episoden-Namen allein wurden nicht als ausreichender Nachweis verwendet. Episode-Matches wurden nur über Import-Key, canonical Pfad+Hash oder den expliziten Namen `obsidian_<import_key>` berücksichtigt.', '', '## A. ALLE AKTUELLEN OBSIDIAN-DATEIEN NICHT IN GRAPHITI', '', '| relative_path | queue_status | attempts | konkreter Grund | Klassifikation |','|---|---|---:|---|---|']
for r in rows:
if r['final_classification']=='NOT_IN_GRAPHITI': report.append(mdrow(r))
report += ['', '## B. ALLE AKTUELLEN DATEIEN MIT ÄLTERER GRAPHITI-VERSION', '', '| relative_path | queue_status | attempts | konkreter Grund | Klassifikation |','|---|---|---:|---|---|']
for r in rows:
if r['final_classification']=='EXACT_OLDER_VERSION': report.append(mdrow(r))
report += ['', '## C. ALLE WAHRSCHEINLICH HISTORISCHEN MATCHES', '', '| relative_path | queue_status | attempts | konkreter Grund | Klassifikation |','|---|---|---:|---|---|']
for r in rows:
if r['final_classification']=='PROBABLE_HISTORICAL': report.append(mdrow(r))
report += ['', '## D. ALLE DUBликATTEN/INKONSISTENZEN', '', '| relative_path | queue_status | attempts | konkreter Grund | Klassifikation |','|---|---|---:|---|---|']
for r in rows:
if r['final_classification']=='DUPLICATE_OR_INCONSISTENT': report.append(mdrow(r))
report += ['', '## E. ALLE GRAPHITI-OBSIDIAN-EPISODEN OHNE AKTUELLE VAULT-DATEI', '', '| uuid | name | source_description | created_at |','|---|---|---|---|']
for e in obs_eps_no_current: report.append('| '+' | '.join(str(e.get(k,'' )).replace('|','\\|').replace('\\n',' ') for k in ['uuid','name','source_description','created_at'])+' |')
report += ['', '## F. ALLE ObsidianDocument-KNOTEN OHNE AKTUELLE VAULT-DATEI', '', '| node_id | relative_path | content_hash | import_key | title | imported_at |','|---|---|---|---|---|---|']
for d in obs_docs_no_current: report.append('| '+' | '.join(str(d.get(k,'' )).replace('|','\\|').replace('\\n',' ') for k in ['node_id','relative_path','content_hash','import_key','title','imported_at'])+' |')
report += ['', '## G. ALLE DUPLIZIERTEN OBSIDIAN-EPISODENNAMEN', '', '| episode_name | Anzahl |','|---|---:|']
for d in duplicate_episode_names: report.append('| '+d['name'].replace('|','/')+' | '+str(d['count'])+' |')
report += ['', '## H. ALLE DUPLIZIERTEN CANONICAL OBSIDIANDOCUMENT-PFADE IN NEO4J', '', '| relative_path | ObsidianDocument-Knoten |','|---|---:|']
for d in duplicate_docs: report.append('| '+d['relative_path'].replace('|','/')+' | '+str(d['count'])+' |')
report += ['', '## Summenprüfung', '', f'SUMME DER FÜNF OBSIDIAN-KLASSEN = {sum(cls.values())}', f'ANZAHL ALLER AKTUELLEN MARKDOWN-DATEIEN IM VAULT = {len(rows)}', '', ('SUMME ERFÜLLT: JA' if sum(cls.values())==len(rows) else 'BERICHT NICHT ABGESCHLOSSEN: NEIN – Summenabweichung.'), '', '## Artefakte und Grenzen', '', f'- Die vollständige Dateimatrix steht in `{OUT}`; vollständige Graph-/Listen-Daten stehen in `{JOUT}`.', '- Keine automatische Bereinigung oder Statuskorrektur.', '- Eine Graphiti-Episode ohne explizite Provenance-Metadaten wird nicht allein über Dateiname als aktuelles Obsidian-Dokument gewertet.', '- Die Zählungen sind Ebenen-spezifisch: Dateien, Importer-Zeilen, Episoden, ObsidianDocument-Knoten, Chunks, Entities und Relationships sind nicht identisch.', '', 'Arbeitsauftrag AA-040 erledigt.']
REPORT.write_text('\n'.join(report)+'\n')
print(json.dumps({'probe_time_utc':PROBE,'matrix_rows':len(rows),'classes':dict(cls),'queue_status_current':dict(current_queue),'status_all':status_counts,'episodes':len(episodes),'obsidian_documents':len(docs),'chunks':label_counts.get('ObsidianChunk',0),'entities':label_counts.get('Entity',0),'relationships':rel_total,'obsidian_episodes_without_current':len(obs_eps_no_current),'obsidian_docs_without_current':len(obs_docs_no_current),'sum_ok':sum(cls.values())==len(rows),'out':str(OUT),'json':str(JOUT),'report':str(REPORT)},ensure_ascii=False))