"""Independent OOXML source reader and oracle; deliberately does not import compare.""" import argparse import hashlib import json import posixpath import re import zipfile from decimal import Decimal from pathlib import Path from xml.etree import ElementTree as ET from openpyxl import load_workbook NS = {'s':'http://schemas.openxmlformats.org/spreadsheetml/2006/main'} def raw_rows(path, sheet, allowed_na_column=None): with zipfile.ZipFile(path) as z: strings = [] if 'xl/sharedStrings.xml' in z.namelist(): strings = [''.join(e.itertext()) for e in ET.fromstring(z.read('xl/sharedStrings.xml'))] relationships = {r.attrib['Id']:r.attrib['Target'] for r in ET.fromstring(z.read('xl/_rels/workbook.xml.rels'))} sheets = ET.fromstring(z.read('xl/workbook.xml')).find('s:sheets', NS) target = next(s for s in sheets if s.attrib['name'] == sheet) rel = target.attrib['{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id'] part = relationships[rel] part = part.lstrip('/') if part.startswith('/') else posixpath.normpath('xl/'+part) result = {} for row in ET.fromstring(z.read(part)).findall('.//s:sheetData/s:row', NS): cells = {} for cell in row: col = re.match(r'[A-Z]+',cell.attrib['r'])[0] typ = cell.get('t'); v = cell.find('s:v', NS) value = v.text if v is not None else None if (typ == 'e' and not (value == '#N/A' and col == allowed_na_column)) or (cell.find('s:f',NS) is not None and value is None): raise AssertionError('source error/missing cache') if typ == 'inlineStr': value = ''.join(n.text or '' for n in cell.findall('.//s:t',NS)) elif typ == 's': value = strings[int(value)] elif typ == 'b': raise AssertionError('boolean source requires review') cells[col] = value result[int(row.attrib['r'])] = cells return result def dec(value): if value is None or str(value).strip() == '': return None d = Decimal(str(value).strip()) assert d.is_finite() and d >= 0, 'invalid numeric source' return d def eq(actual, expected): if isinstance(expected, Decimal): assert actual is not None and abs(Decimal(str(actual))-expected) <= Decimal('1e-12')*max(Decimal(1),abs(expected)), (actual, str(expected)) else: assert actual == (None if expected == '' else expected), (actual, expected) def verify(config_file, input_dir, output_dir): cfg = json.loads(Path(config_file).read_text()); root = Path(input_dir); out = Path(output_dir) vendors = cfg['suppliers']; categories = {}; source_evidence = [] mapped={(str(Path(s['file'])),s['sheet']) for s in cfg['sources']} assert len(mapped)==len(cfg['sources']), 'duplicate source mapping' ignored={(str(Path(s['file'])),s['sheet']) for s in cfg.get('ignored_sheets',[]) if s.get('reason')} for p in (f for f in root.rglob('*') if f.suffix.lower() == '.xlsx'): with zipfile.ZipFile(p) as z: names=[s.attrib['name'] for s in ET.fromstring(z.read('xl/workbook.xml')).find('s:sheets',NS)] assert all((str(p.relative_to(root)),name) in mapped|ignored for name in names), 'unmapped source sheet' for s in cfg['sources']: records = raw_rows(root/s['file'], s['sheet']) header = records[s.get('header_row',1)] mapping = {} for field, title in s['columns'].items(): matches = [c for c, v in header.items() if v == title] assert len(matches) == 1, 'ambiguous header' mapping[field] = matches[0] n = valid = 0 for rownum, cells in records.items(): if rownum <= s.get('header_row',1) or all(v is None for v in cells.values()): continue values = {field:cells.get(col) for field,col in mapping.items()} key = tuple((values[f] or '').strip() for f in ('item','unit','spec')) assert key[0], 'missing item' quotes = categories.setdefault(s['category'],{}).setdefault(key,{}) assert s['supplier'] not in quotes, 'duplicate supplier key' price = dec(values['price']); price = price if price else None quotes[s['supplier']] = (price, dec(values.get('quantity')),dec(values.get('previous')),values.get('previous_date'),values.get('note')) n += 1; valid += price is not None source_evidence.append({'file':s['file'],'sheet':s['sheet'],'supplier':s['supplier'],'category':s['category'], 'rows':n,'valid_quotes':valid,'sha256':hashlib.sha256((root/s['file']).read_bytes()).hexdigest()}) audit = json.loads((out/'coverage.json').read_text()) assert audit['sources'] == source_evidence, 'coverage/hash mismatch' assert {p.name for p in out.glob('*.xlsx')} == {c+'.xlsx' for c in categories} | {'汇总.xlsx'} summary_book = load_workbook(out/'汇总.xlsx',data_only=False) assert summary_book.sheetnames == ['分类最低价最多供应商'] summary = summary_book.active assert list(next(summary.values)) == ['分类','默认供应商','状态','最高次数','分类并列','商品数',*[v+'最低次数' for v in vendors]] assert summary.max_row == len(categories)+1 for ci,(category,items) in enumerate(categories.items(),2): counts = {v:0 for v in vendors}; valid = {v:0 for v in vendors}; minima = {} missing = [] for key,quotes in items.items(): available = [values[0] for values in quotes.values() if values[0] is not None] low = sorted(available)[0] if available else None winners = [v for v in vendors if v in quotes and low is not None and quotes[v][0] == low] for v in winners: counts[v] += 1 for v in vendors: valid[v] += v in quotes and quotes[v][0] is not None minima[key] = (low,winners) absent = [v for v in vendors if v not in quotes] if absent: missing.append({'key':list(key),'missing_suppliers':absent}) assert audit['missing'][category] == missing, 'missing/new-item coverage mismatch' active = [v for v in vendors if valid[v]] leaders = [v for v in active if counts[v] == max(counts.values())] selected = leaders[0] if len(leaders) == 1 else cfg.get('tie_break',{}).get(category) assert not leaders or selected in leaders, 'unresolved tie' status = '单家报价' if len(active)==1 else ('多家比价' if active else '无有效报价') tie = '、'.join(leaders) if len(leaders)>1 else None book = load_workbook(out/(category+'.xlsx'), data_only=False) assert book.sheetnames == ['比较结果','最低价统计'] ws = book['比较结果']; stats=book['最低价统计'] expected_header = ['商品','单位','规格','选定供应商','上期价格','上期价格日期','最终价格','涨跌幅','采购数','小计','备注',*vendors,'最低价供应商','最低报价','并列最低供应商数'] assert list(next(ws.values)) == expected_header assert ws.max_row == len(items)+1 seen=set() for row in list(ws.values)[1:]: key=tuple(v or '' for v in row[:3]); assert key in items and key not in seen; seen.add(key) quotes=items[key]; low,winners=minima[key] quantities={q[1] for q in quotes.values() if q[1] is not None}; assert len(quantities)<=1 previous={q[2] for q in quotes.values() if q[2] is not None}; assert len(previous)<=1 dates={q[3] for q in quotes.values() if q[3] is not None}; assert len(dates)<=1 quantity=next(iter(quantities),None); prev=next(iter(previous),None) final=quotes.get(selected,(None,))[0] notes=[f'{v}: {q[4]}' for v,q in quotes.items() if q[4]] if final is None: notes.append('默认供应商本项无报价' if selected else '全类无有效报价') expected=[*key,selected,prev,next(iter(dates),None),final,(final-prev)/prev if final is not None and prev else None, quantity,final*quantity if final is not None and quantity is not None else None,'；'.join(notes) or None, *[quotes.get(v,(None,))[0] for v in vendors],'、'.join(winners) or None,low,len(winners)] for a,e in zip(row,expected): eq(a,e) assert stats.max_row == len(vendors)+1 for i,v in enumerate(vendors,2): assert [c.value for c in stats[i]] == [v,counts[v],valid[v],selected,status,tie] assert [c.value for c in summary[ci]] == [category,selected,status,max(counts.values()),tie,len(items),*counts.values()] for ws in book: assert not any(c.data_type in ('f','e') for row in ws for c in row), 'unexpected formula/error' return {'verified':True,'categories':len(categories),'sources':len(source_evidence),'rows':sum(map(len,categories.values())), 'oracle':'raw OOXML source cells, independent of analysis and writer'} if __name__ == '__main__': p=argparse.ArgumentParser(); p.add_argument('config'); p.add_argument('input'); p.add_argument('output'); a=p.parse_args() print(json.dumps(verify(a.config,a.input,a.output),ensure_ascii=False,indent=2))