"""报告输出服务：双存根 + VBA 同款文件夹命名 + 印章嵌入 + 报告目录超链接索引
设计依据：reports_migration/notes/03_vba_migration_blueprint.md §7
- §7.1 印章双模式：模式 A 无纸化（PDF 嵌电子章）；模式 B 纸档（xlsx 干净可手改，
PDF 不嵌章供打印+手动盖章）。xlsx 始终为可手改微调档（唯一真相源）。
- §7.2 文件夹命名逐字复用 VBA(Mod_GCreatFloder.新建文件夹)，但**全部由 config.yaml 驱动**
（paths.report_archive.folder 的 pattern/tokens/兜底文本）：
{root}/{ym}计划项目资料/{rq}_{customer}_{project}项目资料
root 统一可配置；外购/自制改用显式 外购类型 列（不再依赖单元格颜色）。
- §7.4 超链接索引：项目文件夹内 报告目录.xlsx，用 HYPERLINK 关联本次所有报告（仿 BaoGaoMuLu）。
"""
from __future__ import annotations
import io
import re
import shutil
from dataclasses import dataclass
from datetime import date, datetime
from pathlib import Path
from typing import Any
from loguru import logger
from certflow.config import settings as app_settings
[文档]
@dataclass
class ReportOutputRequest:
"""单份报告的输出请求（由 xlsx 填充服务产出 xlsx_source 后传入）。"""
report_type: str # 质保书 / 试压报告 / 材质报告 / 双编码合格证 ...
xlsx_source: Path # 已填充的 xlsx（唯一真相源，可手改）
plan_date: date | str # 计划日期（用于 {yyyy年mm月} 与 yyyymmdd）
customer: str # 订货单位（VBA dw）
project: str # 项目名称（VBA xm；空→【空白项目名称】）
is_outsourced: bool = False # 外购/自制（替代 VBA 黄色 Interior.Color）
language: str = "zh" # zh / en
product_no: str = "" # 文件名附加（如生产令号）
variant: str = "" # 模板变体标签（如 spec key）；非空时追加到文件名避免覆盖
report_date: date | str = "" # 报告制作日期（用于文件名 yymmdd 前缀，默认今天）
serial: str = "" # 每日全局流水号（3 位，由 emit 填入，避免同日覆盖）
seal_image: Path | None = None # 本次覆盖印章；None=用全局默认（若配置）/否则不嵌
sales_order_no: str = "" # 销售订单号（用于报告目录 §7.4 索引展示）
records_per_page: int = 0 # 每页记录数（用于报告目录索引，标明页内行数/台数）
variant_label: str = "" # 模板变体中文标签（如「6行」「中英·多」「9行·编号标准」）
[文档]
class ReportOutputService:
"""报告输出编排：建 VBA 风格文件夹 → 双存根(xlsx+pdf) → 嵌章 → 写报告目录。
落点：src/certflow/services/（feature/report-migration 分支）。
"""
def __init__(
self,
root: Path | None = None,
seal_default: Path | None = None,
templates_dir: Path | None = None,
) -> None:
base = app_settings.BASE_DIR
# 保存路径：用户/配置覆盖（report.save_path）优先，回退 paths.report_archive.root
cfg_root = app_settings.get_report_save_path() or app_settings.cfg(
"paths.report_archive.root", "output/项目资料"
)
cfg_seal = app_settings.cfg("paths.report_archive.seal_default", "")
cfg_tpl = app_settings.cfg("paths.report_templates_dir", "templates/reports")
self.root = Path(root) if root else (base / cfg_root)
self.seal_default = (
Path(seal_default)
if seal_default is not None
else (Path(cfg_seal) if cfg_seal else None)
)
self.templates_dir = Path(templates_dir) if templates_dir else (base / cfg_tpl)
self._seal_cfg = app_settings.cfg("paths.report_archive.seal", {}) or {}
self._folder_cfg = app_settings.cfg("paths.report_archive.folder", {}) or {}
# 按订货单位合并到单一工作簿（VBA 风格）；false 回退独立 xlsx 行为
self._consolidate = bool(app_settings.cfg("report.consolidate_by_customer", True))
self._customer_subdir = app_settings.cfg("report.customer_workbook_subdir", "年度质保书")
# ------------------------------------------------------------------ #
# 文件夹命名（§7.2 逐字复用 VBA，全部由 config.yaml 驱动）
# ------------------------------------------------------------------ #
[文档]
def resolve_project_folder(self, plan_date: date | str, customer: str, project: str) -> Path:
"""返回 VBA 同款项目文件夹路径（不存在则创建）。
命名 pattern / 占位符 / 兜底文本全部来自 config.yaml 的
paths.report_archive.folder（蓝图 §7.2），后期只改 yaml 即可调整。
默认：{root}/{ym}计划项目资料/{rq}_{customer}_{project}项目资料
"""
d = self._as_date(plan_date)
cfg = self._folder_cfg
tokens = cfg.get("tokens") or {}
# 占位符求值模板（str.format，支持 {year}{month:02d}{day:02d}）
ym_tpl = tokens.get("ym", "{year}年{month:02d}月")
rq_tpl = tokens.get("rq", "{year}{month:02d}{day:02d}")
ym = ym_tpl.format(year=d.year, month=d.month, day=d.day)
rq = rq_tpl.format(year=d.year, month=d.month, day=d.day)
customer_v = (customer or "").strip()
project_v = (project or "").strip() or cfg.get(
"empty_project_placeholder", "【空白项目名称】"
)
pattern = cfg.get("pattern", "{ym}计划项目资料/{rq}_{customer}_{project}项目资料")
rel = pattern.format(ym=ym, rq=rq, customer=customer_v, project=project_v)
folder = self.root / rel
folder.mkdir(parents=True, exist_ok=True) # B0-4：写时惰性建目录（替代 VBA 预建）
return folder
[文档]
def create_placeholder_folder(
self,
plan_date: date | str,
customer: str,
project: str,
is_outsourced: bool = False,
) -> Path:
"""仅预建 VBA 风格占位项目文件夹（不写任何文件）。
供 B0-4 开关 `import.create_placeholder_dirs=true` 时在导入期调用，
恢复 VBA Mod_GCreatFloder.新建文件夹 的导入期预建行为。命名统一沿用
paths.report_archive.folder 的 pattern/tokens/兜底文本（与下游出报告落点一致）。
Args:
plan_date: 计划日期（用于 {yyyy年mm月} 与 yyyymmdd）
customer: 订货单位（VBA dw）
project: 项目名称（VBA xm；空→【空白项目名称】）
is_outsourced: 外购/自制标识（预留；当前统一命名不拆分根目录，避免 F 路径分类待定问题）
Returns:
Path: 已创建的占位项目文件夹路径
"""
# 直接复用 resolve_project_folder 的惰性 makedirs（同一命名规则）
return self.resolve_project_folder(plan_date, customer, project)
# ------------------------------------------------------------------ #
# 主流程
# ------------------------------------------------------------------ #
[文档]
def emit(self, req: ReportOutputRequest, export_pdf: bool | None = None) -> dict[str, Any]:
"""输出一份报告，返回双存根路径与状态。
两种落点（由 ``report.consolidate_by_customer`` 控制）：
- 合并模式（默认，VBA 风格）：把本报告 sheet 复制进
``{root}/{customer_workbook_subdir}/{customer}{year}.xlsx`` 客户簿，
同名 sheet 先删后写（覆盖）；PDF 按 sheet 派生到订单文件夹。
- 回退模式：每订单每类型独立 xlsx（双存根）到订单文件夹（原行为）。
订单内 ``报告目录.xlsx`` 已废弃：导航统一交给全局总目录 + 客户簿内「目录」sheet。
"""
if export_pdf is None:
export_pdf = app_settings.should_export_report_pdf()
folder = self.resolve_project_folder(req.plan_date, req.customer, req.project)
# 流水号须由上游(report_fill_service.generate)经统一自动编号服务
# (CertificateNumberService.allocate_report_serials)分配，禁止依赖 JSON/数据库双保险。
if not getattr(req, "serial", ""):
raise ValueError(
"ReportOutputRequest.serial 必填：报告流水号须由 report_fill_service.generate "
"经统一自动编号服务(CertificateNumberService)分配。"
)
base_name = self._build_filename(req)
pdf_path = folder / f"{base_name}.pdf"
sealed = False
pdf_ok = False
pdf_note = ""
sheet_name = None
stable_key = None
if self._consolidate:
# —— 合并模式：按订货单位年度合并到单一工作簿 ——
year = self._report_year(req)
xlsx_path = self.resolve_customer_workbook(req.customer, year)
stable_key = self._stable_key(req)
sheet_name = self._merge_sheet_into_workbook(req, xlsx_path)
logger.info(f"合并-xlsx(客户簿): {xlsx_path} [sheet={sheet_name}]")
else:
# —— 回退：每订单每类型独立 xlsx（双存根） ——
xlsx_path = folder / f"{base_name}.xlsx"
shutil.copy2(req.xlsx_source, xlsx_path)
logger.info(f"双存根-xlsx: {xlsx_path}")
if export_pdf:
seal = req.seal_image or self.seal_default
try:
if self._consolidate and sheet_name:
self._export_pdf_of_sheet(xlsx_path, sheet_name, pdf_path)
else:
self._export_pdf(xlsx_path, pdf_path)
pdf_ok = True
if seal and Path(seal).is_file():
self._overlay_seal(pdf_path, seal)
sealed = True
logger.info(f"pdf(已嵌章): {pdf_path}")
else:
# 默认印章图缺失：自动生成占位虚拟章（圆形红章 + 五角星 + 文字），
# 保证 PDF 可正常嵌章且用户可知其为占位。
generated = self._ensure_seal_default()
if generated:
self._overlay_seal(pdf_path, generated)
sealed = True
logger.warning(
f"pdf(已嵌【占位章】): {pdf_path} —— 默认印章图缺失，"
f"已自动生成占位章 {generated}，请替换为正式印章 resources/seals/default.png"
)
else:
logger.warning(
f"pdf(干净, 待手动盖章): {pdf_path} —— 默认印章图缺失且无法生成占位章"
)
except Exception as e: # 无 Excel / COM 不可用等
pdf_note = f"PDF 派生失败（{e}）；xlsx 已生成，可于装 Excel 的机器补导 PDF"
logger.warning(pdf_note)
else:
pdf_note = "PDF 未生成（已配置不同时导出，仅保留 xlsx 可手改档）"
logger.info(pdf_note)
# 全局总目录（绝对链接；xlsx→客户簿/独立文件，pdf/folder→订单文件夹）
try:
self._update_global_catalog(
req, xlsx_path, pdf_path, folder, sealed, pdf_ok, sheet_name, stable_key
)
except Exception as e: # 极端情况下不阻断主产出
logger.warning(f"全局报告总目录更新失败（跳过）: {e}")
return {
"success": True,
"folder": str(folder),
"xlsx": str(xlsx_path),
"sheet_name": sheet_name,
"pdf": str(pdf_path) if pdf_ok else None,
"sealed": sealed,
"pdf_ok": pdf_ok,
"pdf_note": pdf_note,
}
# ------------------------------------------------------------------ #
# 按订货单位合并到单一工作簿（VBA 风格 {customer}{year}.xlsx）
# ------------------------------------------------------------------ #
[文档]
def resolve_customer_workbook(self, customer: str, year: int) -> Path:
"""返回 VBA 风格客户合并簿路径：``{root}/{subdir}/{customer}{year}.xlsx``。"""
base_dir = self.root / self._customer_subdir
base_dir.mkdir(parents=True, exist_ok=True)
safe = re.sub(r'[\\/:*?"<>|]', "_", (customer or "未命名客户").strip()) or "未命名客户"
return base_dir / f"{safe}{year}.xlsx"
@staticmethod
def _report_year(req: ReportOutputRequest) -> int:
"""合并簿按年分簿：优先报告制作年（贴近 VBA「质量证明书2025」），回退计划年。"""
d = ReportOutputService._as_date(req.report_date) if req.report_date else None
if d is None:
d = ReportOutputService._as_date(req.plan_date) if req.plan_date else date.today()
return d.year
@staticmethod
def _stable_key(req: ReportOutputRequest) -> str:
"""报告稳定身份键（不含每日递增的 serial），用于重生成覆盖定位。
组成：订货单位|项目|计划日期|报告类型|变体。同订单同类型重生成时身份相同，
即使 sheet 展示名含新 serial，也能据此找到并删除旧 sheet，避免客户簿无限堆积。
"""
d = ReportOutputService._as_date(req.plan_date) if req.plan_date else None
pd = d.strftime("%Y%m%d") if d else ""
return "|".join(
[
req.customer or "",
req.project or "",
pd,
req.report_type or "",
req.variant or "",
]
)
_META_SHEET = "__meta"
@staticmethod
def _read_meta(dst_wb) -> dict:
if ReportOutputService._META_SHEET not in dst_wb.sheetnames:
return {}
ws = dst_wb[ReportOutputService._META_SHEET]
return {
ws.cell(row=r, column=1).value: ws.cell(row=r, column=2).value
for r in range(2, ws.max_row + 1)
if ws.cell(row=r, column=1).value
}
@staticmethod
def _write_meta(dst_wb, key: str, sheet_name: str) -> None:
title = ReportOutputService._META_SHEET
if title not in dst_wb.sheetnames:
ws = dst_wb.create_sheet(title=title)
ws.cell(row=1, column=1, value="key")
ws.cell(row=1, column=2, value="sheet")
ws.sheet_state = "hidden"
ws = dst_wb[title]
for r in range(2, ws.max_row + 1):
if ws.cell(row=r, column=1).value == key:
ws.cell(row=r, column=2, value=sheet_name)
return
ws.append([key, sheet_name])
def _merge_sheet_into_workbook(self, req: ReportOutputRequest, dst_path: Path) -> str:
"""把 ``req.xlsx_source`` 的唯一数据 sheet 复制进客户簿；按身份键覆盖旧 sheet。
重生成同订单同类型时，serial 会变导致展示 sheet 名不同，故用 stable_key 在
隐藏 ``__meta`` 中查旧 sheet 名并删除，避免客户簿无限堆积。返回新 sheet 名。
"""
from openpyxl import Workbook, load_workbook
src_wb = load_workbook(req.xlsx_source)
src_ws = src_wb.active # 单 sheet 临时簿
sheet_name = src_ws.title
stable_key = self._stable_key(req)
if dst_path.exists():
dst_wb = load_workbook(dst_path)
else:
dst_wb = Workbook()
if "Sheet" in dst_wb.sheetnames: # 移除 openpyxl 默认空 Sheet
dst_wb.remove(dst_wb["Sheet"])
# 覆盖：按 stable_key 找旧 sheet 名并删除（重生成不叠加）
meta = self._read_meta(dst_wb)
old = meta.get(stable_key)
if old and old in dst_wb.sheetnames:
dst_wb.remove(dst_wb[old])
self._copy_sheet_across_books(src_ws, dst_wb, sheet_name)
self._ensure_catalog_sheet(dst_wb)
self._upsert_catalog_row(dst_wb, req, sheet_name, stable_key)
self._write_meta(dst_wb, stable_key, sheet_name)
dst_wb.save(dst_path)
src_wb.close()
return sheet_name
@staticmethod
def _copy_sheet_across_books(src_ws, dst_wb, name: str) -> None: # noqa: C901
"""跨工作簿复制整张 sheet：值+样式+行高列宽+合并+分页符+打印设置。"""
from copy import copy as _copy
dst_ws = dst_wb.create_sheet(title=name)
max_r, max_c = src_ws.max_row, src_ws.max_column
for col, dim in src_ws.column_dimensions.items():
dst_ws.column_dimensions[col].width = dim.width
dst_ws.column_dimensions[col].hidden = dim.hidden
for r, dim in src_ws.row_dimensions.items():
dst_ws.row_dimensions[r].height = dim.height
dst_ws.row_dimensions[r].hidden = dim.hidden
for r in range(1, max_r + 1):
for c in range(1, max_c + 1):
s_cell = src_ws.cell(row=r, column=c)
d_cell = dst_ws.cell(row=r, column=c)
if s_cell.value is not None:
d_cell.value = s_cell.value
d_cell.font = _copy(s_cell.font)
d_cell.border = _copy(s_cell.border)
d_cell.fill = _copy(s_cell.fill)
d_cell.alignment = _copy(s_cell.alignment)
d_cell.number_format = s_cell.number_format
d_cell.protection = _copy(s_cell.protection)
for mr in list(src_ws.merged_cells.ranges):
dst_ws.merge_cells(
start_row=mr.min_row,
start_column=mr.min_col,
end_row=mr.max_row,
end_column=mr.max_col,
)
# 分页符（报告多页，必须复制，否则 PDF 分页错乱）
try:
for brk in src_ws.row_breaks.brk:
dst_ws.row_breaks.append(brk)
except Exception:
pass
try:
for brk in src_ws.col_breaks.brk:
dst_ws.col_breaks.append(brk)
except Exception:
pass
# 打印设置 + 打印区域 + 页面方向/纸张
try:
dst_ws.page_setup = _copy(src_ws.page_setup)
dst_ws.print_options = _copy(src_ws.print_options)
if src_ws.print_area:
dst_ws.print_area = src_ws.print_area
dst_ws.sheet_view.showGridLines = src_ws.sheet_view.showGridLines
except Exception:
pass
@staticmethod
def _ensure_catalog_sheet(dst_wb) -> None:
"""确保客户簿含「目录」sheet（首张），用于内部链接导航。列12 存身份键用于覆盖定位。"""
if "目录" in dst_wb.sheetnames:
return
ws = dst_wb.create_sheet(title="目录", index=0)
headers = [
"序号",
"报告类型",
"订货单位",
"项目名称",
"计划日期",
"制作日期",
"流水",
"变体",
"每页台数",
"语言",
"工作表",
"身份",
]
for c, h in enumerate(headers, 1):
ws.cell(row=1, column=c, value=h)
@staticmethod
def _upsert_catalog_row(
dst_wb, req: ReportOutputRequest, sheet_name: str, stable_key: str
) -> None:
"""在客户簿「目录」sheet 中按身份键 upsert 一行（覆盖重生成，不叠加）。"""
ws = dst_wb["目录"]
existing = None
for r in range(2, ws.max_row + 1):
if ws.cell(row=r, column=12).value == stable_key:
existing = r
break
row = existing or (ws.max_row + 1)
plan_d = ReportOutputService._as_date(req.plan_date) if req.plan_date else None
report_d = (
ReportOutputService._as_date(req.report_date) if req.report_date else date.today()
)
if existing is None:
max_seq = 0
for r in range(2, ws.max_row + 1):
v = ws.cell(row=r, column=1).value
if isinstance(v, int):
max_seq = max(max_seq, v)
ws.cell(row=row, column=1, value=max_seq + 1)
ws.cell(row=row, column=2, value=req.report_type or "")
ws.cell(row=row, column=3, value=req.customer or "")
ws.cell(row=row, column=4, value=req.project or "")
ws.cell(row=row, column=5, value=plan_d.strftime("%Y-%m-%d") if plan_d else "")
ws.cell(row=row, column=6, value=report_d.strftime("%Y-%m-%d"))
ws.cell(row=row, column=7, value=getattr(req, "serial", "") or "")
ws.cell(row=row, column=8, value=getattr(req, "variant_label", "") or "")
ws.cell(row=row, column=9, value=getattr(req, "records_per_page", "") or "")
ws.cell(row=row, column=10, value=getattr(req, "language", "") or "")
# 内部链接跳转到该 sheet（单引号包裹 sheet 名，兼容中文/括号）
ws.cell(row=row, column=11, value=f'=HYPERLINK("#\'{sheet_name}\'!A1", "打开")')
ws.cell(row=row, column=12, value=stable_key)
def _export_pdf_of_sheet(self, xlsx_path: Path, sheet_name: str, pdf_path: Path) -> None:
"""仅把客户簿中的目标 sheet 导出为 PDF（避免整簿多 sheet 误导）。"""
try:
import win32com.client # type: ignore
except ImportError:
raise RuntimeError("win32com 不可用（需 Windows + Excel）") from None
excel = win32com.client.Dispatch("Excel.Application")
excel.Visible = False
wb = excel.Workbooks.Open(str(xlsx_path.resolve()))
try:
ws = wb.Worksheets(sheet_name)
ws.Copy() # 复制到新建工作簿并设为 ActiveWorkbook
new_wb = excel.ActiveWorkbook
try:
new_wb.ExportAsFixedFormat(0, str(pdf_path.resolve()))
finally:
new_wb.Close(False)
finally:
wb.Close(False)
excel.Quit()
[文档]
def emit_batch(self, requests: list[ReportOutputRequest]) -> dict[str, Any]:
results = {"total": len(requests), "success": 0, "failed": 0, "items": []}
for req in requests:
try:
r = self.emit(req)
if r["success"]:
results["success"] += 1
results["items"].append(r)
except Exception as e:
results["failed"] += 1
results["items"].append({"success": False, "error": str(e)})
logger.error(f"批量输出失败: {e}")
return results
# ------------------------------------------------------------------ #
# PDF 派生（§7.1：win32com 最保真；无 Excel 时降级）
# ------------------------------------------------------------------ #
def _export_pdf(self, xlsx_path: Path, pdf_path: Path) -> None:
try:
import win32com.client # type: ignore
except ImportError:
raise RuntimeError("win32com 不可用（需 Windows + Excel）") from None
excel = win32com.client.Dispatch("Excel.Application")
excel.Visible = False
wb = excel.Workbooks.Open(str(xlsx_path.resolve()))
try:
# 0 = xlTypePDF
wb.ExportAsFixedFormat(0, str(pdf_path.resolve()))
finally:
wb.Close(False)
excel.Quit()
def _overlay_seal(self, pdf_path: Path, seal_path: Path) -> None:
"""将电子印章图叠加到 PDF 指定页（默认末页）。依赖 pypdf + Pillow + reportlab。"""
from pypdf import PdfReader, PdfWriter
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
cfg = self._seal_cfg
page_idx = int(cfg.get("page", -1))
x = float(cfg.get("x", 380))
y = float(cfg.get("y", 80))
width = float(cfg.get("width", 140))
packet = io.BytesIO()
c = canvas.Canvas(packet, pagesize=A4)
c.drawImage(
str(seal_path),
x,
y,
width=width,
preserveAspectRatio=True,
mask="auto",
)
c.save()
packet.seek(0)
reader = PdfReader(str(pdf_path))
overlay = PdfReader(packet).pages[0]
writer = PdfWriter()
n = len(reader.pages)
target = page_idx if page_idx >= 0 else (n - 1)
for i, pg in enumerate(reader.pages):
if i == target:
pg.merge_page(overlay)
writer.add_page(pg)
out = io.BytesIO()
writer.write(out)
pdf_path.write_bytes(out.getvalue())
def _ensure_seal_default(self) -> Path | None:
"""默认印章图缺失时，用 Pillow 生成占位虚拟章（圆形红章 + 五角星 + 文字）。
生成路径为配置中的 ``paths.report_archive.seal_default``（默认
``resources/seals/default.png``）。生成成功返回该路径，失败返回 ``None``。
仅作占位，用户应替换为正式印章图。
"""
if self.seal_default is None:
return None
seal_path = Path(self.seal_default)
if seal_path.is_file():
return seal_path
try:
from PIL import Image, ImageDraw
seal_path.parent.mkdir(parents=True, exist_ok=True)
size = 400
img = Image.new("RGBA", (size, size), (255, 255, 255, 0))
d = ImageDraw.Draw(img)
# 圆形红章外框（双环）
margin = 20
d.ellipse(
[margin, margin, size - margin, size - margin],
outline=(200, 0, 0, 255),
width=10,
)
d.ellipse(
[margin + 14, margin + 14, size - margin - 14, size - margin - 14],
outline=(200, 0, 0, 255),
width=3,
)
# 五角星（中心偏上）
cx, cy = size // 2, int(size * 0.38)
r_out, r_in = 70, 28
star = []
import math
for i in range(10):
ang = math.pi / 2 + i * math.pi / 5
r = r_out if i % 2 == 0 else r_in
star.append((cx + r * math.cos(ang), cy - r * math.sin(ang)))
d.polygon(star, fill=(200, 0, 0, 255))
# 弧形文字「测试占位章」
text = "测试占位章"
try:
from PIL import ImageFont
font = ImageFont.truetype(
str(app_settings.BASE_DIR / "resources/fonts/wqy-microhei.ttc"),
size=42,
)
except Exception:
font = ImageFont.load_default()
# 文字沿下半圆分布：逐字旋转摆放
n = len(text)
for idx, ch in enumerate(text):
ang = math.radians(-90 + (idx - (n - 1) / 2) * 26)
tx = cx + (size // 2 - 70) * math.cos(ang)
ty = cy + (size // 2 - 70) * math.sin(ang)
tmp = Image.new("RGBA", (80, 80), (255, 255, 255, 0))
td = ImageDraw.Draw(tmp)
td.text((40, 20), ch, font=font, fill=(200, 0, 0, 255), anchor="mm")
tmp = tmp.rotate(-math.degrees(ang) - 90, expand=True)
img.paste(tmp, (int(tx - tmp.width / 2), int(ty - tmp.height / 2)), tmp)
img.save(seal_path, "PNG")
logger.warning(
f"已自动生成占位印章图: {seal_path}（圆形红章+五角星+「测试占位章」），请替换为正式印章"
)
return seal_path
except Exception as e: # PIL 不可用或字体缺失等
logger.warning(f"生成占位印章图失败: {e}")
return None
# ------------------------------------------------------------------ #
# 报告目录（§7.4：HYPERLINK 索引，仿 BaoGaoMuLu）
# ------------------------------------------------------------------ #
# 报告目录表头（§7.4：HYPERLINK 索引，仿 BaoGaoMuLu）
# 顺序：项目级（计划日期/订货单位/项目名称/销售订单号）→ 报告级（类型/变体/每页台数/语言）→ 元数据（盖章/制作/生成时间）→ 文件链接
_CATALOG_HEADERS: tuple[str, ...] = (
"序号",
"计划日期",
"制作日期",
"订货单位",
"项目名称",
"销售订单号",
"报告类型",
"模板变体",
"每页台数",
"语言",
"是否盖章",
"生成时间",
"xlsx",
"pdf",
)
# 旧 7 列 → 新 14 列 升级映射（仅当升级时使用；按写入顺序索引 1-based）
_CATALOG_OLD_TO_NEW: dict[int, int] = {
1: 1, # 序号 → 序号
2: 7, # 报告类型 → 报告类型
3: 10, # 语言 → 语言
4: 11, # 是否盖章 → 是否盖章
5: 12, # 生成时间 → 生成时间
6: 13, # xlsx → xlsx
7: 14, # pdf → pdf
}
def _update_catalog(
self,
folder: Path,
req: ReportOutputRequest,
xlsx_path: Path,
pdf_path: Path,
sealed: bool,
pdf_ok: bool = False,
) -> None:
from openpyxl import Workbook, load_workbook
catalog = folder / "报告目录.xlsx"
new_headers = list(self._CATALOG_HEADERS)
if catalog.exists():
wb = load_workbook(catalog)
ws = wb.active
# 检测是否旧 7 列表头（第一格为「序号」），若是则升级为新 14 列布局
if (
ws.max_row >= 1
and ws.cell(row=1, column=1).value == "序号"
and ws.max_column < len(new_headers)
):
old_rows: list[list] = []
for r in range(2, ws.max_row + 1):
old_rows.append(
[ws.cell(row=r, column=c).value for c in range(1, ws.max_column + 1)]
)
# 重建 sheet：删旧表，新建同表名 sheet 写新表头
title = ws.title
wb.remove(ws)
ws = wb.create_sheet(title=title, index=0)
for col, h in enumerate(new_headers, 1):
ws.cell(row=1, column=col, value=h)
for _old in old_rows:
ws.append([""] * len(new_headers))
# 迁移旧列到新位置，并重写序号
for idx, old in enumerate(old_rows, start=2):
for old_col, val in enumerate(old, start=1):
new_col = self._CATALOG_OLD_TO_NEW.get(old_col)
if new_col is not None:
ws.cell(row=idx, column=new_col, value=val)
ws.cell(row=idx, column=1, value=idx - 1)
wb.save(catalog)
logger.info(f"报告目录已升级为 {len(new_headers)} 列: {catalog}")
# 升级后本次新行继续 append 在末尾
else:
wb = Workbook()
ws = wb.active
ws.title = "报告目录"
for col, h in enumerate(new_headers, 1):
ws.cell(row=1, column=col, value=h)
# 空表(max_row=1,仅表头)→新行=2；已有数据则续接
row = ws.max_row + 1
fname = xlsx_path.name
pdf_fname = pdf_path.name if pdf_ok else ""
plan_d = self._as_date(req.plan_date) if req.plan_date else None
report_d = self._as_date(req.report_date) if req.report_date else date.today()
ws.cell(row=row, column=1, value=row - 1)
ws.cell(row=row, column=2, value=plan_d.strftime("%Y-%m-%d") if plan_d else "")
ws.cell(row=row, column=3, value=report_d.strftime("%Y-%m-%d"))
ws.cell(row=row, column=4, value=req.customer or "")
ws.cell(row=row, column=5, value=req.project or "")
ws.cell(row=row, column=6, value=req.sales_order_no or "")
ws.cell(row=row, column=7, value=req.report_type or "")
ws.cell(row=row, column=8, value=req.variant_label or "")
ws.cell(row=row, column=9, value=req.records_per_page or "")
ws.cell(row=row, column=10, value=req.language or "")
ws.cell(row=row, column=11, value="是" if sealed else "否")
ws.cell(row=row, column=12, value=datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
ws.cell(row=row, column=13, value=f'=HYPERLINK("{fname}","打开xlsx")')
ws.cell(
row=row,
column=14,
value=(f'=HYPERLINK("{pdf_fname}","打开pdf")' if pdf_fname else "—"),
)
wb.save(catalog)
logger.info(f"报告目录已更新: {catalog}")
# ------------------------------------------------------------------ #
# 全局总目录（提升级别：根目录记录所有订单的全部资料，绝对路径超链接直接跳转）
# ------------------------------------------------------------------ #
_GLOBAL_CATALOG_HEADERS: tuple[str, ...] = (
"序号",
"生成时间",
"计划日期",
"订货单位",
"项目名称",
"销售订单号",
"报告类型",
"模板变体",
"每页台数",
"语言",
"是否盖章",
"项目文件夹",
"xlsx",
"pdf",
"工作表",
"身份",
)
def _update_global_catalog(
self,
req: ReportOutputRequest,
xlsx_path: Path,
pdf_path: Path,
folder: Path,
sealed: bool,
pdf_ok: bool = False,
sheet_name: str | None = None,
stable_key: str | None = None,
) -> None:
"""提升到根目录的全局报告总目录（默认 ``{root}/报告总目录.xlsx``）。
记录所有订单生成的所有资料：项目级信息 + 报告级信息 + 绝对路径超链接，
点击即可直接打开对应文件/文件夹，无需先进入订单文件夹再开订单级目录。
合并模式下 ``sheet_name`` 非空，每行对应客户簿内一张报告 sheet（按 workbook+sheet 去重）；
回退模式下 ``sheet_name`` 为空，按 xlsx 文件去重。导航统一交给本目录 + 客户簿内「目录」sheet。
"""
from openpyxl import Workbook, load_workbook
catalog = self.root / "报告总目录.xlsx"
new_headers = list(self._GLOBAL_CATALOG_HEADERS)
if catalog.exists():
wb = load_workbook(catalog)
ws = wb.active
# 列数不足或表头首格不符则重建（防御性升级）
if ws.max_column < len(new_headers) or ws.cell(row=1, column=1).value != "序号":
title = ws.title
wb.remove(ws)
ws = wb.create_sheet(title=title, index=0)
for col, h in enumerate(new_headers, 1):
ws.cell(row=1, column=col, value=h)
else:
wb = Workbook()
ws = wb.active
ws.title = "全局报告总目录"
for col, h in enumerate(new_headers, 1):
ws.cell(row=1, column=col, value=h)
row = ws.max_row + 1
# 去重：合并模式按 (xlsx, sheet_name) 定位；回退模式按 xlsx 文件名定位。
# 命中则覆盖该行（更新元数据），避免总目录无限累加重复行。
if stable_key:
existing = self._find_existing_row(ws, 16, stable_key)
elif sheet_name:
existing = self._find_existing_row(ws, 15, sheet_name)
else:
existing = self._find_existing_row(ws, 13, xlsx_path.name)
if existing is not None:
row = existing
abs_xlsx = xlsx_path.resolve().as_posix()
abs_pdf = pdf_path.resolve().as_posix() if pdf_ok else None
abs_folder = folder.resolve().as_posix()
plan_d = self._as_date(req.plan_date) if req.plan_date else None
ws.cell(row=row, column=1, value=row - 1)
ws.cell(row=row, column=2, value=datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
ws.cell(row=row, column=3, value=plan_d.strftime("%Y-%m-%d") if plan_d else "")
ws.cell(row=row, column=4, value=req.customer or "")
ws.cell(row=row, column=5, value=req.project or "")
ws.cell(row=row, column=6, value=req.sales_order_no or "")
ws.cell(row=row, column=7, value=req.report_type or "")
ws.cell(row=row, column=8, value=req.variant_label or "")
ws.cell(row=row, column=9, value=req.records_per_page or "")
ws.cell(row=row, column=10, value=req.language or "")
ws.cell(row=row, column=11, value="是" if sealed else "否")
ws.cell(
row=row,
column=12,
value=f'=HYPERLINK("file:///{abs_folder}", "打开文件夹")',
)
ws.cell(row=row, column=13, value=f'=HYPERLINK("{abs_xlsx}", "打开xlsx")')
ws.cell(
row=row,
column=14,
value=(f'=HYPERLINK("{abs_pdf}", "打开pdf")' if abs_pdf else "—"),
)
ws.cell(row=row, column=15, value=sheet_name or "")
ws.cell(row=row, column=16, value=stable_key or "")
wb.save(catalog)
logger.info(f"全局报告总目录已更新: {catalog}")
# ------------------------------------------------------------------ #
# 工具
# ------------------------------------------------------------------ #
@staticmethod
def _find_existing_row(ws, col: int, target: str) -> int | None:
"""在 ws 第 col 列查找已存在行号（去重用）。
- HYPERLINK 公式列：按文件 basename 匹配（如 xlsx/pdf 列）。
- 纯文本列（如「工作表」sheet 名、「身份」stable_key）：直接等值匹配。
"""
import re
pat = re.compile(r'=HYPERLINK\(\s*"([^"]+)"')
for r in range(2, ws.max_row + 1):
v = ws.cell(row=r, column=col).value
if v is None:
continue
if isinstance(v, str) and v.startswith("=HYPERLINK("):
m = pat.search(v)
if not m:
continue
path = m.group(1)
if path.startswith("file:///"):
path = path[len("file:///") :]
if path.replace("\\", "/").split("/")[-1] == target:
return r
elif str(v) == str(target):
return r
return None
@staticmethod
def _as_date(v: date | str) -> date:
if isinstance(v, date):
return v
for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y%m%d"):
try:
return datetime.strptime(v, fmt).date()
except ValueError:
continue
raise ValueError(f"无法解析计划日期: {v}")
@staticmethod
def _build_filename(req: ReportOutputRequest) -> str:
"""按配置 ``report.filename_format`` 组装文件名（占位符缺失→空，非法字符→下划线）。
占位符：``{report_type}`` ``{language}`` ``{product_no}`` ``{customer}``
``{project}`` ``{date}``（计划日期 yyyymmdd）。
"""
from string import Formatter
fmt = app_settings.get_report_filename_format()
d = ReportOutputService._as_date(req.plan_date) if req.plan_date else None
date_str = d.strftime("%Y%m%d") if d else ""
plan_short = d.strftime("%y%m%d") if d else ""
plan_year = d.year if d else ""
plan_month = d.month if d else ""
rd = ReportOutputService._as_date(req.report_date) if req.report_date else date.today()
report_date_str = rd.strftime("%y%m%d")
report_year = rd.year
ctx: dict[str, Any] = {
"report_type": req.report_type or "",
"language": req.language or "",
"product_no": req.product_no or "",
"customer": req.customer or "",
"project": req.project or "",
"date": date_str,
"plan_date_short": plan_short,
"plan_year": plan_year,
"plan_month": plan_month,
"report_date": report_date_str,
"report_year": report_year,
"serial": getattr(req, "serial", "") or "",
}
# 收集格式中引用的占位符，未提供的补空，避免 KeyError
used = {fn for _, fn, _, _ in Formatter().parse(fmt) if fn}
for key in used:
ctx.setdefault(key, "")
try:
name = fmt.format(**ctx)
except (KeyError, IndexError, ValueError):
# 兜底：保持旧行为（类型_语言_产品号）
parts = [req.report_type, req.language]
if req.product_no:
parts.append(req.product_no)
name = "_".join(parts)
# 清理非法文件名字符（Windows: \ / : * ? " < > |）
name = re.sub(r'[\\/:*?"<>|]', "_", name).strip("_ ") or "report"
# 模板变体标签：非空时追加，避免同类型多变体互相覆盖（如 12 模板批量生成）
if getattr(req, "variant", ""):
name = f"{name}_{req.variant}"
return name
__all__ = ["ReportOutputService", "ReportOutputRequest"]