certflow.services.db_source_adapter 源代码

"""DB→DB 导入源适配层（蓝图 §4.3.2：跨库复用导入框架）。

把「查询视图勾选的 SalePlan 列表」包装成与 xlsx 解析结果同构的源记录，
喂给统一「打印合格证」入口 ``QueryController.generate_print_certificates`` 适配层，
实现「销售计划 DB → 合格证清单」的复用，无需经过 Excel 文件。

设计要点（与总原则一致）：
- 纯 SQLAlchemy，不依赖任何 Qt 组件，可单测；
- 视图层只把勾选的 SalePlan id 交给本适配层，适配层负责取回实体并暴露
  ``plan_ids`` / ``records`` / ``to_source_records``；
- 真正的「编号 + 写 Certificate」由统一入口 ``QueryController.generate_print_certificates``
  经 ``CertificatePrintService`` 编排（去重安全），框架零改动，仅新增此适配层 + 一个 view_key。
"""

from __future__ import annotations

from typing import Any

from sqlalchemy.orm import Session

from certflow.models import SalePlan




class DbSourceAdapter:
    """把选中的 SalePlan 包装为 DB→DB 导入源。

    查询视图「打印合格证」动作把勾选的 SalePlan id 交给本适配层，
    适配层从会话取回实体并暴露 ``plan_ids`` / ``records``，供控制器经统一入口
    ``generate_print_certificates`` 生成 Certificate 记录。

    Attributes:
        session: SQLAlchemy 会话
        plan_ids: 选中的销售计划主键列表
    """

    def __init__(self, session: Session, plan_ids: list[int]) -> None:
        self.session = session
        self._plan_ids = list(plan_ids)



    def plan_ids(self) -> list[int]:
        """返回选中的销售计划主键列表（去重保序）。"""
        seen: set[int] = set()
        out: list[int] = []
        for pid in self._plan_ids:
            if pid not in seen:
                seen.add(pid)
                out.append(pid)
        return out




    def get_records(self) -> list[SalePlan]:
        """从会话取回选中的 SalePlan 实体列表（保持传入顺序）。"""
        ids = self.plan_ids()
        if not ids:
            return []
        rows = self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
        order = {pid: i for i, pid in enumerate(ids)}
        rows.sort(key=lambda r: order.get(r.id, 0))
        return rows




    def to_source_records(self) -> list[dict[str, Any]]:
        """导出与 xlsx 解析结果同构的源记录字典列表（供审计/预览）。

        仅包含合格证生成所需的源字段，等价于 ``create_certificates`` 读取的列。
        """
        return [
            {
                "id": sp.id,
                "unique_key": sp.unique_key,
                "production_order_no": sp.production_order_no,
                "sales_order_no": sp.sales_order_no,
                "product_code": sp.product_code,
                "cert_product_name": sp.cert_product_name,
                "cert_product_model": sp.cert_product_model,
                "cert_product_spec": sp.cert_product_spec,
                "product_name": sp.product_name,
                "product_model": sp.product_model,
                "product_spec": sp.product_spec,
                "quantity": sp.quantity,
                "customer": sp.customer,
                "project_name": sp.project_name,
                "test_standard": sp.test_standard,
            }
            for sp in self.get_records()
        ]