certflow.controllers.bom_material_controller 源代码

"""BOM 零件牌号控制器

桥接 BomMaterial 查询视图（View）与 BOM 零件服务（Service）：封装按生产令号
查询、来源统计与材料牌号订正（手动/自学习）等业务逻辑，视图不直接操作 ORM。
模型类与来源常量经服务层间接引用，控制器不直接依赖 ``certflow.models.*``。

架构层次：View (BomMaterialQueryView) → Controller (BomMaterialController)
         → Service (bom_material_service) → Model (BomMaterial)
"""

from __future__ import annotations

from datetime import datetime
from typing import Any

from sqlalchemy import func

from certflow.controllers.base_controller import BaseController
from certflow.services.bom_material_service import (
    MATERIAL_SOURCE_AUTO_LEARN,
    MATERIAL_SOURCE_CHOICES,
    MATERIAL_SOURCE_MANUAL,
    BomMaterial,
    ensure_bom_material_columns,
    scan_and_import,
)




class BomMaterialController(BaseController):
    """BOM 零件牌号控制器 — 查询视图的业务调度层"""

    def __init__(self, db_session=None, db_manager=None, use_shared: bool = True):
        """初始化控制器；同时补齐 bom_materials 存量库新增列。"""
        super().__init__(db_session=db_session, db_manager=db_manager, use_shared=use_shared)
        # 存量库可能缺 material_source/material_note/updated_at 列，幂等补齐
        ensure_bom_material_columns(self.db_manager.engine)

    # ============================================================
    # 查询
    # ============================================================



    def search(self, production_order_no: str, *, fuzzy: bool = False) -> list[BomMaterial]:
        """按生产令号查询其全部零件。

        Args:
            production_order_no: 生产令号（精确匹配或模糊包含）
            fuzzy: True 时使用 LIKE %pon% 模糊匹配

        Returns:
            list[BomMaterial]：按 (生产令号, 序号) 排序的零件记录
        """
        q = self.session.query(BomMaterial)
        if production_order_no:
            if fuzzy:
                q = q.filter(BomMaterial.production_order_no.like(f"%{production_order_no}%"))
            else:
                q = q.filter(BomMaterial.production_order_no == production_order_no)
        return q.order_by(BomMaterial.production_order_no, BomMaterial.seq).all()




    def list_all(self) -> list[BomMaterial]:
        """返回全部 BOM 零件记录（按 生产令号, 序号 排序）。"""
        return (
            self.session.query(BomMaterial)
            .order_by(BomMaterial.production_order_no, BomMaterial.seq)
            .all()
        )




    def get_by_id(self, record_id: int) -> BomMaterial | None:
        """按主键获取记录。"""
        return self.session.get(BomMaterial, record_id)




    def distinct_production_order_nos(self) -> list[str]:
        """返回全部不重复的生产令号（升序），供搜索框自动补全。"""
        return [
            r[0]
            for r in self.session.query(BomMaterial.production_order_no)
            .distinct()
            .order_by(BomMaterial.production_order_no)
            .all()
        ]




    def source_summary(self) -> dict[str, int]:
        """统计各材料牌号来源的记录数（用于状态栏展示）。"""
        rows = (
            self.session.query(BomMaterial.material_source, func.count(BomMaterial.id))
            .group_by(BomMaterial.material_source)
            .all()
        )
        return {src or "import": cnt for src, cnt in rows}


    # ============================================================
    # 订正（手动 / 自学习）
    # ============================================================



    def update_material_grade(
        self,
        record_id: int,
        material_grade: str | None,
        material_source: str = MATERIAL_SOURCE_MANUAL,
        material_note: str | None = None,
    ) -> bool:
        """订正某零件的材料牌号，并记录来源与更新时间。

        Args:
            record_id: 记录主键
            material_grade: 新的材料牌号（空串/None 视为清空）
            material_source: 来源（manual/auto_learn），缺省 manual
            material_note: 补全备注

        Returns:
            bool: 是否更新成功（id 不存在返回 False）
        """
        obj = self.session.get(BomMaterial, record_id)
        if obj is None:
            return False
        obj.material_grade = (material_grade or None) if material_grade else None
        obj.material_source = material_source
        obj.material_note = (material_note or None) if material_note else None
        obj.updated_at = datetime.now()
        self.commit()
        return True




    def auto_fill(
        self,
        record_id: int,
        material_grade: str,
        material_note: str | None = None,
    ) -> bool:
        """自学习补全某零件的材料牌号（来源固定 auto_learn）。"""
        return self.update_material_grade(
            record_id,
            material_grade,
            material_source=MATERIAL_SOURCE_AUTO_LEARN,
            material_note=material_note,
        )


    # ============================================================
    # 静态门面：收口视图层对 service / model 的散落直连
    # （即使 BomMaterialController 未注入实例，视图仍可经类名调用，向后兼容）
    # ============================================================



    @staticmethod
    def material_source_choices() -> list[tuple[str, str]]:
        """材料来源下拉选项（委托 model ``MATERIAL_SOURCE_CHOICES``）。"""
        return list(MATERIAL_SOURCE_CHOICES)




    @staticmethod
    def scan_and_import(folder: str, session: Any, *, mode: str = "skip") -> dict[str, Any]:
        """递归扫描目录并入库（委托 ``bom_material_service.scan_and_import``）。

        Args:
            folder: 待扫描目录
            session: SQLAlchemy 会话
            mode: 导入模式（clear/append/refresh/skip）

        Returns:
            导入统计字典（files/rows/orders/...）
        """
        return scan_and_import(folder, session, mode=mode)