Yuzhiran Dev
9c37c9a574
Phase 4: org_id 注入 JWT/API 过滤/组织管理 CRUD/前端组织列 测试: tests/test_phase_upgrades.py 97项全覆盖 CSS: theme-modern.css 共享 mobile-card-list/status-dot/search-bar 等模式 修复: initial_data.py LLM配置 NOT NULL 约束, TopicResponse 含 org_id
547 lines
21 KiB
Python
547 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
文章配图自动生成器
|
|
基于PIL，根据文章标题、内容自动生成适合各平台的配图
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import datetime
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Dict, List, Tuple, Optional
|
|
from dataclasses import dataclass
|
|
|
|
import yaml
|
|
|
|
# from PIL import Image, ImageDraw, ImageFont
|
|
# 使用系统PIL，确保虚拟环境正确安装
|
|
import sys
|
|
sys.path.insert(0, '/usr/local/lib64/python3.11/site-packages')
|
|
sys.path.insert(0, '/usr/lib64/python3.11/site-packages')
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
import random
|
|
|
|
# 确保项目根目录在路径中
|
|
PROJECT_ROOT = Path('/root/openclaw-workspace/projects/yu-zhi-ran')
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))
|
|
|
|
from db_helper import get_topic_by_id
|
|
from app.models import Article
|
|
from app.database import SessionLocal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 加载配置
|
|
CONFIG_DIR = PROJECT_ROOT / "config"
|
|
with open(CONFIG_DIR / "wecom_config.yaml", 'r', encoding='utf-8') as f:
|
|
wecom_config = yaml.safe_load(f)
|
|
|
|
@dataclass
|
|
class ImageSpec:
|
|
"""图片规格"""
|
|
platform: str
|
|
width: int
|
|
height: int
|
|
format: str = "PNG"
|
|
quality: int = 85
|
|
bg_color: Tuple[int, int, int] = (255, 255, 255) # 白色背景
|
|
accent_color: Tuple[int, int, int] = (76, 175, 80) # 品牌绿色 #4CAF50
|
|
text_color: Tuple[int, int, int] = (51, 51, 51) # 深灰色
|
|
|
|
class ImageGenerator:
|
|
"""图片生成器"""
|
|
|
|
def __init__(self, output_base: Path = None):
|
|
self.output_base = output_base or (PROJECT_ROOT / "automation" / "images" / "generated")
|
|
self.today = datetime.datetime.now().strftime("%Y-%m-%d")
|
|
self.output_dir = self.output_base / self.today
|
|
self.output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 加载平台规格
|
|
self.platform_specs = {}
|
|
for platform, specs in wecom_config["image_specs"].items():
|
|
self.platform_specs[platform] = ImageSpec(
|
|
platform=platform,
|
|
width=specs["width"],
|
|
height=specs["height"],
|
|
format=specs["format"],
|
|
quality=specs["quality"]
|
|
)
|
|
|
|
# 字体路径
|
|
self.font_paths = self._find_chinese_fonts()
|
|
|
|
def _find_chinese_fonts(self) -> List[str]:
|
|
"""查找系统中可用的中文字体"""
|
|
font_paths = [
|
|
"/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", # 文泉驿微米黑
|
|
"/usr/share/fonts/truetype/arphic/uming.ttc", # 文鼎PL中等
|
|
"/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf",
|
|
"/System/Library/Fonts/PingFang.ttc", # macOS
|
|
"/System/Library/Fonts/STHeiti Medium.ttc", # macOS
|
|
"C:\\Windows\\Fonts\\msyh.ttc", # Windows
|
|
"C:\\Windows\\Fonts\\simsun.ttc"
|
|
]
|
|
available = [p for p in font_paths if os.path.exists(p)]
|
|
return available if available else [None] # 回退到默认字体
|
|
|
|
def _get_font(self, size: int, bold: bool = False) -> ImageFont.FreeTypeFont:
|
|
"""获取合适的中文字体"""
|
|
# 优先使用系统中文字体
|
|
chinese_fonts = ["/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", "/usr/share/fonts/zh_CN/SimHei.ttf"]
|
|
for font_path in chinese_fonts + self.font_paths:
|
|
if font_path and os.path.exists(font_path):
|
|
try:
|
|
return ImageFont.truetype(font_path, size)
|
|
except:
|
|
continue
|
|
return ImageFont.load_default()
|
|
|
|
def generate_cover_image(self, title: str, subtitle: str = "", platform: str = "zhihu") -> Path:
|
|
"""生成封面图"""
|
|
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
|
|
|
|
# 创建图片
|
|
img = Image.new('RGB', (spec.width, spec.height), color=spec.bg_color)
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# 添加渐变背景
|
|
for y in range(spec.height):
|
|
# 从顶部到中间的渐变
|
|
ratio = y / (spec.height * 0.6)
|
|
r = int(255 * (1 - ratio) + 230 * ratio)
|
|
g = int(255 * (1 - ratio) + 240 * ratio)
|
|
b = int(255 * (1 - ratio) + 250 * ratio)
|
|
draw.line([(0, y), (spec.width, y)], fill=(r, g, b))
|
|
|
|
# 绘制品牌标识区域（底部条纹）
|
|
stripe_height = 20
|
|
stripe_y = spec.height - stripe_height - 30
|
|
draw.rectangle([0, stripe_y, spec.width, stripe_y + stripe_height], fill=spec.accent_color)
|
|
draw.text((20, stripe_y + 5), "宇之然", fill=(255, 255, 255), font=self._get_font(14))
|
|
|
|
# 标题排版
|
|
title_font = self._get_font(int(spec.height * 0.12), bold=True)
|
|
subtitle_font = self._get_font(int(spec.height * 0.06))
|
|
|
|
# 自动换行处理
|
|
max_width = spec.width * 0.9
|
|
title_lines = self._wrap_text(title, title_font, max_width)
|
|
subtitle_lines = self._wrap_text(subtitle, subtitle_font, max_width) if subtitle else []
|
|
|
|
# 计算总高度
|
|
line_spacing = 1.2
|
|
title_height = len(title_lines) * title_font.size * line_spacing
|
|
subtitle_height = len(subtitle_lines) * subtitle_font.size * line_spacing
|
|
total_text_height = title_height + subtitle_height + 20 # 间距
|
|
|
|
# 居中绘制
|
|
start_y = (spec.height - total_text_height) // 2
|
|
|
|
# 绘制标题
|
|
for i, line in enumerate(title_lines):
|
|
y = start_y + i * (title_font.size * line_spacing)
|
|
self._draw_centered_text(draw, line, y, spec.width, title_font, spec.text_color)
|
|
|
|
# 绘制副标题
|
|
if subtitle_lines:
|
|
subtitle_start_y = start_y + title_height + 10
|
|
for i, line in enumerate(subtitle_lines):
|
|
y = subtitle_start_y + i * (subtitle_font.size * line_spacing)
|
|
self._draw_centered_text(draw, line, y, spec.width, subtitle_font, (102, 102, 102))
|
|
|
|
# 保存图片
|
|
filename = f"cover_{platform}.{spec.format.lower()}"
|
|
output_path = self.output_dir / filename
|
|
img.save(output_path, quality=spec.quality)
|
|
|
|
return output_path
|
|
|
|
def generate_chart_image(self, chart_type: str, data: Dict, title: str, platform: str = "zhihu") -> Path:
|
|
"""生成数据图表"""
|
|
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
|
|
|
|
img = Image.new('RGB', (spec.width, spec.height), color=(255, 255, 255))
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# 绘制标题
|
|
title_font = self._get_font(36, bold=True)
|
|
draw.text((50, 30), title, fill=spec.text_color, font=title_font)
|
|
|
|
# 根据图表类型绘制
|
|
if chart_type == "bar":
|
|
self._draw_bar_chart(draw, data, spec)
|
|
elif chart_type == "pie":
|
|
self._draw_pie_chart(draw, data, spec)
|
|
elif chart_type == "line":
|
|
self._draw_line_chart(draw, data, spec)
|
|
else:
|
|
# 默认显示文本
|
|
text_font = self._get_font(24)
|
|
draw.text((50, 150), f"图表类型: {chart_type}", fill=spec.text_color, font=text_font)
|
|
draw.text((50, 200), f"数据: {json.dumps(data, ensure_ascii=False)}", fill=spec.text_color, font=text_font)
|
|
|
|
# 水印
|
|
watermark_font = self._get_font(14)
|
|
draw.text((spec.width - 150, spec.height - 30), "数据来源: 宇之然", fill=(150, 150, 150), font=watermark_font)
|
|
|
|
filename = f"data_chart_{platform}.png"
|
|
output_path = self.output_dir / filename
|
|
img.save(output_path, quality=spec.quality)
|
|
|
|
return output_path
|
|
|
|
def _draw_bar_chart(self, draw: ImageDraw.Draw, data: Dict, spec: ImageSpec):
|
|
"""绘制柱状图"""
|
|
# 数据格式: {"label1": value1, "label2": value2, ...}
|
|
labels = list(data.keys())
|
|
values = list(data.values())
|
|
max_value = max(values) if values else 1
|
|
|
|
chart_area = {
|
|
"left": 100,
|
|
"top": 120,
|
|
"right": spec.width - 50,
|
|
"bottom": spec.height - 100
|
|
}
|
|
|
|
chart_width = chart_area["right"] - chart_area["left"]
|
|
chart_height = chart_area["bottom"] - chart_area["top"]
|
|
|
|
bar_width = chart_width // (len(values) * 2)
|
|
gap = bar_width
|
|
|
|
# 绘制坐标轴
|
|
draw.line([
|
|
(chart_area["left"], chart_area["top"]),
|
|
(chart_area["left"], chart_area["bottom"])
|
|
], fill=(0, 0, 0), width=2)
|
|
draw.line([
|
|
(chart_area["left"], chart_area["bottom"]),
|
|
(chart_area["right"], chart_area["bottom"])
|
|
], fill=(0, 0, 0), width=2)
|
|
|
|
# 绘制柱子
|
|
for i, (label, value) in enumerate(zip(labels, values)):
|
|
x = chart_area["left"] + i * (bar_width + gap) + gap // 2
|
|
bar_height = (value / max_value) * chart_height
|
|
y_bottom = chart_area["bottom"]
|
|
y_top = chart_area["bottom"] - bar_height
|
|
|
|
# 柱子（渐变色）
|
|
for y in range(int(y_top), int(y_bottom)):
|
|
ratio = (y - y_top) / bar_height if bar_height > 0 else 0
|
|
r = int(76 + (100-76) * ratio)
|
|
g = int(175 + (150-175) * ratio)
|
|
b = int(80 + (120-80) * ratio)
|
|
draw.line([(x, y), (x + bar_width, y)], fill=(r, g, b))
|
|
|
|
# 标签
|
|
label_font = self._get_font(18)
|
|
self._draw_centered_text(draw, label, y_bottom + 10, x + bar_width // 2, label_font, (80, 80, 80))
|
|
|
|
# 数值
|
|
value_font = self._get_font(20, bold=True)
|
|
self._draw_centered_text(draw, f"{value}", y_top - 10, x + bar_width // 2, value_font, spec.accent_color)
|
|
|
|
def _draw_pie_chart(self, draw: ImageDraw.Draw, data: Dict, spec: ImageSpec):
|
|
"""绘制饼图"""
|
|
# 简单实现：绘制圆形扇形
|
|
center_x, center_y = spec.width // 2, spec.height // 2
|
|
radius = min(spec.width, spec.height) // 3
|
|
|
|
total = sum(data.values()) if data else 1
|
|
angle_start = 0
|
|
|
|
# 颜色调色板
|
|
colors = [
|
|
(76, 175, 80), (33, 150, 83), (139, 195, 74),
|
|
(255, 193, 7), (255, 152, 0), (244, 67, 54)
|
|
]
|
|
|
|
for i, (label, value) in enumerate(data.items()):
|
|
angle_extent = (value / total) * 360
|
|
color = colors[i % len(colors)]
|
|
|
|
# 绘制扇形
|
|
draw.arc(
|
|
[center_x - radius, center_y - radius, center_x + radius, center_y + radius],
|
|
angle_start, angle_start + angle_extent,
|
|
fill=color, width=radius * 2
|
|
)
|
|
angle_start += angle_extent
|
|
|
|
# 画中心白圆形成饼图效果
|
|
inner_radius = radius * 0.5
|
|
draw.ellipse(
|
|
[center_x - inner_radius, center_y - inner_radius, center_x + inner_radius, center_y + inner_radius],
|
|
fill=(255, 255, 255)
|
|
)
|
|
|
|
# 绘制图例
|
|
legend_y = spec.height - 80
|
|
legend_x = 100
|
|
for i, (label, value) in enumerate(data.items()):
|
|
color = colors[i % len(colors)]
|
|
# 色块
|
|
draw.rectangle([legend_x, legend_y + i*25, legend_x+20, legend_y+20+i*25], fill=color)
|
|
# 标签
|
|
label_font = self._get_font(16)
|
|
draw.text((legend_x+30, legend_y+i*25), f"{label}: {value}", fill=(60, 60, 60), font=label_font)
|
|
|
|
def _draw_line_chart(self, draw: ImageDraw.Draw, data: Dict, spec: ImageSpec):
|
|
"""绘制折线图"""
|
|
# 简化版：显示文本描述
|
|
title_font = self._get_font(24)
|
|
draw.text((50, 100), "折线图 (数据趋势)", fill=spec.text_color, font=title_font)
|
|
|
|
items = list(data.items())
|
|
if not items:
|
|
draw.text((50, 150), "无可用数据", fill=(100, 100, 100), font=self._get_font(18))
|
|
return
|
|
|
|
# 列出数据
|
|
data_font = self._get_font(16)
|
|
y = 200
|
|
for label, value in items[:10]: # 限制显示数量
|
|
draw.text((50, y), f"{label}: {value}", fill=(80, 80, 80), font=data_font)
|
|
y += 25
|
|
|
|
def generate_concept_image(self, title: str, items: List[str], platform: str = "zhihu") -> Path:
|
|
"""生成概念示意图（用于行动清单等）"""
|
|
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
|
|
|
|
img = Image.new('RGB', (spec.width, spec.height), color=(245, 245, 245))
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# 标题
|
|
title_font = self._get_font(42, bold=True)
|
|
self._draw_centered_text(draw, title, 60, spec.width, title_font, spec.text_color)
|
|
|
|
# 绘制项目列表（带复选框样式）
|
|
item_font = self._get_font(28)
|
|
start_y = 150
|
|
for i, item in enumerate(items[:8]): # 限制8个
|
|
y = start_y + i * 50
|
|
# 复选框
|
|
box_size = 30
|
|
box_x = (spec.width - 400) // 2
|
|
draw.rectangle([box_x, y, box_x + box_size, y + box_size], outline=spec.accent_color, width=3)
|
|
# 勾
|
|
check_font = self._get_font(24)
|
|
draw.text((box_x + 7, y + 2), "✓", fill=spec.accent_color, font=check_font)
|
|
# 文字
|
|
draw.text((box_x + box_size + 20, y + 5), item[:30], fill=(60, 60, 60), font=item_font)
|
|
|
|
filename = f"action_checklist_{platform}.png"
|
|
output_path = self.output_dir / filename
|
|
img.save(output_path, quality=spec.quality)
|
|
|
|
return output_path
|
|
|
|
def generate_equipment_list_image(self, items: List[Dict[str, str]], platform: str = "zhihu") -> Path:
|
|
"""生成装备清单图"""
|
|
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
|
|
|
|
img = Image.new('RGB', (spec.width, spec.height), color=(255, 255, 255))
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# 标题
|
|
title = "装备清单"
|
|
title_font = self._get_font(38, bold=True)
|
|
draw.text((50, 40), title, fill=spec.text_color, font=title_font)
|
|
|
|
# 列头
|
|
headers = ["名称", "用途", "预算"]
|
|
header_font = self._get_font(24, bold=True)
|
|
col_width = spec.width // len(headers)
|
|
for i, header in enumerate(headers):
|
|
x = i * col_width + 20
|
|
draw.text((x, 100), header, fill=(100, 100, 100), font=header_font)
|
|
|
|
# 分隔线
|
|
draw.line([(50, 130), (spec.width-50, 130)], fill=(200, 200, 200), width=2)
|
|
|
|
# 绘制条目
|
|
item_font = self._get_font(20)
|
|
row_height = 40
|
|
y = 150
|
|
for item in items[:10]: # 最多10行
|
|
name = item.get("name", "")[:12]
|
|
purpose = item.get("purpose", "")[:10]
|
|
budget = item.get("budget", "")
|
|
|
|
draw.text((70, y), name, fill=(50, 50, 50), font=item_font)
|
|
draw.text((col_width + 70, y), purpose, fill=(50, 50, 50), font=item_font)
|
|
draw.text((2*col_width + 70, y), budget, fill=(50, 50, 50), font=item_font)
|
|
|
|
y += row_height
|
|
|
|
# 底部总预算
|
|
total_budget = sum([int(item.get("budget", "0").replace("元", "")) for item in items if item.get("budget", "").replace("元", "").isdigit()])
|
|
total_font = self._get_font(22, bold=True)
|
|
draw.text((50, spec.height - 50), f"总预算: {total_budget}元", fill=spec.accent_color, font=total_font)
|
|
|
|
filename = f"equipment_{platform}.png"
|
|
output_path = self.output_dir / filename
|
|
img.save(output_path, quality=spec.quality)
|
|
|
|
return output_path
|
|
|
|
def _wrap_text(self, text: str, font: ImageFont.FreeTypeFont, max_width: int) -> List[str]:
|
|
"""文本自动换行"""
|
|
words = list(text)
|
|
lines = []
|
|
current_line = ""
|
|
|
|
for char in words:
|
|
test_line = current_line + char
|
|
bbox = font.getbbox(test_line)
|
|
width = bbox[2] - bbox[0]
|
|
|
|
if width <= max_width:
|
|
current_line = test_line
|
|
else:
|
|
if current_line:
|
|
lines.append(current_line)
|
|
current_line = char
|
|
|
|
if current_line:
|
|
lines.append(current_line)
|
|
|
|
return lines if lines else [text]
|
|
|
|
def _draw_centered_text(self, draw: ImageDraw.Draw, text: str, y: int, center_x: int, font: ImageFont.FreeTypeFont, color: Tuple[int, int, int]):
|
|
"""绘制居中文本"""
|
|
bbox = font.getbbox(text)
|
|
text_width = bbox[2] - bbox[0]
|
|
x = center_x - text_width // 2
|
|
draw.text((x, y), text, fill=color, font=font)
|
|
|
|
def generate_all_placeholders(self, article_title: str, platform: str = "zhihu") -> Dict[str, Path]:
|
|
"""生成所有占位图片"""
|
|
files = {}
|
|
|
|
# 1. 封面图
|
|
files["cover"] = self.generate_cover_image(article_title, "宇之然 · 可持续生活指南", platform)
|
|
|
|
# 2. 数据图表示例
|
|
files["data_chart"] = self.generate_chart_image("bar", {"选项A": 45, "选项B": 32, "选项C": 23}, "数据对比", platform)
|
|
|
|
# 3. 概念图（行动清单）
|
|
files["action_checklist"] = self.generate_concept_image("立即行动清单", [
|
|
"第一步：记录现状，识别改进空间",
|
|
"第二步：尝试最小可行改变",
|
|
"第三步：评估效果，决定是否继续",
|
|
"第四步：建立习惯，持续改进"
|
|
], platform)
|
|
|
|
# 4. 装备清单图
|
|
files["equipment"] = self.generate_equipment_list_image([
|
|
{"name": "智能插座", "purpose": "定时控制", "budget": "50元"},
|
|
{"name": "土壤传感器", "purpose": "湿度监测", "budget": "80元"},
|
|
{"name": "自动灌溉", "purpose": "浇水", "budget": "120元"},
|
|
{"name": "LED补光灯", "purpose": "光照", "budget": "200元"}
|
|
], platform)
|
|
|
|
return files
|
|
|
|
def generate_for_topic(topic_id: str, platforms: List[str] = None) -> Dict[str, Dict[str, str]]:
|
|
"""为指定选题生成三平台配图，路径存入 articles 表"""
|
|
if platforms is None:
|
|
platforms = ["zhihu", "wechat", "xiaohongshu"]
|
|
|
|
topic = get_topic_by_id(topic_id)
|
|
if not topic:
|
|
raise ValueError(f"Topic {topic_id} not found")
|
|
|
|
title = topic.get("title", "无标题")
|
|
generator = ImageGenerator()
|
|
results = {}
|
|
|
|
for platform in platforms:
|
|
try:
|
|
files = generator.generate_all_placeholders(title, platform)
|
|
cover_path = str(files.get("cover", ""))
|
|
chart_path = str(files.get("data_chart", ""))
|
|
checklist_path = str(files.get("action_checklist", ""))
|
|
|
|
images = {
|
|
"cover": cover_path,
|
|
"chart": chart_path,
|
|
"checklist": checklist_path,
|
|
}
|
|
|
|
# 存入 DB
|
|
save_article_images(topic_id, platform, images)
|
|
|
|
results[platform] = images
|
|
logger.info(f" [{platform}] cover={Path(cover_path).name}" if cover_path else "")
|
|
except Exception as e:
|
|
logger.error(f" [{platform}] 生成失败: {e}")
|
|
results[platform] = {}
|
|
|
|
return results
|
|
|
|
|
|
def save_article_images(topic_id: str, platform: str, images: Dict[str, str]):
|
|
"""将图片路径写入 articles 表的 images 字段"""
|
|
db = SessionLocal()
|
|
try:
|
|
from app.models import Article
|
|
article_id = f"{platform}_{topic_id}"
|
|
article = db.query(Article).filter(Article.id == article_id).first()
|
|
if article:
|
|
existing = article.images or {}
|
|
existing.update(images)
|
|
article.images = existing
|
|
else:
|
|
article = Article(
|
|
id=article_id,
|
|
topic_id=topic_id,
|
|
platform=platform,
|
|
file_path=f"db:{article_id}",
|
|
status="draft",
|
|
images=images,
|
|
)
|
|
db.add(article)
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description='文章配图生成器')
|
|
parser.add_argument('--topic-id', help='选题ID，指定则为选题生成配图')
|
|
args = parser.parse_args()
|
|
|
|
if args.topic_id:
|
|
print(f"为选题 {args.topic_id} 生成配图...")
|
|
results = generate_for_topic(args.topic_id)
|
|
print(json.dumps({"topic_id": args.topic_id, "images": results}, ensure_ascii=False))
|
|
sys.exit(0)
|
|
|
|
"""测试主函数"""
|
|
generator = ImageGenerator()
|
|
|
|
# 测试生成图片
|
|
print(f"开始生成图片到: {generator.output_dir}")
|
|
|
|
# 生成所有类型的占位图
|
|
files = generator.generate_all_placeholders("上海阳台种菜一年：我收获的不仅是蔬菜", "zhihu")
|
|
|
|
print("\n生成的文件:")
|
|
for name, path in files.items():
|
|
print(f" - {name}: {path.name} ({path.stat().st_size // 1024}KB)")
|
|
|
|
print(f"\n✅ 图片生成完成，共 {len(files)} 张")
|
|
|
|
if __name__ == "__main__":
|
|
main() |