"""Explicit XLSX mapping -> category procurement workbooks. No network access.""" import argparse import hashlib import json import re from decimal import Decimal, InvalidOperation from pathlib import Path from openpyxl import Workbook, load_workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter BASE = ['商品','单位','规格','选定供应商','上期价格','上期价格日期','最终价格','涨跌幅','采购数','小计','备注'] TAIL = ['最低价供应商','最低报价','并列最低供应商数'] def number(value, where): if value is None or str(value).strip() == '': return None if isinstance(value, bool): raise ValueError(f'{where}: boolean is not numeric') try: n = Decimal(str(value).strip()) except InvalidOperation: raise ValueError(f'{where}: invalid numeric cell') from None if not n.is_finite() or n < 0: raise ValueError(f'{where}: nonfinite/negative value requires review') return n def read_sources(config, root): unsupported = [p.name for p in root.rglob('*') if p.suffix.lower() in ('.xls','.xlsm','.csv','.pdf','.png','.jpg','.jpeg')] if unsupported: raise ValueError(f'unsupported sources require explicit conversion/review: {unsupported}') vendors = config['suppliers'] if not vendors or len(set(vendors)) != len(vendors): raise ValueError('suppliers must be unique and nonempty') cats, coverage, used, books = {}, [], set(), {} for src in config['sources']: path = (root / src['file']).resolve() if not path.is_relative_to(root.resolve()) or path.suffix.lower() != '.xlsx': raise ValueError('source must be an XLSX below input directory') sheet, vendor, cat = src['sheet'], src['supplier'], src['category'] if vendor not in vendors or not re.fullmatch(r'[^/\\.][^/\\]*', cat) or cat == '汇总': raise ValueError('invalid supplier/category') marker = (path, sheet) if marker in used: raise ValueError('sheet mapped more than once') used.add(marker) if path not in books: books[path] = (load_workbook(path, data_only=False), load_workbook(path, data_only=True)) raw, cached = books[path] ws, values = raw[sheet], cached[sheet] columns = src['columns'] if not {'item','unit','spec','price'} <= columns.keys(): raise ValueError('item/unit/spec/price mappings required') heads = [c.value for c in ws[src.get('header_row', 1)]] positions = {} for field, label in columns.items(): if heads.count(label) != 1: raise ValueError(f'{src["file"]}/{sheet}: header not unique: {label}') positions[field] = heads.index(label) + 1 records = cats.setdefault(cat, {}) count = valid = 0 for row in range(src.get('header_row', 1) + 1, ws.max_row + 1): if all(ws.cell(row, col).value is None for col in range(1, ws.max_column + 1)): continue data = {} for field, col in positions.items(): cell = ws.cell(row, col) value = values.cell(row, col).value if cell.data_type == 'e' or (cell.data_type == 'f' and value is None): raise ValueError(f'{src["file"]}/{sheet}/{cell.coordinate}: error or missing formula cache') data[field] = value key = tuple(str(data[f] or '').strip() for f in ('item','unit','spec')) if not key[0]: raise ValueError(f'{sheet}:{row}: nonempty row without item') item = records.setdefault(key, {'quotes': {}, 'quantity': None, 'previous': None, 'previous_date': None, 'notes': []}) if vendor in item['quotes']: raise ValueError(f'{cat}:{key}: duplicate supplier key; resolve explicitly') price = number(data['price'], f'{sheet}:{row}:price') item['quotes'][vendor] = price if price and price > 0 else None for field in ('quantity','previous'): n = number(data.get(field), f'{sheet}:{row}:{field}') if n is not None: if item[field] is not None and item[field] != n: raise ValueError(f'{cat}:{key}: conflicting {field}') item[field] = n if data.get('previous_date') is not None: date = str(data['previous_date']) if not isinstance(data['previous_date'], str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2}',date): raise ValueError('previous_date must be ISO YYYY-MM-DD text') if item['previous_date'] not in (None, date): raise ValueError('conflicting previous_date') item['previous_date'] = date if data.get('note'): item['notes'].append(f'{vendor}: {data["note"]}') count += 1 valid += bool(price) if count == 0: raise ValueError(f'{sheet}: empty source') coverage.append({'file': src['file'], 'sheet': sheet, 'supplier': vendor, 'category': cat, 'rows': count, 'valid_quotes': valid, 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()}) ignored = {( (root / s['file']).resolve(), s['sheet']) for s in config.get('ignored_sheets', []) if s.get('reason')} for path in (p for p in root.rglob('*') if p.suffix.lower() == '.xlsx'): path = path.resolve() book = books[path][0] if path in books else load_workbook(path, read_only=True) for sheet in book.sheetnames: if (path, sheet) not in used | ignored: raise ValueError(f'unmapped sheet: {path.name}/{sheet}') if not cats: raise ValueError('no sources') return cats, coverage def calculate(config, cats): vendors = config['suppliers'] reports = [] for cat, items in cats.items(): counts = dict.fromkeys(vendors, 0) valid = dict.fromkeys(vendors, 0) minima = {} for key, item in items.items(): quotes = item['quotes'] prices = [v for v in quotes.values() if v is not None] low = min(prices) if prices else None winners = [v for v in vendors if low is not None and quotes.get(v) == low] for v in winners: counts[v] += 1 for v in vendors: valid[v] += quotes.get(v) is not None minima[key] = (low, winners) active = [v for v in vendors if valid[v]] leaders = [v for v in active if counts[v] == max(counts.values())] chosen = leaders[0] if len(leaders) == 1 else None if len(leaders) > 1: rule = config.get('tie_break', {}).get(cat) if rule not in leaders: raise ValueError(f'{cat}: category tie {leaders}; explicit tie_break required') chosen = rule details, missing = [], [] for key, item in items.items(): low, winners = minima[key] final = item['quotes'].get(chosen) quantity, previous = item['quantity'], item['previous'] subtotal = final * quantity if final is not None and quantity is not None else None change = (final - previous) / previous if final is not None and previous else None absent = [v for v in vendors if v not in item['quotes']] if absent: missing.append({'key': key, 'missing_suppliers': absent}) note = list(item['notes']) if final is None: note.append('默认供应商本项无报价' if chosen else '全类无有效报价') details.append([*key, chosen, previous, item['previous_date'], final, change, quantity, subtotal, '；'.join(note) or None, *[item['quotes'].get(v) for v in vendors], '、'.join(winners) or None, low, len(winners)]) reports.append({'category': cat, 'selected': chosen, 'counts': counts, 'valid': valid, 'leaders': leaders, 'status': '单家报价' if len(active) == 1 else ('无有效报价' if not active else '多家比价'), 'details': details, 'missing': missing}) return reports def write_sheet(ws, rows): for row in rows: ws.append([float(v) if isinstance(v, Decimal) else v for v in row]) for cells in ws: for c in cells: if isinstance(c.value, str): c.data_type = 's' # Preserve literal text; never create formulas from supplier text. c.font = Font(name='Arial', size=10, bold=c.row == 1) c.alignment = Alignment(vertical='center', wrap_text=True) c.border = Border(*( [Side(style='thin', color='BFBFBF')] * 4)) if c.row == 1: c.fill = PatternFill('solid', fgColor='D9E2F3') elif isinstance(c.value, (float, int)): c.number_format = '#,##0.00' ws.freeze_panes = 'D2' ws.auto_filter.ref = ws.dimensions ws.row_dimensions[1].height = 34 for col in range(1, ws.max_column + 1): ws.column_dimensions[get_column_letter(col)].width = 24 if col in (1,4,11) else 16 def run(config_file, input_dir, output_dir): config = json.loads(Path(config_file).read_text()) root, output = Path(input_dir).resolve(), Path(output_dir).resolve() if output == root or output.is_relative_to(root): raise ValueError('output must be outside input tree') cats, coverage = read_sources(config, root) reports = calculate(config, cats) output.mkdir(parents=True, exist_ok=True) if list(output.iterdir()): raise ValueError('output must be empty; use a new version directory') vendors = config['suppliers'] summaries = [['分类','默认供应商','状态','最高次数','分类并列','商品数', *[v+'最低次数' for v in vendors]]] for r in reports: wb = Workbook(); ws = wb.active; ws.title = '比较结果' write_sheet(ws, [BASE + vendors + TAIL, *r['details']]) for cell in ws['H'][1:]: cell.number_format = '0.00%' stats = wb.create_sheet('最低价统计') write_sheet(stats, [['供应商','最低次数','有效报价数','默认供应商','状态','分类并列'], *[[v,r['counts'][v],r['valid'][v],r['selected'],r['status'],'、'.join(r['leaders']) if len(r['leaders']) > 1 else None] for v in vendors]]) wb.save(output / (r['category']+'.xlsx')) summaries.append([r['category'],r['selected'],r['status'],max(r['counts'].values()), '、'.join(r['leaders']) if len(r['leaders']) > 1 else None,len(r['details']), *r['counts'].values()]) wb = Workbook(); wb.active.title = '分类最低价最多供应商'; write_sheet(wb.active, summaries); wb.save(output/'汇总.xlsx') (output/'coverage.json').write_text(json.dumps({'sources':coverage,'missing':{r['category']:r['missing'] for r in reports}},ensure_ascii=False,indent=2)) return reports if __name__ == '__main__': p = argparse.ArgumentParser(); p.add_argument('config'); p.add_argument('input'); p.add_argument('output'); a = p.parse_args() run(a.config, a.input, a.output) print('Generated; independent verification is still required.')