530 lines
19 KiB
Python
530 lines
19 KiB
Python
"""
|
||
古诗词意境图生成器（Qwen-Image 云端版）
|
||
将中国古典诗词通过 LLM 分析拆解为多个意境画面，
|
||
再使用 Qwen-Image API（通过 SiliconFlow）逐一生成高质量图片。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
from io import BytesIO
|
||
from pathlib import Path
|
||
|
||
import requests as http_requests
|
||
import yaml
|
||
from openai import OpenAI
|
||
from PIL import Image
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 配置加载
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def load_config(config_path: str = "config_qwen.yaml") -> dict:
|
||
with open(config_path, "r", encoding="utf-8") as f:
|
||
cfg = yaml.safe_load(f)
|
||
|
||
api_key = os.environ.get("LLM_API_KEY") or cfg["llm"].get("api_key", "")
|
||
cfg["llm"]["api_key"] = api_key
|
||
|
||
img_api_key = (
|
||
os.environ.get("IMAGE_API_KEY")
|
||
or cfg["image"].get("api_key", "")
|
||
or api_key
|
||
)
|
||
cfg["image"]["api_key"] = img_api_key
|
||
return cfg
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# LLM 古诗词分析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
SYSTEM_PROMPT = """\
|
||
你是一位精通中国古典文学与视觉艺术的大师，同时深谙文生图 AI 的 prompt 工程。\
|
||
你的任务是分析用户提供的古诗词，将其意境拆解为若干幅独立的画面，\
|
||
每幅画面对应诗词中一个完整的意象或场景。
|
||
|
||
## 核心原则：信、雅、达
|
||
|
||
1. **信**（忠实）：画面内容必须忠于原诗的意象、情感和时代背景，不可凭空臆造。\
|
||
诗中有月则画月，诗中无人则不强加人物。
|
||
2. **雅**（优美）：画面描述应体现中国传统美学，注重意境营造、留白与含蓄之美。
|
||
3. **达**（通畅）：prompt 要清晰、具体、富有画面感，\
|
||
能被文生图模型准确理解并生成高质量图像。
|
||
|
||
## 诗词体裁识别与风格匹配
|
||
|
||
请先识别诗词的体裁（唐诗/宋词/元曲/其他），再根据题材选择最合适的中国传统画风。\
|
||
以下是可选的风格菜单，请根据诗意灵活选取，同一首诗的不同画面可以使用不同风格：
|
||
|
||
| 风格 | prompt 关键词 | 适用场景 |
|
||
|------|-------------|---------|
|
||
| 水墨写意 | 水墨写意，淡墨晕染，留白 | 山水、边塞、禅意、抒情 |
|
||
| 青绿山水 | 青绿山水，石青石绿，金碧辉煌 | 春夏山水、游记、壮丽河山 |
|
||
| 工笔花鸟 | 工笔花鸟，细腻勾勒，精细渲染 | 花卉、仕女、宫廷、精致细腻 |
|
||
| 工笔重彩 | 工笔重彩，浓墨重色，华丽精细 | 华丽、富贵、节庆、历史叙事 |
|
||
| 没骨画法 | 没骨画法，不勾轮廓，直接点染 | 花卉、蔬果、清新淡雅 |
|
||
| 文人画 | 文人画风格，诗书画印，意趣高远 | 隐逸、高洁、书卷气 |
|
||
| 泼墨大写意 | 泼墨大写意，墨色淋漓，气势磅礴 | 豪放、苍茫、雄壮 |
|
||
| 界画/建筑 | 界画，工整精细，楼台亭阁 | 楼阁、宫殿、城市场景 |
|
||
| 浅绛山水 | 浅绛山水，赭石淡彩，萧疏清远 | 秋冬山水、萧瑟、怀古 |
|
||
|
||
### 体裁特点提示
|
||
- **唐诗**（尤其五七言律绝）：意境开阔，气象宏大，多配水墨写意或青绿山水。
|
||
- **宋词**：情感细腻，意象精致，婉约派多配工笔花鸟/没骨，豪放派可配泼墨写意。
|
||
- **边塞诗**：苍凉壮阔，适合泼墨大写意或浅绛山水。
|
||
- **田园诗**：恬淡自然，适合青绿山水或文人画。
|
||
- **咏物诗/闺怨词**：精致细腻，适合工笔花鸟或没骨画法。
|
||
|
||
## 分析步骤
|
||
|
||
1. 识别诗词的标题、作者、体裁、题材和情感基调。
|
||
2. 逐句/逐联理解字面意思与深层意境。
|
||
3. 判断需要多少幅画来完整呈现意境（通常每一联或每一句对应一幅，\
|
||
但意境连贯的句子可以合并为一幅）。
|
||
4. 为每幅画从上方风格菜单中选择最匹配的画风。
|
||
5. 为每幅画撰写**中文 prompt** 和**英文 prompt**，均采用「正向描述」策略（只描述要画什么，\
|
||
不描述不要什么），包含：
|
||
- 画面主体（人物、景物、动作、姿态）
|
||
- 环境氛围（季节、天气、光线、时辰、色调）
|
||
- 选定的艺术风格关键词
|
||
- 构图与视角（远景/中景/特写，俯视/平视等）
|
||
- 画面质感（绢本/纸本/留白/墨色浓淡等细节）
|
||
- 画面氛围（清冷/温暖/苍茫/静谧等情感色彩）
|
||
|
||
### 中文 prompt 要求
|
||
- 使用中国传统绘画的专业术语（如水墨写意、工笔重彩、留白等）。
|
||
- 具体且富有画面感，避免抽象空泛的概念。
|
||
|
||
### 英文 prompt 要求
|
||
- 中文 prompt 的忠实翻译与适配，保持相同的画面内容和风格意图。
|
||
- 使用英文中对应的艺术术语（如 ink wash painting, meticulous brushwork, negative space 等）。
|
||
- 自然流畅的英文表达，而非逐字翻译。
|
||
|
||
## 重要提示
|
||
- 文生图模型（Qwen-Image）对中英文 prompt 均有优秀支持，中文表现尤为突出。
|
||
- 支持 negative prompt：请为每幅画面生成针对性的 negative_prompt，排除与目标画风冲突的元素。
|
||
- 每个 prompt 建议 80-200 字（中文）/ 50-150 词（英文），确保细节充分。
|
||
- 必须同时输出中文和英文两个版本的 prompt。
|
||
|
||
### negative_prompt 编写要点
|
||
- 针对所选画风排除冲突风格（如：水墨写意应排除"照片写实, 3D渲染, 油画质感"；\
|
||
工笔花鸟应排除"粗犷笔触, 抽象风格, 泼墨"）。
|
||
- 排除常见 AI 生成瑕疵（如：肢体变形, 手指畸形, 面部模糊, 文字乱码）。
|
||
- 排除与诗词意境不符的元素（如：悲秋诗不应出现"鲜艳色彩, 欢快氛围"）。
|
||
- 简洁有效，20-60 字（中文），以逗号分隔。
|
||
|
||
## 输出格式
|
||
|
||
严格按照以下 JSON 格式输出，不要包含任何其他文字：
|
||
|
||
```json
|
||
{
|
||
"title": "诗词标题",
|
||
"author": "作者",
|
||
"dynasty": "朝代",
|
||
"genre": "体裁（如：五言绝句、七言律诗、词·水调歌头等）",
|
||
"analysis": "对整首诗意境的简要分析（中文，2-3句话）",
|
||
"images": [
|
||
{
|
||
"scene": "这幅画对应的诗句（原文）",
|
||
"description": "画面内容的中文描述",
|
||
"style": "选用的画风（中文名称）",
|
||
"prompt": "详细的中文文生图提示词，80-200字，仅使用正向描述...",
|
||
"prompt_en": "Detailed English text-to-image prompt, 50-150 words, positive description only...",
|
||
"negative_prompt": "针对该画面的负向提示词，排除与画风冲突的元素和常见瑕疵，20-60字..."
|
||
}
|
||
]
|
||
}
|
||
```\
|
||
"""
|
||
|
||
|
||
def analyze_poetry(poem: str, cfg: dict) -> dict:
|
||
"""调用 LLM 分析古诗词，返回结构化的图片生成方案。"""
|
||
llm_cfg = cfg["llm"]
|
||
|
||
client = OpenAI(
|
||
base_url=llm_cfg["base_url"],
|
||
api_key=llm_cfg["api_key"],
|
||
)
|
||
|
||
print(f"\n{'='*60}")
|
||
print("正在调用 LLM 分析古诗词意境...")
|
||
print(f"模型: {llm_cfg['model']}")
|
||
print(f"{'='*60}\n")
|
||
|
||
response = client.chat.completions.create(
|
||
model=llm_cfg["model"],
|
||
temperature=llm_cfg.get("temperature", 0.7),
|
||
max_tokens=llm_cfg.get("max_tokens", 4096),
|
||
messages=[
|
||
{"role": "system", "content": SYSTEM_PROMPT},
|
||
{"role": "user", "content": f"请分析以下古诗词并生成图片方案：\n\n{poem}"},
|
||
],
|
||
)
|
||
|
||
content = response.choices[0].message.content.strip()
|
||
|
||
json_match = re.search(r"```(?:json)?\s*(.*?)```", content, re.DOTALL)
|
||
if json_match:
|
||
content = json_match.group(1).strip()
|
||
|
||
try:
|
||
result = json.loads(content)
|
||
except json.JSONDecodeError:
|
||
json_match = re.search(r"\{.*\}", content, re.DOTALL)
|
||
if json_match:
|
||
result = json.loads(json_match.group())
|
||
else:
|
||
print("LLM 返回内容无法解析为 JSON：")
|
||
print(content)
|
||
sys.exit(1)
|
||
|
||
return result
|
||
|
||
|
||
def display_analysis(analysis: dict) -> None:
|
||
"""友好地展示 LLM 的分析结果。"""
|
||
print(f"\n{'='*60}")
|
||
title = analysis.get("title", "未知")
|
||
author = analysis.get("author", "未知")
|
||
dynasty = analysis.get("dynasty", "")
|
||
genre = analysis.get("genre", "")
|
||
print(f" {title} — {dynasty} · {author} [{genre}]")
|
||
print(f"{'='*60}")
|
||
print(f"\n 意境分析：{analysis.get('analysis', '')}\n")
|
||
|
||
for i, img in enumerate(analysis["images"], 1):
|
||
print(f"{'─'*50}")
|
||
print(f" 第 {i} 幅 | {img['scene']}")
|
||
print(f" 画风选择：{img.get('style', '未指定')}")
|
||
print(f" 中文描述：{img['description']}")
|
||
print(f" Prompt(zh)：{img['prompt'][:120]}...")
|
||
if img.get("prompt_en"):
|
||
print(f" Prompt(en)：{img['prompt_en'][:120]}...")
|
||
if img.get("negative_prompt"):
|
||
print(f" Negative ：{img['negative_prompt'][:120]}")
|
||
|
||
print(f"\n共 {len(analysis['images'])} 幅画面\n")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 尺寸预设（适配 Qwen-Image 推荐分辨率）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
SIZE_PRESETS: dict[str, str] = {
|
||
"square": "1328x1328", # 1:1
|
||
"phone": "928x1664", # 9:16
|
||
"phone_hd": "1056x1584", # 2:3（接近 9:16 高清）
|
||
"desktop": "1664x928", # 16:9
|
||
"desktop_hd": "1584x1056", # 3:2（接近 16:9 高清）
|
||
"landscape": "1472x1140", # 4:3
|
||
"portrait": "1140x1472", # 3:4
|
||
}
|
||
|
||
|
||
def resolve_image_size(img_cfg: dict) -> str:
|
||
"""根据 size_preset 或 height/width 配置，返回 'WIDTHxHEIGHT' 字符串。"""
|
||
preset = img_cfg.get("size_preset", "").strip().lower()
|
||
if preset and preset != "custom" and preset in SIZE_PRESETS:
|
||
return SIZE_PRESETS[preset]
|
||
w = img_cfg.get("width", 1328)
|
||
h = img_cfg.get("height", 1328)
|
||
return f"{w}x{h}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Qwen-Image API 图片生成
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _call_image_api(
|
||
prompt: str,
|
||
cfg: dict,
|
||
seed: int | None = None,
|
||
negative_prompt: str = "",
|
||
) -> tuple[str, int]:
|
||
"""调用 SiliconFlow Qwen-Image API，返回 (image_url, seed)。
|
||
|
||
图片 URL 有效期为 1 小时，调用方应及时下载。
|
||
negative_prompt: 每幅画面专属的负向提示词，会与配置中的全局 negative_prompt 合并。
|
||
"""
|
||
img_cfg = cfg["image"]
|
||
base_url = img_cfg.get("base_url", "https://api.siliconflow.cn/v1").rstrip("/")
|
||
api_key = img_cfg["api_key"]
|
||
|
||
url = f"{base_url}/images/generations"
|
||
headers = {
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
payload: dict = {
|
||
"model": img_cfg.get("model", "Qwen/Qwen-Image"),
|
||
"prompt": prompt,
|
||
"image_size": resolve_image_size(img_cfg),
|
||
}
|
||
|
||
steps = img_cfg.get("num_inference_steps")
|
||
if steps is not None:
|
||
payload["num_inference_steps"] = steps
|
||
|
||
guidance = img_cfg.get("guidance_scale")
|
||
if guidance is not None:
|
||
payload["guidance_scale"] = guidance
|
||
|
||
cfg_scale = img_cfg.get("cfg")
|
||
if cfg_scale is not None:
|
||
payload["cfg"] = cfg_scale
|
||
|
||
global_negative = img_cfg.get("negative_prompt", "").strip()
|
||
scene_negative = negative_prompt.strip()
|
||
parts = [p for p in (scene_negative, global_negative) if p]
|
||
merged_negative = ", ".join(parts)
|
||
if merged_negative:
|
||
payload["negative_prompt"] = merged_negative
|
||
|
||
if seed is not None and seed >= 0:
|
||
payload["seed"] = seed
|
||
|
||
max_retries = img_cfg.get("max_retries", 3)
|
||
timeout = img_cfg.get("request_timeout", 180)
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
resp = http_requests.post(
|
||
url, headers=headers, json=payload, timeout=timeout
|
||
)
|
||
|
||
if resp.status_code == 429:
|
||
wait = min(60, 5 * (attempt + 1))
|
||
print(f" API 限流 (429)，等待 {wait}s 后重试...")
|
||
time.sleep(wait)
|
||
continue
|
||
|
||
if resp.status_code != 200:
|
||
error_detail = resp.text[:500]
|
||
print(f" API 返回错误 [{resp.status_code}]: {error_detail}")
|
||
if attempt < max_retries - 1:
|
||
time.sleep(3)
|
||
continue
|
||
resp.raise_for_status()
|
||
|
||
data = resp.json()
|
||
img_url = data["images"][0]["url"]
|
||
returned_seed = data.get("seed", seed if seed and seed >= 0 else 0)
|
||
return img_url, returned_seed
|
||
|
||
except http_requests.exceptions.Timeout:
|
||
print(f" 请求超时 ({timeout}s)，" + (
|
||
f"重试 ({attempt+1}/{max_retries})..." if attempt < max_retries - 1 else "已达最大重试次数"
|
||
))
|
||
if attempt < max_retries - 1:
|
||
time.sleep(3)
|
||
continue
|
||
raise
|
||
|
||
except http_requests.exceptions.ConnectionError as e:
|
||
print(f" 连接失败: {e}")
|
||
if attempt < max_retries - 1:
|
||
time.sleep(5)
|
||
continue
|
||
raise
|
||
|
||
raise RuntimeError("API 调用失败，已达最大重试次数")
|
||
|
||
|
||
def _download_image(url: str, save_path: Path, timeout: int = 120) -> None:
|
||
"""下载图片并保存到本地。"""
|
||
resp = http_requests.get(url, timeout=timeout, stream=True)
|
||
resp.raise_for_status()
|
||
img = Image.open(BytesIO(resp.content))
|
||
img.save(save_path)
|
||
|
||
|
||
def generate_images(analysis: dict, cfg: dict) -> list[Path]:
|
||
"""根据分析结果逐一调用 Qwen-Image API 生成图片，返回保存路径列表。"""
|
||
img_cfg = cfg["image"]
|
||
out_cfg = cfg["output"]
|
||
|
||
output_dir = Path(out_cfg.get("dir", "./output"))
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
prefix = out_cfg.get("filename_prefix", "poem")
|
||
image_size = resolve_image_size(img_cfg)
|
||
seed = img_cfg.get("seed", -1)
|
||
prompt_lang = img_cfg.get("prompt_language", "zh")
|
||
images_per_prompt = max(1, min(4, img_cfg.get("images_per_prompt", 1)))
|
||
|
||
print(f"\n{'='*60}")
|
||
print("Qwen-Image API 图片生成")
|
||
print(f"模型: {img_cfg.get('model', 'Qwen/Qwen-Image')}")
|
||
print(f"图片尺寸: {image_size}")
|
||
print(f"Prompt 语言: {prompt_lang}")
|
||
if images_per_prompt > 1:
|
||
print(f"每个 prompt 生成 {images_per_prompt} 张图（不同种子）")
|
||
print(f"{'='*60}\n")
|
||
|
||
saved_paths = []
|
||
total = len(analysis["images"])
|
||
|
||
for i, img_info in enumerate(analysis["images"], 1):
|
||
if prompt_lang == "en" and img_info.get("prompt_en"):
|
||
prompt = img_info["prompt_en"]
|
||
else:
|
||
prompt = img_info["prompt"]
|
||
|
||
scene_negative = img_info.get("negative_prompt", "")
|
||
|
||
print(f"\n[{i}/{total}] 正在生成: {img_info['scene']}")
|
||
print(f" 画风: {img_info.get('style', '未指定')}")
|
||
print(f" Prompt({prompt_lang}): {prompt[:120]}...")
|
||
if scene_negative:
|
||
print(f" Negative: {scene_negative[:100]}")
|
||
|
||
for j in range(images_per_prompt):
|
||
variant_offset = i * 100 + j
|
||
if seed >= 0:
|
||
actual_seed = seed + variant_offset
|
||
else:
|
||
actual_seed = (int(time.time() * 1000) % (10**10)) + variant_offset
|
||
|
||
suffix = chr(ord("a") + j) if images_per_prompt > 1 else ""
|
||
if images_per_prompt > 1:
|
||
print(f" --- 第 {j+1}/{images_per_prompt} 张 (seed={actual_seed}) ---")
|
||
|
||
start_time = time.time()
|
||
|
||
try:
|
||
img_url, returned_seed = _call_image_api(
|
||
prompt, cfg, seed=actual_seed, negative_prompt=scene_negative
|
||
)
|
||
except Exception as e:
|
||
print(f" 生成失败: {e}")
|
||
continue
|
||
|
||
elapsed_api = time.time() - start_time
|
||
print(f" API 响应完成，耗时 {elapsed_api:.1f}s")
|
||
|
||
img_path = output_dir / f"{prefix}_{i:02d}{suffix}.png"
|
||
try:
|
||
_download_image(img_url, img_path)
|
||
saved_paths.append(img_path)
|
||
print(f" 已保存: {img_path}")
|
||
except Exception as e:
|
||
print(f" 图片下载失败: {e}")
|
||
print(f" URL（1小时内有效）: {img_url}")
|
||
|
||
if out_cfg.get("save_prompts", True):
|
||
txt_path = output_dir / f"{prefix}_{i:02d}_prompt.txt"
|
||
prompt_zh = img_info["prompt"]
|
||
prompt_en = img_info.get("prompt_en", "")
|
||
txt_path.write_text(
|
||
f"Scene: {img_info['scene']}\n"
|
||
f"Style: {img_info.get('style', '')}\n"
|
||
f"Description: {img_info['description']}\n"
|
||
f"Prompt(zh): {prompt_zh}\n"
|
||
f"Prompt(en): {prompt_en}\n"
|
||
f"Negative: {scene_negative}\n"
|
||
f"Used({prompt_lang}): {prompt}\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
return saved_paths
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主流程
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="古诗词意境图生成器 — 基于 LLM 分析 + Qwen-Image API 生成"
|
||
)
|
||
parser.add_argument(
|
||
"-c", "--config",
|
||
default="config_qwen.yaml",
|
||
help="配置文件路径（默认: config_qwen.yaml）",
|
||
)
|
||
parser.add_argument(
|
||
"-p", "--poem",
|
||
type=str,
|
||
default=None,
|
||
help="直接传入古诗词文本（如不指定则交互式输入）",
|
||
)
|
||
parser.add_argument(
|
||
"--analyze-only",
|
||
action="store_true",
|
||
help="仅进行 LLM 分析，不生成图片",
|
||
)
|
||
parser.add_argument(
|
||
"-o", "--output",
|
||
type=str,
|
||
default=None,
|
||
help="覆盖输出目录",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
cfg = load_config(args.config)
|
||
|
||
if args.output:
|
||
cfg["output"]["dir"] = args.output
|
||
else:
|
||
now = datetime.now()
|
||
date_dir = now.strftime("%Y-%m-%d")
|
||
time_dir = now.strftime("%H-%M-%S")
|
||
cfg["output"]["dir"] = str(
|
||
Path(cfg["output"].get("dir", "./output")) / date_dir / time_dir
|
||
)
|
||
|
||
if args.poem:
|
||
poem = args.poem
|
||
else:
|
||
print("请输入古诗词（输入空行结束）：")
|
||
lines = []
|
||
while True:
|
||
line = input()
|
||
if line.strip() == "":
|
||
break
|
||
lines.append(line)
|
||
poem = "\n".join(lines)
|
||
|
||
if not poem.strip():
|
||
print("未输入任何内容，退出。")
|
||
sys.exit(0)
|
||
|
||
print(f"\n输入的诗词：\n{poem}")
|
||
|
||
analysis = analyze_poetry(poem, cfg)
|
||
display_analysis(analysis)
|
||
|
||
output_dir = Path(cfg["output"].get("dir", "./output"))
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
analysis_path = output_dir / "analysis.json"
|
||
analysis_path.write_text(
|
||
json.dumps(analysis, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
print(f"分析结果已保存: {analysis_path}")
|
||
|
||
if args.analyze_only:
|
||
print("\n已完成分析（--analyze-only 模式），跳过图片生成。")
|
||
return
|
||
|
||
saved = generate_images(analysis, cfg)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"全部完成！共生成 {len(saved)} 幅图片：")
|
||
for p in saved:
|
||
print(f" {p}")
|
||
print(f"{'='*60}\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|