"""Controlled Phase-1 retrieval repair.
Read-only against knowledge DB, Obsidian and Graphiti.
No entity merges, source writes, Graphiti writes or index changes.
"""
from __future__ import annotations
import json, os, re, sqlite3, urllib.request
from pathlib import Path

DB=os.environ.get('KNOWLEDGE_DB','/opt/struktur/youtube-research/knowledge.db')
VAULT=Path('/opt/obsidian-vault')
GRAPH=os.environ.get('GRAPHITI_URL','http://127.0.0.1:8644')
FLAGS={
 'entity_resolver': os.environ.get('RETRIEVAL_ENTITY_RESOLVER','1')!='0',
 'intent_gating': os.environ.get('RETRIEVAL_INTENT_GATING','1')!='0',
 'graphiti_provenance_gate': os.environ.get('RETRIEVAL_GRAPHITI_PROVENANCE_GATE','1')!='0',
 'entity_first_fts': os.environ.get('RETRIEVAL_ENTITY_FIRST_FTS','1')!='0',
 'cross_layer_dedupe': os.environ.get('RETRIEVAL_CROSS_LAYER_DEDUPE','1')!='0',
 'evidence_dedupe': os.environ.get('EVIDENCE_DEDUPE','1')!='0',
 'assertion_scope': os.environ.get('ASSERTION_SCOPE','1')!='0',
 'entity_youtube_gate': os.environ.get('ENTITY_YOUTUBE_GATE','1')!='0',
 'retrieval_metadata_v2': os.environ.get('RETRIEVAL_METADATA_V2','1')!='0',
}
ALIASES=[
 'lüftungsprofi mallorca','lüftungsprofi','lueftungsprofi mallorca','lueftungsprofi',
 'mallorca airservices','mallorca air services'
]
CANONICAL='company:lueftungsprofi-mallorca'
CANONICAL_NAME='Lüftungsprofi Mallorca / Mallorca AirServices'
PROJECT_RE=re.compile(r'\b(?:eisen|heinz)\b',re.I)
STOP={'der','die','das','ein','eine','einer','einem','einen','und','oder','für','von','mit','auf','aus','wie','was','ist','sind','wir','uns','the','and','or','for','with','what','how','which','our','have','into','about','their','does','that','this','über','den','dem','des'}
EXPANSION={'autonome':['agent','autonomous','agenten','long-running','reliability','monitoring','recovery','context','token'],'autonomen':['agent','autonomous','agenten','long-running','reliability','monitoring','recovery','context','token'],'agenten':['agent','autonomous','subagent','skills','workflow','monitoring'],'ki-agenten':['agent','autonomous','subagent','skills','workflow','monitoring'],'langfristig':['long-running','dauerbetrieb','monitoring','recovery','reliability'],'kosten':['cost','token','budget','rate','modellwahl'],'nachvollziehbarkeit':['provenienz','source','evidence','audit','traceability'],'wissen':['wissensbasis','knowledge','rag','obsidian','retrieval'],'ausfällen':['failure','recovery','monitoring','retry','reliability'],'fehlverhalten':['guardrail','validation','error','recovery']}
DOC_WEIGHT={'AUFTRAGSBESTAETIGUNG':1.35,'RECHNUNG':1.25,'BESTELLUNG':1.15,'PROJEKTDATEN':1.05,'TECHNISCHE_UNTERLAGE':1.0,'ANGEBOT':.9,'INBETRIEBNAHME':1.1,'EMAIL':.75,'SONSTIG':.6}

def ro(): return sqlite3.connect(f'file:{DB}?mode=ro',uri=True,timeout=10)
def tokens(q): return [x for x in re.findall(r'[\wÄÖÜäöüß-]{3,}',q.lower()) if x not in STOP][:20]
def expanded(q):
 b=tokens(q); e=[]
 for t in b:e.extend(EXPANSION.get(t,[]))
 return list(dict.fromkeys(b+e))[:32]
def phrase_expr(values): return ' OR '.join('"'+x.replace('"','""')+'"' for x in values) or '"wissen"'

def resolve_entity(q):
 low=q.casefold(); hits=[]
 for a in sorted(ALIASES,key=len,reverse=True):
  if a in low:hits.append(a)
 project=bool(PROJECT_RE.search(q))
 if hits:
  matched=hits[0]
  ambiguous=(matched in {'lüftungsprofi','lueftungsprofi'} and len(low.split())<=4 and not any(x in low for x in ('mallorca','unternehmen','firma','geschäft','website','firma')))
  return {'canonical_entity_id':CANONICAL,'canonical_name':CANONICAL_NAME,'matched_alias':matched,'match_type':'exact_alias' if matched in ALIASES else 'token_alias','confidence':'medium' if ambiguous else 'high','ambiguity':ambiguous,'project_hint':project}
 return {'canonical_entity_id':None,'canonical_name':None,'matched_alias':None,'match_type':'none','confidence':'none','ambiguity':False,'project_hint':project}

def classify(q,ent):
 low=q.casefold()
 ent['project_evidence_requested']=any(x in low for x in ('projektunterlagen','projektpraxis','projektbeleg','durch konkrete projekt','belegt'))
 if any(x in low for x in ('typischerweise','typisch pro','durchschnitt','durchschnittlich','meistverkauft','häufigste','übliche projektgröße')):
  return 'statistical'
 if ent.get('project_hint') or any(x in low for x in ('im projekt','im auftrag','angebot','auftragsbestätigung','auftragsbestaetigung','bestellung','rechnung','verkauft','stück')):
  return 'project'
 if ent.get('canonical_entity_id') and (any(x in low for x in ('was weiß','was wisst','firma','unternehmen','geschäft','website','leistungen','partner','lüftungsprofi','mallorca air'))): return 'company'
 if any(x in low for x in ('smartfan','lüftungsanlage','lüftungsgerät','luftqualität','feuchtigkeit')): return 'product'
 if any(x in low for x in ('agent','ki','autonom','rag','second brain','wissens')): return 'research'
 return 'unclear'

def _entity_text(x):
 s=' '.join(str(x.get(k) or '') for k in ('title','summary','content','statement','fact','source_identity','original_path','path')).casefold()
 return any(a in s for a in ALIASES) or 'mallorca-airservices' in s

def fts(q,ent,intent):
 c=ro(); c.row_factory=sqlite3.Row; out=[]
 try:
  # Entity/phrase-first: exact/alias candidates are restricted to entity-bearing records.
  vals=ALIASES if FLAGS['entity_first_fts'] and intent=='company' else expanded(q)
  expr=phrase_expr(vals)
  rows=c.execute('''SELECT v.id,v.youtube_id,v.title,v.channel,v.summary,v.published_at,(-bm25(videos_fts)) score FROM videos_fts JOIN videos v ON videos_fts.rowid=v.id WHERE videos_fts MATCH ? ORDER BY score DESC LIMIT 30''',(expr,)).fetchall()
  for r in rows:
   d=dict(r); d['rank']=len(out)+1; d['source_kind']='SOURCE'; d['entity_match']=_entity_text(d)
   if FLAGS['entity_youtube_gate'] and intent in ('company','project','statistical') and not d['entity_match']: continue
   out.append(d)
  return out
 except sqlite3.Error:return []
 finally:c.close()

def kfts(q,ent,intent):
 c=ro(); c.row_factory=sqlite3.Row; out=[]
 try:
  vals=expanded(q)
  if intent=='company': vals=ALIASES
  expr=phrase_expr(vals)
  rows=c.execute('''SELECT rowid,doc_id,source_identity,project,document_type,original_path,content,(-bm25(kdata_fts)) score FROM kdata_fts WHERE kdata_fts MATCH ? ORDER BY score DESC LIMIT 30''',(expr,)).fetchall()
  for r in rows:
   d=dict(r); d['entity_match']=_entity_text(d)
   project_source='eisen, heinz' in str(d.get('source_identity','')).casefold()
   if intent=='company' and (d.get('project') or project_source or '/eisen, heinz/' in str(d.get('original_path','')).casefold()) and not ent.get('project_evidence_requested'):
    continue
   if intent=='research' and (project_source or '/eisen, heinz/' in str(d.get('original_path','')).casefold()):
    continue
   if intent=='company' and not d['entity_match']: continue
   if intent!='project' and d.get('project') and not d['entity_match']: continue
   d['rank']=len(out)+1; d['source_kind']='SOURCE'; out.append(d)
  return out
 except sqlite3.Error:return []
 finally:c.close()

def kus(q,ent,intent):
 c=ro(); c.row_factory=sqlite3.Row; out=[]
 try:
  vals=expanded(q); clauses=' OR '.join('statement LIKE ?' for _ in vals) or '1=1'
  rows=c.execute(f'''SELECT id,statement,source_identity,source_version_id,source_locator,evidence_class,review_state,gold_state,legacy_placeholder,content_version_hash FROM knowledge_units WHERE legacy_placeholder=0 AND review_state NOT IN ('REJECTED','LEGACY') AND ({clauses})''',[f'%{x}%' for x in vals]).fetchall()
  for r in rows:
   d=dict(r); d['source_kind']='SOURCE' if str(d.get('source_identity') or '').startswith(('k-daten:','youtube:')) else 'DERIVED'
   d['quality_weight']={'GESICHERT':1.3,'BESTÄTIGT':1.2,'PLAUSIBEL':1.0,'UNSICHER':.65,'WIDERSPRÜCHLICH':.7}.get(d.get('evidence_class'),.7)
   d['entity_match']=_entity_text(d)
   project_source='eisen, heinz' in str(d.get('source_identity','')).casefold()
   if intent in ('company','research') and project_source and not (intent=='company' and ent.get('project_evidence_requested')): continue
   if intent in ('company','product') and not d['entity_match']: continue
   d['relevance']=sum((d.get('statement') or '').casefold().count(t) for t in vals)
   out.append(d)
  out.sort(key=lambda x:(x['relevance']*x['quality_weight'],x['id']),reverse=True)
  for i,d in enumerate(out[:30],1):d['rank']=i
  return out[:30]
 except sqlite3.Error:return []
 finally:c.close()

def graph(q,ent,intent):
 req=urllib.request.Request(GRAPH+'/search',json.dumps({'query':q,'limit':20}).encode(),{'Content-Type':'application/json'},method='POST')
 try:
  with urllib.request.urlopen(req,timeout=30) as r:d=json.loads(r.read())
  out=[]
  for i,x in enumerate(d.get('results',[]),1):
   # Never synthesize identity from fact text. Retain raw diagnostic only.
   item={'fact':x.get('fact',''),'score':x.get('score'),'source':'graphiti:/search','rank':i,'source_kind':'UNKNOWN','provenance_status':'unattributed_graph_fact','entity_match':False}
   item['attribution_present']=bool(x.get('episode_uuid') or x.get('episode') or x.get('source_identity') or x.get('source_description') or x.get('timestamp') or x.get('original_path'))
   out.append(item)
  return out
 except Exception:return []

def obs(q,ent,intent):
 out=[]; search_roots=[VAULT/'LP-CONTEXT',VAULT/'03-Projekte',VAULT/'Projects',VAULT/'Email-Agent',VAULT/'AG-CONTEXT']
 for root in search_roots:
  if not root.exists():continue
  for p in root.rglob('*.md'):
   try:t=p.read_text(encoding='utf8',errors='replace')
   except OSError:continue
   low=t.casefold()
   if not any(a in low for a in ALIASES) and not any(x in low for x in ('lueftungsprofi','mallorca-airservices')):continue
   if intent=='research':continue
   out.append({'source_identity':'obsidian:'+str(p.relative_to(VAULT)),'path':str(p),'text':t[:5000],'relevance':sum(low.count(x) for x in (ALIASES if intent=='company' else tokens(q))),'source_kind':'SOURCE','entity_match':True,'rank':0})
 out.sort(key=lambda x:(x['relevance'],x['source_identity']),reverse=True)
 for i,x in enumerate(out[:20],1):x['rank']=i
 return out[:20]

def identity(x):
 if x.get('source_identity'):return x['source_identity']
 if x.get('youtube_id'):return 'youtube:'+x['youtube_id']
 if x.get('fact'):return None
 return 'unknown:'+str(x.get('rowid') or x.get('rank'))
def evidence_id(x):
 sid=x.get('source_identity')
 if sid and sid.startswith('k-daten:'): return sid
 if sid and sid.startswith('obsidian:'): return sid
 if sid and sid.startswith('youtube:'): return sid
 return None

def evidence_family(x):
 sid=str(x.get('source_identity') or '')
 path=(sid+' '+str(x.get('original_path') or '')).casefold()
 # Explicit document-family rule for the AG2502 confirmation variants.
 if 'eisen, heinz' in path and 'auftragsbestaetigung_projekt_eisen_ag2502' in path:
  return 'family:projekt-eisen:auftragsbestaetigung-ag2502'
 if x.get('evidence_id'): return 'family:'+str(x['evidence_id'])
 return None

def assertion_scope(x,intent):
 sid=str(x.get('source_identity') or '').casefold(); text=str(x.get('statement') or x.get('content') or '').casefold()
 project=('eisen, heinz' in sid or '/eisen, heinz/' in sid or 'eisen, heinz' in text)
 historical=project and (x.get('document_type') in ('ANGEBOT','AUFTRAGSBESTAETIGUNG','BESTELLUNG','RECHNUNG','PROJEKTDATEN'))
 if project: return ['PROJECT_ONLY']+(['HISTORICAL'] if historical else [])
 if x.get('layer')=='obsidian' or x.get('source_kind')=='DERIVED': return ['COMPANY_GENERAL' if intent=='company' else 'CURRENT_UNVERIFIED']
 if x.get('layer')=='fts5': return ['CURRENT_UNVERIFIED']
 return ['CURRENT_UNVERIFIED']

def family_version_label(x):
 sid=str(x.get('source_identity') or '')
 low=sid.casefold()
 if 'auftragsbestaetigung_projekt_eisen_ag2502' in low:
  return 'AG2502-Auftragsbestätigung'
 return sid
def source_meta(identity_value):
 try:
  c=ro(); r=c.execute('select document_type from kdata_documents where source_identity=?',(identity_value,)).fetchone(); c.close(); return {'document_type':r[0] if r else None}
 except Exception:return {}

def env(): return os.environ.get('OPENROUTER_API_KEY','')

def synth(q,ctx,n):
 if not n:return 'Die vorhandenen Wissensquellen ermöglichen keine belastbare Antwort.','insufficient_evidence'
 key=env()
 if not key:return 'Relevante Quellen wurden gefunden, aber die LLM-Synthese ist nicht verfügbar.','llm_not_configured'
 body=json.dumps({'model':'qwen/qwen3.7-flash','temperature':0.1,'max_tokens':5000,'messages':[{'role':'system','content':'Beantworte auf Deutsch streng quellengebunden. Nutze nur den erkannten Intent und die attribuierten Quellen. Trenne COMPANY_GENERAL/COMPANY_SUPPORTED von PROJECT_ONLY und HISTORICAL. Projektquellen sind starke Praxisbelege, beweisen aber allein kein allgemeines Portfolio. Stelle einzelne Projektpreise nie als allgemeine Standardpreise dar. Bei wenigen Projekten keine typischen, durchschnittlichen oder meistverkauften Werte ableiten; sage ausdrücklich, wenn die Datenbasis zu klein ist. Wenn Dokumente abweichen und kein Grund dokumentiert ist, sage nur: Die Dokumente zeigen unterschiedliche Darstellungen; der genaue Grund ergibt sich aus den vorliegenden Quellen nicht. Behaupte keine rechtliche Entity-Gleichheit ohne Beleg. Erfinde nichts.'},{'role':'user','content':f'Frage: {q}\\n\\nMaterial:\\n{ctx}'}]},ensure_ascii=False).encode()
 req=urllib.request.Request('https://openrouter.ai/api/v1/chat/completions',body,{'Content-Type':'application/json','Authorization':'Bearer '+key,'HTTP-Referer':'http://127.0.0.1:8655','X-Title':'Hermes Knowledge Query'},method='POST')
 try:
  with urllib.request.urlopen(req,timeout=90) as r:d=json.loads(r.read())
  content=d['choices'][0]['message'].get('content')
  if isinstance(content,list):content=''.join(x.get('text','') if isinstance(x,dict) else str(x) for x in content)
  return str(content or '').strip(),None
 except Exception as e:return 'Quellen wurden gefunden, aber die LLM-Synthese ist fehlgeschlagen; es wird keine unbelegte Antwort ausgegeben.',type(e).__name__

def ask(question):
 q=(question or '').strip()
 if not q: raise ValueError('question darf nicht leer sein')
 ent=resolve_entity(q) if FLAGS['entity_resolver'] else {'canonical_entity_id':None,'canonical_name':None,'matched_alias':None,'match_type':'disabled','confidence':'none','ambiguity':False,'project_hint':False}
 intent=classify(q,ent) if FLAGS['intent_gating'] else 'unclear'
 raw={'fts5':fts(q,ent,intent),'kdata_fts5':kfts(q,ent,intent),'knowledge_units':kus(q,ent,intent),'graphiti':graph(q,ent,intent),'obsidian':obs(q,ent,intent)}
 allrows=[]; rejected={'project_without_project_intent':0,'unattributed_graph_fact':0,'non_entity_source':0}
 for layer,rows in raw.items():
  seen=set()
  for x in rows:
   x=dict(x); ident=identity(x)
   if ident in seen:continue
   seen.add(ident)
   if layer=='graphiti' and FLAGS['graphiti_provenance_gate'] and not x.get('attribution_present'):
    rejected['unattributed_graph_fact']+=1; continue
   if intent=='company' and layer in ('kdata_fts5','knowledge_units') and ('eisen, heinz' in str(x.get('source_identity','')).casefold() or x.get('project')=='Eisen, Heinz') and not ent.get('project_evidence_requested'):
    rejected['project_without_project_intent']+=1; continue
   if intent=='company' and layer=='fts5' and not x.get('entity_match'):
    rejected['non_entity_source']+=1; continue
   x['layer']=layer; x['source_identity']=ident; x['entity_resolution']=ent; x['intent']=intent
   x['evidence_id']=evidence_id(x); x['document_type']=x.get('document_type') or source_meta(ident).get('document_type')
   x['evidence_family']=evidence_family(x)
   x['assertion_scope']=assertion_scope(x,intent)
   x['version_label']=family_version_label(x)
   x['rrf_score']=1/(60+float(x.get('rank',999)))
   x['quality_multiplier']=.2 if x.get('source_kind')=='SYNTHETIC_TEST' else DOC_WEIGHT.get(x.get('document_type'),1.0)
   x['fusion_score']=x['rrf_score']*x['quality_multiplier']; allrows.append(x)
 if FLAGS['cross_layer_dedupe'] or FLAGS['evidence_dedupe']:
  best={}
  family_members={}
  for x in allrows:
   k=(x.get('evidence_family') if FLAGS['evidence_dedupe'] else None) or x.get('evidence_id') or ('graph:'+str(x.get('rank'))+':'+str(x.get('layer')))
   family_members.setdefault(k,[]).append(x)
   if k not in best or x['fusion_score']>best[k]['fusion_score']:best[k]=x
  context=sorted(best.values(),key=lambda x:(x['fusion_score'],x.get('quality_weight',1),-x.get('rank',999)),reverse=True)[:20]
  for x in context:
   members=family_members.get((x.get('evidence_family') if FLAGS['evidence_dedupe'] else None) or x.get('evidence_id') or ('graph:'+str(x.get('rank'))+':'+str(x.get('layer'))),[])
   x['family_version_count']=len(set(str(m.get('source_identity')) for m in members if m.get('source_identity')))
   x['family_members']=[m.get('source_identity') for m in members if m.get('source_identity')]
   x['representation_count']=len(members)
 else: context=sorted(allrows,key=lambda x:x['fusion_score'],reverse=True)[:20]
 sources=[]
 for x in context:
  sources.append({k:x.get(k) for k in ('layer','source_identity','source_kind','document_type','rank','score','rrf_score','fusion_score','evidence_id','evidence_family','family_version_count','family_members','representation_count','assertion_scope','version_label','title','original_path','path','source_locator','statement','fact','provenance_status')})
 blocks=[]
 for x in context:
  if x['layer']=='fts5': payload=f"{x.get('title')} | {x.get('summary') or ''}"
  elif x['layer']=='kdata_fts5': payload=(x.get('content') or '')[:3000]
  elif x['layer']=='knowledge_units': payload=x.get('statement') or ''
  elif x['layer']=='obsidian': payload=x.get('text') or ''
  else: payload=x.get('fact') or ''
  blocks.append(f"[{x['layer']}|{x.get('source_identity')}|family={x.get('evidence_family')}|kind={x.get('source_kind')}|doc={x.get('document_type')}|scope={','.join(x.get('assertion_scope') or [])}|rank={x.get('rank')}|evidence_id={x.get('evidence_id')}]\n{payload}")
 answer,warning=synth(q,'\n\n'.join(blocks)[:24000],len(context))
 source_count=len(set(x.get('source_identity') for x in context if x.get('source_identity')))
 family_count=len(set(x.get('evidence_family') or x.get('evidence_id') or x.get('source_identity') for x in context))
 representation_count=sum(x.get('representation_count',1) for x in context)
 layer_counts={k:sum(1 for x in context if x.get('layer')==k) for k in ('fts5','kdata_fts5','knowledge_units','graphiti','obsidian')}
 metadata={'source_count':source_count,'evidence_family_count':family_count,'representation_count':representation_count,'duplicate_representations':max(0,representation_count-source_count),'final_context_count':len(context),**layer_counts,'fts5_count':layer_counts['fts5'],'kdata_fts5_count':layer_counts['kdata_fts5'],'knowledge_unit_count':layer_counts['knowledge_units'],'graphiti_count':layer_counts['graphiti'],'obsidian_count':layer_counts['obsidian']}
 return {'answer':answer,'confidence':'insufficient' if warning else ('supported' if source_count>=3 else 'medium'),'warning':warning,'entity_resolution':ent,'intent':intent,'rejected_candidates':rejected,'evidence':raw,'sources':sources,'retrieval':{'question':q,'tokens':tokens(q),'expanded_tokens':expanded(q),'raw_candidate_counts':{k:len(v) for k,v in raw.items()},'final_context_count':len(context),'distinct_source_count':source_count,'final_context':sources,'feature_flags':FLAGS,'metadata_v2':metadata,'source_count':source_count,'evidence_family_count':family_count,'representation_count':representation_count}}