# src/certflow/handlers/scan_handler.py
"""合格证扫描件处理器
负责扫描件的底层绘制和文件生成：
- 图片绘制 (PIL)
- 背景图片处理
- 文字渲染
- JPG/PDF 导出
- 布局计算
- 文件命名
"""
from __future__ import annotations
import os
from datetime import datetime
from PIL import Image, ImageDraw, ImageFont
from certflow.utils.logger import logger
[文档]
class ScanImageHandler:
"""扫描件图片处理器（底层绘制）"""
# 渲染分辨率：扫描件按 300 DPI 输出。背景图资源即按 300 DPI 制作
# （如 full_chinese_bg.png = 709×1187 ≈ 60×100mm@300DPI）。此前用 96 DPI
# （3.78 px/mm）会把高清背景下采样到约 227×378 像素，导致发虚；对齐 300 DPI
# 后背景 1:1 绘制、文字按 FONT_SCALE 同步放大，清晰度提升且版式比例不变。
RENDER_DPI = 300
PX_PER_MM = RENDER_DPI / 25.4 # ≈ 11.81 px/mm
# 字段坐标中的 font_size 以 96 DPI 设计稿为基准，渲染到 RENDER_DPI 时按此比例
# 放大，确保相对版式与该 96 DPI 预览一致（仅像素密度提升，不放大错位）。
FONT_SCALE = RENDER_DPI / 96.0
MM_TO_PX = PX_PER_MM # 兼容别名
def __init__(self) -> None:
"""初始化扫描件图片处理器（建立字体缓存）"""
self._font_cache = {}
# 字体候选默认值（与 paths.fonts.candidates 配置一致；配置缺失时回退此列表）。
# 内置 CJK 资源优先，其次系统路径，最后 DejaVu 回退；优先含中文形的字体避免方块(tofu)。
_FONT_CANDIDATES_DEFAULT = [
"resources/fonts/wqy-microhei.ttc",
"/usr/share/fonts/truetype/wqy/wqy-microhei.ttc",
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
"/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttf",
"simsun.ttc", # 宋体
"simhei.ttf", # 黑体
"msyh.ttc", # 微软雅黑
"C:/Windows/Fonts/simsun.ttc",
"C:/Windows/Fonts/simhei.ttf",
"C:/Windows/Fonts/msyh.ttc",
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
]
def _font_candidates(self) -> list[str]:
"""构造字体候选路径列表（配置驱动，见 ``paths.fonts.candidates``）。
相对路径（如 ``resources/fonts/...``）解析为项目根下的绝对路径；
内置资源字体（``resources/fonts/wqy-microhei.ttc``）始终置于最前，保证仓库自包含优先。
"""
from certflow.config.paths_override import cfg
from certflow.utils.path_utils import get_project_root
raw = cfg("paths.fonts.candidates", self._FONT_CANDIDATES_DEFAULT)
root = get_project_root()
internal = str(root / "resources" / "fonts" / "wqy-microhei.ttc")
candidates: list[str] = [internal] # 内置资源字体最高优先级（T8 配置化前的硬编码行为）
for p in raw:
p = str(p)
if not os.path.isabs(p):
# 相对路径优先按项目根解析，再保留原值兜底
candidates.append(str(root / p))
candidates.append(p)
return candidates
def _resolve_font_path(self) -> str | None:
"""解析首个可用的字体路径（结果缓存，避免每次重建字体探测开销）"""
if getattr(self, "_resolved_font", None):
return self._resolved_font
for path in self._font_candidates():
try:
ImageFont.truetype(path, 10)
self._resolved_font = path
return path
except OSError:
continue
self._resolved_font = None
return None
self._resolved_font = None
return None
def _get_font(self, font_size: int) -> ImageFont.FreeTypeFont:
"""获取字体（带缓存）。优先 CJK 字体以正确渲染中文，避免方块(tofu)。"""
key = int(font_size)
if key in self._font_cache:
return self._font_cache[key]
path = self._resolve_font_path()
try:
font = ImageFont.truetype(path, key) if path else ImageFont.load_default()
except OSError:
font = ImageFont.load_default()
self._font_cache[key] = font
return font
[文档]
def mm_to_px(self, mm: float) -> int:
"""毫米转像素
Args:
mm: 毫米值
Returns:
像素值
"""
return int(mm * self.PX_PER_MM)
# 色彩空间标记（ICC profile 头 16 字节内）到 PIL 图像 mode 的对应
_ICC_SPACE_TO_MODE = {"RGB": "RGB", "GRAY": "L", "CMYK": "CMYK", "LAB": "LAB"}
[文档]
@classmethod
def load_background_rgb(cls, background_path: str) -> Image.Image:
"""加载背景图并归一化到干净 sRGB（修复俄英文背景在 Linux/IDE 偏色）。
问题根因（实测 ``ru_en_bg.png``）：该 PNG 像素为 **RGB** 模式，却内嵌了约
557KB 的 **CMYK 打印机 profile**（``Japan Color 2001 Coated``, device
class=prtr, space=CMYK）——profile 与像素模式**不匹配**。支持色彩管理的
看图器（Linux/IDE 预览）会试图按该 CMYK profile 解读 RGB 像素 → 顶栏/印章
偏色；不做色彩管理的看图器则正常。全中文/中英文背景无 ICC，故不受影响。
解法：
- profile 的色彩空间与图像 mode **匹配**（如 RGB profile 配 RGB 图）→ 用
``ImageCms`` 正常转换到 sRGB；
- profile 与 mode **不匹配**（本例 CMYK profile 配 RGB 图，无法转换）→
判定为错误标签，**剥离 profile**，直接使用 RGB 像素值；
- 无 ICC → 仅确保 mode == RGB。
最终一律**清除 ``icc_profile``**，使生成结果在任意看图器下渲染一致（消除
用户所述「Linux 环境 / IDE 打开导致偏色」的不确定性）。
Args:
background_path: 背景图片路径。
Returns:
无 ICC profile 的 sRGB `Image.Image`。
"""
bg = Image.open(background_path)
icc = bg.info.get("icc_profile")
if icc:
# ICC profile 头第 16-20 字节为数据色彩空间标记（如 "RGB ""GRAY""CMYK"）
profile_space = icc[16:20].decode("ascii", "ignore").strip()
expected_mode = cls._ICC_SPACE_TO_MODE.get(profile_space)
if expected_mode == bg.mode:
try:
import io
from PIL import ImageCms
src_profile = ImageCms.ImageCmsProfile(io.BytesIO(icc))
dst_profile = ImageCms.createProfile("sRGB")
bg = ImageCms.profileToProfile(bg, src_profile, dst_profile, outputMode="RGB")
logger.debug(
"背景图 ICC({}) → sRGB 转换完成: {}", profile_space, background_path
)
except Exception as e: # 转换失败回退普通 RGB
logger.warning("背景图 ICC 转换失败，回退 convert(RGB): {}", e)
bg = bg.convert("RGB")
else:
# profile 与像素模式不匹配（错误标签），剥离 profile 直接用像素值
logger.warning(
"背景图 ICC profile({}) 与图像模式({}) 不匹配，已剥离: {}",
profile_space or "?",
bg.mode,
background_path,
)
bg = bg.convert("RGB")
elif bg.mode != "RGB":
bg = bg.convert("RGB")
# 一律清除 icc_profile，保证输出为干净 sRGB、跨看图器一致
bg.info.pop("icc_profile", None)
return bg
[文档]
def create_canvas(
self,
width_mm: float,
height_mm: float,
background_path: str | None = None,
) -> Image.Image:
"""创建画布
Args:
width_mm: 宽度（毫米）
height_mm: 高度（毫米）
background_path: 背景图片路径（可选）
Returns:
PIL Image 对象
"""
width_px = self.mm_to_px(width_mm)
height_px = self.mm_to_px(height_mm)
img = Image.new("RGB", (width_px, height_px), "white")
if background_path and os.path.exists(background_path):
try:
bg = self.load_background_rgb(background_path)
bg = bg.resize((width_px, height_px), Image.Resampling.LANCZOS)
img.paste(bg, (0, 0))
logger.debug(f"已加载背景图片: {background_path}")
except Exception as e:
logger.warning(f"加载背景图片失败: {e}")
return img
[文档]
def draw_text(
self,
img: Image.Image,
text: str,
x_mm: float,
y_mm: float,
font_size: int = 9,
color: str = "black",
) -> None:
"""在画布上绘制文字（单段，不自动换行；框内换行请用 draw_field）。
Args:
img: PIL Image 对象
text: 文字内容
x_mm: X 坐标（毫米）
y_mm: Y 坐标（毫米）
font_size: 字号（像素）
color: 颜色
Returns:
None: 直接在 img 上绘制，无返回值
"""
self.draw_field(img, text, x_mm, y_mm, font_size, color=color)
[文档]
def draw_field(
self,
img: Image.Image,
value: str,
x_mm: float,
y_mm: float,
font_size: int = 9,
width_mm: float | None = None,
height_mm: float | None = None,
align: str = "left",
v_align: str = "top",
color: str = "black",
) -> None:
"""绘制字段：按框宽自动换行、超框高自动缩小字号、按对齐绘制。
解决阶段5扫描件两类渲染缺陷（BUG-006）：
- 长字符串（如产品型号 ``BESDZY-320``）按框宽折行，不再溢出整行；
- 字号按基准 ``font_size`` 给出，不再被放大 2 倍；
- 内容超高时自动缩小字号以适配框高；
- 支持水平(align)/垂直(v_align)对齐，使用坐标配置中的 width/height。
Args:
img: PIL Image 对象
value: 字段文字内容
x_mm, y_mm: 字段框左上角坐标（毫米）
font_size: 基准字号（像素）
width_mm, height_mm: 字段框宽/高（毫米）；提供后启用换行与缩放
align: 水平对齐 left/center/right
v_align: 垂直对齐 top/center/bottom
color: 文字颜色
Returns:
None: 直接在 img 上绘制，无返回值
"""
if not value or not str(value).strip():
return
draw = ImageDraw.Draw(img)
x_px = self.mm_to_px(x_mm)
y_px = self.mm_to_px(y_mm)
width_px = self.mm_to_px(width_mm) if width_mm else None
height_px = self.mm_to_px(height_mm) if height_mm else None
text = str(value)
# 自适应字号：从 font_size 起，换行后总高超出框高则逐步缩小至下限，
# 确保长内容（含多行产品名称）不溢出框。
min_size = max(6, int(font_size * 0.6 * self.FONT_SCALE))
size = int(font_size * self.FONT_SCALE)
font = self._get_font(size)
lines = self._wrap_text(text, font, width_px)
line_height = int(font.size * 1.25)
while height_px and line_height * len(lines) > height_px and size > min_size:
size -= 1
font = self._get_font(size)
lines = self._wrap_text(text, font, width_px)
line_height = int(font.size * 1.25)
total_height = line_height * len(lines)
ty = y_px
if height_px and v_align == "center":
ty = y_px + max(0, (height_px - total_height) // 2)
elif height_px and v_align == "bottom":
ty = y_px + max(0, height_px - total_height)
for i, line in enumerate(lines):
line_w = draw.textlength(line, font=font)
if align == "center" and width_px:
tx = x_px + max(0, (width_px - line_w) // 2)
elif align == "right" and width_px:
tx = x_px + max(0, width_px - line_w)
else:
tx = x_px
draw.text((tx, ty + i * line_height), line, fill=color, font=font)
def _wrap_text(
self, text: str, font: ImageFont.FreeTypeFont, max_width: int | None
) -> list[str]:
"""按最大像素宽度折行：CJK 逐字断行，ASCII 尽量在空格处断词。
Args:
text: 原始文本
font: 当前字体（用于测量宽度）
max_width: 最大像素宽度；为 None 时仅按显式换行符分段
Returns:
折行后的文本行列表
"""
if max_width is None:
return text.split("\n")
out: list[str] = []
for paragraph in text.split("\n"):
if paragraph == "":
out.append("")
continue
cur = ""
for ch in paragraph:
if font.getlength(cur + ch) <= max_width:
cur += ch
else:
# 优先在空格处断词，保持 ASCII 单词完整；否则逐字断行
if " " in cur:
head, tail = cur.rsplit(" ", 1)
out.append(head)
cur = tail + ch
else:
if cur:
out.append(cur)
cur = ch
out.append(cur)
return out
[文档]
def save_jpg(self, img: Image.Image, file_path: str, quality: int = 95) -> str:
"""保存为 JPG 文件
Args:
img: PIL Image 对象
file_path: 文件路径
quality: 质量 (1-100)
Returns:
文件路径
"""
os.makedirs(os.path.dirname(file_path), exist_ok=True)
# 写入 DPI 元数据，使看图器按物理尺寸显示（避免被默认按 96 DPI 缩成小图）
img.save(file_path, "JPEG", quality=quality, dpi=(self.RENDER_DPI, self.RENDER_DPI))
logger.debug(
f"已保存 JPG: {file_path} ({img.size[0]}x{img.size[1]} @ {self.RENDER_DPI}DPI)"
)
return file_path
[文档]
def save_pdf(self, img: Image.Image, file_path: str) -> str:
"""保存为 PDF 文件
Args:
img: PIL Image 对象
file_path: 文件路径
Returns:
文件路径
"""
os.makedirs(os.path.dirname(file_path), exist_ok=True)
img.save(file_path, "PDF", resolution=float(self.RENDER_DPI))
logger.debug(f"已保存 PDF: {file_path}")
return file_path
[文档]
class ScanLayoutHandler:
"""扫描件布局处理器（负责位置计算）"""
def __init__(self, rows: int = 1, cols: int = 1) -> None:
"""初始化布局处理器
Args:
rows: 每页行数（合格证张数），默认 1
cols: 每页列数（合格证张数），默认 1
"""
self.rows = rows
self.cols = cols
self.per_page = rows * cols
[文档]
def calculate_canvas_size(
self,
cert_width_mm: float,
cert_height_mm: float,
) -> tuple[int, int]:
"""计算画布尺寸
Args:
cert_width_mm: 单张合格证宽度（毫米）
cert_height_mm: 单张合格证高度（毫米）
Returns:
(width_px, height_px) 像素尺寸
"""
mm_to_px = ScanImageHandler.PX_PER_MM
width_px = int(cert_width_mm * self.cols * mm_to_px)
height_px = int(cert_height_mm * self.rows * mm_to_px)
return width_px, height_px
[文档]
def calculate_canvas_size_mm(
self,
cert_width_mm: float,
cert_height_mm: float,
) -> tuple[float, float]:
"""计算画布尺寸（毫米）
Args:
cert_width_mm: 单张合格证宽度（毫米）
cert_height_mm: 单张合格证高度（毫米）
Returns:
(width_mm, height_mm) 毫米尺寸
"""
width_mm = cert_width_mm * self.cols
height_mm = cert_height_mm * self.rows
return width_mm, height_mm
[文档]
def get_position(
self,
index: int,
cert_width_mm: float,
cert_height_mm: float,
) -> tuple[float, float, int, int]:
"""获取指定索引的位置
Args:
index: 合格证索引（0-based）
cert_width_mm: 单张宽度（毫米）
cert_height_mm: 单张高度（毫米）
Returns:
(x_offset_mm, y_offset_mm, row, col)
"""
row = index // self.cols
col = index % self.cols
x_offset_mm = col * cert_width_mm
y_offset_mm = row * cert_height_mm
return x_offset_mm, y_offset_mm, row, col
[文档]
def is_multi_page(self) -> bool:
"""是否为多张拼接模式
Args:
无
Returns:
bool: 当 rows*cols > 1（多张拼接）时为 True
"""
return self.per_page > 1
[文档]
def get_total_pages(self, total_certs: int) -> int:
"""计算总页数
Args:
total_certs: 总合格证数量
Returns:
总页数
"""
if self.per_page <= 1:
return total_certs
return (total_certs + self.per_page - 1) // self.per_page
[文档]
def effective_grid(self, n: int) -> tuple[int, int]:
"""返回实际占用网格 ``(used_rows, used_cols)``，用于最后一页（非满）裁剪画布。
满页（``n == per_page``）时返回完整 ``(rows, cols)``，行为不变；非满页时
去除多余空白单元格：
- 仅当所有单元落在**单行**（``used_rows == 1``）时，按实际列数横向裁剪，
彻底消除尾部空白；
- 否则（上方已有满行，仅末行不满）保留整列宽，仅裁剪纵向空行。
这样最后一页不会生成整块 180×200 的空白合格证，画布尺寸贴合实际内容。
Args:
n: 本页实际单元（单台编号）数量。
Returns:
(used_rows, used_cols) 实际参与排版的行列数。
"""
if n <= 0:
return (0, 0)
used_rows = (n + self.cols - 1) // self.cols
used_cols = n if used_rows == 1 else self.cols
return (used_rows, used_cols)
[文档]
class ScanExportHandler:
"""扫描件导出处理器（负责文件命名和组织）"""
def __init__(self, output_dir: str, output_format: str = "JPG") -> None:
"""初始化扫描件导出处理器
Args:
output_dir: 导出文件输出目录
output_format: 导出格式，"JPG" 或 "PDF"，默认 "JPG"（自动转大写）
"""
self.output_dir = output_dir
self.output_format = output_format.upper()
[文档]
def ensure_output_dir(self) -> None:
"""确保输出目录存在
Args:
无
Returns:
None: 递归创建 output_dir，已存在则跳过
"""
os.makedirs(self.output_dir, exist_ok=True)
[文档]
def generate_filename_single(self, certificate_no: str) -> str:
"""生成单张模式文件名
Args:
certificate_no: 合格证编号
Returns:
完整文件路径
"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
ext = self.output_format.lower()
# 清理文件名中的非法字符
safe_name = self._sanitize_filename(certificate_no or "certificate")
filename = f"{safe_name}_{timestamp}.{ext}"
return os.path.join(self.output_dir, filename)
[文档]
def generate_filename_multi(self, page_num: int, total_pages: int) -> str:
"""生成多张拼接模式文件名
Args:
page_num: 当前页码
total_pages: 总页数
Returns:
完整文件路径
"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
ext = self.output_format.lower()
filename = f"scan_page{page_num:03d}_of_{total_pages:03d}_{timestamp}.{ext}"
return os.path.join(self.output_dir, filename)
[文档]
def generate_filename(self, base_name: str, page_num: int = 1, total_pages: int = 1) -> str:
"""生成文件名（兼容接口）
Args:
base_name: 基础名称
page_num: 页码
total_pages: 总页数
Returns:
完整文件路径
"""
if total_pages == 1:
return self.generate_filename_single(base_name)
return self.generate_filename_multi(page_num, total_pages)
@staticmethod
def _sanitize_filename(filename: str) -> str:
"""清理文件名中的非法字符"""
illegal_chars = '<>:"/\\|?*'
for char in illegal_chars:
filename = filename.replace(char, "_")
return filename[:100] # 限制长度