"""Explicit teaching adapters. File preparation is separate from Foggy execution.""" from pathlib import Path from decimal import Decimal from datetime import datetime import csv,json,sqlite3,hashlib def digest(path):return hashlib.sha256(Path(path).read_bytes()).hexdigest() def load(path):return json.loads(Path(path).read_text(encoding='utf-8-sig')) def save(path,value): path=Path(path);path.parent.mkdir(parents=True,exist_ok=True) path.write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n',encoding='utf-8') def decimal(value): d=Decimal(str(value)) if not d.is_finite() or d!=d.quantize(Decimal('.01')):raise ValueError('金额不是有限的两位小数') return d def normalize_dirty(source,allow_quarantine=False): import openpyxl source=Path(source);wb=openpyxl.load_workbook(source,read_only=True,data_only=False);ws=wb['原始报表'] records=[];audit=[];pending=[];seen=set() for n,cells in enumerate(ws.iter_rows(min_row=4,max_col=7),start=4): values=[c.value for c in cells];iid,code,date,country,amount,unit,note=values if all(v is None for v in values):audit.append({'row':n,'action':'skip_blank'});continue if iid=='合计':audit.append({'row':n,'action':'skip_declared_total','amount':str(amount)});continue try: if cells[4].data_type=='f':raise ValueError('公式不是固定值；需确认计算结果后重新提交') iid=str(iid).strip() if not iid or iid in seen:raise ValueError('编号缺失或重复') if isinstance(code,(int,float)): if int(code)!=code:raise ValueError('客户编号不是整数') code=str(int(code)).zfill(5) else:code=str(code).strip() if len(code)!=5 or not code.isdigit():raise ValueError('客户编号不符合已确认的五位规则') if isinstance(date,datetime):date=date.strftime('%Y-%m-%d %H:%M:%S') elif isinstance(date,str) and len(date)==10 and date[4] in '-/' and date[7]==date[4]: date=datetime.strptime(date,'%Y-%m-%d' if date[4]=='-' else '%Y/%m/%d').strftime('%Y-%m-%d %H:%M:%S') else:raise ValueError('日期存在歧义，不能猜月日顺序') value=Decimal(str(amount)) if unit=='分':value/=100 elif unit!='元':raise ValueError('金额单位待确认') value=decimal(value) record={'invoice_id':iid,'customer_code':code,'invoice_date':date,'billing_country':str(country),'invoice_amount':str(value),'source_file':source.name,'source_row':n} records.append(record);seen.add(iid) audit.append({'row':n,'action':'accepted','sourceValues':[str(v) for v in values[:6]],'normalized':record}) except (ValueError,TypeError) as error:pending.append({'file':source.name,'row':n,'invoiceId':str(iid),'reason':str(error)}) wb.close() report={'source':source.name,'sourceSha256':digest(source),'accepted':len(records),'pending':pending,'audit':audit,'amount':str(sum((Decimal(r['invoice_amount']) for r in records),Decimal(0))),'unit':'教学元；不是 Chinook 原币种声明'} if pending and not allow_quarantine:raise ValueError('有待确认记录，未发布任何数据：'+json.dumps(pending,ensure_ascii=False)) return records,report def write_invoice_db(path,records,report): path=Path(path);path.parent.mkdir(parents=True,exist_ok=True) with sqlite3.connect(path) as db: db.execute('create table if not exists file_invoices(invoice_id TEXT PRIMARY KEY, customer_code TEXT NOT NULL, invoice_date DATETIME NOT NULL, billing_country TEXT NOT NULL, invoice_amount DECIMAL(18,2) NOT NULL, source_file TEXT NOT NULL, source_row INTEGER NOT NULL)') db.execute('delete from file_invoices') db.executemany('insert into file_invoices values(?,?,?,?,?,?,?)',[(r['invoice_id'],r['customer_code'],r['invoice_date'],r['billing_country'],r['invoice_amount'],r['source_file'],r['source_row']) for r in records]) db.execute('create table if not exists import_manifest(id INTEGER PRIMARY KEY, manifest_json TEXT NOT NULL)') db.execute('delete from import_manifest');db.execute('insert into import_manifest values(1,?)',(json.dumps(report,ensure_ascii=False),)) def merge_month_files(manifest_path): path=Path(manifest_path);manifest=load(path);state={};ledger=[];seen_months=set();duplicates=0;updated=0;added=0 for item in manifest['files']: source=path.parent/item['path'] if not source.exists():raise ValueError('输入文件缺失：'+item['path']) if item['kind']=='snapshot':seen_months.add(item['month']) with source.open(encoding='utf-8-sig',newline='') as inp: reader=csv.DictReader(inp);mapping=item.get('columns',{k:k for k in ['invoice_id','customer_code','invoice_date','billing_country','invoice_amount']}) if not set(mapping.values())<=set(reader.fieldnames or []):raise ValueError('必需列缺失：'+source.name) for n,row in enumerate(reader,start=2): r={key:row[value] for key,value in mapping.items()};r['invoice_amount']=str(decimal(r['invoice_amount'])) datetime.strptime(r['invoice_date'],'%Y-%m-%d %H:%M:%S') if len(r['customer_code'])!=5 or not r['customer_code'].isdigit():raise ValueError('客户编码规则不符') iid=r['invoice_id'];old=state.get(iid);content={k:r[k] for k in mapping} if old: if all(old[k]==r[k] for k in mapping):duplicates+=1;ledger.append({'file':source.name,'row':n,'id':iid,'action':'same_record'});continue if item['kind']!='correction' or item['revision']<=old['_revision']:raise ValueError('同键不同内容，没有有效更正声明：'+iid) updated+=1;action='corrected' else: if item['kind']=='correction':raise ValueError('更正指向不存在的编号：'+iid) added+=1;action='inserted' state[iid]={**r,'source_file':source.name,'source_row':n,'_revision':item['revision']} ledger.append({'file':source.name,'row':n,'id':iid,'action':action,'revision':item['revision']}) ledger.append({'file':source.name,'sourceSha256':digest(source),'action':'file_loaded'}) if set(manifest['expectedMonths'])!=seen_months:raise ValueError('月度文件不完整') records=[state[k] for k in sorted(state)] report={'manifest':path.name,'manifestSha256':digest(path),'rows':len(records),'amount':str(sum((Decimal(r['invoice_amount']) for r in records),Decimal(0))),'duplicateRows':duplicates,'corrections':updated,'insertedKeys':added,'ledger':ledger,'rule':'按声明的快照/追加/更正清单重建，非文件修改时间优先'} return records,report def create_legacy(path,original): path=Path(path);path.parent.mkdir(parents=True,exist_ok=True) with sqlite3.connect(path) as db: db.execute('create table if not exists legacy_bill(id0 TEXT PRIMARY KEY,c0 TEXT,d0 DATETIME,v0 INTEGER,v1 DECIMAL(18,2),s0 TEXT,n0 TEXT)') db.execute('create table if not exists legacy_customer_map(old_code TEXT PRIMARY KEY,customer_code TEXT NOT NULL UNIQUE)') db.execute('delete from legacy_bill');db.execute('delete from legacy_customer_map') db.executemany('insert into legacy_bill values(?,?,?,?,?,?,?)',[(r['invoice_id'],'K'+r['customer_code'],r['invoice_date'],int(Decimal(r['invoice_amount'])*100),str(Decimal(r['invoice_amount'])*10),'V',r['billing_country']) for r in original]) db.executemany('insert into legacy_bill values(?,?,?,?,?,?,?)',[('TEACH-VOID','K00026','2012-12-31 00:00:00',199,19.90,'X','USA'),('TEACH-DRAFT','K00026','2012-12-31 00:00:00',300,30.00,'D','USA')]) db.executemany('insert into legacy_customer_map values(?,?)',[('K'+c,c) for c in sorted({r['customer_code'] for r in original})]) return {'rows':85,'validRows':83,'validAmount':'477.53','unit':'v0 存分，视图除以 100 后为教学元','status':{'V':'有效','X':'作废','D':'草稿'},'obsoleteField':'v1 是人为设定的旧口径金额，不参与分析'} def read_targets(source,expected_version): import openpyxl wb=openpyxl.load_workbook(source,read_only=True,data_only=False);ws=wb['目标'] rows=list(ws.iter_rows(min_col=1,max_col=5));assert [c.value for c in rows[0]]==['客户编号','目标年份','目标版本','目标金额','单位'] result=[];keys=set() for n,cells in enumerate(rows[1:],start=2): code,period,version,amount,unit=[c.value for c in cells] if any(c.data_type=='f' for c in cells):raise ValueError('目标文件公式需先确认固定值') if not isinstance(code,str) or len(code)!=5:raise ValueError('目标客户编号必须是五位文本') period=str(period) if version!=expected_version or period!='2012' or unit!='元':raise ValueError('目标期间、版本或单位不符') key=(code,period,version) if key in keys:raise ValueError('目标业务键重复：'+str(key)) keys.add(key);result.append({'customer_code':code,'period':period,'version':version,'target_amount':str(decimal(amount)),'source_file':Path(source).name,'source_row':n}) wb.close();return result def write_alignment_db(path,actual,targets): if len({(r['customer_code'],r['period'],r['version']) for r in targets})!=len(targets):raise ValueError('目标业务键重复') path=Path(path);path.parent.mkdir(parents=True,exist_ok=True) with sqlite3.connect(path) as db: db.execute('create table if not exists actual_invoices(invoice_id TEXT PRIMARY KEY,customer_code TEXT,invoice_date DATETIME,billing_country TEXT,invoice_amount DECIMAL(18,2))') db.execute('create table if not exists targets(customer_code TEXT,period TEXT,version TEXT,target_amount DECIMAL(18,2),source_file TEXT,source_row INTEGER,PRIMARY KEY(customer_code,period,version))') db.execute('delete from actual_invoices');db.execute('delete from targets') db.executemany('insert into actual_invoices values(?,?,?,?,?)',[(r['invoiceId'],r['customerCode'],r['invoiceDate'],r['billingCountry'],r['invoiceAmount']) for r in actual]) db.executemany('insert into targets values(?,?,?,?,?,?)',[(r['customer_code'],r['period'],r['version'],r['target_amount'],r['source_file'],r['source_row']) for r in targets])