import argparse import json import re from decimal import Decimal from pathlib import Path from openpyxl import Workbook from openpyxl.styles import Font, PatternFill HEADERS = ["编号", "商品", "数量", "单价（元）", "金额（元）", "备注"] def safe_text(value, required=False): if value is None and not required: return None if not isinstance(value, str) or not value.strip() or len(value) > 200: raise ValueError("Text must be a short nonempty string or null") if value.lstrip().startswith(("=", "+", "-", "@")): raise ValueError("Formula-like text is rejected") return value def money(value): if value is None: return None if not isinstance(value, str) or not re.fullmatch(r"\d{1,8}\.\d{2}", value): raise ValueError("Amounts must be decimal strings with two places or null") return Decimal(value) def validate(data): if not isinstance(data, dict) or data.get("headers") != HEADERS or data.get("currency") != "CNY": raise ValueError("Expected the six documented headers and currency CNY") document_id = safe_text(data.get("document_id"), True) rows = data.get("rows") if not isinstance(rows, list) or not 1 <= len(rows) <= 500: raise ValueError("Local example accepts one to 500 rows") ids, output, issues, quantity_sum, amount_sum = set(), [], [], 0, Decimal("0.00") for index, row in enumerate(rows): if not isinstance(row, list) or len(row) != 6: raise ValueError(f"Row {index + 1} must contain six cells") row_id, product, quantity, unit_price, amount, note = row row_id = safe_text(row_id, True) if row_id in ids: raise ValueError("Duplicate row ID") ids.add(row_id) product = safe_text(product) note = safe_text(note) if note not in (None, "") else None if quantity is not None and (type(quantity) is not int or not 0 <= quantity <= 1_000_000): raise ValueError("Quantity must be a bounded integer or null") unit_price, amount = money(unit_price), money(amount) unknown = [HEADERS[i] for i, value in enumerate([row_id, product, quantity, unit_price, amount]) if value is None] if unknown: issues.append({"row_id": row_id, "reason": "Unreadable or missing required cells", "columns": unknown}) if quantity is not None and unit_price is not None and amount is not None: if (Decimal(quantity) * unit_price).quantize(Decimal("0.01")) != amount: issues.append({"row_id": row_id, "reason": "Quantity multiplied by unit price does not match amount", "columns": [HEADERS[2], HEADERS[3], HEADERS[4]]}) if quantity is not None: quantity_sum += quantity if amount is not None: amount_sum += amount output.append([row_id, product, quantity, unit_price, amount, note]) expected_quantity = data.get("total_quantity") if type(expected_quantity) is not int or expected_quantity < 0: raise ValueError("total_quantity must be an integer") expected_amount = money(data.get("total_amount")) if expected_amount is None: raise ValueError("total_amount must be a decimal string") if quantity_sum != expected_quantity: issues.append({"row_id": "TOTAL", "reason": "Visible total quantity differs from extracted rows", "columns": [HEADERS[2]]}) if amount_sum != expected_amount: issues.append({"row_id": "TOTAL", "reason": "Visible total amount differs from extracted rows", "columns": [HEADERS[4]]}) return document_id, output, issues, quantity_sum, amount_sum def main(): parser = argparse.ArgumentParser() parser.add_argument("json_file", type=Path) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--document-id") args = parser.parse_args() if args.json_file.stat().st_size > 2 * 1024 * 1024: raise ValueError("Local example JSON limit: 2 MiB") if args.output.exists(): raise ValueError("Output already exists") data = json.loads(args.json_file.read_text(encoding="utf-8")) if args.document_id is not None: safe_text(args.document_id, True) if not isinstance(data, dict): raise ValueError("Expected a JSON object") if data.get("document_id") not in (None, args.document_id): raise ValueError("Provided document ID conflicts with the JSON") data = {**data, "document_id": args.document_id} document_id, rows, issues, total_quantity, total_amount = validate(data) wb = Workbook() ws = wb.active ws.title = "识别待核对" ws.append(HEADERS) for row in rows: ws.append([float(value) if isinstance(value, Decimal) else value for value in row]) for cell in ws[1]: cell.fill = PatternFill("solid", fgColor="513074") cell.font = Font(color="FFFFFF", bold=True) for row in ws.iter_rows(min_row=2): row[0].number_format = "@" row[3].number_format = row[4].number_format = "0.00" for cell in row[:5]: if cell.value is None: cell.fill = PatternFill("solid", fgColor="F5E1A4") for col, width in {"A": 12, "B": 24, "C": 12, "D": 18, "E": 18, "F": 16}.items(): ws.column_dimensions[col].width = width ws.freeze_panes = "A2" ws.auto_filter.ref = f"A1:F{len(rows) + 1}" review = wb.create_sheet("待确认") review.append(["document_id", document_id]) review.append(["status", "needs_review" if issues else "arithmetic_checked_not_visually_accepted"]) review.append(["document_id_source", "provided_by_operator" if args.document_id is not None else "input_json"]) review.append(["row_id", "reason", "columns"]) for issue in issues: review.append([issue["row_id"], issue["reason"], ", ".join(issue["columns"])]) wb.save(args.output) print(json.dumps({"document_id": document_id, "document_id_source": "provided_by_operator" if args.document_id is not None else "input_json", "rows": len(rows), "issues": issues, "total_quantity": total_quantity, "total_amount": str(total_amount), "output": str(args.output), "visual_review_completed": False}, ensure_ascii=False)) if __name__ == "__main__": main()