# src/certflow/controllers/scan_controller.py
"""扫描件控制器 - UI 交互层
提供扫描件生成的异步控制，封装 ScanWorker 线程和信号机制。
通过信号槽实现非阻塞扫描件生成。
"""
from __future__ import annotations
import io
from typing import TYPE_CHECKING, Any
from PySide6.QtCore import QObject, QThread, Signal
from certflow.services.output_guard import (
FIELD_LABELS,
collect_incomplete,
family_requires_pn_unit,
scan_required_fields_for,
)
from certflow.services.pn_service import PNService
from certflow.services.printer.template_field_formatter import family_of
from certflow.services.printer.template_manager import get_template_manager
from certflow.services.sale_plan_service import SalePlanService
from certflow.services.scan_service import ScanService
from certflow.utils.logger import logger
if TYPE_CHECKING:
from certflow.models.certificate import Certificate
[文档]
class ScanWorker(QThread):
"""扫描件生成工作线程"""
progress = Signal(int, int, str) # current, total, message
finished = Signal(dict) # result
error = Signal(str) # error_message
def __init__(self, service: ScanService, params: dict[str, Any]) -> None:
"""初始化工作线程
Args:
service: ScanService 实例
params: 生成参数字典
"""
super().__init__()
self.service = service
self.params = params
self._is_cancelled = False
[文档]
def cancel(self) -> None:
"""取消任务
Returns:
None
"""
self._is_cancelled = True
[文档]
def run(self) -> None:
"""线程主执行方法
调用 ScanService 生成扫描件，成功后发出 finished 信号，
异常时发出 error 信号。
Returns:
None
"""
try:
result = self.service.generate_scans(
**self.params,
progress_callback=self._on_progress,
)
if not self._is_cancelled:
self.finished.emit(result)
else:
self.finished.emit({"success": 0, "failed": 0, "files": [], "message": "已取消"})
except Exception as e:
logger.error(f"扫描件生成线程异常: {e}")
if not self._is_cancelled:
self.error.emit(str(e))
def _on_progress(self, current: int, total: int, message: str) -> None:
"""进度回调
Args:
current: 当前已处理数量
total: 总数量
message: 进度描述信息
"""
if not self._is_cancelled:
self.progress.emit(current, total, message)
[文档]
class ScanController(QObject):
"""扫描件控制器
管理扫描件生成任务，通过信号槽实现 UI 与业务逻辑的解耦。
Signals:
progress: 进度信号 (current, total, message)
completed: 完成信号 (result: dict)
error: 错误信号 (error_message: str)
"""
progress = Signal(int, int, str)
completed = Signal(dict)
error = Signal(str)
def __init__(self, session: Any = None, db_manager: Any = None) -> None:
"""初始化扫描件控制器
Args:
session: SQLAlchemy 会话对象（优先使用）
db_manager: 数据库管理器（session 为 None 时经其创建会话）
"""
super().__init__()
if session is None and db_manager is not None:
session = db_manager.get_session()
self._service = ScanService(session)
self._worker: ScanWorker | None = None
self._current_task_id: int = 0
# ------------------------------------------------------------------
# 视图层穿透收口（阶段 C）：ScanView 原先直连 services/handlers，
# 现统一经本控制器委托，消除视图对底层服务/工具的散落依赖。
# ------------------------------------------------------------------
[文档]
def strip_unit(self, pn_display: str) -> tuple[str, str]:
"""从展示用压力串拆出 ``(数值, 单位)``
收口 ``ScanView`` 原先直连 ``PNService.strip_unit`` 的静态调用，
使压力单位解析逻辑的唯一入口收敛到 Controller 层。
Args:
pn_display: 打印界面展示的压力串（如 ``25MPa`` / ``150Lb``）
Returns:
tuple[str, str]: (数值, 单位)；无法拆分时返回 ``("", "")``
"""
return PNService.strip_unit(pn_display)
# ------------------------------------------------------------------
# 静态门面：收口 ScanView 对 service / handler / model 的散落直连
# （即使 ScanController 未注入实例，View 仍可经类名调用，向后兼容）
# ------------------------------------------------------------------
[文档]
@staticmethod
def family_of(template_key: str | None) -> str:
"""由模板键推导语言族（委托 ``template_field_formatter.family_of``）。"""
return family_of(template_key)
[文档]
@staticmethod
def scan_required_fields(family: str, base: list[str]) -> list[str]:
"""按语言族计算扫描必填字段（委托 ``output_guard.scan_required_fields_for``）。"""
return scan_required_fields_for(family, base)
[文档]
@staticmethod
def family_requires_pn_unit(family: str) -> bool:
"""判断语言族是否要求压力单位（委托 ``output_guard.family_requires_pn_unit``）。"""
return family_requires_pn_unit(family)
[文档]
@staticmethod
def collect_incomplete(cdata: dict[str, Any], required: list[str]) -> list[str]:
"""收集缺失必填字段（委托 ``output_guard.collect_incomplete``）。"""
return collect_incomplete(cdata, required)
[文档]
@staticmethod
def field_label(field: str) -> str:
"""字段中文标签（委托 ``output_guard.FIELD_LABELS``）。"""
return FIELD_LABELS.get(field, field)
[文档]
@staticmethod
def list_enabled_templates() -> list[tuple[str, str, str]]:
"""列出已启用模板（委托 ``template_manager.get_template_manager``）。"""
return get_template_manager().list_enabled_templates()
[文档]
@staticmethod
def load_dual_defaults(session: Any, sale_plan_id: Any) -> dict[str, str]:
"""读取关联 SalePlan 的 SN/KKS 基准值（委托 ``SalePlanService.find_by_id``）。
查询失败或不存在时返回空串字典，不影响基础字段。
"""
sn, kks = "", ""
try:
if sale_plan_id is not None and session is not None:
plan = SalePlanService(session).find_by_id(sale_plan_id)
if plan is not None:
sn = getattr(plan, "sn_code", "") or ""
kks = getattr(plan, "kks_code", "") or ""
except Exception: # noqa: BLE001 - 查询失败不影响基础字段
pass
return {"sn_code": sn, "kks_code": kks}
[文档]
def individual_codes(
self, cert: Certificate, override: dict[str, Any] | None = None
) -> list[str]:
"""展开逐台编号（委托 ``ScanService._individual_codes``）。"""
return self._service._individual_codes(cert, override)
[文档]
def load_coordinates(self, template_type: str) -> dict[str, Any]:
"""加载模板坐标（委托 ``ScanService._load_coordinates``）。"""
return self._service._load_coordinates(template_type)
[文档]
def build_cert_data(
self,
cert: Certificate,
template_type: str,
serial_number: str | None = None,
overrides: dict[str, Any] | None = None,
dual_code: bool = False,
) -> dict[str, Any]:
"""构建单台合格证字段数据（委托 ``ScanService._build_cert_data``）。"""
return self._service._build_cert_data(
cert,
template_type,
serial_number=serial_number,
overrides=overrides,
dual_code=dual_code,
)
[文档]
def render_preview_png(
self,
cert: Certificate,
template_type: str,
width_mm: float,
height_mm: float,
serial_number: str | None = None,
overrides: dict[str, Any] | None = None,
dual_code: bool = False,
) -> bytes:
"""渲染扫描件预览并返回 PNG 字节流（框架无关，不含 PySide6）。
收口 ScanView 原先对 ``ScanService``（坐标/数据）与 ``ScanImageHandler``
（画布/绘制）的直连：预览 == 产出，统一经 Controller 管道。
"""
coords = self._service._load_coordinates(template_type)
bg = coords.get("background_path", "")
fields = coords.get("fields", {})
handler = self._service.image_handler
img = handler.create_canvas(width_mm, height_mm, bg)
data = self._service._build_cert_data(
cert,
template_type,
serial_number=serial_number,
overrides=overrides,
dual_code=dual_code,
)
for field_key, pos in fields.items():
value = data.get(field_key, "")
if value:
handler.draw_field(
img,
value,
pos.get("x", 0),
pos.get("y", 0),
pos.get("font_size", 9),
width_mm=pos.get("width"),
height_mm=pos.get("height"),
align=pos.get("align", "left"),
)
buf = io.BytesIO()
img.save(buf, "PNG")
return buf.getvalue()
[文档]
def load_certificates(self, certificate_ids: list[int]) -> list[Certificate]:
"""加载合格证记录
Args:
certificate_ids: 合格证 ID 列表
Returns:
合格证列表
"""
return self._service.load_certificates(certificate_ids)
[文档]
def generate_scans(
self,
certificate_ids: list[int],
output_dir: str,
template_type: str = "全中文",
output_format: str = "JPG",
rows: int = 1,
cols: int = 1,
paper_width_mm: float = 60,
paper_height_mm: float = 100,
data_overrides: dict[int, dict] | None = None,
dual_code: bool = False,
) -> int:
"""异步生成扫描件
Args:
certificate_ids: 合格证 ID 列表
output_dir: 输出目录
template_type: 模板类型 (全中文/中英文/俄英文)
output_format: 输出格式 (JPG/PDF)
rows: 每页行数
cols: 每页列数
paper_width_mm: 纸张宽度 (mm)
paper_height_mm: 纸张高度 (mm)
Returns:
任务 ID
Examples:
>>> controller = ScanController(session)
>>> from certflow.config.paths_override import cfg
>>> task_id = controller.generate_scans([1, 2, 3], cfg("paths.output.scans", "output/scans"), "全中文")
>>> print(task_id)
"""
# 取消当前正在执行的任务
self.cancel()
params = {
"certificate_ids": certificate_ids,
"output_dir": output_dir,
"template_type": template_type,
"output_format": output_format,
"rows": rows,
"cols": cols,
"paper_width_mm": paper_width_mm,
"paper_height_mm": paper_height_mm,
"data_overrides": data_overrides,
"dual_code": dual_code,
}
self._current_task_id += 1
task_id = self._current_task_id
self._worker = ScanWorker(self._service, params)
self._worker.progress.connect(self.progress)
self._worker.finished.connect(self._on_finished)
self._worker.error.connect(self.error)
self._worker.start()
logger.info(
f"开始生成扫描件 [任务 {task_id}]: "
f"{len(certificate_ids)} 份合格证, 模板={template_type}"
)
return task_id
[文档]
def cancel(self) -> None:
"""取消当前生成任务"""
if self._worker and self._worker.isRunning():
self._worker.cancel()
self._worker.quit()
self._worker.wait(3000) # 等待最多 3 秒
logger.info("扫描件生成任务已取消")
self._worker = None
def _on_finished(self, result: dict[str, Any]) -> None:
"""任务完成回调
Args:
result: 生成结果字典，含 success/failed/files/message 等键
"""
self.completed.emit(result)
self._worker = None
logger.info(f"扫描件生成完成: 成功 {result.get('success', 0)} 个文件")
[文档]
def is_running(self) -> bool:
"""检查是否有任务正在运行
Returns:
是否有任务正在运行
"""
return self._worker is not None and self._worker.isRunning()