from __future__ import annotations
import collections, hashlib, json, os, sqlite3
from pathlib import Path
from jarvis_finance.services.household_import import preview_household_import

DB=Path('/home/agent/jarvis_runtime/finance-system/data/finance.sqlite3')
ROOT=Path('/home/agent/jarvis_runtime/finance-system/sprint16.2-analysis')
PAYLOAD=Path('/home/agent/jarvis_runtime/finance-system/sprint16.1-uat/real-preview-payload-private.json')
RESPONSE=ROOT/'sprint16.2-local-preview-private.json'
EVIDENCE=ROOT/'sprint16.2-local-preview-evidence-private.json'
SOURCES=[Path('/home/agent/.hermes/private/finance/sprint16')/x for x in ('raiffeisen.csv','akb_transactions.csv','visa.csv','migros.csv')]
def sha_file(p):
 h=hashlib.sha256()
 with p.open('rb') as f:
  for chunk in iter(lambda:f.read(1024*1024),b''):h.update(chunk)
 return h.hexdigest()
def table_digest(c,t):
 h=hashlib.sha256();h.update(t.encode())
 for row in c.execute(f'SELECT * FROM "{t}" ORDER BY rowid'):h.update(json.dumps(list(row),ensure_ascii=False,default=str,separators=(',',':')).encode())
 return h.hexdigest()
def snapshot():
 c=sqlite3.connect(f'file:{DB}?mode=ro',uri=True)
 tables=[r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")]
 s={'tables':{t:{'count':c.execute(f'SELECT COUNT(*) FROM "{t}"').fetchone()[0],'digest':table_digest(c,t)} for t in tables},'fk':list(c.execute('PRAGMA foreign_key_check')),'integrity':c.execute('PRAGMA integrity_check').fetchone()[0]};c.close();return s
def private_write(path,data):
 path.parent.mkdir(parents=True,exist_ok=True);tmp=path.with_suffix(path.suffix+'.tmp');tmp.write_text(json.dumps(data,ensure_ascii=False,indent=2));os.chmod(tmp,0o600);os.replace(tmp,path);os.chmod(path,0o600)
def main():
 before=snapshot();source_before={str(p):sha_file(p) for p in SOURCES}
 conn=sqlite3.connect(f'file:{DB}?mode=ro',uri=True);conn.row_factory=sqlite3.Row
 result=preview_household_import(conn,json.loads(PAYLOAD.read_text()));conn.close()
 after=snapshot();source_after={str(p):sha_file(p) for p in SOURCES}
 private_write(RESPONSE,result)
 rows=result.get('rows',[]);clusters=result.get('merchant_clusters',[])
 unresolved=[r for r in rows if r.get('user_state')=='decision_needed' and r.get('disposition') not in ('duplicate_file','duplicate_source_row','pending','superseded_pending')]
 income=[r for r in unresolved if r.get('transaction_semantics')=='income']
 cause=collections.Counter(str(r.get('classification_v2',{}).get('income_kind') or r.get('classification') or 'unresolved') for r in income)
 evidence={'source_preview_payload_sha256':hashlib.sha256(PAYLOAD.read_bytes()).hexdigest(),'db_before':before,'db_after':after,'changed_tables':[t for t in before['tables'] if before['tables'][t]!=after['tables'][t]],'source_hashes_unchanged':source_before==source_after,'source_hashes_before':source_before,'source_hashes_after':source_after,'counts':result.get('counts'),'review_threshold':result.get('review_threshold'),'readiness_checks':result.get('readiness_checks'),'technically_confirmable':result.get('technically_confirmable'),'business_ready_for_confirm':result.get('business_ready_for_confirm'),'errors':result.get('errors'),'unresolved_count':len(unresolved),'unresolved_income_count':len(income),'income_private_kinds':dict(cause),'merchant_cluster_count':len(clusters),'confirm_called':False,'real_import_performed':False}
 private_write(EVIDENCE,evidence)
 print(json.dumps({'classification_version':result.get('classification_version'),'counts':result.get('counts'),'review_threshold':result.get('review_threshold'),'technically_confirmable':result.get('technically_confirmable'),'business_ready_for_confirm':result.get('business_ready_for_confirm'),'errors':result.get('errors'),'unresolved_income_count':len(income),'db_changed_tables':evidence['changed_tables'],'source_hashes_unchanged':evidence['source_hashes_unchanged'],'fk_before_after':[len(before['fk']),len(after['fk'])],'integrity_before_after':[before['integrity'],after['integrity']],'response':str(RESPONSE),'evidence':str(EVIDENCE),'modes':[oct(RESPONSE.stat().st_mode&0o777),oct(EVIDENCE.stat().st_mode&0o777)],'confirm_called':False},ensure_ascii=False,indent=2))
if __name__=='__main__':main()