"""已发货导出处理器模块
负责三层数据生命周期中 Layer 2/3 的归档和 Excel 导出操作。
将 Excel 写入、记录拷贝等底层操作从 Service 层分离到 Handler 层。
架构:
Service → ShippedExportHandler → Model / pandas
"""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING, Any
import pandas as pd
from loguru import logger
from sqlalchemy.orm import Session
if TYPE_CHECKING:
from certflow.models import SalePlan, SalePlanShipped
# 延迟导入以避免循环依赖 (models → sale_plan → handlers → shipped_export_handler → models)
# 实际导入在方法内部执行: from certflow.models import SalePlan, SalePlanShipped
# SalePlan 所有业务列的字段名（排除 id 等自动生成字段，用于 insert）
_SALEPLAN_COPY_FIELDS = [
"unique_key",
"original_order",
"sort_group",
"sort_order",
"created_at",
"updated_at",
"source_file",
"source_sheet",
"import_batch_id",
"import_date",
"contract_no",
"sales_order_no",
"production_order_no",
"plan_no",
"plan_date",
"planned_delivery_date",
"contract_delivery_date",
"customer",
"project_name",
"project_unit",
"product_name",
"product_model",
"product_spec",
"quantity",
"weight",
"category",
"unit_price",
"total_price",
"business_dept",
"affiliated_dept",
"payment_method",
"supply_type",
"tech_requirements",
"equipment_code",
"product_code",
"sn_code",
"kks_code",
"production_status",
"execution_status",
"execution_date",
"execution_inferred",
"certificate_done",
"nameplate_done",
"test_report_done",
"warranty_done",
"cert_scan_done",
"certificate_time",
"nameplate_time",
"test_report_time",
"warranty_time",
"cert_scan_time",
"progress_percent",
"certificate_remarks",
"sales_plan_remarks",
"order_no",
"specification",
"material",
"certificate_number",
"cert_product_name",
"cert_product_model",
"cert_product_spec",
"test_standard",
"format_status",
"font_colors",
"background_colors",
"comments",
"is_hidden",
]
[文档]
class ShippedExportHandler:
"""已发货导出处理器
负责:
- 将 SalePlan 记录拷贝到 SalePlanShipped 归档表
- 导出完整字段 Excel 文件
- 按列定义导出选中行到 Excel
Examples:
>>> handler = ShippedExportHandler(session)
>>> count = handler.archive_to_shipped(plans)
>>> result = handler.export_full_excel(plans, "/output")
"""
def __init__(self, session: Session):
"""初始化处理器
Args:
session: SQLAlchemy 数据库会话对象
"""
self.session = session
# ============================================================
# 归档操作
# ============================================================
[文档]
def copy_to_shipped(
self, plan: SalePlan, shipped_by: str, shipped_at: datetime | None = None
) -> SalePlanShipped:
"""将单条 SalePlan 记录拷贝为 SalePlanShipped 归档记录
不执行 session.add，由调用方控制事务。
Args:
plan: 源 SalePlan 记录
shipped_by: 归档来源标识 ("archive" 或 "export_delete")
shipped_at: 归档时间，默认当前时间
Returns:
SalePlanShipped: 归档记录对象
"""
from certflow.models import SalePlanShipped
now = shipped_at or datetime.now()
shipped = SalePlanShipped(shipped_at=now, shipped_by=shipped_by)
for field_name in _SALEPLAN_COPY_FIELDS:
if hasattr(plan, field_name):
setattr(shipped, field_name, getattr(plan, field_name))
shipped.production_status = "已发货"
return shipped
[文档]
def batch_archive(self, plans: list[SalePlan], shipped_by: str) -> int:
"""批量归档 SalePlan → SalePlanShipped 并从主表删除
Args:
plans: 要归档的 SalePlan 记录列表
shipped_by: 归档来源标识
Returns:
int: 归档记录数
"""
now = datetime.now()
count = 0
for plan in plans:
shipped = self.copy_to_shipped(plan, shipped_by, shipped_at=now)
self.session.add(shipped)
self.session.delete(plan)
count += 1
self.session.commit()
logger.info(f"归档完成: {count} 条 → sale_plans_shipped (shipped_by={shipped_by})")
return count
# ============================================================
# Excel 导出
# ============================================================
[文档]
def export_full_excel(self, plans: list[SalePlan], output_dir: str) -> dict[str, Any]:
"""导出 SalePlan 完整字段到 Excel 文件
Args:
plans: 要导出的 SalePlan 记录列表
output_dir: 输出目录
Returns:
{"count": N, "filepath": "..."}
"""
data = []
for p in plans:
data.append({fn: getattr(p, fn, "") for fn in _SALEPLAN_COPY_FIELDS})
now = datetime.now()
ts = now.strftime("%Y%m%d_%H%M%S")
filepath = f"{output_dir}/已发货导出_{ts}.xlsx"
pd.DataFrame(data).to_excel(filepath, index=False)
logger.info(f"导出完整 Excel: {len(plans)} 条 → {filepath}")
return {"count": len(plans), "filepath": filepath}
[文档]
def export_selected_to_excel(
self,
records: list[Any],
columns: list[dict[str, str]],
file_path: str,
) -> int:
"""按列定义导出选中记录到 Excel
Args:
records: ORM 记录列表（SalePlan 对象）
columns: 列定义列表 [{"field": "...", "label": "..."}, ...]
file_path: 目标文件路径
Returns:
int: 导出记录数
"""
data = []
for record in records:
row_data = {}
for col_def in columns:
value = getattr(record, col_def["field"], "")
row_data[col_def["label"]] = value
data.append(row_data)
pd.DataFrame(data).to_excel(file_path, index=False)
logger.info(f"导出选中行: {len(data)} 条 → {file_path}")
return len(data)
[文档]
def export_query_result_to_excel(
self,
records: list[Any],
columns: list[dict[str, str]],
file_path: str,
) -> int:
"""按列定义导出查询结果到 Excel（带 plan_date 格式化）
Args:
records: ORM 记录列表
columns: 列定义列表
file_path: 目标文件路径
Returns:
int: 导出记录数
"""
from certflow.utils.date_utils import normalize_plan_date
data = []
for record in records:
row_data = {}
for col_def in columns:
value = getattr(record, col_def["field"], "")
if col_def["field"] == "plan_date":
value = normalize_plan_date(value) or ""
row_data[col_def["label"]] = value
data.append(row_data)
pd.DataFrame(data).to_excel(file_path, index=False)
logger.info(f"导出查询结果: {len(data)} 条 → {file_path}")
return len(data)