from __future__ import annotations import hashlib import re import zipfile from dataclasses import dataclass, field from datetime import date, datetime from decimal import Decimal from pathlib import Path from typing import Iterable from fastapi import HTTPException from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.pdfbase import pdfmetrics from reportlab.pdfbase.cidfonts import UnicodeCIDFont from reportlab.pdfgen import canvas from sqlalchemy import case, func, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, aliased from app.models.document_archive import DocumentArchive from app.models.master_data import Item, Warehouse from app.models.org import Employee, User from app.models.operations import ( CompletionReceipt, InventoryTxn, ProductionBatchLedger, ProductionBatchLedgerTxn, PurchaseOrder, PurchaseOrderItem, PurchaseReceipt, PurchaseReceiptItem, StockLot, Supplier, WarehouseLocation, WorkOrder, ) from app.models.sales import Customer, SalesOrder, SalesOrderItem from app.schemas.document_archives import DocumentArchiveGenerateResult DOCUMENT_TYPE_SALES_ORDER = "销售订单" DOCUMENT_TYPE_PURCHASE_ORDER = "采购订单" DOCUMENT_TYPE_PURCHASE_RECEIPT = "到货入库单" DOCUMENT_TYPE_QUALITY_INSPECTION = "质量校验单" DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT = "生产领料出库单" DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT = "生产入库结算单" DOCUMENT_TYPE_WAREHOUSE_OPERATION = "仓库出入库单" ARCHIVE_STATUS_READY = "已归档" ARCHIVE_STATUS_FAILED = "归档失败" ARCHIVE_STATUS_MISSING = "未生成" TEMPLATE_VERSION = "单据纸面V1" FILE_FORMAT_PDF = "PDF" ARCHIVE_TABLE_X = 42 ARCHIVE_TABLE_RIGHT_MARGIN = 42 ARCHIVE_TABLE_TOP_Y_OFFSET = 190 ARCHIVE_TABLE_BOTTOM_Y = 132 ARCHIVE_MIN_ROW_HEIGHT = 24 ARCHIVE_CELL_HORIZONTAL_PADDING = 8 ARCHIVE_CELL_VERTICAL_PADDING = 8 ARCHIVE_CELL_FONT_SIZE = 8 ARCHIVE_METADATA_FONT_SIZE = 10 ARCHIVE_METADATA_LEFT_X = 42 ARCHIVE_METADATA_RIGHT_X = 340 ARCHIVE_METADATA_COLUMN_GAP = 10 ARCHIVE_METADATA_LEFT_WIDTH = ARCHIVE_METADATA_RIGHT_X - ARCHIVE_METADATA_LEFT_X - ARCHIVE_METADATA_COLUMN_GAP ARCHIVE_METADATA_RIGHT_WIDTH = A4[0] - ARCHIVE_TABLE_RIGHT_MARGIN - ARCHIVE_METADATA_RIGHT_X ARCHIVE_METADATA_TOP_Y_OFFSET = 92 ARCHIVE_METADATA_MIN_ROW_HEIGHT = 20 ARCHIVE_METADATA_ROW_PADDING = 7 ARCHIVE_METADATA_TABLE_GAP = 18 ARCHIVE_STATUS_LABELS = { "ACTIVE": "启用", "ENABLED": "启用", "INACTIVE": "停用", "DISABLED": "停用", "OPEN": "未开始", "NEW": "新建", "DRAFT": "草稿", "SUBMITTED": "已提交", "APPROVED": "已审批", "PARTIAL": "部分完成", "PARTIAL_RECEIVED": "部分收货", "CLOSED": "已关闭", "CANCELED": "已取消", "CANCELLED": "已取消", "ORDERED": "已下单", "RECEIVED": "已收货", "RELEASED": "已下达", "IN_PROGRESS": "进行中", "READY_TO_RECEIPT": "待成品入库", "COMPLETED": "已完工", "SETTLED": "已结单", "DONE": "已完成", "PENDING": "待处理", "PENDING_QC": "待质检", "PASS": "已放行", "PASSED": "已放行", "REJECT": "已拒收", "REJECTED": "已拒收", "POSTED": "已过账", "AVAILABLE": "可用", "LOCKED": "已锁单", "DEPLETED": "已耗尽", "CONFIRMED": "已确认", "OVERRIDDEN": "已覆盖", "CREATED": "已创建", "IDLE": "空闲", "READY": "就绪", "ERROR": "异常", "FAILED": "失败", "EXPORTED": "已导出", "IMPORTED": "已导入", "SUPERSEDED": "已替代", "MINIAPP_ONLY": "小程序清单", "HAS_RETURN": "存在退货", } PRODUCTION_INBOUND_TXN_LABELS = { "FG_IN": "成品入库", "PRODUCTION_FINISHED_IN": "成品入库", "PRODUCTION_SURPLUS_IN": "生产余料入库", "PRODUCTION_SCRAP_IN": "生产废料入库", } WAREHOUSE_OPERATION_TXN_LABELS = { "CUSTOMER_SUPPLIED": "客料入库", "CUSTOMER_SUPPLIED_IN": "客料入库", "OPENING_IN": "期初入库", "PURCHASE_IN": "采购入库", "PRODUCTION_SURPLUS_IN": "生产余料入库", "OUTSOURCING_SURPLUS_IN": "委外余料入库", "PRODUCTION_SCRAP_IN": "生产废料入库", "REWORK_SCRAP_IN": "返工废料入库", "OUTSOURCING_SCRAP_IN": "委外废料入库", "RETURN_SCRAP_IN": "退货废料入库", "WIP_IN": "产中入库", "WIP_OUT": "产中出库", "OUTSOURCING_IN": "委外入库", "OUTSOURCING_OUT": "委外出库", "PRODUCTION_FINISHED_IN": "生产入库", "FG_IN": "成品入库", "REWORK_FINISHED_IN": "返工入库", "PRODUCTION_OUT": "生产出库", "MATERIAL_ISSUE": "生产领料出库", "SCRAP_SALE_OUT": "售卖出库", "PURCHASE_RETURN_OUT": "退货出库", "SPECIAL_IN": "特殊入库", "SPECIAL_OUT": "特殊出库", "SALES_OUT": "销售出库", "SCRAP_IN": "废料入库", "SCRAP_OUT": "废料出库", "RETURN_IN": "退货入库", "RETURN_OUT": "退货出库", "AUX_IN": "辅料入库", "AUX_OUT": "辅料出库", "RAW_IN": "原材料入库", "RAW_OUT": "原材料出库", "SEMI_FINISHED_IN": "半成品入库", "SEMI_FINISHED_OUT": "半成品出库", "FINISHED_IN": "成品入库", "FINISHED_OUT": "成品出库", "TRANSFER_IN": "调拨入库", "TRANSFER_OUT": "调拨出库", "STOCKTAKE_GAIN": "盘盈入库", "STOCKTAKE_LOSS": "盘亏出库", } PRODUCTION_DOCUMENT_BATCH_PREFIX = "单据批次号：" def production_document_batch_no_from_text(value: str | None) -> str | None: text = str(value or "").strip() if not text: return None for part in re.split(r"[；;]", text): current = part.strip() if current.startswith(PRODUCTION_DOCUMENT_BATCH_PREFIX): batch_no = current[len(PRODUCTION_DOCUMENT_BATCH_PREFIX):].strip() return batch_no or None return None @dataclass(slots=True) class ArchiveLine: line_no: int item_code: str item_name: str specification: str | None quantity: Decimal | float | int | None delivered_or_received_quantity: Decimal | float | int | None unit_price: Decimal | float | int | None line_amount: Decimal | float | int | None promised_or_expected_date: date | datetime | str | None = None remark: str | None = None @dataclass(slots=True) class ArchiveContext: document_type: str business_id: int document_no: str title: str partner_label: str partner_name: str document_date: date | datetime | str | None due_date_label: str due_date: date | datetime | str | None status: str tax_rate: Decimal | float | int | None total_amount: Decimal | float | int | None contact_name: str | None = None contact_phone: str | None = None address_label: str = "地址" address: str | None = None remark: str | None = None lines: list[ArchiveLine] = field(default_factory=list) prepared_by: str | None = None operator_name: str | None = None def default_archive_root() -> Path: return Path(__file__).resolve().parents[2] / "storage" / "document_archives" def safe_filename(value: str) -> str: safe = re.sub(r'[\\/:*?"<>|\s]+', "_", value.strip()) safe = re.sub(r"_+", "_", safe).strip("._") return safe or "document" def sha256_file(path: str | Path) -> str: digest = hashlib.sha256() with Path(path).open("rb") as file: for chunk in iter(lambda: file.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def decimal_text(value: Decimal | float | int | None, places: int = 2) -> str: if value is None: return "" number = Decimal(str(value)) quant = Decimal("1") if places <= 0 else Decimal("1").scaleb(-places) text = f"{number.quantize(quant):f}" return text.rstrip("0").rstrip(".") if "." in text else text def date_text(value: date | datetime | str | None) -> str: if value is None: return "" if isinstance(value, datetime): return value.strftime("%Y-%m-%d") if isinstance(value, date): return value.strftime("%Y-%m-%d") return str(value) def archive_status_text(value: str | None) -> str: text = str(value or "").strip() if not text: return "" normalized = text.upper() if normalized in ARCHIVE_STATUS_LABELS: return ARCHIVE_STATUS_LABELS[normalized] if re.fullmatch(r"[A-Z0-9_]+", text): return "未识别状态" return text def resolve_archive_user_label(db: Session, user_id: int | None) -> str | None: if user_id is None: return None row = db.execute( select(User, Employee) .outerjoin(Employee, Employee.id == User.employee_id) .where(User.id == user_id) ).first() if row is None: return None user, employee = row for value in (user.nickname, employee.employee_name if employee else None, user.username): text = str(value or "").strip() if text: return text return None def apply_archive_operator_context(db: Session, context: ArchiveContext, created_by: int | None) -> ArchiveContext: operator_label = resolve_archive_user_label(db, created_by) if operator_label: context.prepared_by = operator_label context.operator_name = context.operator_name or operator_label return context def get_latest_archive(db: Session, document_type: str, business_id: int) -> DocumentArchive | None: return db.execute( select(DocumentArchive) .where( DocumentArchive.document_type == document_type, DocumentArchive.business_id == business_id, ) .order_by(DocumentArchive.archive_version.desc(), DocumentArchive.id.desc()) .limit(1) ).scalar_one_or_none() def next_archive_version(db: Session, document_type: str, business_id: int) -> int: latest_version = db.execute( select(func.max(DocumentArchive.archive_version)).where( DocumentArchive.document_type == document_type, DocumentArchive.business_id == business_id, DocumentArchive.file_format == FILE_FORMAT_PDF, ) ).scalar_one() return int(latest_version or 0) + 1 def next_available_archive_version( db: Session, document_type: str, business_id: int, preferred_version: int, ) -> int: version = max(1, int(preferred_version)) used_versions = set( db.execute( select(DocumentArchive.archive_version).where( DocumentArchive.document_type == document_type, DocumentArchive.business_id == business_id, DocumentArchive.file_format == FILE_FORMAT_PDF, DocumentArchive.archive_version >= version, ) ).scalars() ) while version in used_versions: version += 1 return version def collect_sales_order_archive_context(db: Session, sales_order_id: int) -> ArchiveContext: row = db.execute( select(SalesOrder, Customer) .join(Customer, Customer.id == SalesOrder.customer_id) .where(SalesOrder.id == sales_order_id) ).one_or_none() if row is None: raise HTTPException(status_code=404, detail="销售订单不存在") order, customer = row item_rows = db.execute( select(SalesOrderItem, Item) .join(Item, Item.id == SalesOrderItem.product_item_id) .where(SalesOrderItem.sales_order_id == sales_order_id) .order_by(SalesOrderItem.line_no) ).all() lines = [ ArchiveLine( line_no=order_item.line_no, item_code=item.item_code, item_name=item.item_name, specification=item.specification, quantity=order_item.order_qty, delivered_or_received_quantity=order_item.delivered_qty, unit_price=order_item.unit_price, line_amount=order_item.line_amount, promised_or_expected_date=order_item.promised_date, remark=order_item.customer_part_no, ) for order_item, item in item_rows ] return ArchiveContext( document_type=DOCUMENT_TYPE_SALES_ORDER, business_id=order.id, document_no=order.order_no, title="销售订单归档", partner_label="客户", partner_name=customer.customer_name, document_date=order.order_date, due_date_label="承诺日期", due_date=order.promised_date, status=archive_status_text(order.status), tax_rate=order.tax_rate, total_amount=order.total_amount, contact_name=customer.contact_name, contact_phone=customer.contact_phone, address=order.delivery_address or customer.address, lines=lines, ) def collect_purchase_order_archive_context(db: Session, purchase_order_id: int) -> ArchiveContext: row = db.execute( select(PurchaseOrder, Supplier) .join(Supplier, Supplier.id == PurchaseOrder.supplier_id) .where(PurchaseOrder.id == purchase_order_id) ).one_or_none() if row is None: raise HTTPException(status_code=404, detail="采购订单不存在") order, supplier = row item_rows = db.execute( select(PurchaseOrderItem, Item) .join(Item, Item.id == PurchaseOrderItem.material_item_id) .where(PurchaseOrderItem.purchase_order_id == purchase_order_id) .order_by(PurchaseOrderItem.line_no) ).all() is_raw_purchase = str(order.target_warehouse_type or "").upper() == "RAW" lines = [ ArchiveLine( line_no=order_item.line_no, item_code=item.item_code, item_name=item.item_name, specification=item.specification, quantity=order_item.order_weight_kg if is_raw_purchase else order_item.order_qty, delivered_or_received_quantity=order_item.received_weight_kg if is_raw_purchase else order_item.received_qty, unit_price=order_item.unit_price, line_amount=order_item.line_amount, promised_or_expected_date=None, remark=f"kg: {order_item.remark}" if is_raw_purchase and order_item.remark else ("kg" if is_raw_purchase else order_item.remark), ) for order_item, item in item_rows ] return ArchiveContext( document_type=DOCUMENT_TYPE_PURCHASE_ORDER, business_id=order.id, document_no=order.po_no, title="采购订单归档", partner_label="供应商", partner_name=supplier.supplier_name, document_date=order.order_date, due_date_label="预计到货", due_date=order.expected_date, status=archive_status_text(order.status), tax_rate=order.tax_rate, total_amount=order.total_amount, contact_name=supplier.contact_name, contact_phone=supplier.contact_phone, address=supplier.address, remark=order.remark, lines=lines, ) def collect_purchase_receipt_archive_context(db: Session, purchase_receipt_id: int) -> ArchiveContext: row = db.execute( select(PurchaseReceipt, PurchaseOrder, Supplier, Warehouse) .join(PurchaseOrder, PurchaseOrder.id == PurchaseReceipt.purchase_order_id) .join(Supplier, Supplier.id == PurchaseOrder.supplier_id) .join(Warehouse, Warehouse.id == PurchaseReceipt.warehouse_id) .where(PurchaseReceipt.id == purchase_receipt_id) ).one_or_none() if row is None: raise HTTPException(status_code=404, detail="到货入库单不存在") receipt, order, supplier, warehouse = row item_rows = db.execute( select(PurchaseReceiptItem, Item) .join(Item, Item.id == PurchaseReceiptItem.material_item_id) .where(PurchaseReceiptItem.receipt_id == purchase_receipt_id) .order_by(PurchaseReceiptItem.line_no) ).all() is_raw_receipt = str(order.target_warehouse_type or "").upper() != "AUX" lines = [] total_amount = Decimal("0") for receipt_item, item in item_rows: received_value = receipt_item.received_weight_kg if is_raw_receipt else receipt_item.received_qty unit_cost = Decimal(str(receipt_item.unit_cost or 0)) line_amount = Decimal(str(received_value or 0)) * unit_cost total_amount += line_amount remark_parts = [f"批次：{receipt_item.lot_no}", f"状态：{archive_status_text(receipt_item.status)}"] if receipt_item.remark: remark_parts.append(str(receipt_item.remark)) lines.append( ArchiveLine( line_no=receipt_item.line_no, item_code=item.item_code, item_name=item.item_name, specification=item.specification, quantity=received_value, delivered_or_received_quantity=receipt_item.accepted_weight_kg if is_raw_receipt else receipt_item.accepted_qty, unit_price=receipt_item.unit_cost, line_amount=line_amount, promised_or_expected_date=None, remark="；".join(remark_parts), ) ) remark_parts = [] if receipt.logistics_waybill_no: remark_parts.append(f"运单号：{receipt.logistics_waybill_no}") if receipt.logistics_freight_amount is not None: remark_parts.append(f"运费：{decimal_text(receipt.logistics_freight_amount, 2)}") if receipt.logistics_photo_url: remark_parts.append("辅助照片：已上传") if receipt.remark: remark_parts.append(str(receipt.remark)) return ArchiveContext( document_type=DOCUMENT_TYPE_PURCHASE_RECEIPT, business_id=receipt.id, document_no=receipt.receipt_no, title="到货入库单归档", partner_label="供应商", partner_name=supplier.supplier_name, document_date=receipt.receipt_date, due_date_label="采购订单", due_date=order.po_no, status=archive_status_text(receipt.status), tax_rate=order.tax_rate, total_amount=total_amount, contact_name=supplier.contact_name, contact_phone=supplier.contact_phone, address=warehouse.warehouse_name, remark="；".join(remark_parts), lines=lines, ) def collect_quality_inspection_archive_context(db: Session, receipt_item_id: int) -> ArchiveContext: row = db.execute( select(PurchaseReceiptItem, PurchaseReceipt, PurchaseOrder, Supplier, Warehouse, Item) .join(PurchaseReceipt, PurchaseReceipt.id == PurchaseReceiptItem.receipt_id) .join(PurchaseOrder, PurchaseOrder.id == PurchaseReceipt.purchase_order_id) .join(Supplier, Supplier.id == PurchaseOrder.supplier_id) .join(Warehouse, Warehouse.id == PurchaseReceipt.warehouse_id) .join(Item, Item.id == PurchaseReceiptItem.material_item_id) .where(PurchaseReceiptItem.id == receipt_item_id) ).one_or_none() if row is None: raise HTTPException(status_code=404, detail="质量校验明细不存在") receipt_item, receipt, order, supplier, warehouse, item = row lot = db.scalar( select(StockLot).where( StockLot.source_doc_type == "PURCHASE_RECEIPT", StockLot.source_doc_id == receipt.id, StockLot.source_line_id == receipt_item.id, ) ) if not lot and receipt_item.merge_to_lot_id: lot = db.get(StockLot, receipt_item.merge_to_lot_id) is_raw_receipt = str(order.target_warehouse_type or "").upper() != "AUX" received_value = receipt_item.received_weight_kg if is_raw_receipt else receipt_item.received_qty accepted_value = receipt_item.accepted_weight_kg if is_raw_receipt else receipt_item.accepted_qty line_amount = Decimal(str(accepted_value or 0)) * Decimal(str(receipt_item.unit_cost or 0)) inspection_result = archive_status_text((lot.quality_status if lot else None) or receipt_item.status) inspector_name = str(lot.remark or "").strip() if lot and lot.remark else "" line_remark_parts = [f"库存批次号：{receipt_item.lot_no}", f"质检结论：{inspection_result}"] if receipt_item.remark: line_remark_parts.append(str(receipt_item.remark)) remark_parts = [f"采购订单：{order.po_no}", f"入库仓库：{warehouse.warehouse_name}"] if inspector_name: remark_parts.append(f"质检人员：{inspector_name}") if receipt.remark: remark_parts.append(str(receipt.remark)) return ArchiveContext( document_type=DOCUMENT_TYPE_QUALITY_INSPECTION, business_id=receipt_item.id, document_no=f"质检-{receipt.receipt_no}-{receipt_item.line_no}", title="质量校验单归档", partner_label="供应商", partner_name=supplier.supplier_name, document_date=getattr(receipt_item, "updated_at", None) or receipt.receipt_date, due_date_label="入库单", due_date=receipt.receipt_no, status=archive_status_text(receipt_item.status), tax_rate=order.tax_rate, total_amount=line_amount, contact_name=supplier.contact_name, contact_phone=supplier.contact_phone, address=warehouse.warehouse_name, remark="；".join(remark_parts), lines=[ ArchiveLine( line_no=receipt_item.line_no, item_code=item.item_code, item_name=item.item_name, specification=item.specification, quantity=received_value, delivered_or_received_quantity=accepted_value, unit_price=receipt_item.unit_cost, line_amount=line_amount, promised_or_expected_date=None, remark="；".join(line_remark_parts), ) ], ) def collect_production_material_out_archive_context(db: Session, production_ledger_id: int) -> ArchiveContext: material_item = aliased(Item) product_item = aliased(Item) row = db.execute( select(ProductionBatchLedger, StockLot, material_item, product_item, Warehouse) .join(StockLot, StockLot.id == ProductionBatchLedger.material_lot_id) .join(material_item, material_item.id == ProductionBatchLedger.material_item_id) .join(product_item, product_item.id == ProductionBatchLedger.product_item_id) .join(Warehouse, Warehouse.id == StockLot.warehouse_id) .where(ProductionBatchLedger.id == production_ledger_id) ).one_or_none() if row is None: raise HTTPException(status_code=404, detail="生产台账不存在") ledger, material_lot, material, product, warehouse = row issued_weight = Decimal(str(ledger.total_issued_weight_kg or 0)) unit_cost = Decimal(str(material_lot.unit_cost or 0)) total_amount = issued_weight * unit_cost issue_count = db.scalar( select(func.count(ProductionBatchLedgerTxn.id)).where( ProductionBatchLedgerTxn.production_ledger_id == ledger.id, ProductionBatchLedgerTxn.txn_type == "生产出库", ) ) or 0 line_remark_parts = [ f"库存批次号：{ledger.material_lot_no}", f"累计生产出库次数：{issue_count}", f"库外未闭环重量：{decimal_text(ledger.outside_weight_kg, 6)}kg", ] if ledger.remark: line_remark_parts.append(str(ledger.remark)) remark_parts = [ f"材料库存批次号：{ledger.material_lot_no}", f"生产产品：{product.item_name}", ] if ledger.remark: remark_parts.append(str(ledger.remark)) return ArchiveContext( document_type=DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT, business_id=ledger.id, document_no=f"生产领料-{ledger.material_lot_no}-{product.item_name}", title="生产领料出库单归档", partner_label="生产产品", partner_name=product.item_name, document_date=ledger.last_issue_time or ledger.first_issue_time or ledger.updated_at, due_date_label="生产台账", due_date=ledger.id, status=archive_status_text(ledger.status), tax_rate=0, total_amount=total_amount, address=warehouse.warehouse_name, remark="；".join(remark_parts), lines=[ ArchiveLine( line_no=1, item_code=material.item_code, item_name=material.item_name, specification=material.specification, quantity=ledger.total_issued_weight_kg, delivered_or_received_quantity=ledger.outside_weight_kg, unit_price=material_lot.unit_cost, line_amount=total_amount, promised_or_expected_date=ledger.last_issue_time or ledger.first_issue_time, remark="；".join(line_remark_parts), ) ], ) def collect_production_inbound_settlement_archive_context(db: Session, inventory_txn_id: int) -> ArchiveContext: main_txn = db.get(InventoryTxn, inventory_txn_id) if main_txn is None: raise HTTPException(status_code=404, detail="生产入库库存流水不存在") if main_txn.txn_type not in PRODUCTION_INBOUND_TXN_LABELS: raise HTTPException(status_code=400, detail="该库存流水不是生产入库结算流水") ledger_row = None if main_txn.source_doc_type == "生产台账": product_item = aliased(Item) ledger_row = db.execute( select(ProductionBatchLedger, product_item) .join(product_item, product_item.id == ProductionBatchLedger.product_item_id) .where(ProductionBatchLedger.id == main_txn.source_doc_id) ).one_or_none() ledger = ledger_row[0] if ledger_row else None product = ledger_row[1] if ledger_row else None work_order_id: int | None = None if ledger is None and main_txn.source_doc_type == "FG_RECEIPT": receipt = db.get(CompletionReceipt, main_txn.source_doc_id) work_order_id = int(receipt.work_order_id) if receipt else None elif ledger is None and main_txn.source_doc_type == "WORK_ORDER": work_order_id = int(main_txn.source_doc_id or 0) or None if ledger is None and work_order_id: work_order_row = db.execute( select(WorkOrder, Item) .join(Item, Item.id == WorkOrder.product_item_id) .where(WorkOrder.id == work_order_id) ).one_or_none() product = work_order_row[1] if work_order_row else None txn_order = case( (InventoryTxn.txn_type.in_(("FG_IN", "PRODUCTION_FINISHED_IN")), 1), (InventoryTxn.txn_type == "PRODUCTION_SURPLUS_IN", 2), (InventoryTxn.txn_type == "PRODUCTION_SCRAP_IN", 3), else_=99, ) related_stmt = ( select(InventoryTxn, Item, Warehouse, StockLot) .join(Item, Item.id == InventoryTxn.item_id) .join(Warehouse, Warehouse.id == InventoryTxn.warehouse_id) .outerjoin(StockLot, StockLot.id == InventoryTxn.lot_id) .order_by(txn_order, InventoryTxn.id.asc()) ) batch_no = production_document_batch_no_from_text(main_txn.remark) if batch_no: related_stmt = related_stmt.where( InventoryTxn.remark.like(f"%{PRODUCTION_DOCUMENT_BATCH_PREFIX}{batch_no}%"), InventoryTxn.txn_type.in_(tuple(PRODUCTION_INBOUND_TXN_LABELS)), ) elif work_order_id and main_txn.source_doc_type in {"FG_RECEIPT", "WORK_ORDER"}: receipt_ids = db.scalars(select(CompletionReceipt.id).where(CompletionReceipt.work_order_id == work_order_id)).all() source_filters = [ (InventoryTxn.source_doc_type == "WORK_ORDER") & (InventoryTxn.source_doc_id == work_order_id), ] if receipt_ids: source_filters.append( (InventoryTxn.source_doc_type == "FG_RECEIPT") & (InventoryTxn.source_doc_id.in_([int(row_id) for row_id in receipt_ids])) ) related_stmt = related_stmt.where( or_(*source_filters), InventoryTxn.biz_time == main_txn.biz_time, InventoryTxn.txn_type.in_(tuple(PRODUCTION_INBOUND_TXN_LABELS)), ) else: related_stmt = related_stmt.where( InventoryTxn.source_doc_type == main_txn.source_doc_type, InventoryTxn.source_doc_id == main_txn.source_doc_id, InventoryTxn.biz_time == main_txn.biz_time, InventoryTxn.txn_type.in_(tuple(PRODUCTION_INBOUND_TXN_LABELS)), ) related_rows = db.execute(related_stmt).all() lines: list[ArchiveLine] = [] total_amount = Decimal("0") related_warehouse_names: list[str] = [] for line_no, (txn, item, warehouse, lot) in enumerate(related_rows, start=1): total_amount += Decimal(str(txn.amount or 0)) if warehouse.warehouse_name and warehouse.warehouse_name not in related_warehouse_names: related_warehouse_names.append(warehouse.warehouse_name) lot_no = lot.lot_no if lot else "" line_remark_parts = [ f"入库类型：{PRODUCTION_INBOUND_TXN_LABELS.get(txn.txn_type, txn.txn_type)}", f"入库仓库：{warehouse.warehouse_name}", ] if lot_no: line_remark_parts.append(f"库存批次号：{lot_no}") if ledger: line_remark_parts.append(f"原材料库存批次号：{ledger.material_lot_no}") if txn.remark: line_remark_parts.append(str(txn.remark)) lines.append( ArchiveLine( line_no=line_no, item_code=item.item_code, item_name=item.item_name, specification=item.specification, quantity=txn.qty_change if txn.txn_type in ("FG_IN", "PRODUCTION_FINISHED_IN") else txn.weight_change_kg, delivered_or_received_quantity=txn.weight_change_kg, unit_price=txn.unit_cost, line_amount=txn.amount, promised_or_expected_date=txn.biz_time, remark="；".join(line_remark_parts), ) ) remark_parts = [ f"库存流水号：{main_txn.txn_no}", f"入库类型：{PRODUCTION_INBOUND_TXN_LABELS.get(main_txn.txn_type, main_txn.txn_type)}", ] if ledger: remark_parts.append(f"材料库存批次号：{ledger.material_lot_no}") if related_warehouse_names: remark_parts.append(f"涉及仓库：{'、'.join(related_warehouse_names)}") if main_txn.remark: remark_parts.append(str(main_txn.remark)) due_date_label = "生产台账" if ledger else "生产工单" if work_order_id else "库存流水" due_date_value = str(ledger.id if ledger else work_order_id if work_order_id else main_txn.id) return ArchiveContext( document_type=DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT, business_id=main_txn.id, document_no=f"生产入库结算-{main_txn.txn_no}", title="生产入库结算单归档", partner_label="生产产品", partner_name=product.item_name if product else "", document_date=main_txn.biz_time, due_date_label=due_date_label, due_date=due_date_value, status=PRODUCTION_INBOUND_TXN_LABELS.get(main_txn.txn_type, archive_status_text(main_txn.txn_type)), tax_rate=0, total_amount=total_amount, address_label="涉及仓库", address="、".join(related_warehouse_names), remark="；".join(remark_parts), lines=lines, ) def warehouse_operation_txn_label(txn_type: str | None) -> str: text = str(txn_type or "").strip() if not text: return "仓库操作" normalized = text.upper() label = WAREHOUSE_OPERATION_TXN_LABELS.get(normalized) if label: return label if re.fullmatch(r"[A-Za-z0-9_]+", text): return "未识别仓库操作" return text def warehouse_operation_document_title(txn_type: str | None) -> str: label = warehouse_operation_txn_label(txn_type) if label.endswith("单"): return label if label.endswith(("入库", "出库", "操作")): return f"{label}单" return f"{label}单" def warehouse_operation_document_no(title: str, txn_no: str) -> str: title_prefix = title[:-1] if title.endswith("单") else title return f"{title_prefix}-{txn_no}" def _warehouse_operation_quantity(txn: InventoryTxn) -> Decimal: qty_change = Decimal(str(txn.qty_change or 0)) weight_change = Decimal(str(txn.weight_change_kg or 0)) value = qty_change if qty_change != 0 else weight_change return abs(value) def collect_warehouse_operation_archive_context(db: Session, inventory_txn_id: int) -> ArchiveContext: source_lot = aliased(StockLot) row = db.execute( select(InventoryTxn, Item, Warehouse, WarehouseLocation, StockLot, source_lot) .join(Item, Item.id == InventoryTxn.item_id) .join(Warehouse, Warehouse.id == InventoryTxn.warehouse_id) .outerjoin(WarehouseLocation, WarehouseLocation.id == InventoryTxn.location_id) .outerjoin(StockLot, StockLot.id == InventoryTxn.lot_id) .outerjoin(source_lot, source_lot.id == StockLot.source_material_lot_id) .where(InventoryTxn.id == inventory_txn_id) ).one_or_none() if row is None: raise HTTPException(status_code=404, detail="仓库库存流水不存在") txn, item, warehouse, location, lot, source_lot_row = row document_title = warehouse_operation_document_title(txn.txn_type) operation_label = warehouse_operation_txn_label(txn.txn_type) location_name = location.location_name if location else None lot_no = lot.lot_no if lot else None source_lot_no = None if source_lot_row: source_lot_no = source_lot_row.lot_no elif lot and lot.source_material_sub_batch_no: source_lot_no = lot.source_material_sub_batch_no quantity = _warehouse_operation_quantity(txn) received_or_delivered = abs(Decimal(str(txn.weight_change_kg or 0))) or quantity line_remark_parts = [ f"库存流水号：{txn.txn_no}", f"操作类型：{operation_label}", ] if lot_no: line_remark_parts.append(f"库存批次号：{lot_no}") if source_lot_no: line_remark_parts.append(f"来源库存批次号：{source_lot_no}") if location_name: line_remark_parts.append(f"库位：{location_name}") if txn.logistics_waybill_no: line_remark_parts.append(f"运单号：{txn.logistics_waybill_no}") if txn.logistics_freight_amount is not None: line_remark_parts.append(f"运费：{decimal_text(txn.logistics_freight_amount, 2)}") if txn.logistics_photo_url: line_remark_parts.append("辅助照片：已上传") if txn.remark: line_remark_parts.append(str(txn.remark)) remark_parts = [ f"库存流水号：{txn.txn_no}", f"操作类型：{operation_label}", f"仓库：{warehouse.warehouse_name}", ] if lot_no: remark_parts.append(f"库存批次号：{lot_no}") if source_lot_no: remark_parts.append(f"来源库存批次号：{source_lot_no}") if location_name: remark_parts.append(f"库位：{location_name}") if txn.source_doc_type: remark_parts.append(f"来源单据：{txn.source_doc_type}-{txn.source_doc_id}") if txn.remark: remark_parts.append(str(txn.remark)) address_parts = [warehouse.warehouse_name] if location_name: address_parts.append(location_name) return ArchiveContext( document_type=DOCUMENT_TYPE_WAREHOUSE_OPERATION, business_id=txn.id, document_no=warehouse_operation_document_no(document_title, txn.txn_no), title=f"{document_title}归档", partner_label="仓库", partner_name=warehouse.warehouse_name, document_date=txn.biz_time, due_date_label="库存流水", due_date=txn.txn_no, status=operation_label, tax_rate=0, total_amount=txn.amount, address=" / ".join(address_parts), remark="；".join(remark_parts), lines=[ ArchiveLine( line_no=1, item_code=item.item_code, item_name=item.item_name, specification=item.specification, quantity=quantity, delivered_or_received_quantity=received_or_delivered, unit_price=txn.unit_cost, line_amount=txn.amount, promised_or_expected_date=txn.biz_time, remark="；".join(line_remark_parts), ) ], ) def collect_archive_context(db: Session, document_type: str, business_id: int) -> ArchiveContext: if document_type == DOCUMENT_TYPE_SALES_ORDER: return collect_sales_order_archive_context(db, business_id) if document_type == DOCUMENT_TYPE_PURCHASE_ORDER: return collect_purchase_order_archive_context(db, business_id) if document_type == DOCUMENT_TYPE_PURCHASE_RECEIPT: return collect_purchase_receipt_archive_context(db, business_id) if document_type == DOCUMENT_TYPE_QUALITY_INSPECTION: return collect_quality_inspection_archive_context(db, business_id) if document_type == DOCUMENT_TYPE_PRODUCTION_MATERIAL_OUT: return collect_production_material_out_archive_context(db, business_id) if document_type == DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT: return collect_production_inbound_settlement_archive_context(db, business_id) if document_type == DOCUMENT_TYPE_WAREHOUSE_OPERATION: return collect_warehouse_operation_archive_context(db, business_id) raise HTTPException(status_code=400, detail="不支持的单据类型") def _register_archive_font() -> tuple[str, str]: try: pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light")) return "STSong-Light", "STSong-Light" except Exception: return "Helvetica", "Helvetica-Bold" def _cell_line_height(size: int) -> float: return size + 3 def _wrap_text_to_lines(text: str, font_name: str, size: int, max_width: float) -> list[str]: if max_width <= 0: return [""] value = str(text or "") if value == "": return [""] lines: list[str] = [] for paragraph in value.replace("\r\n", "\n").replace("\r", "\n").split("\n"): if paragraph == "": lines.append("") continue current = "" for char in paragraph: candidate = f"{current}{char}" if current and pdfmetrics.stringWidth(candidate, font_name, size) > max_width: lines.append(current) current = char else: current = candidate lines.append(current) return lines or [""] def _cell_required_height(text: str, font_name: str, size: int, width: float) -> float: lines = _wrap_text_to_lines(text, font_name, size, max(0, width - ARCHIVE_CELL_HORIZONTAL_PADDING)) return max(ARCHIVE_MIN_ROW_HEIGHT, len(lines) * _cell_line_height(size) + ARCHIVE_CELL_VERTICAL_PADDING) def _row_required_height(values: list[str], columns: list[tuple[str, int, str]], font_name: str, size: int) -> float: return max( _cell_required_height(value, font_name, size, col_w) for value, (_, col_w, _) in zip(values, columns, strict=True) ) def _draw_cell( pdf: canvas.Canvas, x: float, y: float, width: float, height: float, text: str, font_name: str, size: int = 8, align: str = "left", ) -> None: pdf.rect(x, y - height, width, height, stroke=1, fill=0) pdf.setFont(font_name, size) lines = _wrap_text_to_lines(text, font_name, size, max(0, width - ARCHIVE_CELL_HORIZONTAL_PADDING)) line_height = _cell_line_height(size) text_height = len(lines) * line_height top_padding = max(4, (height - text_height) / 2) text_y = y - top_padding - size for line in lines: if align == "right": pdf.drawRightString(x + width - 4, text_y, line) elif align == "center": pdf.drawCentredString(x + width / 2, text_y, line) else: pdf.drawString(x + 4, text_y, line) text_y -= line_height def _archive_metadata_rows(context: ArchiveContext) -> list[tuple[tuple[str, str], tuple[str, str]]]: contact_text = f"{context.contact_name or ''} 电话: {context.contact_phone or ''}" tax_text = f"{decimal_text((context.tax_rate or 0) * 100)}%" return [ (("单据编号", context.document_no), ("单据日期", date_text(context.document_date))), ((context.partner_label, context.partner_name), (context.due_date_label, date_text(context.due_date))), (("联系人", contact_text), ("状态", context.status)), ((context.address_label or "地址", context.address or ""), ("税率", tax_text)), ] def _archive_metadata_line_groups( context: ArchiveContext, font_name: str, size: int = ARCHIVE_METADATA_FONT_SIZE, ) -> list[tuple[list[str], list[str]]]: groups: list[tuple[list[str], list[str]]] = [] for (left_label, left_value), (right_label, right_value) in _archive_metadata_rows(context): left_text = f"{left_label}: {left_value or ''}" right_text = f"{right_label}: {right_value or ''}" groups.append( ( _wrap_text_to_lines(left_text, font_name, size, ARCHIVE_METADATA_LEFT_WIDTH), _wrap_text_to_lines(right_text, font_name, size, ARCHIVE_METADATA_RIGHT_WIDTH), ) ) return groups def _metadata_row_height(line_count: int, size: int = ARCHIVE_METADATA_FONT_SIZE) -> float: return max(ARCHIVE_METADATA_MIN_ROW_HEIGHT, line_count * _cell_line_height(size) + ARCHIVE_METADATA_ROW_PADDING) def _archive_table_y_for_context(context: ArchiveContext, font_name: str) -> float: _, height = A4 y = height - ARCHIVE_METADATA_TOP_Y_OFFSET for left_lines, right_lines in _archive_metadata_line_groups(context, font_name): y -= _metadata_row_height(max(len(left_lines), len(right_lines), 1)) return min(height - ARCHIVE_TABLE_TOP_Y_OFFSET, y - ARCHIVE_METADATA_TABLE_GAP) def _draw_metadata_lines( pdf: canvas.Canvas, x: float, y: float, lines: list[str], line_height: float, ) -> None: for line in lines: pdf.drawString(x, y, line) y -= line_height def _archive_table_columns() -> list[tuple[str, int, str]]: return [ ("行", 26, "center"), ("物料编码", 64, "left"), ("物料名称", 94, "left"), ("规格", 60, "left"), ("数量", 50, "right"), ("已交/收", 52, "right"), ("单价", 48, "right"), ("金额", 55, "right"), ("日期/备注", 62, "left"), ] def _draw_archive_page_frame( pdf: canvas.Canvas, context: ArchiveContext, font_name: str, bold_font_name: str, page_no: int, total_pages: int, ) -> float: width, height = A4 pdf.setFillColorRGB(0.99, 0.975, 0.92) pdf.rect(0, 0, width, height, stroke=0, fill=1) pdf.setStrokeColor(colors.black) pdf.setFillColor(colors.black) pdf.rect(28, 28, width - 56, height - 56, stroke=1, fill=0) pdf.setFont(bold_font_name, 22) title = context.title if page_no == 1 else f"{context.title}（续页）" pdf.drawCentredString(width / 2, height - 58, title) pdf.setFont(font_name, 10) pdf.drawRightString(width - 42, height - 44, "ERP留存联") pdf.drawRightString(width - 42, height - 62, "PDF归档") pdf.drawRightString(width - 42, height - 80, f"第 {page_no}/{total_pages} 页") pdf.setFont(font_name, ARCHIVE_METADATA_FONT_SIZE) line_height = _cell_line_height(ARCHIVE_METADATA_FONT_SIZE) metadata_y = height - ARCHIVE_METADATA_TOP_Y_OFFSET for left_lines, right_lines in _archive_metadata_line_groups(context, font_name): _draw_metadata_lines(pdf, ARCHIVE_METADATA_LEFT_X, metadata_y, left_lines, line_height) _draw_metadata_lines(pdf, ARCHIVE_METADATA_RIGHT_X, metadata_y, right_lines, line_height) metadata_y -= _metadata_row_height(max(len(left_lines), len(right_lines), 1)) pdf.setFont(bold_font_name, 26) pdf.setStrokeColorRGB(0.55, 0.05, 0.05) pdf.setFillColorRGB(0.65, 0.05, 0.05) pdf.rotate(12) pdf.drawString(390, 500, "PDF归档") pdf.rotate(-12) pdf.setStrokeColor(colors.black) pdf.setFillColor(colors.black) return min(height - ARCHIVE_TABLE_TOP_Y_OFFSET, metadata_y - ARCHIVE_METADATA_TABLE_GAP) def _draw_archive_table_header(pdf: canvas.Canvas, y: float, bold_font_name: str) -> None: x = ARCHIVE_TABLE_X for header, col_w, align in _archive_table_columns(): _draw_cell(pdf, x, y, col_w, ARCHIVE_MIN_ROW_HEIGHT, header, bold_font_name, ARCHIVE_CELL_FONT_SIZE, align) x += col_w def _archive_line_values(line: ArchiveLine) -> list[str]: return [ str(line.line_no), line.item_code, line.item_name, line.specification or "", decimal_text(line.quantity, 6), decimal_text(line.delivered_or_received_quantity, 6), decimal_text(line.unit_price, 4), decimal_text(line.line_amount, 2), line.remark or date_text(line.promised_or_expected_date), ] def _draw_archive_table_rows( pdf: canvas.Canvas, lines: list[ArchiveLine], y: float, font_name: str, minimum_blank_rows: int, ) -> float: columns = _archive_table_columns() for line in lines: values = _archive_line_values(line) row_h = _row_required_height(values, columns, font_name, ARCHIVE_CELL_FONT_SIZE) x = ARCHIVE_TABLE_X for value, (_, col_w, align) in zip(values, columns, strict=True): _draw_cell(pdf, x, y, col_w, row_h, value, font_name, ARCHIVE_CELL_FONT_SIZE, align) x += col_w y -= row_h for _ in range(max(0, minimum_blank_rows - len(lines))): x = ARCHIVE_TABLE_X for _, col_w, align in columns: _draw_cell(pdf, x, y, col_w, ARCHIVE_MIN_ROW_HEIGHT, "", font_name, ARCHIVE_CELL_FONT_SIZE, align) x += col_w y -= ARCHIVE_MIN_ROW_HEIGHT return y def _draw_archive_footer( pdf: canvas.Canvas, context: ArchiveContext, y: float, font_name: str, bold_font_name: str, ) -> None: width, _ = A4 pdf.setFont(bold_font_name, 10) pdf.drawRightString(width - 42, y - 20, f"合计金额: {decimal_text(context.total_amount, 2)}") pdf.setFont(font_name, 9) remark_lines = _wrap_text_to_lines(f"备注: {context.remark or ''}", font_name, 9, width - 84) remark_y = 104 for line in remark_lines: pdf.drawString(42, remark_y, line) remark_y -= _cell_line_height(9) signature_y = min(70, remark_y - 4) signature_y = max(46, signature_y) pdf.drawString(42, signature_y, f"制单: {context.prepared_by or ''}") pdf.drawString(176, signature_y, f"经办人: {context.operator_name or context.prepared_by or ''}") pdf.drawString(310, signature_y, "审核: ") pdf.drawString(444, signature_y, f"归档模板: {TEMPLATE_VERSION}") def _archive_line_height(line: ArchiveLine, font_name: str) -> float: return _row_required_height(_archive_line_values(line), _archive_table_columns(), font_name, ARCHIVE_CELL_FONT_SIZE) def _line_pages(lines: list[ArchiveLine], max_body_height: float, font_name: str) -> list[list[ArchiveLine]]: if not lines: return [[]] pages: list[list[ArchiveLine]] = [] page_lines: list[ArchiveLine] = [] used_height = 0.0 for line in lines: row_height = _archive_line_height(line, font_name) if page_lines and used_height + row_height > max_body_height: pages.append(page_lines) page_lines = [] used_height = 0.0 page_lines.append(line) used_height += row_height if page_lines: pages.append(page_lines) return pages def render_pdf_archive(context: ArchiveContext, output_path: str | Path) -> None: output = Path(output_path) output.parent.mkdir(parents=True, exist_ok=True) font_name, bold_font_name = _register_archive_font() pdf = canvas.Canvas(str(output), pagesize=A4) table_y = _archive_table_y_for_context(context, font_name) row_h = ARCHIVE_MIN_ROW_HEIGHT max_body_height = table_y - row_h - ARCHIVE_TABLE_BOTTOM_Y pages = _line_pages(context.lines, max_body_height, font_name) for page_index, page_lines in enumerate(pages, start=1): table_y = _draw_archive_page_frame(pdf, context, font_name, bold_font_name, page_index, len(pages)) _draw_archive_table_header(pdf, table_y, bold_font_name) y = _draw_archive_table_rows( pdf, page_lines, table_y - row_h, font_name, minimum_blank_rows=6 if page_index == len(pages) else 0, ) _draw_archive_footer(pdf, context, y, font_name, bold_font_name) pdf.showPage() pdf.save() def _archive_output_path(root: Path, context: ArchiveContext, archive_version: int) -> Path: type_dir = safe_filename(context.document_type) file_name = safe_filename(f"{context.document_no}_V{archive_version}") + ".pdf" return root / type_dir / file_name def _failed_archive( *, document_type: str, business_id: int, document_no: str, archive_version: int, error_message: str, created_by: int | None, ) -> DocumentArchive: return DocumentArchive( document_type=document_type, business_id=business_id, document_no=document_no, archive_version=archive_version, template_version=TEMPLATE_VERSION, file_format=FILE_FORMAT_PDF, file_name="", file_path="", file_hash=None, status=ARCHIVE_STATUS_FAILED, error_message=error_message[:2000], created_by=created_by, ) def _fallback_failed_archive_version(db: Session, document_type: str, business_id: int) -> int: try: return next_available_archive_version(db, document_type, business_id, 1) except Exception: db.rollback() return 1 def _persist_failed_archive_with_retry( db: Session, *, document_type: str, business_id: int, document_no: str, preferred_version: int, error_message: str, created_by: int | None, max_attempts: int = 3, ) -> tuple[int, str]: archive_version = preferred_version last_error = error_message for _ in range(max_attempts): try: archive_version = next_available_archive_version(db, document_type, business_id, archive_version) failed = _failed_archive( document_type=document_type, business_id=business_id, document_no=document_no, archive_version=archive_version, error_message=error_message, created_by=created_by, ) db.add(failed) db.commit() return failed.archive_version, failed.error_message or error_message except IntegrityError as exc: db.rollback() archive_version += 1 last_error = f"{error_message}; failed archive version collision: {exc}" except Exception as exc: db.rollback() return archive_version, f"{error_message}; failed archive persist error: {exc}" return archive_version, last_error def generate_document_archive( db: Session, document_type: str, business_id: int, created_by: int | None = None, archive_root: str | Path | None = None, ) -> DocumentArchiveGenerateResult: archive_version: int | None = None document_no = f"{document_type}-{business_id}" try: archive_version = next_archive_version(db, document_type, business_id) context = collect_archive_context(db, document_type, business_id) apply_archive_operator_context(db, context, created_by) document_no = context.document_no root = Path(archive_root) if archive_root is not None else Path(default_archive_root()) for _ in range(3): archive_version = next_available_archive_version(db, document_type, business_id, archive_version) output_path = _archive_output_path(root, context, archive_version) render_pdf_archive(context, output_path) archive = DocumentArchive( document_type=document_type, business_id=business_id, document_no=context.document_no, archive_version=archive_version, template_version=TEMPLATE_VERSION, file_format=FILE_FORMAT_PDF, file_name=output_path.name, file_path=str(output_path), file_hash=sha256_file(output_path), status=ARCHIVE_STATUS_READY, error_message=None, created_by=created_by, ) db.add(archive) try: db.commit() db.refresh(archive) return DocumentArchiveGenerateResult( business_id=business_id, document_type=document_type, document_no=context.document_no, archive_status=ARCHIVE_STATUS_READY, archive_version=archive.archive_version, archive_error_message=None, ) except IntegrityError: db.rollback() archive_version += 1 raise RuntimeError("archive version collision after retries") except HTTPException: db.rollback() raise except Exception as exc: db.rollback() failed_archive_version, failed_error_message = _persist_failed_archive_with_retry( db, document_type=document_type, business_id=business_id, document_no=document_no, preferred_version=archive_version or _fallback_failed_archive_version(db, document_type, business_id), error_message=str(exc), created_by=created_by, ) return DocumentArchiveGenerateResult( business_id=business_id, document_type=document_type, document_no=document_no, archive_status=ARCHIVE_STATUS_FAILED, archive_version=failed_archive_version, archive_error_message=failed_error_message, ) def build_archive_zip( db: Session, document_type: str, business_ids: Iterable[int], output_path: str | Path, ) -> Path: output = Path(output_path) output.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive_zip: for business_id in business_ids: latest = get_latest_archive(db, document_type, business_id) if latest is None or latest.status != ARCHIVE_STATUS_READY: continue file_path = Path(latest.file_path) if not file_path.exists(): continue archive_zip.write(file_path, arcname=latest.file_name or file_path.name) return output