"""A concrete sample pipeline: candidate snapshot, governed checks, then publish pointer.""" from pathlib import Path from datetime import datetime from decimal import Decimal from collections import defaultdict import argparse,json,sys,shutil,sqlite3 PROJECT=Path(__file__).resolve().parents[1] LAB=PROJECT.parent sys.path.insert(0,str(LAB/'checks')) from file_series_support import * import files_support as common from series_data import read_targets,write_alignment_db def run(target,version): stamp=datetime.now().strftime('%Y%m%d-%H%M%S-%f') folder=PROJECT/'series/chapter-34/runs'/stamp;e=folder/'evidence';e.mkdir(parents=True) common.E=e pointer=PROJECT/'series/chapter-34/latest-accepted.json' report={'run':stamp,'status':'running','phase':'input-validation','targetFile':str(target),'targetSha256':sha(target),'version':version,'published':False,'normalExecutions':0} before_pointer=sha(pointer) if pointer.exists() else None old_ds=None try: # Candidate validation must not redefine the baseline to fit the answer. targets=read_targets(target,version) target_total=sum((Decimal(t['target_amount']) for t in targets),Decimal(0)) expected_target={'v1':Decimal(85),'v2':Decimal(90)}[version] if target_total!=expected_target:raise ValueError('目标合计与已审核版本基线不符；需要人工审核后更新基线') report['phase']='extract-actual' source_folder=folder/'source' payload={'columns':['invoiceId','customerCode','invoiceDate','billingCountry','invoiceAmount'],'orderBy':['invoiceId'],'limit':100} source_file=source_folder/'query.json';save(source_file,payload) call('source-validate','query','validate','LegacyBillQueryModel','--payload',source_file,ns='file-legacy') actual_response=call('source-execute','query','execute','LegacyBillQueryModel','--payload',source_file,ns='file-legacy');actual=actual_response['data']['items'];report['normalExecutions']+=1 counts=source_folder/'count.json';save(counts,{'columns':['count(invoiceId) as bills'],'limit':1}) call('source-count-validate','query','validate','LegacyBillQueryModel','--payload',counts,ns='file-legacy') count=call('source-count-execute','query','execute','LegacyBillQueryModel','--payload',counts,ns='file-legacy')['data']['items'][0]['bills'];report['normalExecutions']+=1 if len(actual)!=count:raise ValueError('实际提取被截断；先补齐分页，不能发布不完整快照') with sqlite3.connect((LAB/'data/Chinook_Sqlite.sqlite').as_uri()+'?mode=ro',uri=True) as db: original=[(str(i),str(c).zfill(5),d,country,Decimal(str(a))) for i,c,d,country,a in db.execute('select InvoiceId,CustomerId,InvoiceDate,BillingCountry,Total from Invoice order by InvoiceId') if '2012-01-01'<=d<'2013-01-01'] expected_actual=[{'invoiceId':i,'customerCode':c,'invoiceDate':d,'billingCountry':country,'invoiceAmount':a} for i,c,d,country,a in original] same(actual,expected_actual) report['phase']='candidate-prepare' snapshot=folder/'data/analysis.sqlite';write_alignment_db(snapshot,actual,targets) namespace='file-run-'+stamp.lower() templates=PROJECT/'series/chapter-33/models' def writer(model_dir): shutil.copytree(templates,model_dir,dirs_exist_ok=True) return ['AnalysisInvoiceQueryModel','AnalysisTargetQueryModel'] register_case(folder,namespace,snapshot,['actual_invoices','targets'],writer) report['phase']='candidate-regression' by_customer=defaultdict(Decimal) for _,code,_,_,amount in original:by_customer[code]+=amount query_case(folder,'34-actual-total',namespace,'AnalysisInvoiceQueryModel',{'columns':['count(invoiceId) as bills','sum(invoiceAmount) as amount'],'limit':1},[{'bills':83,'amount':Decimal('477.53')}]);report['normalExecutions']+=1 target_payload={'columns':['sum(targetAmount) as target'],'slice':[{'field':'period','op':'=','value':'2012'},{'field':'version','op':'=','value':version}],'limit':1} query_case(folder,'34-target-total',namespace,'AnalysisTargetQueryModel',target_payload,[{'target':expected_target}]);report['normalExecutions']+=1 expected=[] for t in targets: code=t['customer_code'];amount=Decimal(t['target_amount']);r={'target_code':code,'target_amount':amount} if code in by_customer:r.update(actual_code=code,actual_amount=by_customer[code],amount_difference=(by_customer[code]-amount).quantize(Decimal('.01')),**({'achievement':by_customer[code]/amount} if amount else {})) expected.append(r) compose_case(folder,'34-achievement',namespace,alignment_script(version),expected);report['normalExecutions']+=1 compose_case(folder,'34-actual-only',namespace,alignment_script(version,'actual-only'),[{'actual_code':c,'actual_amount':by_customer[c]} for c in sorted(by_customer) if c not in {t['customer_code'] for t in targets}]);report['normalExecutions']+=1 compose_case(folder,'34-target-only',namespace,alignment_script(version,'target-only'),[{'target_code':'00999','target_amount':Decimal(10)}]);report['normalExecutions']+=1 report['phase']='publish' old_ds=call('current-binding','datasources','binding','--namespace','file-regression-current',ns='file-regression-current')['data']['dataSource'] new_ds=namespace+'-sqlite' try: call('publish-binding','datasources','bind','--namespace','file-regression-current','--data-source',new_ds,ns='file-regression-current') for model in ['AnalysisInvoiceQueryModel','AnalysisTargetQueryModel']: call('publish-refresh-'+model,'models','refresh','--model',model,ns='file-regression-current') call('publish-describe-'+model,'models','describe',model,ns='file-regression-current') query_case(folder,'34-published-target','file-regression-current','AnalysisTargetQueryModel',target_payload,[{'target':expected_target}]);report['normalExecutions']+=1 except Exception: call('restore-previous-binding','datasources','bind','--namespace','file-regression-current','--data-source',old_ds,ns='file-regression-current') for model in ['AnalysisInvoiceQueryModel','AnalysisTargetQueryModel']:call('restore-refresh-'+model,'models','refresh','--model',model,ns='file-regression-current') raise report.update(status='passed',phase='complete',published=True,publishedNamespace='file-regression-current',namespace=namespace,datasource=new_ds,database=str(snapshot),databaseSha256=sha(snapshot),actualRows=83,actualAmount='477.53',targetAmount=str(expected_target),modelHashes={str(p.relative_to(folder)):sha(p) for p in (folder/'models').rglob('*') if p.is_file()},sourceQuerySha256=sha(source_file),sourceModel='LegacyBillQueryModel',unit='教学元',template='第 33 章已审核模型',boundary='样例回归，不是任意文件的自动正确性证明') save(folder/'report.json',report) temp=pointer.with_suffix('.pending.json');save(temp,{'run':stamp,'version':version,'report':str(folder/'report.json'),'datasource':new_ds,'database':str(snapshot),'targetAmount':str(expected_target)}) temp.replace(pointer) except Exception as error: if old_ds is not None and report.get('phase')=='complete': # A failed report/pointer write after the smoke must restore the prior binding too. call('restore-after-record-failure','datasources','bind','--namespace','file-regression-current','--data-source',old_ds,ns='file-regression-current') for model in ['AnalysisInvoiceQueryModel','AnalysisTargetQueryModel']:call('restore-record-refresh-'+model,'models','refresh','--model',model,ns='file-regression-current') report['published']=False report.update(status='failed',error=str(error),latestAcceptedUnchanged=(sha(pointer) if pointer.exists() else None)==before_pointer) save(folder/'report.json',report) print(json.dumps({'status':'failed','report':str(folder/'report.json'),'error':str(error)},ensure_ascii=True),flush=True) return 1 print(json.dumps({'status':'passed','report':str(folder/'report.json'),'executions':report['normalExecutions'],'version':version},ensure_ascii=True),flush=True) return 0 if __name__=='__main__': parser=argparse.ArgumentParser();parser.add_argument('--target',type=Path,required=True);parser.add_argument('--version',choices=['v1','v2'],required=True) args=parser.parse_args();raise SystemExit(run(args.target,args.version))