"""Run new file-series cases through actual public CLI, with independent raw-row oracles."""
from file_series_support import *
import sys,sqlite3,os
from collections import defaultdict
sys.path.insert(0,str(PROJECT/'tools'))
from series_data import normalize_dirty,write_invoice_db,merge_month_files,read_targets,write_alignment_db

original=load(PROJECT/'inputs/source-records.json')['rows']
oracle={r['invoice_id']:r for r in original};amounts=defaultdict(Decimal);countries=defaultdict(Decimal)
for r in original:amounts[r['customer_code']]+=Decimal(r['invoice_amount']);countries[r['billing_country']]+=Decimal(r['invoice_amount'])
protected=[LAB/'data/Chinook_Sqlite.sqlite',LAB/'runtime/launcher-start-current.json',LAB/'runtime/work/runtime-bundle-registry.json',LAB/'runtime/work/runtime-datasource-registry.json',PROJECT/'data/files.sqlite',PROJECT/'inputs/ChinookInvoices2012.xlsx',PROJECT/'inputs/ChinookInvoices2012.csv',*(PROJECT/'models').rglob('*')]
before={str(p):sha(p) for p in protected if p.is_file()};save(SERIES/'protected-before.json',before)
START=int(os.environ.get('FILES_FROM_CHAPTER','30'))
summary={str(ch):load(SERIES/f'chapter-{ch}'/'evidence/acceptance.json') for ch in range(30,START)}
total={'columns':['count(invoiceId) as bills','sum(invoiceAmount) as amount'],'limit':1}

if START<=30:
    # 30: audit first, explicitly accept only the confirmed subset, retain pending rows.
    e=chapter_evidence(30);folder=e.parent;ns='file-cleaning';source=PROJECT/'series-inputs/Invoices2012Messy.xlsx'
    try:normalize_dirty(source)
    except ValueError as error:save(e/'strict-rejection.json',{'rejected':True,'published':False,'message':str(error)})
    else:raise AssertionError('Unresolved rows were silently accepted')
    records,audit=normalize_dirty(source,allow_quarantine=True)
    assert len(records)==83 and len(audit['pending'])==3,(len(records),audit['pending'])
    assert sum(a['action']=='skip_blank' for a in audit['audit'])==1
    assert sum(a['action']=='skip_declared_total' for a in audit['audit'])==1
    for record in records:
        assert all(Decimal(record[k])==Decimal(oracle[record['invoice_id']][k]) if k=='invoice_amount' else record[k]==oracle[record['invoice_id']][k] for k in ['invoice_id','customer_code','invoice_date','billing_country','invoice_amount'])
    save(e/'normalization-audit.json',audit);db=folder/'data/analysis.sqlite';write_invoice_db(db,records,audit)
    register_case(folder,ns,db,['file_invoices'],lambda m:[write_invoice_models(m,scope_description='仅包含已确认接入的部分；待确认记录未计入，回答完整性时应说明范围并核对整理报告。')])
    query_case(folder,'30-total',ns,'FileInvoiceQueryModel',total,[{'bills':83,'amount':Decimal('477.53')}])
    query_case(folder,'30-countries',ns,'FileInvoiceQueryModel',{'columns':['billingCountry','sum(invoiceAmount) as amount'],'groupBy':['billingCountry'],'orderBy':['billingCountry'],'limit':100},[{'billingCountry':c,'amount':countries[c]} for c in sorted(countries)])
    detail=[{'invoiceId':r['invoice_id'],'customerCode':r['customer_code'],'invoiceDate':r['invoice_date'],'billingCountry':r['billing_country'],'invoiceAmount':Decimal(r['invoice_amount']),'sourceFile':source.name,'sourceRow':r['source_row']} for r in sorted(records,key=lambda r:r['invoice_id'])]
    query_case(folder,'30-traceable-details',ns,'FileInvoiceQueryModel',{'columns':['invoiceId','customerCode','invoiceDate','billingCountry','invoiceAmount','sourceFile','sourceRow'],'orderBy':['invoiceId'],'limit':100},detail)
    query_case(folder,'30-leading-zero',ns,'FileInvoiceQueryModel',{'columns':['customerCode','sum(invoiceAmount) as amount'],'slice':[{'field':'customerCode','op':'=','value':'00026'}],'groupBy':['customerCode'],'limit':10},[{'customerCode':'00026','amount':Decimal('25.84')}])
    summary['30']={'executions':4,'preparationRejections':1,'acceptedRows':83,'pendingRows':3,'amount':'477.53','unit':'教学元','sourceUnchanged':sha(source)==audit['sourceSha256']}
    save(e/'acceptance.json',summary['30'])
    

if START<=31:
    # 31: a manifest defines snapshot/append/correction semantics; never mtime wins.
    e=chapter_evidence(31);folder=e.parent;ns='file-monthly';db=folder/'data/analysis.sqlite';root=PROJECT/'series-inputs/monthly'
    rows,report=merge_month_files(root/'manifest-v1.json');assert len(rows)==83 and report['amount']=='477.53' and report['duplicateRows']==1
    write_invoice_db(db,rows,report);save(e/'manifest-v1-report.json',report)
    register_case(folder,ns,db,['file_invoices'],lambda m:[write_invoice_models(m)])
    query_case(folder,'31-initial',ns,'FileInvoiceQueryModel',total,[{'bills':83,'amount':Decimal('477.53')}])
    again,repeated=merge_month_files(root/'manifest-v1.json');write_invoice_db(db,again,repeated)
    query_case(folder,'31-repeat-initial',ns,'FileInvoiceQueryModel',total,[{'bills':83,'amount':Decimal('477.53')}])
    digest=sha(db);rejects=[]
    for name in ['manifest-conflict.json','manifest-missing.json']:
        try:merge_month_files(root/name)
        except ValueError as error:rejects.append({'manifest':name,'message':str(error),'databaseUnchanged':sha(db)==digest})
        else:raise AssertionError(name+' should reject')
    assert all(r['databaseUnchanged'] for r in rejects);save(e/'preparation-rejections.json',rejects)
    for version,bills,amount in [('v2',84,'487.53'),('v3',84,'488.53')]:
        rows,report=merge_month_files(root/('manifest-'+version+'.json'));write_invoice_db(db,rows,report);save(e/('manifest-'+version+'-report.json'),report)
        query_case(folder,'31-'+version,ns,'FileInvoiceQueryModel',total,[{'bills':bills,'amount':Decimal(amount)}])
    rows,report=merge_month_files(root/'manifest-v3.json');write_invoice_db(db,rows,report)
    query_case(folder,'31-repeat-v3',ns,'FileInvoiceQueryModel',total,[{'bills':84,'amount':Decimal('488.53')}])
    query_case(folder,'31-corrected-record',ns,'FileInvoiceQueryModel',{'columns':['invoiceId','invoiceAmount','sourceFile'],'slice':[{'field':'invoiceId','op':'=','value':'250'}],'limit':1},[{'invoiceId':'250','invoiceAmount':Decimal('14.86'),'sourceFile':'correction.csv'}])
    updated=[dict(r) for r in original];updated[0]['invoice_amount']='14.86'
    updated.append({**original[0],'invoice_id':'TEACH-NEW-2012','customer_code':'00026','invoice_date':'2012-12-31 00:00:00','invoice_amount':'10.00'})
    query_case(folder,'31-all-updated-details',ns,'FileInvoiceQueryModel',{'columns':['invoiceId','customerCode','invoiceDate','billingCountry','invoiceAmount'],'orderBy':['invoiceId'],'limit':100},[{'invoiceId':r['invoice_id'],'customerCode':r['customer_code'],'invoiceDate':r['invoice_date'],'billingCountry':r['billing_country'],'invoiceAmount':Decimal(r['invoice_amount'])} for r in sorted(updated,key=lambda r:r['invoice_id'])])
    summary['31']={'executions':7,'preparationRejections':2,'initial':[83,'477.53'],'afterAppend':[84,'487.53'],'afterCorrection':[84,'488.53'],'duplicateRows':1,'corrections':1,'repeatDoesNotAppend':True,'allUpdatedRowsIndependentlyChecked':True};save(e/'acceptance.json',summary['31'])
    

if START<=32:
    # 32: model-owned read-only SQL view normalizes documented legacy fields.
    e=chapter_evidence(32);folder=e.parent;ns='file-legacy';db=PROJECT/'series-data/legacy.sqlite';legacy_before=sha(db)
    view="SELECT b.id0 AS invoice_id,m.customer_code,b.d0 AS invoice_date,b.n0 AS billing_country,b.v0 / 100.0 AS invoice_amount FROM legacy_bill b JOIN legacy_customer_map m ON b.c0=m.old_code WHERE b.s0='V'"
    register_case(folder,ns,db,['legacy_bill','legacy_customer_map'],lambda m:[write_invoice_models(m,table='legacy_bill',name='LegacyBill',source_columns=False,view=view)])
    raw=call('32-raw-status-counts','sql','query','--data-source',ns+'-sqlite','--sql','select s0,count(*) as rows,sum(v0) as cents from legacy_bill group by s0 order by s0','--max-rows','10',ns=ns)
    query_case(folder,'32-valid-total',ns,'LegacyBillQueryModel',total,[{'bills':83,'amount':Decimal('477.53')}])
    query_case(folder,'32-customers',ns,'LegacyBillQueryModel',{'columns':['customerCode','sum(invoiceAmount) as amount'],'groupBy':['customerCode'],'orderBy':['customerCode'],'limit':100},[{'customerCode':c,'amount':amounts[c]} for c in sorted(amounts)])
    query_case(folder,'32-excluded-statuses',ns,'LegacyBillQueryModel',{'columns':['invoiceId'],'slice':[{'field':'invoiceId','op':'in','value':['TEACH-VOID','TEACH-DRAFT']}],'limit':10},[])
    bad=folder/'queries/32-obsolete-field.json';save(bad,{'columns':['v1'],'limit':1})
    rejected=call('32-obsolete-field-validate','query','validate','LegacyBillQueryModel','--payload',bad,ns=ns,required=False)
    assert not rejected['success'];assert sha(db)==legacy_before
    summary['32']={'executions':3,'queryRejections':1,'rawRows':85,'validRows':83,'amount':'477.53','oldSourceUnchanged':True,'viewSql':view,'wwiUsed':False};save(e/'acceptance.json',summary['32'])
    

if START<=33:
    # 33: query actual through its governed QM, then prepare a common analysis snapshot.
    e=chapter_evidence(33);folder=e.parent;ns='file-target-analysis'
    expected_actual=[{'invoiceId':r['invoice_id'],'customerCode':r['customer_code'],'invoiceDate':r['invoice_date'],'billingCountry':r['billing_country'],'invoiceAmount':Decimal(r['invoice_amount'])} for r in original]
    actual=query_case(folder,'33-extract-governed-actual','file-legacy','LegacyBillQueryModel',{'columns':['invoiceId','customerCode','invoiceDate','billingCountry','invoiceAmount'],'orderBy':['invoiceId'],'limit':100},expected_actual)
    targets=read_targets(PROJECT/'series-inputs/Targets2012-v1.xlsx','v1');assert len(targets)==5
    db=folder/'data/analysis.sqlite';write_alignment_db(db,actual,targets);save(e/'prepared-sources.json',{'actualModel':'LegacyBillQueryModel','actualRows':83,'targetsSource':'Targets2012-v1.xlsx','targetsSha256':sha(PROJECT/'series-inputs/Targets2012-v1.xlsx'),'targetRows':5,'unit':'共同教学元','targetVersion':'v1','period':'2012','method':'governed actual extraction plus external file preparation into one SQLite'})
    def alignment_models(m):
        qm=write_invoice_models(m,table='actual_invoices',name='AnalysisInvoice',source_columns=False);write_target_models(m);return [qm,'AnalysisTargetQueryModel']
    register_case(folder,ns,db,['actual_invoices','targets'],alignment_models)
    query_case(folder,'33-actual-total',ns,'AnalysisInvoiceQueryModel',total,[{'bills':83,'amount':Decimal('477.53')}])
    query_case(folder,'33-target-total',ns,'AnalysisTargetQueryModel',{'columns':['sum(targetAmount) as target'],'slice':[{'field':'period','op':'=','value':'2012'},{'field':'version','op':'=','value':'v1'}],'limit':1},[{'target':Decimal('85')}])
    expected=[]
    for t in targets:
        c=t['customer_code'];v=Decimal(t['target_amount']);r={'target_code':c,'target_amount':v}
        if c in amounts:r.update(actual_code=c,actual_amount=amounts[c],amount_difference=(amounts[c]-v).quantize(Decimal('.01')),**({'achievement':amounts[c]/v} if v else {}))
        expected.append(r)
    compose_case(folder,'33-achievement',ns,alignment_script(),expected)
    compose_case(folder,'33-actual-without-target',ns,alignment_script(mode='actual-only'),[{'actual_code':c,'actual_amount':amounts[c]} for c in sorted(amounts) if c not in {t['customer_code'] for t in targets}])
    compose_case(folder,'33-target-without-actual',ns,alignment_script(mode='target-only'),[{'target_code':'00999','target_amount':Decimal(10)}])
    db_before=sha(db)
    try:write_alignment_db(db,actual,targets+[dict(targets[0])])
    except ValueError as error:save(e/'duplicate-target-rejection.json',{'message':str(error),'databaseUnchanged':sha(db)==db_before})
    else:raise AssertionError('Duplicate target should reject before writing')
    assert sha(db)==db_before
    summary['33']={'executions':6,'preparationRejections':1,'actualCustomers':47,'targetCustomers':5,'matched':4,'actualOnly':43,'targetOnly':1,'zeroTargetCustomer':'00002','zeroTargetActual':'0.99','zeroTargetAchievement':None,'actualAmount':'477.53','targetAmount':'85','unit':'教学元','directFederatedJoin':False};save(e/'acceptance.json',summary['33'])
assert all(sha(Path(p))==h for p,h in before.items())
save(SERIES/'acceptance-30-33.json',{'launcher':'0.1.42','cli':'0.1.25','runtime':URL,'chapters':summary,'originalTeachingFilesUnchanged':True,'newIssuesSubmitted':False,'cleanSeparateCodexConversationTested':False})
print(json.dumps(summary,ensure_ascii=True),flush=True)