"""核对模拟表的键和数值，再清洗与 m:1 合并；逐行保留处置记录，不静默选冲突值。""" from pathlib import Path from collections import defaultdict import argparse,csv,json,re p=argparse.ArgumentParser();p.add_argument('--input-dir',default='raw');p.add_argument('--output-dir',default='rerun/python');a=p.parse_args();src=Path(a.input_dir);out=Path(a.output_dir);out.mkdir(parents=True,exist_ok=True) def load(name): with (src/name).open(encoding='utf-8',newline='') as f:return list(csv.DictReader(f)) def save(name,rows,fields): with (out/name).open('w',encoding='utf-8',newline='') as f: w=csv.DictWriter(f,fieldnames=fields,extrasaction='ignore',lineterminator='\n');w.writeheader();w.writerows(rows) number=re.compile(r'(?:[0-9]+|[0-9]{1,3}(?:,[0-9]{3})+)(?:[.][0-9]+)?') def parse(value):return float(value.replace(',','')) if value else None def clean(rows,lookup=False): audit=[];valid=[] for row in rows: row={k:v.strip() for k,v in row.items()};status='kept' if not re.fullmatch(r'[0-9]{5}',row['firm_id']) or (not lookup and not re.fullmatch(r'[0-9]{4}',row['year'])):status='invalid_key' elif lookup and not row['region']:status='invalid_value' elif not lookup and ((row['revenue'] and not number.fullmatch(row['revenue'])) or (row['employees'] and not re.fullmatch(r'[0-9]+',row['employees']))):status='invalid_value' if not lookup: for field in ['year','revenue','employees']:row[field+'_raw']=row[field] if status=='kept': if not lookup:row.update(year=int(row['year']),revenue=parse(row['revenue']),employees=int(row['employees']) if row['employees'] else None) valid.append(row) row['row_status']=status;audit.append(row) # 同一规范化内容只保留最早来源行；来源行号不参与“内容相同”的判断。 seen=set();unique=[] fields=['firm_id','region'] if lookup else ['firm_id','year','revenue','employees'] for row in valid: signature=tuple(row[k] for k in fields) if signature in seen:row['row_status']='duplicate_removed' else:seen.add(signature);unique.append(row) grouped=defaultdict(list) for row in unique:grouped[(row['firm_id'],) if lookup else (row['firm_id'],row['year'])].append(row) for group in grouped.values(): if len(group)>1: for row in group:row['row_status']='key_conflict' kept=[r for r in unique if r['row_status']=='kept'] assert len(audit)==len(kept)+sum(r['row_status']!='kept' for r in audit) return audit,kept master_audit,master=clean(load('firm-year.csv'));lookup_audit,lookup=clean(load('firm-region.csv'),True) assert len({(r['firm_id'],r['year']) for r in master})==len(master) assert len({r['firm_id'] for r in lookup})==len(lookup) byid={r['firm_id']:r for r in lookup};used=set();merged=[] for row in sorted(master,key=lambda r:(r['firm_id'],r['year'])): match=byid.get(row['firm_id']);used.add(row['firm_id']) merged.append(dict(row,region=match['region'] if match else '',lookup_row=match['lookup_row'] if match else '',merge_status='matched' if match else 'master_only')) for row in sorted(lookup,key=lambda r:r['firm_id']): if row['firm_id'] not in used:merged.append(dict(row,source_row='',year='',revenue=None,employees=None,merge_status='lookup_only')) merged.sort(key=lambda r:(r['firm_id'],int(r['year']) if r['year'] else 9999)) master_fields=['source_row','firm_id','year','revenue','employees'];lookup_fields=['lookup_row','firm_id','region'];merged_fields=master_fields+['lookup_row','region','merge_status'] save('master-audit.csv',master_audit,['source_row','firm_id','year_raw','revenue_raw','employees_raw','row_status']);save('lookup-audit.csv',lookup_audit,lookup_fields+['row_status']) save('clean-master.csv',sorted(master,key=lambda r:(r['firm_id'],r['year'])),master_fields);save('clean-lookup.csv',sorted(lookup,key=lambda r:r['firm_id']),lookup_fields) save('merged-all.csv',merged,merged_fields);save('analysis-ready.csv',[r for r in merged if r['merge_status']!='lookup_only'],merged_fields) counts={} for name,audit,kept in [('master',master_audit,master),('lookup',lookup_audit,lookup)]: counts[name+'_raw']=len(audit) counts[name+'_invalid']=sum(r['row_status'].startswith('invalid') for r in audit) counts[name+'_duplicate_removed']=sum(r['row_status']=='duplicate_removed' for r in audit) counts[name+'_conflict_rows']=sum(r['row_status']=='key_conflict' for r in audit) counts[name+'_clean']=len(kept) for status in ['matched','master_only','lookup_only']:counts[status]=sum(r['merge_status']==status for r in merged) counts.update(merged_all=len(merged),analysis_ready=len(master),missing_revenue=sum(r['revenue'] is None for r in master),missing_employees=sum(r['employees'] is None for r in master)) assert counts['matched']+counts['master_only']==len(master) save('counts.csv',[{'metric':k,'value':v} for k,v in counts.items()],['metric','value']) report={'simulated':True,'counts':counts,'rules':'strip surrounding whitespace; preserve text IDs and source rows; quarantine invalid keys/values; remove exact duplicate content; quarantine all conflicting keys; do not impute missing values; retain every merge status'} (out/'report.json').write_text(json.dumps(report,indent=2)) # 图只展示本次计算的数量；来源和冲突处理仍以逐行审计表为准。 labels=[('Master raw',counts['master_raw']),('Invalid',counts['master_invalid']),('Exact duplicates removed',counts['master_duplicate_removed']),('Conflicting rows held',counts['master_conflict_rows']),('Master clean',counts['master_clean']),('Matched observations',counts['matched']),('Master-only observations',counts['master_only']),('Lookup-only records',counts['lookup_only'])] svg=['Data cleaning and merge auditSimulated inputs · retain exceptions and unmatched records'] for i,(label,value) in enumerate(labels): y=139+i*53;svg.append(f'{label}{value}') svg.append('Merge union: 11 rows · analysis-ready master: 10 observations · no missing-value imputation') (out/'summary.svg').write_text(''.join(svg)) print(json.dumps({'stage':'complete','marker':'CLEAN_MERGE_COMPLETED','counts':counts}))