Files
wordcloud/backend/core/pipeline.py
T
broccoliandClaude Sonnet 5 1d17b5e20d Rework layout engine around exact-glyph collision, add tests and docs sync
Replace the old bbox/heuristic placement (scale search rounds, large-font
capping, stratified sampling, fill-retry ladders) with an area-model font
sizing pass feeding a C++ exact-glyph collision engine (centroid-biased
spiral + random probing, HD clearance refinement, density/hole
optimization). Simplify the frontend advanced-params panel and JobParams
type to match the surviving config surface, add a layout-constraints test
suite and a repeatable benchmark tool, and bring docs/*.md back in sync
with current code (plus new TESTING.md and DEPLOYMENT.md).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-26 18:32:25 +08:00

837 lines
34 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import logging
import math
import os
import sqlite3
import sys
import time
from pathlib import Path
import numpy as np
from PIL import Image, ImageDraw, ImageFont
import pandas as pd
from . import config
from .fonts import get_cached_font
from .layout import OptimizedEfficientWordCloud
from .mask import analyze_mask, apply_safe_padding, calculate_dynamic_dimensions, prepare_mask
from .render import (
compute_fill_ratio_fast,
count_layout_overlap_pixels,
largest_empty_square_size,
refine_layout_with_hd_clearance,
scale_layout_for_hd,
)
from .weights import (
calculate_font_by_area_model,
extract_weights_from_df,
get_stroke_complexity_batch,
merge_weight_maps,
)
log = logging.getLogger("core.pipeline")
def run_generation_pass(
names,
frequencies_data,
name_weights_map,
mask_hd,
real_hd_w,
real_hd_h,
_collision_margin=0,
):
log.info("[run_generation_pass] 开始")
log.info(" 输入: %d 词 | HD尺寸: %dx%d", len(names), real_hd_w, real_hd_h)
w_small = max(1, int(real_hd_w * config.WORK_SCALE))
h_small = max(1, int(real_hd_h * config.WORK_SCALE))
img_small = Image.fromarray(mask_hd).resize((w_small, h_small), Image.NEAREST)
mask_small = np.array(img_small)
apply_safe_padding(mask_small)
log.info(" 运算网格: %dx%d (WORK_SCALE=%.4f)", w_small, h_small, config.WORK_SCALE)
log.info(" mask_small 统计: 总像素=%d, 空闲像素=%d, 空闲率=%.4f",
mask_small.size, int(np.sum(mask_small == 0)),
int(np.sum(mask_small == 0)) / mask_small.size if mask_small.size else 0)
if config.SAVE_DEBUG_IMAGES:
debug_dir = Path(config.DEBUG_OUTPUT_DIR)
debug_dir.mkdir(parents=True, exist_ok=True)
Image.fromarray(mask_hd).save(str(debug_dir / "mask_hd.png"))
Image.fromarray(mask_small).save(str(debug_dir / "mask_small.png"))
print(f"最终输出: {real_hd_w}x{real_hd_h} | 运算网格: {w_small}x{h_small}")
readable_min_font = max(
config.MIN_FONT_FLOOR,
int(math.ceil(config.MIN_READABLE_HEIGHT_PX * config.WORK_SCALE)),
)
total_target = len(names) * config.N_REPETITIONS
final_wc = None
final_scale = 1.0
base_layout_seed = config.LAYOUT_SEED if config.LAYOUT_SEED is not None else config.SEED
final_layout_seed = base_layout_seed
base_min_font, base_max_font = calculate_font_by_area_model(
mask_small,
names,
name_weights_map,
config.TARGET_FILL_RATIO,
config.SIZE_RATIO,
config.PACKING_EFFICIENCY,
config.N_REPETITIONS,
)
explicit_min_font = config.USER_MIN_FONT_SIZE is not None
hard_min_font = readable_min_font
if explicit_min_font:
hard_min_font = max(config.MIN_FONT_FLOOR, int(round(config.USER_MIN_FONT_SIZE)))
hard_max_font = None
if config.USER_MAX_FONT_SIZE is not None:
hard_max_font = max(config.MIN_FONT_FLOOR, int(round(config.USER_MAX_FONT_SIZE)))
if explicit_min_font and hard_max_font < hard_min_font:
raise ValueError(
f"字号硬约束冲突: USER_MAX_FONT_SIZE={hard_max_font} "
f"小于最小允许字号 {hard_min_font}"
)
if not explicit_min_font:
# A user-specified maximum outranks the automatic readability
# suggestion. It remains an exact ceiling rather than causing an
# artificial conflict with a value the user never requested.
hard_min_font = min(hard_min_font, hard_max_font)
equal_size_mode = math.isclose(float(config.SIZE_RATIO), 1.0, rel_tol=0.0, abs_tol=1e-9)
base_min_font = max(hard_min_font, int(base_min_font))
base_max_font = max(base_min_font, int(base_max_font))
if hard_max_font is not None:
base_min_font = min(base_min_font, hard_max_font)
base_max_font = min(base_max_font, hard_max_font)
if equal_size_mode:
# Keep one scalar throughout every retry. This is what makes
# SIZE_RATIO=1 exact even after automatic batch scaling.
equal_font = min(base_min_font, base_max_font)
base_min_font = equal_font
base_max_font = equal_font
def scaled_bounds(scale):
if equal_size_mode:
size = max(hard_min_font, int(round(base_min_font * scale)))
if hard_max_font is not None:
size = min(size, hard_max_font)
return size, size
min_font = max(hard_min_font, int(round(base_min_font * scale)))
max_font = max(min_font, int(round(base_max_font * scale)))
if hard_max_font is not None:
min_font = min(min_font, hard_max_font)
max_font = min(max_font, hard_max_font)
return min_font, max(min_font, max_font)
def try_place(scale, layout_seed=base_layout_seed):
min_font, max_font = scaled_bounds(scale)
wc = OptimizedEfficientWordCloud(
width=w_small,
height=h_small,
mask=mask_small,
font_path=config.WC_FONT_PATH,
max_words=total_target,
min_font_size=min_font,
max_font_size=max_font,
background_color=config.get_output_background(),
prefer_horizontal=0.82,
# HD clearance is applied after scaling. Keeping the coarse-grid
# margin at zero avoids turning 1 HD pixel into 5-6 output pixels.
margin=_collision_margin,
)
wc.layout_seed = layout_seed
wc.generate_from_frequencies(frequencies_data)
return wc, len(wc.layout_), min_font, max_font
print(f"--- 5. 启动生成 (目标: {total_target} 词) ---")
log.info("--- 5. 启动生成 ---")
log.info(" 目标词数: %d (names=%d * N_REPETITIONS=%d)", total_target, len(names), config.N_REPETITIONS)
log.info(
" 基准字号: [%d, %d], 硬边界: [%d, %s], 等字号=%s",
base_min_font,
base_max_font,
hard_min_font,
hard_max_font if hard_max_font is not None else "∞",
equal_size_mode,
)
# Every attempt is a fresh, whole-cloud layout. No word may silently
# receive a smaller fallback size. Usually the area model succeeds on
# attempt one; two adaptive retries cover fragmentation-heavy masks.
best_wc = None
best_count = 0
best_scale = 1.0
scale = 1.0
tried_layouts = set()
failed_scales = []
for attempt in range(1, 4):
bounds = scaled_bounds(scale)
layout_key = (bounds, base_layout_seed)
if layout_key in tried_layouts:
break
tried_layouts.add(layout_key)
wc, placed_count, min_font, max_font = try_place(scale)
print(
f" 整批布局 #{attempt}: scale={scale:.3f}, "
f"字号=[{min_font}, {max_font}] -> {placed_count}/{total_target}"
)
log.info(
" 整批布局 #%d scale=%.3f 字号=[%d,%d] -> %d/%d",
attempt,
scale,
min_font,
max_font,
placed_count,
total_target,
)
if placed_count > best_count:
best_wc = wc
best_count = placed_count
best_scale = scale
if placed_count >= total_target:
final_wc = wc
final_scale = scale
final_layout_seed = base_layout_seed
break
failed_scales.append(scale)
placed_ratio = placed_count / max(1, total_target)
# Required box area is roughly proportional to size². The extra
# safety margin absorbs fragmentation without wasting a binary search.
shrink = 0.62 if placed_ratio <= 0 else min(0.92, max(0.58, math.sqrt(placed_ratio) * 0.92))
scale *= shrink
if final_wc is None:
final_wc = best_wc
final_scale = best_scale
if final_wc is None:
return {
"wc": None,
"fill_ratio": 0.0,
"occ_fast": None,
"w_small": w_small,
"h_small": h_small,
"base_min_font": base_min_font,
"base_max_font": base_max_font,
"mask_small": mask_small,
"size_scale": final_scale,
"layout_seed": final_layout_seed,
"collision_margin": _collision_margin,
"hd_overlap_pixels": 0,
"hd_layout": [],
"hd_clearance": None,
}
fill_ratio, occ_fast = compute_fill_ratio_fast(final_wc.layout_, mask_small, config.WC_FONT_PATH)
largest_empty_square = largest_empty_square_size(occ_fast, mask_small)
def has_character_sized_hole(wc, hole_size):
if not equal_size_mode or wc is None or not wc.layout_:
return False
font_size = int(wc.layout_[0][1])
return hole_size >= max(2, int(math.ceil(font_size * 1.25)))
complete_candidates = []
if len(final_wc.layout_) >= total_target:
complete_candidates.append(
(final_wc, final_scale, final_layout_seed, fill_ratio, occ_fast)
)
print(f"填充率: {fill_ratio:.3f}")
log.info("[填充率] 初始填充率: %.4f", fill_ratio)
log.info(" layout_ 词数: %d", len(final_wc.layout_))
if config.SAVE_DEBUG_IMAGES and occ_fast is not None:
debug_dir = Path(config.DEBUG_OUTPUT_DIR)
debug_dir.mkdir(parents=True, exist_ok=True)
Image.fromarray((occ_fast * 255).astype(np.uint8)).save(str(debug_dir / "occ_fast.png"))
# Probe larger whole-cloud layouts and keep the largest complete one. If
# an earlier batch was too large, search the discrete interval between the
# complete and failed scales instead of accepting an over-aggressive
# shrink. Every probe rebuilds the entire cloud with one shared scale.
if (
len(final_wc.layout_) >= total_target
and fill_ratio > 0
and (
fill_ratio < config.TARGET_FILL_RATIO * 0.98
or has_character_sized_hole(final_wc, largest_empty_square)
)
):
upper_scale = min(
(failed for failed in failed_scales if failed > final_scale),
default=None,
)
for density_attempt in range(1, 5):
if upper_scale is not None:
grow_scale = (final_scale + upper_scale) / 2.0
elif equal_size_mode:
current_size, _ = scaled_bounds(final_scale)
grow_scale = (current_size + 1) / max(1, base_min_font)
else:
desired_growth = min(
1.12,
math.sqrt(config.TARGET_FILL_RATIO / fill_ratio) * 0.98,
)
if desired_growth <= 1.005:
break
grow_scale = final_scale * desired_growth
grow_bounds = scaled_bounds(grow_scale)
if grow_bounds == scaled_bounds(final_scale):
break
grow_min, grow_max = grow_bounds
layout_key = (grow_bounds, base_layout_seed)
attempted_layout = False
if layout_key not in tried_layouts:
tried_layouts.add(layout_key)
wc, placed_count, _, _ = try_place(grow_scale)
attempted_layout = True
print(
f" 密度优化 #{density_attempt}: scale={grow_scale:.3f}, "
f"字号=[{grow_min}, {grow_max}] -> {placed_count}/{total_target}"
)
else:
wc, placed_count = None, -1
selected_seed = base_layout_seed
if placed_count < total_target and base_layout_seed is not None:
# The reference library samples a fresh legal position order.
# One bounded whole-cloud re-layout recovers dense solutions
# without per-word shrinking or an unbounded random search.
candidate_seed = (int(base_layout_seed) * 3 + 3) % (2**31 - 1)
retry_key = (grow_bounds, candidate_seed)
if retry_key not in tried_layouts:
tried_layouts.add(retry_key)
retry_wc, retry_count, _, _ = try_place(grow_scale, candidate_seed)
attempted_layout = True
print(
f" 密度优化 #{density_attempt} 整批重排: "
f"seed={candidate_seed}, 字号=[{grow_min}, {grow_max}] -> "
f"{retry_count}/{total_target}"
)
if retry_count > placed_count:
wc = retry_wc
placed_count = retry_count
selected_seed = candidate_seed
if not attempted_layout:
break
if placed_count < total_target:
upper_scale = grow_scale
continue
new_fill, new_occ = compute_fill_ratio_fast(wc.layout_, mask_small, config.WC_FONT_PATH)
if new_fill <= fill_ratio:
break
final_wc = wc
final_scale = grow_scale
final_layout_seed = selected_seed
fill_ratio = new_fill
occ_fast = new_occ
largest_empty_square = largest_empty_square_size(occ_fast, mask_small)
complete_candidates.append(
(final_wc, final_scale, final_layout_seed, fill_ratio, occ_fast)
)
if (
fill_ratio >= config.TARGET_FILL_RATIO * 0.98
and not has_character_sized_hole(final_wc, largest_empty_square)
):
break
# At the largest complete equal-size tier, compare a small deterministic
# set of whole-cloud reorderings and retain the one with the smallest
# character-scale void. This never changes an individual font size.
if (
equal_size_mode
and len(final_wc.layout_) >= total_target
and has_character_sized_hole(final_wc, largest_empty_square)
and base_layout_seed is not None
):
if total_target < 100:
hole_attempt_budget = 3
elif total_target <= 300:
hole_attempt_budget = 2
else:
hole_attempt_budget = 1
modulus = 2**31 - 1
for hole_attempt in range(1, hole_attempt_budget + 1):
candidate_seed = (
int(base_layout_seed) ^ ((0x9E3779B9 * hole_attempt) & 0x7FFFFFFF)
) % modulus
bounds = scaled_bounds(final_scale)
layout_key = (bounds, candidate_seed)
if layout_key in tried_layouts:
continue
tried_layouts.add(layout_key)
candidate_wc, placed_count, _, _ = try_place(final_scale, candidate_seed)
if placed_count < total_target:
continue
candidate_fill, candidate_occ = compute_fill_ratio_fast(
candidate_wc.layout_, mask_small, config.WC_FONT_PATH
)
candidate_hole = largest_empty_square_size(candidate_occ, mask_small)
print(
f" 空洞优化 #{hole_attempt}: seed={candidate_seed}, "
f"最大空白={candidate_hole}px -> {placed_count}/{total_target}"
)
if candidate_hole >= largest_empty_square:
continue
final_wc = candidate_wc
final_layout_seed = candidate_seed
fill_ratio = candidate_fill
occ_fast = candidate_occ
largest_empty_square = candidate_hole
complete_candidates.append(
(final_wc, final_scale, final_layout_seed, fill_ratio, occ_fast)
)
hd_layout = None
hd_clearance = None
raw_hd_layout = []
for candidate_index, candidate in enumerate(reversed(complete_candidates), start=1):
candidate_wc, candidate_scale, candidate_seed, candidate_fill, candidate_occ = candidate
raw_hd_layout = scale_layout_for_hd(candidate_wc.layout_, config.WORK_SCALE)
refined_layout = None
clearance_stats = None
clearance_modes = (1, 0)
if total_target >= 100 and candidate_index > 1:
clearance_modes = (0,)
for clearance_px in clearance_modes:
priority_layout = raw_hd_layout
priority_restarts = 0
max_priority_attempts = 2 if clearance_px == 0 or total_target < 100 else 1
for priority_attempt in range(max_priority_attempts):
refined_layout, clearance_stats = refine_layout_with_hd_clearance(
priority_layout,
mask_hd,
config.WC_FONT_PATH,
clearance=clearance_px,
)
if refined_layout is not None:
break
failed_word = clearance_stats["failed_word"]
failed_index = next(
(index for index, item in enumerate(priority_layout) if item[0] == failed_word),
None,
)
if (
failed_index is None
or failed_index == 0
or priority_attempt + 1 >= max_priority_attempts
):
break
failed_item = priority_layout[failed_index]
priority_layout = [failed_item, *priority_layout[:failed_index], *priority_layout[failed_index + 1:]]
priority_restarts += 1
if refined_layout is not None:
clearance_stats["clearance_px"] = clearance_px
clearance_stats["priority_restarts"] = priority_restarts
break
clearance_stats["clearance_px"] = clearance_px
clearance_stats["priority_restarts"] = priority_restarts
if refined_layout is None:
log.warning(
"高清候选 #%d 隔离精修失败: word=%s, shifted=%d, max_shift=%d",
candidate_index,
clearance_stats["failed_word"],
clearance_stats["shifted_words"],
clearance_stats["max_shift"],
)
hd_clearance = clearance_stats
continue
final_wc = candidate_wc
final_scale = candidate_scale
final_layout_seed = candidate_seed
fill_ratio = candidate_fill
occ_fast = candidate_occ
largest_empty_square = largest_empty_square_size(occ_fast, mask_small)
hd_layout = refined_layout
hd_clearance = clearance_stats
break
if hd_layout is None:
hd_overlap_pixels = -1
hd_layout = raw_hd_layout
if hd_clearance is None:
hd_clearance = {
"shifted_words": 0,
"max_shift": 0,
"clearance_px": None,
"priority_restarts": 0,
"failed_word": None,
}
else:
hd_overlap_pixels = count_layout_overlap_pixels(
hd_layout,
(real_hd_h, real_hd_w),
config.WC_FONT_PATH,
)
print(
f"最终填充率: {fill_ratio:.3f} | 高清重叠像素: {hd_overlap_pixels} | "
f"精修位移: {hd_clearance['shifted_words']} 词, 最大 {hd_clearance['max_shift']}px | "
f"隔离带: {hd_clearance['clearance_px']}px"
)
return {
"wc": final_wc,
"fill_ratio": fill_ratio,
"occ_fast": occ_fast,
"w_small": w_small,
"h_small": h_small,
"base_min_font": base_min_font,
"base_max_font": base_max_font,
"mask_small": mask_small,
"size_scale": final_scale,
"layout_seed": final_layout_seed,
"collision_margin": _collision_margin,
"hd_overlap_pixels": hd_overlap_pixels,
"hd_layout": hd_layout,
"hd_clearance": hd_clearance,
"largest_empty_square_work_px": largest_empty_square,
"largest_empty_square_font_ratio": (
largest_empty_square / max(1, int(final_wc.layout_[0][1]))
if final_wc.layout_ and equal_size_mode
else None
),
}
def main():
t_start = time.time()
print("--- 1. 读取数据 ---")
log.info("=" * 60)
log.info("[Pipeline] main() 开始")
log.info(" EXCEL_PATH = %s", config.EXCEL_PATH)
log.info(" DATA_COL = %d", config.DATA_COL_INDEX)
log.info(" MODE = %s", config.MODE)
log.info(" FILL_ON = %s", config.FILL_ON)
log.info(" WORK_SCALE = %.4f", config.WORK_SCALE)
log.info(" SEED = %s", config.SEED)
names = []
df = None
if os.path.exists(config.EXCEL_PATH):
try:
df = pd.read_excel(config.EXCEL_PATH)
raw_names = df.iloc[:, config.DATA_COL_INDEX].dropna().astype(str)
if config.REMOVE_DUPLICATES:
names = raw_names.unique().tolist()
print(f"模式: 去重 | 数量: {len(names)}")
else:
names = raw_names.tolist()
print(f"模式: 保留重复 | 数量: {len(names)}")
except Exception as e:
print(f"读取 Excel 失败: {e}")
log.error("读取 Excel 失败: %s", e)
sys.exit(1)
else:
count = 12000
print(f"未找到Excel,使用测试数据: {count}条")
log.info("未找到 Excel,使用测试数据: %d 条", count)
names = [f"测试_{i % 100}" for i in range(count)]
input_count = len(names)
log.info("[阶段1] 读取完成: input_count=%d, 去重=%s", input_count, config.REMOVE_DUPLICATES)
if names:
sample = names[:min(10, len(names))]
log.info(" 前10个名字: %s", sample)
print("--- 2. 智能画幅计算 ---")
log.info("--- 阶段2: 智能画幅计算 ---")
avg_len = sum(len(n) for n in names) / len(names) if names else 3
log.info(" 平均名字长度: %.2f 字符", avg_len)
log.info(" BASE_HD: %dx%d", config.BASE_HD_WIDTH, config.BASE_HD_HEIGHT)
# 只生成一次掩膜:先 probe 尺寸,再按需扩展后复用(避免二次 LANCZOS)
probe_mask_hd, (probe_w, probe_h), _ = prepare_mask(config.BASE_HD_WIDTH, config.BASE_HD_HEIGHT)
probe_stats = analyze_mask(probe_mask_hd)
log.info(" Probe mask: %dx%d, free_ratio=%.4f, bbox_fill_ratio=%.4f",
probe_w, probe_h, probe_stats['free_ratio'], probe_stats['bbox_fill_ratio'])
if probe_stats.get('bbox'):
log.info(" Probe bbox: %s", probe_stats['bbox'])
print(f"[Mask Probe] 可填充比例={probe_stats['free_ratio']:.3f}")
hd_w, hd_h = calculate_dynamic_dimensions(probe_w, probe_h, len(names), avg_len, probe_stats)
log.info(" 动态画幅计算结果: %dx%d", hd_w, hd_h)
print("--- 3. 生成掩膜 (High Quality & Edge Fix) ---")
log.info("--- 阶段3: 生成掩膜 ---")
if (hd_w, hd_h) == (probe_w, probe_h):
mask_hd, real_hd_w, real_hd_h = probe_mask_hd, probe_w, probe_h
mask_stats = probe_stats
else:
mask_hd, (real_hd_w, real_hd_h), _ = prepare_mask(hd_w, hd_h)
mask_stats = analyze_mask(mask_hd)
log.info(" mask_hd: %dx%d", real_hd_w, real_hd_h)
log.info(" free_area=%d, free_ratio=%.6f", mask_stats['free_area'], mask_stats['free_ratio'])
log.info(" bbox_fill_ratio=%.6f", mask_stats['bbox_fill_ratio'])
if mask_stats.get('bbox'):
log.info(" bbox=%s", mask_stats['bbox'])
print(f"[Mask Final] 可填充比例={mask_stats['free_ratio']:.3f}")
print("--- 4. 计算权重 ---")
log.info("--- 阶段4: 计算权重 ---")
t_weights = time.time()
if config.ENABLE_STROKE_WEIGHTS:
stroke_weights_map = get_stroke_complexity_batch(names, config.WC_FONT_PATH)
log.info(" 笔画权重计算完成: %d 个词, 耗时=%.3fs", len(stroke_weights_map), time.time() - t_weights)
else:
stroke_weights_map = {}
print("笔画权重已关闭")
log.info(" 笔画权重已关闭")
# 打印权重分布统计
if stroke_weights_map:
w_vals = list(stroke_weights_map.values())
log.info(" 笔画权重分布: min=%.1f, max=%.1f, avg=%.1f, median=%.1f",
min(w_vals), max(w_vals), sum(w_vals)/len(w_vals),
sorted(w_vals)[len(w_vals)//2])
sample_items = list(stroke_weights_map.items())[:5]
log.info(" 笔画权重样本: %s", sample_items)
excel_weights_map = extract_weights_from_df(df, names) if df is not None else {}
if excel_weights_map:
print(f"Excel 权重生效: {len(excel_weights_map)} 个词")
log.info(" Excel 权重生效: %d 个词", len(excel_weights_map))
ew_vals = list(excel_weights_map.values())
log.info(" Excel 权重分布: min=%.1f, max=%.1f, avg=%.1f",
min(ew_vals), max(ew_vals), sum(ew_vals)/len(ew_vals))
elif config.WEIGHT_COL_NAME is not None or config.WEIGHT_COL_INDEX is not None:
fallback = "笔画权重" if config.ENABLE_STROKE_WEIGHTS else "均等权重"
print(f"Excel 权重不可用,已回退{fallback}")
log.info(" Excel 权重不可用,已回退%s", fallback)
name_weights_map = merge_weight_maps(
names,
stroke_weights_map if config.ENABLE_STROKE_WEIGHTS else {},
excel_weights_map,
)
frequencies_data = name_weights_map if config.REMOVE_DUPLICATES else [(name, name_weights_map.get(name, 10)) for name in names]
canvas_retry_round = 0
generation_result = None
t_gen = time.time()
while True:
log.info("[画布] 第%d轮生成 pass, 当前画布: %dx%d", canvas_retry_round + 1, real_hd_w, real_hd_h)
generation_result = run_generation_pass(
names,
frequencies_data,
name_weights_map,
mask_hd,
real_hd_w,
real_hd_h,
)
wc = generation_result.get("wc")
placed = len(wc.layout_) if wc is not None else 0
target = len(names) * config.N_REPETITIONS
# 名单完整性是硬约束。填充率优化不得以漏掉姓名为代价。
hd_overlap_pixels = int(generation_result.get("hd_overlap_pixels", 0))
placement_ok = wc is not None and placed >= target and hd_overlap_pixels == 0
if placement_ok or canvas_retry_round >= config.CANVAS_RETRY_MAX_ROUNDS:
if not placement_ok:
print(
f"生成失败:放置 {placed}/{target},高清重叠像素 {hd_overlap_pixels}"
"未满足完整名单与零碰撞约束"
)
log.error(
"生成失败:放置 %d/%d, 高清重叠像素=%d (已重试 %d 轮)",
placed,
target,
hd_overlap_pixels,
canvas_retry_round,
)
sys.exit(1)
log.info(" 生成完成 placed=%d/%d fill=%.4f retry=%d",
placed, target, generation_result["fill_ratio"], canvas_retry_round)
break
canvas_retry_round += 1
next_w = int(real_hd_w * config.CANVAS_RETRY_GROWTH)
next_h = int(real_hd_h * config.CANVAS_RETRY_GROWTH)
print(f"[画布重试#{canvas_retry_round}] {real_hd_w}x{real_hd_h} -> {next_w}x{next_h} (放置 {placed}/{target})")
log.info("[画布重试#%d] %dx%d -> %dx%d (growth=%.2f)",
canvas_retry_round, real_hd_w, real_hd_h, next_w, next_h, config.CANVAS_RETRY_GROWTH)
mask_hd, (real_hd_w, real_hd_h), _ = prepare_mask(next_w, next_h)
mask_stats = analyze_mask(mask_hd)
final_wc = generation_result["wc"]
fill_ratio = generation_result["fill_ratio"]
w_small = generation_result["w_small"]
h_small = generation_result["h_small"]
log.info("[阶段5完成] 生成耗时=%.2fs, fill_ratio=%.4f, size_scale=%.4f",
time.time() - t_gen, fill_ratio, generation_result["size_scale"])
print("--- 6. 高清渲染 ---")
log.info("--- 阶段6: 高清渲染 ---")
t_render = time.time()
hd_layout = generation_result["hd_layout"]
log.info(" HD layout 词数: %d", len(hd_layout))
log.info(" HD 画布: %dx%d", real_hd_w, real_hd_h)
if hd_layout:
sample = hd_layout[:3]
for s in sample:
log.info(" 样本: text='%s', size=%d, pos=(%d,%d), orient=%s, color=%s",
s[0], s[1], s[2][1], s[2][0], s[3], s[4])
final_wc.layout_ = hd_layout
final_wc.width = real_hd_w
final_wc.height = real_hd_h
base_img = final_wc.to_image().convert("RGB")
# 更快的 PNG 写出(压缩等级 1,视觉无损)
base_img.save(config.OUTPUT_PNG, compress_level=1)
print(f"已保存: {config.OUTPUT_PNG}")
log.info(" PNG 已保存: %s (%.2f MB)", config.OUTPUT_PNG,
Path(config.OUTPUT_PNG).stat().st_size / 1024 / 1024 if Path(config.OUTPUT_PNG).exists() else 0)
# 一次构建路径,同时写出 fill / stroke 两份 SVG
stroke_svg = final_wc.export_svgs(config.OUTPUT_SVG)
print(f"已保存: {config.OUTPUT_SVG}")
print(f"已保存: {stroke_svg}")
log.info(" SVG 已保存: %s", config.OUTPUT_SVG)
log.info(" SVG(stroke) 已保存: %s", stroke_svg)
log.info(" 渲染耗时: %.2fs", time.time() - t_render)
log.info("--- 阶段7: 写入数据库 ---")
t_db = time.time()
try:
conn = sqlite3.connect(config.DB_PATH)
cursor = conn.cursor()
cursor.execute("DROP TABLE IF EXISTS word_locations")
cursor.execute("""
CREATE TABLE word_locations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT,
x INTEGER,
y INTEGER,
font_size INTEGER,
color TEXT,
orientation TEXT,
box_x INTEGER,
box_y INTEGER,
box_width INTEGER,
box_height INTEGER
)
""")
# Cache full font bearings as well as dimensions so search highlights
# match the actual PIL-rendered glyph position.
bbox_cache = {}
_measure = ImageDraw.Draw(Image.new("L", (1, 1)))
db_data = []
for name, font_size, (y, x), orient, color in final_wc.layout_:
orientation = "vertical" if orient else "horizontal"
key = (name, int(font_size), bool(orient))
box = bbox_cache.get(key)
if box is None:
font = get_cached_font(config.WC_FONT_PATH, max(1, int(font_size)))
if orient:
font = ImageFont.TransposedFont(font, orientation=orient)
bb = _measure.textbbox((0, 0), name, font=font)
box = (bb[0], bb[1], bb[2] - bb[0], bb[3] - bb[1])
bbox_cache[key] = box
bx, by, bw, bh = box
db_data.append(
(
name,
x + bx,
y + by,
font_size,
color,
orientation,
x,
y,
bw,
bh,
)
)
cursor.executemany(
"""
INSERT INTO word_locations
(name, x, y, font_size, color, orientation, box_x, box_y, box_width, box_height)
VALUES (?,?,?,?,?,?,?,?,?,?)
""",
db_data,
)
conn.commit()
conn.close()
log.info(" DB 写入完成: %s, %d 行, 耗时=%.3fs", config.DB_PATH, len(db_data), time.time() - t_db)
except sqlite3.Error as e:
print(f"DB Error: {e}")
log.error(" DB 写入失败: %s", e)
sys.exit(1)
elapsed = time.time() - t_start
placed_count = len(final_wc.layout_)
metrics = {
"seed": config.SEED,
"layout_seed": generation_result.get("layout_seed", config.LAYOUT_SEED),
"input_count": input_count,
"placed_count": placed_count,
"fill_ratio": fill_ratio,
"largest_empty_square_work_px": int(
generation_result.get("largest_empty_square_work_px", 0)
),
"largest_empty_square_font_ratio": generation_result.get(
"largest_empty_square_font_ratio"
),
"hd_overlap_pixels": int(generation_result.get("hd_overlap_pixels", -1)),
"hd_clearance": generation_result.get("hd_clearance"),
"elapsed_seconds": round(elapsed, 4),
"font_info": {
"layout_font_path": config.WC_FONT_PATH,
"mask_font_path": config.MASK_FONT_PATH,
"palette": list(config.get_output_palette()),
"background": config.get_output_background(),
},
"mask_info": {
"free_ratio": round(mask_stats["free_ratio"], 6),
"bbox_fill_ratio": round(mask_stats["bbox_fill_ratio"], 6),
"canvas_retry_rounds": canvas_retry_round,
},
"canvas_info": {
"hd_width": real_hd_w,
"hd_height": real_hd_h,
"work_width": w_small,
"work_height": h_small,
"work_scale": config.WORK_SCALE,
},
"output_paths": {
"png": config.OUTPUT_PNG,
"svg": config.OUTPUT_SVG,
"db": config.DB_PATH,
"metrics": config.METRICS_FILE,
"debug_dir": config.DEBUG_OUTPUT_DIR,
},
"config_snapshot": {
"mode": config.MODE,
"excel_path": config.EXCEL_PATH,
"mask_image_path": config.MASK_IMAGE_PATH,
"output_dir": config.OUTPUT_DIR,
"output_prefix": config.OUTPUT_PREFIX,
"min_font_size": config.MIN_FONT_SIZE,
"fill_on": config.FILL_ON,
"layout_seed": generation_result.get("layout_seed", config.LAYOUT_SEED),
}
}
config.write_metrics(metrics)
print(f"\n✅ 完成! 总耗时: {elapsed:.2f}s")
log.info("=" * 60)
log.info("[Pipeline] 全流程完成!")
log.info(" 总耗时: %.2fs", elapsed)
log.info(" 输入: %d 词 -> 放置: %d 词", input_count, placed_count)
log.info(" 填充率: %.4f", fill_ratio)
log.info(" 画布: %dx%d (运算: %dx%d)", real_hd_w, real_hd_h, w_small, h_small)
log.info(" 输出: PNG=%s", config.OUTPUT_PNG)
log.info(" 输出: SVG=%s", config.OUTPUT_SVG)
log.info(" 输出: DB=%s", config.DB_PATH)
log.info("=" * 60)