"""Check cohort selection, detail rows and separate date scopes using raw Chinook rows."""
from composition_support import *
from decimal import Decimal
from collections import defaultdict
import sqlite3

e=new_run('chapter28'); q=PROJECT/'queries/chapter-28';fixture=load(PROJECT/'fixture.json')
before={str(p):sha(p) for p in [Path(fixture['database']),*(PROJECT/'models').rglob('*')] if p.is_file()}
cap=call(e,'capabilities','capabilities')
assert cap['data']['securityMode']=='none-dev-test-only'
assert all(cap['data']['capabilities']['compose.'+mode]=='supported' for mode in ['validate','preview','execute'])
binding=call(e,'binding','datasources','binding','--namespace',NS)
assert binding['data']['dataSource']==fixture['datasource']
validation=call(e,'models-validate','models','validate','--models-dir',PROJECT/'models')
assert not validation['data'].get('invalidFiles',0),validation
model='ChinookInvoiceQueryModel'
desc=call(e,'describe','models','describe',model)
assert all(f in desc['data']['data']['fields'] for f in ['customerId','invoiceId','invoiceDate','invoiceAmount'])
md=call(e,'markdown','models','describe',model,'--format','markdown')
(e/(model+'.md')).write_text(md['data']['content'],encoding='utf-8')
main=(q/'03-top-customers-bills.fsscript').read_text(encoding='utf-8')
totals=main.split('const selected =')[0]
selected=main.split('const bills =')[0]
bills='const bills ='+main.split('const bills =',1)[1].split('const result =')[0]
year="['2012-01-01', '2013-01-01']"; next_year="['2013-01-01', '2014-01-01']"
cross=main.replace(bills,bills.replace(year,next_year))
# Alternate base TopN keeps exactly the same customer ordering and output fields.
selected_base=totals.replace('const totals =','const selected =').replace("groupBy: ['customerId']", "groupBy: ['customerId'],\n    orderBy: ['-selection_amount', 'customer_id'], limit: 3")
direct=selected_base+'const bills ='+main.split('const bills =',1)[1]
scripts={
 '01-customer-totals':totals+"return {plans:totals.query({columns:['customer_id','selection_amount'],orderBy:['customer_id'],limit:100})};\n",
 '02-top-three-customers':selected+"return {plans:selected};\n",
 '03-top-customers-bills':main,
 '04-final-limit-three':main.replace("orderBy: ['customer_id', 'invoice_date', 'invoice_id'], limit: 100", "orderBy: ['customer_id', 'invoice_date', 'invoice_id'], limit: 3"),
 '05-three-largest-invoices':bills+"return {plans:bills.query({columns:['bill_customer_id','invoice_id','invoice_date','bill_amount'],orderBy:['-bill_amount','invoice_id'],limit:3})};\n",
 '06-selected-next-year':cross,
 '07-selected-next-year-left':cross.replace("selected.join(bills, 'inner'", "selected.join(bills, 'left'"),
 '08-empty-selection':main.replace("columns: ['customer_id', 'selection_amount'],\n    orderBy:", "columns: ['customer_id', 'selection_amount'],\n    slice: [{field:'selection_amount',op:'>=',value:1000}],\n    orderBy:"),
 '09-base-top-three':direct,
 '10-recheck-selected-totals':main.replace('return { plans: result };',"return {plans:result.query({columns:['customer_id','sum(bill_amount) as detail_amount','count(invoice_id) as bill_count'],groupBy:['customer_id'],orderBy:['customer_id'],limit:100})};")}
for name,text in scripts.items():(q/(name+'.fsscript')).write_text(text,encoding='utf-8')

# Independent oracle: read source Invoice records, sum and select in Python, no engine SQL reuse.
with sqlite3.connect(Path(fixture['database']).as_uri()+'?mode=ro',uri=True) as db:
    raw=[(iid,cid,date,Decimal(str(amount))) for iid,cid,date,amount in db.execute('select InvoiceId,CustomerId,InvoiceDate,Total from Invoice')]
amounts=defaultdict(Decimal)
for iid,cid,date,amount in raw:
    if '2012-01-01'<=date<'2013-01-01':amounts[cid]+=amount
top=sorted(amounts,key=lambda cid:(-amounts[cid],cid))[:3]
assert top==[26,34,13]

def detail(start,end,left=False):
    out=[]
    for cid in sorted(top):
        found=sorted([row for row in raw if row[1]==cid and start<=row[2]<end],key=lambda row:(row[2],row[0]))
        out.extend([{'customer_id':cid,'invoice_id':iid,'invoice_date':date,'bill_amount':amount} for iid,_,date,amount in found])
        if not found and left:out.append({'customer_id':cid})
    return out
current=detail('2012-01-01','2013-01-01');following=detail('2013-01-01','2014-01-01')
largest=sorted([row for row in raw if '2012-01-01'<=row[2]<'2013-01-01'],key=lambda row:(-row[3],row[0]))[:3]
expected={
 '01-customer-totals':[{'customer_id':cid,'selection_amount':amounts[cid]} for cid in sorted(amounts)],
 '02-top-three-customers':[{'customer_id':cid,'selection_amount':amounts[cid]} for cid in top],
 '03-top-customers-bills':current,'04-final-limit-three':current[:3],
 '05-three-largest-invoices':[{'bill_customer_id':cid,'invoice_id':iid,'invoice_date':date,'bill_amount':amount} for iid,cid,date,amount in largest],
 '06-selected-next-year':following,'07-selected-next-year-left':detail('2013-01-01','2014-01-01',True),
 '08-empty-selection':[], '09-base-top-three':current,
 '10-recheck-selected-totals':[{'customer_id':cid,'detail_amount':amounts[cid],'bill_count':sum(r['customer_id']==cid for r in current)} for cid in sorted(top)]}

def same(actual,wanted):
    assert len(actual)==len(wanted),(len(actual),len(wanted))
    for a,b in zip(actual,wanted):
        for key in set(a)|set(b):
            av,bv=a.get(key),b.get(key)
            if isinstance(bv,Decimal):assert av is not None and abs(Decimal(str(av))-bv)<Decimal('1e-9'),(key,av,str(bv))
            else:assert av==bv,(key,av,bv)
passed=[];results={}
for name,wanted in expected.items():
    script=q/(name+'.fsscript')
    v=call(e,name+'-validate','compose','validate','--script',script);assert v['data']['valid'],v
    p=call(e,name+'-preview','compose','preview','--script',script);assert p['data']['valid'],p
    x=call(e,name+'-execute','compose','execute','--script',script);assert x['data']['valid'],x
    actual=x['data']['value']['plans'];same(actual,wanted);results[name]=actual
    passed.append({'case':name,'rows':len(actual),'independentMatch':True,'requestSha256':sha(script)})
    print(name,len(actual),flush=True)
assert all(sha(Path(p))==h for p,h in before.items())
assert all(sha(Path(p))==h for p,h in fixture['originalFiles'].items())
report={'launcher':LAUNCHER,'cli':'0.1.25','runtime':URL,'namespace':NS,'datasource':fixture['datasource'],'identity':'anonymous local dev/test; Console/FAP disabled','evidenceDirectory':str(e),'passed':passed,'selection':{'period':['2012-01-01','2013-01-01'],'idsInRankOrder':top,'customers':len(top),'amount':str(sum((amounts[c] for c in top),Decimal(0)))},'detail':{'rows':len(current),'amount':str(sum((r['bill_amount'] for r in current),Decimal(0))),'invoiceIds':[r['invoice_id'] for r in current]},'followingYear':{'rows':len(following),'amount':str(sum((r['bill_amount'] for r in following),Decimal(0))),'customerIds':sorted({r['customer_id'] for r in following}),'selectionUnchanged':True},'dataObservedRange':[min(r[2] for r in raw),max(r[2] for r in raw)],'originalFilesUnchanged':True,'teachingModelsAndDataUnchanged':True,'cleanSeparateCodexConversationTested':False}
save(e/'acceptance.json',report);save(ROOT/'chapter28-acceptance.json',report);save(e/'results.json',results)
save(e/'request-sha256.json',{p.name:sha(p) for p in q.glob('*.fsscript')})
print(json.dumps({'passed':len(passed),'selection':report['selection'],'detail':report['detail'],'followingYear':report['followingYear'],'evidenceDirectory':str(e)},ensure_ascii=True))