certflow.controllers.output_controller 源代码

# src/certflow/controllers/output_controller.py
"""统一输出控制器（§11.8 阶段4 控制器收敛）。

聚合打印 / 扫描两控制器，对外提供单一调度入口 ``OutputController.run(target, action, **kwargs)``，
避免 ``OutputView`` / ``main_window`` 直接散落多处控制器引用。各子目标仍由既有
``PrinterController`` / ``ScanController`` 承担实际工作（蓝图 §11.8 阶段4『或保留
PrinterController/ScanController 被 dispatch 调』），本类只做统一路由 + 显式 dispatch 表。

设计要点：
- 打印机控制器（PrinterController）跨打印 / 扫描共享一份（它内部以 worker 列表承载并发，
  共享安全）；扫描控制器（ScanController）由 *前台 ScanView* 持有，PrintView 的后台扫描
  仍使用其独立实例（避免与前台任务互相 cancel / 信号交叉）。
- dispatch 表为目标驱动的显式映射，便于后续接入 ``output.targets.<t>.run`` 配置键。
"""

from __future__ import annotations

from collections.abc import Callable
from typing import Any

from certflow.controllers.printer_controller import PrinterController
from certflow.controllers.scan_controller import ScanController




class OutputController:
    """打印 / 扫描统一调度控制器（§11.8 阶段4）。"""

    def __init__(
        self,
        session: Any = None,
        printer_name: str | None = None,
        printer_controller: PrinterController | None = None,
        scan_controller: ScanController | None = None,
    ) -> None:
        """初始化统一输出控制器。

        Args:
            session: SQLAlchemy 会话（供 ScanController 使用；为 None 时 scan 不可用）。
            printer_name: 指定打印机名；None 则使用系统默认（仅当未注入 printer_controller）。
            printer_controller: 注入既有打印机控制器（共享）；None 则新建。
            scan_controller: 注入既有扫描控制器（共享）；None 则按需新建（需 session）。
        """
        self._session = session
        self._printer_controller = printer_controller or PrinterController()
        if printer_name and printer_controller is None:
            self._printer_controller.set_printer(printer_name)
        self._scan_controller = scan_controller or (
            ScanController(session) if session is not None else None
        )

    # ---- 显式 dispatch 表：target -> {action: bound method} ----
    @property
    def _dispatch(self) -> dict[str, dict[str, Callable[..., Any]]]:
        table: dict[str, dict[str, Callable[..., Any]]] = {
            "print": {
                "print": self._printer_controller.print_certificate,
            },
        }
        if self._scan_controller is not None:
            table["scan"] = {
                "generate": self._scan_controller.generate_scans,
                "load": self._scan_controller.load_certificates,
            }
        return table



    def run(self, target: str, action: str = "print", **kwargs: Any) -> Any:
        """按 (target, action) 显式 dispatch 到对应控制器方法。

        Args:
            target: "print" / "scan"
            action: 目标内的动作键（见 ``_dispatch`` 表）
            **kwargs: 透传给具体控制器方法

        Returns:
            被调方法的返回值（generate_scans 返回 task_id；print_certificate 为 None）

        Raises:
            KeyError: target/action 不在 dispatch 表中
        """
        table = self._dispatch
        if target not in table or action not in table[target]:
            raise KeyError(f"未支持的 output 调度: target={target!r} action={action!r}")
        return table[target][action](**kwargs)


    @property
    def printer_controller(self) -> PrinterController:
        return self._printer_controller

    @property
    def scan_controller(self) -> ScanController:
        if self._scan_controller is None:
            raise RuntimeError("OutputController 未绑定 session，无法使用扫描控制器")
        return self._scan_controller



    def get_controller(self, target: str) -> Any:
        """按目标取对应控制器（统一出口）。"""
        if target == "scan":
            return self.scan_controller
        return self._printer_controller


    # 打印机状态查询统一出口（供设置页 / 状态栏复用）


    def get_printers(self) -> list[str]:
        return self._printer_controller.get_printers()


    @property
    def current_printer(self) -> str:
        return self._printer_controller.current_printer



    def get_printer_status(self) -> bool:
        return self._printer_controller.get_printer_status()