jinye_huang/TravelContentCreator
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

进行了模块解耦

Browse Source
This commit is contained in:
jinye_huang 2025-04-22 13:58:08 +08:00
parent 549223ab53
commit 2577d44a9c
5 changed files with 562 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
README.md
View File

@ -8,6 +8,7 @@
- **内容创作**：基于选题自动生成文字内容，包括标题和正文
- **海报制作**：结合景点图片和生成的文字内容，自动创建精美的宣传海报
- **批量处理**：支持一次性生成多个选题和多个变体内容
- **模块化设计**：各功能模块独立，可单独使用或组合使用
## 安装
@ -50,8 +51,8 @@ pip install numpy pandas opencv-python pillow
## 使用方法
1. 准备景点资源信息，放入`resource/Object/`目录中
2. 准备景点图片资源
3. 配置`poster_gen_config.json`文件（或直接修改`main.py`中的配置）
2. 准备景点图片资源，放入图片目录中
3. 复制`example_config.json`为`poster_gen_config.json`并进行配置
4. 运行主程序：
```bash
@ -60,9 +61,8 @@ python main.py
## 配置说明
主要配置项说明：
必须配置项：
- `date`: 日期标记
- `num`: 生成选题数量
- `model`: 使用的大语言模型
- `api_url`: API调用地址
@ -71,8 +71,19 @@ python main.py
- `prompts_dir`: 提示词目录
- `output_dir`: 输出结果保存路径
- `variants`: 每个选题生成的变体数量
- `topic_temperature`: 选题生成的创意度参数
- `content_temperature`: 内容生成的创意度参数
- `image_base_dir`: 图片资源根目录
- `topic_system_prompt`: 选题生成系统提示词路径
- `topic_user_prompt`: 选题生成用户提示词路径
- `content_system_prompt`: 内容生成系统提示词路径
可选配置项：
- `date`: 日期标记（默认为当前日期）
- `topic_temperature`: 选题生成的创意度参数（默认0.2）
- `content_temperature`: 内容生成的创意度参数（默认0.3）
- `camera_image_subdir`: 相机图片子目录名（默认"相机"）
- `modify_image_subdir`: 修改图片子目录名（默认"modify"）
- `poster_target_size`: 海报尺寸 [宽, 高]（默认[900, 1200]）
项目提供了一个示例配置文件 `example_config.json`，可以复制并根据需要修改：
@ -84,6 +95,50 @@ cp example_config.json poster_gen_config.json
nano poster_gen_config.json
```
配置文件示例：
```json
{
"date": "5月15日",
"num": 5,
"model": "qwen",
"api_url": "http://your-api-endpoint/v1/completions",
"api_key": "your-api-key",
"topic_system_prompt": "./SelectPrompt/systemPrompt.txt",
"topic_user_prompt": "./SelectPrompt/userPrompt.txt",
"content_system_prompt": "./genPrompts/systemPrompt.txt",
"resource_dir": [
{
"type": "Object",
"num": 3,
"file_path": [
"./resource/Object/景点信息-泰宁古城.txt",
"./resource/Object/景点信息-尚书第.txt",
"./resource/Object/景点信息-明清园.txt"
]
}
],
"prompts_dir": "./genPrompts",
"output_dir": "./result",
"image_base_dir": "/path/to/your/image/directory",
"camera_image_subdir": "相机",
"modify_image_subdir": "modify",
"variants": 3,
"topic_temperature": 0.2,
"content_temperature": 0.3,
"poster_target_size": [900, 1200]
}
```
## 项目流程
整个内容生成流程分为以下几个独立步骤：
1. **选题生成**：使用AI根据景点信息生成多个选题方向
2. **内容生成**：根据选题生成具体的旅游推文内容
3. **海报生成**：使用景点图片和生成的文本内容创建海报
每个步骤都有独立的函数处理，也可以作为整体流程一次执行。
## 开始使用
### 1. 准备景点信息文件
@ -99,9 +154,32 @@ nano poster_gen_config.json
### 2. 准备图片资源
将景点的高质量图片保存在指定图片目录中。
将景点的高质量图片保存在配置的图片目录中，符合以下结构：
### 3. 运行生成流程
```
image_base_dir/
├── 相机/ (camera_image_subdir)
│ ├── 泰宁古城/
│ │ ├── 图片1.jpg
│ │ ├── 图片2.jpg
│ │ └── description.txt (可选描述文件)
│ └── 其他景点/
└── modify/ (modify_image_subdir)
├── 泰宁古城/
│ ├── 图片1.jpg
│ ├── 图片2.jpg
│ └── ...
└── 其他景点/
```
### 3. 配置文件
复制并修改`example_config.json`为`poster_gen_config.json`，确保:
- 设置正确的API信息
- 配置景点资源文件路径
- 设置图片资源根目录
### 4. 运行生成流程
执行主程序，系统将自动生成选题、内容和海报：
@ -109,14 +187,15 @@ nano poster_gen_config.json
python main.py
```
生成的结果将保存在`result/`目录中。
生成的结果将保存在配置的`output_dir`目录中。
## 示例
可以查看`examples`目录中的示例代码，了解如何使用本工具生成海报。
查看`examples`目录中的示例代码，了解如何使用本工具的各个组件。
## 注意事项
- 确保已安装所有依赖库
- 图片目录结构需要按照配置文件中的规范组织
- AI生成内容质量取决于提供的景点信息质量和提示词设计
- AI生成内容质量取决于提供的景点信息质量和提示词设计
- 检查您的API密钥和URL配置是否正确

35
example_config.json Normal file
View File

@ -0,0 +1,35 @@
{
"date": "5月15日",
"num": 5,
"model": "qwen",
"api_url": "http://your-api-endpoint/v1/completions",
"api_key": "your-api-key",
"topic_system_prompt": "./SelectPrompt/systemPrompt.txt",
"topic_user_prompt": "./SelectPrompt/userPrompt.txt",
"content_system_prompt": "./genPrompts/systemPrompt.txt",
"resource_dir": [
{
"type": "Object",
"num": 3,
"file_path": [
"./resource/Object/景点信息-泰宁古城.txt",
"./resource/Object/景点信息-尚书第.txt",
"./resource/Object/景点信息-明清园.txt"
]
},
{
"type": "Product",
"num": 0,
"file_path": []
}
],
"prompts_dir": "./genPrompts",
"output_dir": "./result",
"image_base_dir": "/path/to/your/image/directory",
"camera_image_subdir": "相机",
"modify_image_subdir": "modify",
"variants": 3,
"topic_temperature": 0.2,
"content_temperature": 0.3,
"poster_target_size": [900, 1200]
}

53
examples/README.md Normal file
View File

@ -0,0 +1,53 @@
# 旅游内容创作工具示例
本目录包含使用旅游内容创作工具各个组件的示例代码。通过这些示例，您可以了解如何调用系统的各个模块，或测试整个工作流程。
## 测试工作流程
`test_workflow.py` 是一个完整的测试脚本，它展示了如何测试整个内容生成流程，包括：
1. 完整流程测试：执行从选题生成到海报制作的全过程
2. 分步骤测试：将流程拆分为独立的步骤，模拟在不同时间或机器上执行的场景
### 使用方法
确保已经正确配置了 `poster_gen_config.json` 文件（可以从项目根目录的 `example_config.json` 复制并修改）。
```bash
# 测试完整工作流程（从选题生成到海报制作的全过程）
python examples/test_workflow.py --mode full
# 测试分步骤工作流程（模拟分阶段执行的场景）
python examples/test_workflow.py --mode steps
# 两种测试模式都执行（默认）
python examples/test_workflow.py
```
## 海报生成示例
`generate_poster.py` 是一个单独的海报生成示例，它展示了如何使用系统的海报生成模块（不依赖于选题和内容生成）。
### 使用方法
```bash
# 使用默认参数生成海报
python examples/generate_poster.py
# 指定输入图片和输出路径
python examples/generate_poster.py --input_image /path/to/image.jpg --output_path /path/to/output.jpg
```
## 其他示例
未来将添加更多示例，演示如何独立使用系统的各个组件，例如：
- 仅使用选题生成功能
- 仅使用内容生成功能
- 自定义AI模型和提示词
## 注意事项
- 这些示例依赖于主项目中的配置和资源，确保已正确设置 `poster_gen_config.json`
- 测试脚本会自动调节某些参数（如生成数量）以加快测试速度
- 实际使用时，您可能需要调整参数以获得更好的效果

155
examples/test_workflow.py Normal file
View File

@ -0,0 +1,155 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试脚本：验证旅游内容创作工具的完整工作流程
"""
import os
import sys
import json
import time
# 添加项目根目录到Python路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 导入所需模块
from main import load_config, generate_topics_step, generate_content_and_posters_step
def test_full_workflow():
"""测试完整的工作流程，从选题生成到海报制作"""
print("测试完整工作流程...")
try:
# 1. 加载配置
print("步骤 1: 加载配置...")
config = load_config()
print(f"配置加载成功: {len(config)} 个配置项")
# 2. 执行选题生成
print("\n步骤 2: 生成选题...")
run_id, tweet_topic_record = generate_topics_step(config)
if not run_id or not tweet_topic_record:
print("选题生成失败，测试终止。")
return False
print(f"选题生成成功，Run ID: {run_id}")
print(f"生成了 {len(tweet_topic_record.topics_list)} 个选题")
# 3. 执行内容和海报生成
print("\n步骤 3: 生成内容和海报...")
generate_content_and_posters_step(config, run_id, tweet_topic_record)
# 4. 验证输出
output_dir = os.path.join(config["output_dir"], run_id)
if os.path.exists(output_dir):
items = os.listdir(output_dir)
print(f"\n输出目录 {output_dir} 中有 {len(items)} 个项目")
# 检查是否有生成的海报
poster_count = 0
for item in items:
if item.startswith(("1_", "2_", "3_", "4_", "5_")): # 假设选题编号从1开始
poster_dir = os.path.join(output_dir, item, "poster")
if os.path.exists(poster_dir):
poster_files = [f for f in os.listdir(poster_dir) if f.endswith(".jpg")]
poster_count += len(poster_files)
print(f"共生成 {poster_count} 个海报")
return poster_count > 0
else:
print(f"输出目录 {output_dir} 不存在")
return False
except Exception as e:
print(f"测试过程中出错: {e}")
import traceback
traceback.print_exc()
return False
def test_steps_separately():
"""分步测试工作流程"""
print("分步测试工作流程...")
try:
# 1. 加载配置
print("步骤 1: 加载配置...")
config = load_config()
# 将variants设置为较小的值以加快测试
test_config = config.copy()
test_config["variants"] = 1 # 每个选题只生成1个变体
test_config["num"] = 2 # 只生成2个选题
print(f"测试配置已准备: variants={test_config['variants']}, num={test_config['num']}")
# 2. 仅执行选题生成
print("\n步骤 2: 仅测试选题生成...")
run_id, tweet_topic_record = generate_topics_step(test_config)
if not run_id or not tweet_topic_record:
print("选题生成失败，测试终止。")
return False
# 保存生成的选题数据用于后续测试
topics_file = os.path.join(test_config["output_dir"], run_id, "tweet_topic.json")
print(f"选题结果已保存到: {topics_file}")
# 3. 模拟间隔后继续处理
print("\n步骤 3: 模拟间隔后处理选题结果...")
print("（在真实场景中，可能是在不同的时间或机器上进行后续处理）")
time.sleep(2) # 模拟短暂延迟
# 4. 从保存的选题文件加载数据，并手动执行内容和海报生成
print("\n步骤 4: 加载保存的选题，执行内容和海报生成...")
# 这部分通常是由main函数中的流程自动处理的
# 这里为了演示分段流程，模拟手动加载数据并处理
from core.topic_parser import TopicParser
if os.path.exists(topics_file):
with open(topics_file, 'r', encoding='utf-8') as f:
topics_data = json.load(f)
loaded_topic_record = TopicParser.from_json(topics_data)
print(f"从 {topics_file} 加载了 {len(loaded_topic_record.topics_list)} 个选题")
# 执行内容和海报生成步骤
generate_content_and_posters_step(test_config, run_id, loaded_topic_record)
return True
else:
print(f"选题文件 {topics_file} 不存在")
return False
except Exception as e:
print(f"测试过程中出错: {e}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
print("==== 旅游内容创作工具工作流程测试 ====\n")
# 判断测试模式
import argparse
parser = argparse.ArgumentParser(description='测试旅游内容创作工具的工作流程')
parser.add_argument('--mode', type=str, choices=['full', 'steps', 'both'],
default='both', help='测试模式: full=完整流程, steps=分步测试, both=两种模式都测试')
args = parser.parse_args()
success = True
if args.mode in ['full', 'both']:
print("\n==== 测试完整工作流程 ====")
success = test_full_workflow() and success
if args.mode in ['steps', 'both']:
print("\n==== 测试分步工作流程 ====")
success = test_steps_separately() and success
if success:
print("\n==== 所有测试通过! ====")
sys.exit(0)
else:
print("\n==== 测试失败! ====")
sys.exit(1)

399
main.py
View File

@ -4,6 +4,7 @@ from datetime import datetime
import argparse
import sys
import traceback
import json
from core.ai_agent import AI_Agent
from core.topic_parser import TopicParser
@ -16,182 +17,240 @@ import random
TEXT_POSBILITY = 0.3
def main():
config_file = {
"date": "4月24日",
"num": 10,
"model": "qwenQWQ",
"api_url": "vllm",
"api_key": "EMPTY",
"topic_system_prompt": "/root/autodl-tmp/TravelContentCreator/SelectPrompt/systemPrompt.txt",
"topic_user_prompt": "/root/autodl-tmp/TravelContentCreator/SelectPrompt/userPrompt.txt",
"content_system_prompt": "/root/autodl-tmp/TravelContentCreator/genPrompts/systemPrompt.txt",
"resource_dir": [{
"type": "Object",
"num": 4,
"file_path": ["/root/autodl-tmp/TravelContentCreator/resource/Object/景点信息-尚书第.txt",
"/root/autodl-tmp/TravelContentCreator/resource/Object/景点信息-明清园.txt",
"/root/autodl-tmp/TravelContentCreator/resource/Object/景点信息-泰宁古城.txt",
"/root/autodl-tmp/TravelContentCreator/resource/Object/景点信息-甘露寺.txt"
]},
{
"type": "Product",
"num": 0,
"file_path": []
}
],
"prompts_dir": "/root/autodl-tmp/TravelContentCreator/genPrompts",
"output_dir": "/root/autodl-tmp/TravelContentCreator/result",
"variants": 10,
"topic_temperature": 0.2,
"content_temperature": 0.3
}
def load_config(config_path="poster_gen_config.json"):
"""Loads configuration from a JSON file."""
if not os.path.exists(config_path):
print(f"Error: Configuration file '{config_path}' not found.")
print("Please copy 'example_config.json' to 'poster_gen_config.json' and customize it.")
sys.exit(1)
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
# Basic validation (can be expanded)
required_keys = ["api_url", "model", "api_key", "resource_dir", "prompts_dir", "output_dir", "num", "variants", "topic_system_prompt", "topic_user_prompt", "content_system_prompt", "image_base_dir"]
if not all(key in config for key in required_keys):
print(f"Error: Config file '{config_path}' is missing one or more required keys.")
print(f"Required keys are: {required_keys}")
sys.exit(1)
# Resolve relative paths based on config location or a defined base path if necessary
# For simplicity, assuming paths in config are relative to project root or absolute
return config
except json.JSONDecodeError:
print(f"Error: Could not decode JSON from '{config_path}'. Check the file format.")
sys.exit(1)
except Exception as e:
print(f"Error loading configuration from '{config_path}': {e}")
sys.exit(1)
if True:
# 1. 首先生成选题
ai_agent, system_prompt, user_prompt, output_dir = prepare_topic_generation(
config_file["date"], config_file["num"], config_file["topic_system_prompt"], config_file["topic_user_prompt"],
config_file["api_url"], config_file["model"], config_file["api_key"], config_file["prompts_dir"], config_file["resource_dir"], config_file["output_dir"]
)
run_id, tweet_topic_record = generate_topics(
ai_agent, system_prompt, user_prompt, config_file["output_dir"],
config_file["topic_temperature"], 0.5, 1.5
)
output_dir = os.path.join(config_file["output_dir"], run_id)
os.makedirs(output_dir, exist_ok=True)
tweet_topic_record.save_topics(os.path.join(output_dir, "tweet_topic.json"))
tweet_topic_record.save_prompt(os.path.join(output_dir, "tweet_prompt.txt"))
def generate_topics_step(config):
"""Generates topics based on the configuration."""
print("Step 1: Generating Topics...")
ai_agent, system_prompt, user_prompt, base_output_dir = prepare_topic_generation(
config.get("date", datetime.now().strftime("%Y-%m-%d")), # Use current date if not specified
config["num"], config["topic_system_prompt"], config["topic_user_prompt"],
config["api_url"], config["model"], config["api_key"], config["prompts_dir"],
config["resource_dir"], config["output_dir"]
)
run_id, tweet_topic_record = generate_topics(
ai_agent, system_prompt, user_prompt, config["output_dir"],
config.get("topic_temperature", 0.2), config.get("topic_top_p", 0.5), config.get("topic_max_tokens", 1.5) # Added defaults for safety
)
if not run_id or not tweet_topic_record:
print("Topic generation failed. Exiting.")
ai_agent.close()
# raise Exception("选题生成失败，退出程序")
if not run_id or not tweet_topic_record:
print("选题生成失败，退出程序")
return
# 2. 然后生成内容
print("\n开始根据选题生成内容...")
# 加载内容生成的系统提示词
content_system_prompt = ResourceLoader.load_system_prompt(config_file["content_system_prompt"])
return None, None
if not content_system_prompt:
print("内容生成系统提示词为空，使用选题生成的系统提示词")
content_system_prompt = system_prompt
# 直接使用同一个AI Agent实例
for i in range(len(tweet_topic_record.topics_list)):
tweet_content_list = []
for j in range(config_file["variants"]):
time.sleep(random.random())
ai_agent = AI_Agent(config_file["api_url"], config_file["model"], config_file["api_key"])
output_dir = os.path.join(config["output_dir"], run_id)
os.makedirs(output_dir, exist_ok=True)
tweet_topic_record.save_topics(os.path.join(output_dir, "tweet_topic.json"))
tweet_topic_record.save_prompt(os.path.join(output_dir, "tweet_prompt.txt"))
ai_agent.close()
print(f"Topics generated successfully. Run ID: {run_id}")
return run_id, tweet_topic_record
def generate_content_and_posters_step(config, run_id, tweet_topic_record):
"""Generates content and posters based on generated topics."""
if not run_id or not tweet_topic_record:
print("Missing run_id or topics data. Skipping content and poster generation.")
return
print("Step 2: Generating Content and Posters...")
base_output_dir = config["output_dir"]
output_dir = os.path.join(base_output_dir, run_id) # Directory for this specific run
# Load content generation system prompt
content_system_prompt = ResourceLoader.load_system_prompt(config["content_system_prompt"])
if not content_system_prompt:
print("Warning: Content generation system prompt is empty. Using default logic if available or might fail.")
# Potentially load topic system prompt as fallback if needed, or handle error
# content_system_prompt = ResourceLoader.load_system_prompt(config["topic_system_prompt"])
image_base_dir = config.get("image_base_dir", None)
if not image_base_dir:
print("Error: 'image_base_dir' not specified in config. Cannot locate images.")
return
camera_image_subdir = config.get("camera_image_subdir", "相机") # Default '相机'
modify_image_subdir = config.get("modify_image_subdir", "modify") # Default 'modify'
for i, topic in enumerate(tweet_topic_record.topics_list):
topic_index = i + 1
print(f"Processing Topic {topic_index}/{len(tweet_topic_record.topics_list)}: {topic.get('title', 'N/A')}")
tweet_content_list = []
# --- Content Generation Loop ---
for j in range(config["variants"]):
variant_index = j + 1
print(f" Generating Variant {variant_index}/{config['variants']}...")
time.sleep(random.random()) # Keep the random delay? Okay for now.
ai_agent = AI_Agent(config["api_url"], config["model"], config["api_key"])
try:
tweet_content, gen_result = generate_single_content(
ai_agent, content_system_prompt, tweet_topic_record.topics_list[i],
config_file["prompts_dir"], config_file["resource_dir"],
output_dir, run_id, i+1, j+1, config_file["content_temperature"]
ai_agent, content_system_prompt, topic,
config["prompts_dir"], config["resource_dir"],
output_dir, run_id, topic_index, variant_index, config.get("content_temperature", 0.3) # Added default
)
tweet_content_list.append(tweet_content.get_json_file())
ai_agent.close()
if not tweet_content:
print(f"生成第{i+1}篇文章的第{j+1}个变体失败，跳过")
continue
object_name = tweet_topic_record.topics_list[i]["object"]
try:
object_name = object_name.split(".")[0]
except:
pass
try:
object_name = object_name.split("景点信息-")[1]
except:
pass
# 处理对象名称中可能包含的"+"等特殊字符
# if "+" in object_name:
# # 取第一个景点名称作为主要对象
# object_name = object_name.split("+")[0].strip()
# print(f"对象名称包含多个景点，使用第一个景点：{object_name}")
# 检查图片路径是否存在
img_dir_path = f"/root/autodl-tmp/sanming_img/modify/{object_name}"
if not os.path.exists(img_dir_path):
print(f"图片目录不存在：{img_dir_path}，跳过该对象")
continue
info_directory = [
f"/root/autodl-tmp/sanming_img/相机/{object_name}/description.txt"
]
# 检查描述文件是否存在
if not os.path.exists(info_directory[0]):
print(f"描述文件不存在：{info_directory[0]}，使用生成的内容替代")
info_directory = []
poster_num = config_file["variants"]
input_dir = img_dir_path # 使用前面检查过的目录路径
target_size = (900, 1200)
result_path = []
content_gen = contentGen.ContentGenerator()
response = content_gen.run(info_directory, poster_num, tweet_content_list)
print(response)
try:
poster_config_summary = posterGen.PosterConfig(response)
for j_index in range(config_file["variants"]):
poster_config = poster_config_summary.get_config_by_index(j_index)
img_dir = os.path.join(output_dir, f"{i+1}_{j_index+1}")
try:
# 创建输出目录
collage_output_dir = os.path.join(img_dir, "collage_img")
os.makedirs(collage_output_dir, exist_ok=True)
poster_output_dir = os.path.join(img_dir, "poster")
os.makedirs(poster_output_dir, exist_ok=True)
# 处理图片目录
img_list = simple_collage.process_directory(
input_dir,
target_size=target_size,
output_count=1,
output_dir=collage_output_dir
)
print(img_list)
if not img_list or len(img_list) == 0:
print(f"未能生成拼贴图片，跳过海报生成")
continue
# 生成海报
poster_gen = posterGen.PosterGenerator()
if random.random() < TEXT_POSBILITY:
text_data = {
"title": f"{poster_config['main_title']}",
"subtitle": "",
"additional_texts": [
{"text": f"{poster_config['texts'][0]}", "position": "bottom", "size_factor": 0.5},
{"text": f"{poster_config['texts'][1]}", "position": "bottom", "size_factor": 0.5}
]
}
else:
text_data = {
"title": f"{poster_config['main_title']}",
"subtitle": "",
"additional_texts": [
{"text": f"{poster_config['texts'][0]}", "position": "bottom", "size_factor": 0.5},
# {"text": f"{poster_config['texts'][1]}", "position": "bottom", "size_factor": 0.5}
]
}
print(text_data)
img_path = img_list[0]['path']
print(f"使用图片路径: {img_path}")
output_path = os.path.join(poster_output_dir, f"poster.jpg")
result_path.append(poster_gen.create_poster(img_path, text_data, output_path))
except Exception as e:
print(f"海报生成过程中出错: {e}")
traceback.print_exc() # 打印完整的堆栈跟踪信息
continue
if tweet_content:
tweet_content_list.append(tweet_content.get_json_file()) # Assuming this returns the structured data needed later
else:
print(f" Failed to generate content for Topic {topic_index}, Variant {variant_index}. Skipping.")
except Exception as e:
print(f"配置解析失败，跳过当前项目: {e}")
traceback.print_exc() # 打印完整的堆栈跟踪信息
continue
print(f" Error during content generation for Topic {topic_index}, Variant {variant_index}: {e}")
traceback.print_exc()
finally:
ai_agent.close() # Ensure agent is closed
if not tweet_content_list:
print(f" No content generated for Topic {topic_index}. Skipping poster generation.")
continue
# --- Poster Generation Setup ---
object_name = topic.get("object", "")
if not object_name:
print(" Warning: Topic object name is missing. Cannot determine image path.")
continue
# Clean object name (consider making this a utility function)
try:
object_name = object_name.split(".")[0]
if "景点信息-" in object_name:
object_name = object_name.split("景点信息-")[1]
# Handle cases like "景点A+景点B"? Needs clearer logic if required.
except Exception as e:
print(f" Warning: Could not fully clean object name '{object_name}': {e}")
# Construct and check image paths using config base dir
# Path for collage/poster input images (e.g., from 'modify' dir)
input_img_dir_path = os.path.join(image_base_dir, modify_image_subdir, object_name)
# Path for potential description file (e.g., from '相机' dir)
camera_img_dir_path = os.path.join(image_base_dir, camera_image_subdir, object_name)
description_file_path = os.path.join(camera_img_dir_path, "description.txt")
if not os.path.exists(input_img_dir_path) or not os.path.isdir(input_img_dir_path):
print(f" Image directory not found or not a directory: '{input_img_dir_path}'. Skipping poster generation for this topic.")
continue
info_directory = []
if os.path.exists(description_file_path):
info_directory = [description_file_path]
print(f" Using description file: {description_file_path}")
else:
print(f" Description file not found: '{description_file_path}'. Using generated content for poster text.")
# --- Generate Text Configurations for Posters ---
content_gen = contentGen.ContentGenerator()
try:
# Assuming tweet_content_list contains the JSON data needed by content_gen.run
poster_text_configs_raw = content_gen.run(info_directory, config["variants"], tweet_content_list)
print(f" Raw poster text configs: {poster_text_configs_raw}") # For debugging
poster_config_summary = posterGen.PosterConfig(poster_text_configs_raw)
except Exception as e:
print(f" Error running ContentGenerator or parsing poster configs: {e}")
traceback.print_exc()
continue # Skip poster generation for this topic
# --- Poster Generation Loop ---
poster_num = config["variants"] # Same as content variants
target_size = tuple(config.get("poster_target_size", [900, 1200])) # Add default size
for j_index in range(poster_num):
variant_index = j_index + 1
print(f" Generating Poster {variant_index}/{poster_num}...")
try:
poster_config = poster_config_summary.get_config_by_index(j_index)
if not poster_config:
print(f" Warning: Could not get poster config for index {j_index}. Skipping.")
continue
variant_output_dir = os.path.join(output_dir, f"{topic_index}_{variant_index}")
collage_output_dir = os.path.join(variant_output_dir, "collage_img")
poster_output_dir = os.path.join(variant_output_dir, "poster")
os.makedirs(collage_output_dir, exist_ok=True)
os.makedirs(poster_output_dir, exist_ok=True)
# --- Image Collage ---
img_list = simple_collage.process_directory(
input_img_dir_path,
target_size=target_size,
output_count=1, # Assuming 1 collage image per poster variant
output_dir=collage_output_dir
)
# print(f" Collage image list: {img_list}") # Debugging
if not img_list or len(img_list) == 0 or not img_list[0].get('path'):
print(f" Failed to generate collage image for Variant {variant_index}. Skipping poster.")
continue
collage_img_path = img_list[0]['path']
print(f" Using collage image: {collage_img_path}")
# --- Create Poster ---
poster_gen_instance = posterGen.PosterGenerator() # Renamed to avoid conflict
# Prepare text data (Simplified logic, adjust TEXT_POSBILITY if needed)
# Consider moving text data preparation into posterGen or a dedicated function
text_data = {
"title": poster_config.get('main_title', 'Default Title'),
"subtitle": "", # Subtitle seems unused?
"additional_texts": []
}
texts = poster_config.get('texts', [])
if texts:
text_data["additional_texts"].append({"text": texts[0], "position": "bottom", "size_factor": 0.5})
if len(texts) > 1 and random.random() < TEXT_POSBILITY: # Apply possibility check
text_data["additional_texts"].append({"text": texts[1], "position": "bottom", "size_factor": 0.5})
# print(f" Text data for poster: {text_data}") # Debugging
final_poster_path = os.path.join(poster_output_dir, "poster.jpg")
result_path = poster_gen_instance.create_poster(collage_img_path, text_data, final_poster_path)
if result_path:
print(f" Successfully generated poster: {result_path}")
else:
print(f" Poster generation function did not return a valid path.")
except Exception as e:
print(f" Error during poster generation for Variant {variant_index}: {e}")
traceback.print_exc()
continue # Continue to next variant
def main():
# No argparse for now, directly load default config
config = load_config() # Load from poster_gen_config.json
# Execute steps sequentially
run_id, tweet_topic_record = generate_topics_step(config)
if run_id and tweet_topic_record:
generate_content_and_posters_step(config, run_id, tweet_topic_record)
else:
print("Exiting due to issues in topic generation.")
if __name__ == "__main__":
main()