"""报告控制器模块
处理报告生成的业务逻辑，协调「数据库取数 → 填充 Excel 模板 → 落盘交付」。
底层使用 ``certflow.services.report_fill_service``（openpyxl 填充 + ReportOutputService），
取代旧的报告生成器（ReportLab 程序化硬编码，与 12 模板不可维护，已排除）。
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from loguru import logger
from certflow.config import settings as app_settings
from certflow.controllers.base_controller import BaseController
from certflow.services.report_fill_service import generate as fill_generate
[文档]
class ReportController(BaseController):
"""报告控制器 - 处理报告生成的业务逻辑
提供质保书 / 试压报告 / 材质报告的生成，以及模板查询、文件打开等功能。
"""
def __init__(self, db_session=None) -> None:
"""初始化报告控制器
Args:
db_session: 数据库会话对象，如果为None则自动创建
"""
super().__init__(db_session)
# ------------------------------------------------------------------ #
# 统一入口
# ------------------------------------------------------------------ #
[文档]
def generate_report(
self,
sale_plan_ids: list[int],
report_kind: str = "quality",
language: str = "zh",
records_per_page: int = 6,
export_pdf: bool | None = None,
) -> dict:
"""生成报告（统一入口）。
Args:
sale_plan_ids: 选中的销售计划 ID 列表。
report_kind: ``quality``(质保书) / ``pressure``(试压报告) / ``material``(材质报告)。
language: ``zh`` / ``en``。
records_per_page: 每页记录数（挑选模板变体：行数/双语）。
export_pdf: 是否同时派生 PDF（None→读配置，默认 False）。
Returns:
dict: ReportOutputService.emit 的结果（含 success/folder/xlsx/pdf/sealed）。
"""
if not sale_plan_ids:
return {"success": False, "error": "请选择要生成报告的数据"}
try:
return fill_generate(
session=self.session,
sale_plan_ids=sale_plan_ids,
report_kind=report_kind,
language=language,
records_per_page=records_per_page,
export_pdf=export_pdf,
)
except Exception as e: # 顶层兜底，避免 UI 线程崩溃
logger.exception(f"报告生成失败: {e}")
return {"success": False, "error": str(e)}
# ------------------------------------------------------------------ #
# 兼容旧调用签名（report_view 历史入口）
# ------------------------------------------------------------------ #
[文档]
def generate_pressure_test_report(
self,
sale_plan_ids: list[int],
template: str = "standard",
language: str = "zh",
records_per_page: int = 5,
) -> dict:
"""生成试压报告（兼容旧签名）。"""
return self.generate_report(
sale_plan_ids=sale_plan_ids,
report_kind="pressure",
language=language,
records_per_page=records_per_page,
)
[文档]
def generate_material_report(
self,
data_ids: list[int],
report_type: str = "material",
language: str = "zh",
records_per_page: int = 6,
) -> dict:
"""生成材质报告或质保书（兼容旧签名）。
Args:
data_ids: 销售计划 ID 列表。
report_type: ``material``(材质报告) / ``quality``(质保书)。
"""
kind = "quality" if report_type == "quality" else "material"
return self.generate_report(
sale_plan_ids=data_ids,
report_kind=kind,
language=language,
records_per_page=records_per_page,
)
[文档]
def get_available_templates(self) -> list[dict]:
"""获取可用的报告模板列表（用于 UI 展示）。"""
return [
{
"id": "quality",
"name": "质保书",
"type": "quality",
"languages": ["zh", "en"],
"page_options": [5, 6, 9],
"description": "产品质量保证书（5/6/9 行模板）",
},
{
"id": "pressure",
"name": "试压报告",
"type": "pressure",
"languages": ["zh", "en"],
"page_options": [13, 25],
"description": "产品压力试验检测报告（中/少/中英）",
},
{
"id": "material",
"name": "材质报告",
"type": "material",
"languages": ["zh", "en"],
"page_options": [1, 2, 3],
"description": "产品材质证明书（1/2/3 行）",
},
]
[文档]
def open_report_file(self, filepath: str) -> bool:
"""打开生成的报告文件（系统默认程序）。"""
try:
path = Path(filepath)
if not path.exists():
logger.error(f"文件不存在: {filepath}")
return False
import os
import sys
if sys.platform == "win32":
os.startfile(str(path))
elif sys.platform == "darwin":
os.system(f"open '{path}'")
else:
os.system(f"xdg-open '{path}'")
logger.info(f"打开报告文件: {filepath}")
return True
except Exception as e:
logger.error(f"打开文件失败: {e}")
return False
[文档]
def open_report_folder(self, folder: str) -> bool:
"""打开报告目录（文件夹）；不存在时惰性创建，确保可打开。"""
try:
import os
import sys
path = Path(folder)
# 解析为绝对路径（相对路径基于 BASE_DIR）
if not path.is_absolute():
path = app_settings.BASE_DIR / path
path.mkdir(parents=True, exist_ok=True)
if sys.platform == "win32":
os.startfile(str(path))
elif sys.platform == "darwin":
os.system(f"open '{path}'")
else:
os.system(f"xdg-open '{path}'")
logger.info(f"打开报告目录: {path}")
return True
except Exception as e:
logger.error(f"打开目录失败: {e}")
return False
[文档]
def get_report_statistics(self) -> dict:
"""获取报告统计信息（轻量占位，历史记录由报告目录.xlsx 承载）。"""
return {"total_reports": 0, "by_type": {}, "recent_reports": []}
# ------------------------------------------------------------------ #
# 数据选择辅助（收口 View 对 services.query_service / services.report_bom_enricher 的直连）
# ------------------------------------------------------------------ #
[文档]
def query(
self,
conditions: dict | None = None,
page: int = 1,
page_size: int = 200,
or_filters: Any = None,
) -> dict:
"""复用查询引擎对全库检索（报告数据选择）。
收口 ``report_view`` 对 ``QueryService`` 的直连调用，使取数统一经 Controller。
Args:
conditions: 过滤条件字典（同 QueryService.query）
page: 页码，从 1 开始
page_size: 每页条数
or_filters: 或条件（同 QueryService.query）
Returns:
dict: 含 records/total 等键的查询结果
"""
from certflow.services.query_service import QueryService
return QueryService(self.session).query(
conditions=conditions, page=page, page_size=page_size, or_filters=or_filters
)
[文档]
def get_distinct_field_combos(self, fields: list[str]) -> list[tuple]:
"""查询可选的字段组合（供报告数据选择下拉去重）。
收口 ``report_view`` 对 ``QueryService`` 的直连调用。
Args:
fields: 需要去重组合的字段名列表
Returns:
list[tuple]: 去重后的字段值组合
"""
from certflow.services.query_service import QueryService
return QueryService(self.session).get_distinct_field_combos(fields)
[文档]
def enrich_bom(self, pno: str) -> dict[str, Any]:
"""按产品型号富化 BOM 材质信息。
收口 ``report_view`` 对 ``report_bom_enricher.enrich`` 的直连调用，
避免 View 直接持有 ``controller.session`` 喂给 Service 层函数。
Args:
pno: 产品型号
Returns:
dict: 富化后的 BOM 材质字典
"""
from certflow.services.report_bom_enricher import enrich
return enrich(pno, self.session)