"""Raw uploaded workbooks -> detected mapping -> audited category comparison.""" import argparse import hashlib import json import re from pathlib import Path from openpyxl import Workbook, load_workbook from compare import calculate, read_sources, run, write_sheet, number from verify import verify from verify_intake import verify_intake from delivery import build as build_delivery ALIASES = { 'item': ['商品','商品名称','物料','物料名称','品名','产品名称'], 'price': ['单价','报价','价格','含税单价','供货价'], 'quantity': ['数量','采购数','采购数量','采购量'], 'unit': ['单位','计量单位'], 'spec': ['规格','规格型号','包装规格'], 'note': ['备注','说明'], 'supplier': ['供应商','供应商名称','商户'], 'category': ['分类','品类','类别','商品分类'], 'previous': ['上期价格','上期单价'], 'previous_date': ['上期价格日期'], } FIELDS = ['item','unit','spec','price','quantity','note','previous','previous_date'] HEADERS = ['商品','单位','规格','报价','数量','备注','上期价格','上期价格日期'] GENERIC = {'明细','报价','报价单','sheet1','sheet2','sheet3','工作表1'} def clean_category(value): text = re.sub(r'\s+', '', str(value or '')) text = re.sub(r'\d+$', '', text) for token in ('询比价','报价单','报价','询价单','询价'): text = text.replace(token, '') text = text.strip(' _-—类') text = re.sub(r'补充$', '', text).strip(' _-—类') if re.fullmatch(r'(?:sheet|工作表)\d+',text,re.I): return None return text if text and text.lower() not in GENERIC else None def detect(ws): candidates = [] for row in ws.iter_rows(min_row=1,max_row=min(30,ws.max_row)): labels = [str(c.value or '').strip() for c in row] mapping = {} for field, names in ALIASES.items(): hits = [i+1 for i, value in enumerate(labels) if value in names] if len(hits) > 1: mapping[field] = None elif hits: mapping[field] = hits[0] if mapping.get('item') and mapping.get('price'): if any(v is None for v in mapping.values()): raise ValueError('ambiguous duplicate headers') candidates.append((row[0].row, mapping)) if len(candidates) != 1: raise ValueError('cannot uniquely identify item/price header') return candidates[0] def unique_value(ws, col, first_row): vals = {str(ws.cell(n,col).value).strip() for n in range(first_row+1,ws.max_row+1) if ws.cell(n,col).value is not None} if len(vals) != 1: raise ValueError('multiple/empty supplier or category column; split or map explicitly') return vals.pop() def execute(input_dir, output_dir, overrides=None): root, output = Path(input_dir).resolve(), Path(output_dir).resolve() if not root.is_dir() or output == root or output.is_relative_to(root): raise ValueError('input directory required; output must be outside input tree') if output.exists(): raise ValueError('use a new output directory') output.mkdir(parents=True) work = output/'核验依据'; work.mkdir() normalized = work/'normalized'; normalized.mkdir() overrides = overrides or {} entries, issues, sources, vendors = [], [], [], [] # Only uploaded top-level workbooks, never descend into old output directories. inputs = sorted(p for p in root.iterdir() if p.is_file() and not p.name.startswith('~$') and p.suffix.lower() in ('.xlsx','.xlsm','.xls')) if not inputs: raise ValueError('no Excel inputs') for path in inputs: if path.suffix.lower() == '.xls': issues.append({'file':path.name,'sheet':None,'category':None,'status':'blocked','reason':'legacy .xls needs lossless conversion and review'}) continue raw = load_workbook(path,data_only=False); cached = load_workbook(path,data_only=True) for ws in raw: identity = path.name+'::'+ws.title override = overrides.get('sheets',{}).get(identity,{}) entry = {'file':path.name,'sheet':ws.title,'sha256':hashlib.sha256(path.read_bytes()).hexdigest(),'category':None,'status':'blocked'} entries.append(entry) try: if override.get('ignore_reason'): entry.update(status='ignored',reason=override['ignore_reason']);continue header, columns = detect(ws) parts = [p.strip() for p in re.split(r'[-_—]+',path.stem) if p.strip()] supplier = override.get('supplier') or (unique_value(cached[ws.title],columns['supplier'],header) if 'supplier' in columns else (parts[0] if len(parts)>1 else None)) category = override.get('category') or (unique_value(cached[ws.title],columns['category'],header) if 'category' in columns else None) if not category: category = clean_category(ws.title) if not category and len(parts)>1: category = next((c for token in reversed(parts[1:]) if (c:=clean_category(token))),None) if not supplier or not category: raise ValueError('supplier/category ambiguous; use source evidence to set sheet override') category = overrides.get('category_aliases',{}).get(category,category) if re.search(r'[\\/:*?"<>|]',category) or category in ('.','..','汇总'): raise ValueError('unsafe/reserved category name') entry.update(supplier=supplier,category=category,header_row=header,columns=columns) if supplier not in vendors: vendors.append(supplier) wb=Workbook(); dest=wb.active; dest.title='报价';dest.append(HEADERS) row_map=[]; events=[] for rownum in range(header+1,ws.max_row+1): if all(c.value is None for c in ws[rownum]): continue values={} for field,col in columns.items(): cell=ws.cell(rownum,col); value=cached[ws.title].cell(rownum,col).value if field=='price' and value=='#N/A': value=None events.append({'row':rownum,'cell':cell.coordinate,'raw':'#N/A','action':'missing_quote'}) elif cell.data_type=='e' or (cell.data_type=='f' and value is None): raise ValueError(f'{cell.coordinate}: source error or missing formula cache') if field in ('price','quantity','previous'): number(value,f'{cell.coordinate}:{field}') values[field]=value if values.get('item') is None or not str(values['item']).strip(): raise ValueError(f'row {rownum}: nonempty row without item') values['item']=str(values['item']).strip() if 'spec' not in columns: matches=re.findall(r'【([^】]+)】',values['item']);values['spec']=matches[-1] if matches else None # No independent unit column -> blank, never infer or convert. if events and events[-1]['row']==rownum: note=f'原始报价#N/A按缺报价留空（{path.name}/{ws.title}/{events[-1]["cell"]}）' values['note']='；'.join(str(v) for v in (values.get('note'),note) if v) dest.append([values.get(f) for f in FIELDS]) for cell in dest[dest.max_row]: if isinstance(cell.value,str):cell.data_type='s' row_map.append({'original_row':rownum,'normalized_row':dest.max_row}) if not row_map: raise ValueError('empty quote sheet') filename=f'source-{len(sources)+1:04d}.xlsx'; wb.save(normalized/filename) entry.update(status='normalized',normalized_file=filename,row_map=row_map,events=events,rows=len(row_map)) sources.append({'file':filename,'sheet':'报价','supplier':supplier,'category':category,'columns':dict(zip(FIELDS,HEADERS))}) except (ValueError,KeyError) as exc: entry['reason']=str(exc) issues.append({'file':path.name,'sheet':ws.title,'category':entry.get('category'),'status':'blocked','reason':str(exc)}) (work/'intake.json').write_text(json.dumps({'entries':entries,'issues':issues},ensure_ascii=False,indent=2)) input_proof = verify_intake(root,normalized,entries) # Any failed source with a known category blocks that entire category, avoiding partial-vendor decisions. blocked={i['category'] for i in issues if i.get('category')} # Unclassified source could belong to any category: no selections until mapping is resolved. unknown=any(not i.get('category') for i in issues) config={'suppliers':vendors,'sources':sources,'tie_break':overrides.get('tie_break',{}),'ignored_sheets':[]} for category in dict.fromkeys(s['category'] for s in sources): if category in blocked: continue subset={**config,'sources':[s for s in sources if s['category']==category]} subset['ignored_sheets']=[{'file':s['file'],'sheet':'报价','reason':'other category preflight'} for s in sources if s['category']!=category] try: cats,_=read_sources(subset,normalized);calculate(subset,cats) except ValueError as exc: blocked.add(category);issues.append({'file':None,'sheet':None,'category':category,'status':'blocked','reason':str(exc)}) if unknown: blocked.update(s['category'] for s in sources) config['sources']=[s for s in sources if s['category'] not in blocked] config['ignored_sheets']=[{'file':s['file'],'sheet':'报价','reason':'blocked category; original records retained in audit'} for s in sources if s['category'] in blocked] config_path=work/'config.json';config_path.write_text(json.dumps(config,ensure_ascii=False,indent=2)) reports=[]; verified={'verified':False,'reason':'no complete category'} if config['sources']: reports=run(config_path,normalized,output/'比价结果') verified=verify(config_path,normalized,output/'比价结果') review=Workbook();ws=review.active;ws.title='处理记录' rows=[['来源文件','Sheet','分类','状态','说明']] rows.extend([[i.get('file'),i.get('sheet'),i.get('category'),'待确认',i['reason']] for i in issues]) rows.extend([[e['file'],e['sheet'],e['category'],'已按缺报价处理',f'{v["cell"]}: #N/A → 留空；原始文件未修改'] for e in entries for v in e.get('events',[])]) if len(rows)==1:rows.append([None,None,None,'通过','无待确认事项']) write_sheet(ws,rows) # Keep every blocked-category row visible, including duplicate records, without using row order as identity. ws=review.create_sheet('未出表分类原始行');rows=[['分类','供应商','文件','Sheet','原始行',*HEADERS]] for e in entries: if e.get('normalized_file') and e['category'] in blocked: sheet=load_workbook(normalized/e['normalized_file'],data_only=True).active for m in e['row_map']: rows.append([e['category'],e['supplier'],e['file'],e['sheet'],m['original_row'],*[c.value for c in sheet[m['normalized_row']]]]) write_sheet(ws,rows) ws=review.create_sheet('来源识别清单') write_sheet(ws,[['文件','Sheet','供应商','分类','表头行','读取行数','识别状态'], *[[e['file'],e['sheet'],e.get('supplier'),e.get('category'),e.get('header_row'),e.get('rows'),e['status']] for e in entries]]) review.save(work/'处理与待确认.xlsx') state={'status':'partial' if issues else 'complete','raw_workbooks':len(inputs),'raw_sheets':len(entries),'completed_categories':[r['category'] for r in reports], 'blocked_categories':sorted(blocked),'issues':issues,'na_as_missing':sum(len(e.get('events',[])) for e in entries), 'input_verification':input_proof,'output_verification':verified} for e in entries: assert hashlib.sha256((root/e['file']).read_bytes()).hexdigest()==e['sha256'],'source changed during processing' (work/'内部处理说明.md').write_text('# 原始 Excel 自动比价结果\n\n状态：'+state['status']+'。已完成分类：'+('、'.join(state['completed_categories']) or '无')+'。\n\n待确认分类：'+('、'.join(state['blocked_categories']) or '无')+'。请查看“处理与待确认.xlsx”，未出表分类的原始行和全部原始文件均保留。\n\n报价列 #N/A 已按无有效报价留空并保留来源；其他错误不猜价。无单位列则留空。最终结果保留小计公式，修改采购数由Excel重算；修改报价或选商规则需重跑。\n\n识别依据与逐行回读证据在“核验依据”，采购数据不可提交共享Git。\n') if reports: state['delivery']=build_delivery(output/'比价结果',output/'结果文件') (work/'result.json').write_text(json.dumps(state,ensure_ascii=False,indent=2)) return state if __name__=='__main__': p=argparse.ArgumentParser();p.add_argument('input');p.add_argument('output');p.add_argument('--overrides');a=p.parse_args() print(json.dumps(execute(a.input,a.output,json.loads(Path(a.overrides).read_text()) if a.overrides else None),ensure_ascii=False,indent=2))