修改了poster_gen的目录路径

This commit is contained in:
jinye_huang 2025-04-25 10:11:45 +08:00
parent 59bb2d9865
commit b554e13025
14 changed files with 754 additions and 220 deletions

340
README.md
View File

@ -1,260 +1,184 @@
# 旅游内容创作工具 (Travel Content Creator)
# TravelContentCreator
这是一个基于AI的旅游内容自动生成工具可以根据景点信息自动生成高质量的旅游推文和宣传海报。
TravelContentCreator是一个用于自动化生成旅游内容和宣传海报的系统。该系统利用AI技术生成景点选题、撰写文章内容并制作相应的宣传海报。
## 功能特点
- **自动选题生成**:根据提供的景点信息和配置的提示词模板,自动生成吸引人的旅游选题
- **内容创作**:基于选题和配置的提示词模板,自动生成文字内容(标题、正文)
- **海报制作**:结合景点图片和生成的文字内容,自动创建精美的宣传海报
- **批量处理**:支持一次性生成多个选题和多个变体内容
- **模块化设计**核心功能配置加载、提示词管理、AI交互、选题、内容生成、海报制作分离方便维护和扩展
- **配置驱动**:通过配置文件集中管理所有运行参数
- 自动生成旅游景点选题(包括目标受众、写作风格等)
- 根据选题生成详细的旅游文章内容
- 为生成的文章制作美观的宣传海报
- 支持多种写作风格和目标受众需求
- 文件模糊匹配功能,增强系统健壮性
- 模块化设计,可单独使用选题、内容生成或海报生成功能
## 新功能: 流式输出处理
## 目录结构
TravelContentCreator 现已支持三种流式输出处理方法,提供了更灵活的 AI 文本生成体验:
```
TravelContentCreator/
├── core/ # 核心功能模块
├── utils/ # 工具类和辅助函数
├── genPrompts/ # 生成提示词
│ ├── Style/ # 风格提示词
│ ├── Demand/ # 需求提示词
│ └── Refer/ # 参考提示词
├── SelectPrompt/ # 选题提示词
├── resource/ # 资源文件
├── main.py # 主程序
├── test_topic_content.py # 选题和内容生成测试脚本
├── test_poster.py # 海报生成测试脚本
├── topic_content_config.json # 选题和内容生成配置
├── poster_config.json # 海报生成配置
└── poster_gen_config.json # 主程序配置
```
- **同步流式响应**: 使用流式 API 但返回完整响应
- **回调式流式响应**: 通过回调函数处理每个文本块
- **异步流式响应**: 使用异步生成器返回文本流
这些功能大大提升了长文本生成的用户体验和系统响应性。
详细文档请参阅:
- [流式处理文档](docs/streaming.md)
- [流式处理演示](examples/test_stream.py)
## 快速开始
### 1. 环境准备
## 安装
1. 克隆仓库
```bash
# 克隆项目
git clone https://github.com/yourusername/TravelContentCreator.git
git clone [仓库URL]
cd TravelContentCreator
# 安装依赖 (假设有requirements.txt文件)
# pip install -r requirements.txt
# 或者手动安装
pip install numpy pandas opencv-python pillow requests tqdm
```
### 2. 配置设置
2. 安装依赖
```bash
pip install -r requirements.txt
```
3. 配置环境
- 确保安装了Python 3.6+
- 配置好AI模型API本系统默认使用QwenAPI
- 准备好必要的素材和资源文件
## 使用说明
### 1. 主程序
主程序可以执行完整的流程,包括选题生成、内容生成和海报生成:
```bash
# 复制示例配置(选择一个或从基础开始)
cp configs/basic_config.json poster_gen_config.json
# 编辑配置文件
vim poster_gen_config.json
# 必须修改api_url, api_key, image_base_dir
python main.py [--config CONFIG_PATH] [--run_id RUN_ID] [--topics_file TOPICS_FILE] [--debug]
```
### 3. 运行系统
参数说明:
- `--config`: 配置文件路径,默认为`poster_gen_config.json`
- `--run_id`: 自定义运行ID用于区分不同批次的生成结果
- `--topics_file`: 预生成的选题文件路径,如果提供则跳过选题生成步骤
- `--debug`: 启用调试级别日志
### 2. 测试选题和内容生成
使用专门的测试脚本运行选题和内容生成模块:
```bash
# 完整流程(从选题到海报生成)
python main.py
# 或分阶段执行 (使用默认配置)
python examples/run_step1_topics.py
# 记下输出的Run ID
python examples/run_step2_content_posters.py YOUR_RUN_ID
# 使用特定配置运行
# python main.py --config configs/social_media_config.json
python test_topic_content.py [--config CONFIG_PATH] [--run_id RUN_ID] [--topics_file TOPICS_FILE] [--debug]
```
### 4. 查看结果
参数说明:
- `--config`: 配置文件路径,默认为`topic_content_config.json`
- `--run_id`: 自定义运行ID
- `--topics_file`: 预生成的选题文件路径,如果提供则跳过选题生成
- `--debug`: 启用调试级别日志
### 3. 测试海报生成
使用专门的测试脚本运行海报生成模块:
```bash
# 结果保存在配置的output_dir目录下默认为./result/
ls -la ./result/最新的Run_ID/
python test_poster.py --topics_file TOPICS_FILE [--config CONFIG_PATH] [--topic_index TOPIC_INDEX] [--run_id RUN_ID] [--debug]
```
## 核心组件说明
参数说明:
- `--topics_file`: 必需的选题JSON文件路径用于获取海报生成的主题数据
- `--config`: 配置文件路径,默认为`poster_config.json`
- `--topic_index`: 要生成海报的特定选题索引,如果未提供则为所有选题生成海报
- `--run_id`: 自定义运行ID
- `--debug`: 启用调试级别日志
项目采用模块化设计,主要包含以下组件:
## 配置文件说明
- **主流程协调器** (`main.py`): 负责加载配置并协调执行整个生成流程
- **AI交互模块** (`core/ai_agent.py`): 封装与大语言模型的通信
- **选题生成器** (`utils/tweet_generator.py`): 生成旅游选题
- **内容生成器** (`core/contentGen.py`): 处理内容创作
- **海报制作器** (`core/posterGen.py`): 合成文字和图片,生成最终海报
## 资源准备指南
### 1. 景点信息文件
`resource/Object/` 目录创建景点信息文件,示例格式:
```
景点名称:泰宁古城
位置:福建省三明市泰宁县
简介:泰宁古城始建于宋代...
特色:古城墙、古街巷...
历史:泰宁古城有着悠久的历史...
适合游客:喜欢历史文化的游客、摄影爱好者
建议游览时间2-3小时
最佳季节:春季和秋季
```
> **提示**:景点信息越详细,生成的内容质量越高
### 2. 图片资源结构
图片资源应按以下结构组织(可通过配置自定义目录名):
```
<image_base_dir>/ # 配置中的图片根目录
├── 相机/ # 存放原始照片 (camera_image_subdir)
│ ├── 泰宁古城/
│ │ ├── 图片1.jpg
│ │ ├── 图片2.jpg
│ │ └── description.txt (可选的图片描述)
│ └── 其他景点/
└── modify/ # 存放处理后的图片 (modify_image_subdir)
├── 泰宁古城/
│ ├── 图片1.jpg
│ └── ...
└── 其他景点/
```
> **重要**:确保每个景点的图片目录名与景点信息文件中的名称匹配。海报生成默认从 `modify/` 目录选取图片。
## 配置文件详解
`poster_gen_config.json` 是系统的核心配置文件,包含以下主要配置项:
### 基础配置
### 选题和内容生成配置 (topic_content_config.json)
```json
{
"date": "5月15日", // 日期标记,用于提示词
"num": 5, // 生成选题数量
"variants": 3 // 每个选题生成的变体数量
}
```
### AI模型配置
```json
"date": "5月15日, 5月16日, 5月17日, 6月1日", // 选题日期
"num": 2, // 生成选题数量
"variants": 1, // 每个选题的变体数量
"topic_temperature": 0.2, // 选题生成的temperature参数
"content_temperature": 0.3, // 内容生成的temperature参数
"model": "qwenQWQ", // 使用的AI模型
"api_url": "http://localhost:8000/v1/", // API地址
"api_key": "EMPTY", // API密钥
"topic_system_prompt": "./SelectPrompt/systemPrompt.txt", // 选题系统提示词路径
"topic_user_prompt": "./SelectPrompt/userPrompt.txt", // 选题用户提示词路径
"content_system_prompt": "./genPrompts/systemPrompt.txt", // 内容系统提示词路径
"prompts_config": [ // 提示词配置
{
"model": "qwen", // 使用的模型名称
"api_url": "http://localhost:8000/v1/", // API端点
"api_key": "YOUR_API_KEY", // API密钥
"topic_temperature": 0.2, // 选题生成的随机性
"content_temperature": 0.3 // 内容生成的随机性
}
```
### 资源路径配置
```json
"type": "Style", // 风格提示词
"file_path": [...] // 风格提示词文件路径列表
},
{
"resource_dir": [ // 景点信息资源
"type": "Demand", // 需求提示词
"file_path": [...] // 需求提示词文件路径列表
},
...
],
"resource_dir": [ // 资源目录配置
{
"type": "Object",
"num": 3,
"file_path": [
"./resource/Object/景点信息-泰宁古城.txt",
"./resource/Object/景点信息-尚书第.txt"
"type": "Object", // 对象类型
"file_path": [...] // 对象文件路径列表
},
...
]
}
],
"image_base_dir": "/path/to/your/image/directory", // 图片根目录
"camera_image_subdir": "相机", // 原始照片子目录
"modify_image_subdir": "modify" // 处理后图片子目录
}
```
### 提示词配置
```json
{
"topic_system_prompt": "./SelectPrompt/systemPrompt.txt",
"topic_user_prompt": "./SelectPrompt/userPrompt.txt",
"content_system_prompt": "./genPrompts/systemPrompt.txt",
"prompts_dir": "./genPrompts"
}
```
### 输出配置
### 海报生成配置 (poster_config.json)
```json
{
"variants": 1, // 每个选题的变体数量
"model": "qwenQWQ", // 使用的AI模型
"api_url": "http://localhost:8000/v1/", // API地址
"api_key": "EMPTY", // API密钥
"poster_content_system_prompt": "./genPrompts/poster_content_systemPrompt.txt", // 海报内容系统提示词路径
"resource_dir": [...], // 资源目录配置
"output_dir": "./result", // 输出目录
"poster_target_size": [900, 1200], // 海报尺寸
"text_possibility": 0.3 // 文字元素出现概率
"image_base_dir": "...", // 图片基础目录
"poster_assets_base_dir": "...", // 海报素材基础目录
"poster_target_size": [900, 1200], // 海报目标尺寸
"text_possibility": 0.3, // 文本可能性
"img_frame_possibility": 0.7, // 图像框可能性
"text_bg_possibility": 0 // 文本背景可能性
}
```
## 配置示例
## 结果输出
本项目提供了多种预设配置文件,适用于不同场景。这些配置文件位于 `configs/` 目录下
生成的结果保存在配置文件中指定的`output_dir`目录下,按照`run_id`组织。每次运行的结果包括
- **基础配置** (`configs/basic_config.json`): 适合初次使用和测试
- **OpenAI配置** (`configs/openai_config.json`): 使用OpenAI API的配置
- **高质量配置** (`configs/high_quality_config.json`): 更高质量的生成设置
- **批量处理配置** (`configs/batch_processing_config.json`): 处理大量景点信息
- **社交媒体配置** (`configs/social_media_config.json`): 针对多个社交平台优化
- **本地LLM配置** (`configs/local_llm_config.json`): 使用本地部署的LLM模型
- 选题文件:`tweet_topic_{run_id}.json`
- 选题使用的提示词:`tweet_prompt_{run_id}.txt`
- 文章内容:分目录保存在`{run_id}/{topic_index}_{variant_index}/article.json`
- 海报图像:保存在`{run_id}/{topic_index}_{variant_index}/poster/poster.jpg`
- 拼贴图像:保存在`{run_id}/{topic_index}_{variant_index}/collage_img/`
使用示例配置:
## 自定义扩展
```bash
# 复制适合您场景的配置
cp configs/social_media_config.json poster_gen_config.json
系统各部分设计为模块化,您可以:
# 按需修改配置
vim poster_gen_config.json
```
1. 添加新的风格提示词到`genPrompts/Style/`目录
2. 添加新的需求提示词到`genPrompts/Demand/`目录
3. 添加新的景点资源到`resource/Object/`目录
4. 修改AI模型参数以适应不同生成需求
5. 自定义海报生成的尺寸和样式
详细说明请参阅 `configs/README.md` 文件。
## 注意事项
## 高级使用指南
- 确保提示词文件和资源文件的编码为UTF-8
- API密钥应妥善保管建议使用环境变量或外部配置
- 图像生成需要足够的系统资源,建议在性能良好的设备上运行
- 文件模糊匹配功能可以处理一些文件名不完全匹配的情况,但建议尽量保持文件名规范
### 自定义提示词
## 贡献
编辑 `SelectPrompt/``genPrompts/` 目录下的提示词文件,可自定义生成内容的风格和侧重点。
### 调整生成参数
- 增加 `variants` 值可获得更多内容变体
- 调整 `temperature` 参数可以改变生成内容的创造性
- 修改 `poster_target_size` 可以设置不同的海报尺寸
### 分布式执行
利用分阶段执行功能,可在不同机器上完成选题生成和内容生成:
1. 机器A执行选题生成 (`run_step1_topics.py`),将结果保存到共享存储
2. 机器B从共享存储读取选题 (`run_step2_content_posters.py <run_id>`),执行计算密集的内容和海报生成
## 常见问题
1. **生成内容质量不高?**
- 尝试提供更详细的景点信息
- 调整提示词模板
- 降低 `temperature` 参数以减少随机性
2. **找不到景点图片?**
- 确保图片目录名与景点信息匹配
- 检查配置文件中的 `image_base_dir` 路径是否正确
3. **API调用失败**
- 验证 API Key 和 URL 是否正确
- 检查网络连接和防火墙设置
## 示例
查看 `examples/` 目录中的示例脚本及其 `README.md` 文件,了解更多使用方法。
## 贡献指南
欢迎提交 Pull Request 或 Issue 来帮助改进本项目。
## 许可证
本项目采用 MIT 许可证。
欢迎提交问题报告和功能建议,或直接提交代码改进。

View File

@ -1,7 +1,7 @@
from .ai_agent import AI_Agent
from .topic_parser import TopicParser
from .contentGen import ContentGenerator
from .posterGen import PosterGenerator
from .poster_gen import PosterGenerator
from .simple_collage import process_directory
__all__ = ['AI_Agent', 'TopicParser', 'ContentGenerator', 'PosterGenerator', 'process_directory']

Binary file not shown.

View File

@ -21,7 +21,7 @@ if PROJECT_ROOT not in sys.path:
# 导入所需的图像处理模块
try:
import core.simple_collage as simple_collage
import core.posterGen as posterGen
import core.poster_gen as poster_gen
from utils.resource_loader import ResourceLoader
except ImportError as e:
logging.critical(f"导入模块失败: {e}")
@ -188,7 +188,7 @@ def test_poster_generation(config, output_dir, collage_dir=None):
# 创建海报生成器
try:
poster_generator = posterGen.PosterGenerator(
poster_generator = poster_gen.PosterGenerator(
poster_save_dir=poster_output_dir,
assets_base_dir=poster_assets_dir,
poster_size=tuple(config.get("poster_target_size", [900, 1200]))

View File

@ -21,7 +21,7 @@ project_root = str(Path(__file__).parent.parent.absolute())
if project_root not in sys.path:
sys.path.append(project_root)
from core import posterGen
from core import poster_gen
def parse_arguments():
"""解析命令行参数"""
@ -126,7 +126,7 @@ def main():
# 初始化PosterGenerator
print("初始化PosterGenerator...")
poster_generator = posterGen.PosterGenerator(base_dir=base_dir)
poster_generator = poster_gen.PosterGenerator(base_dir=base_dir)
# 确定底图路径
image_path = args.image

View File

@ -10,7 +10,7 @@ import logging
from core.ai_agent import AI_Agent
# from core.topic_parser import TopicParser # No longer needed directly in main?
import core.contentGen as contentGen
import core.posterGen as posterGen
import core.poster_gen as poster_gen
import core.simple_collage as simple_collage
from utils.resource_loader import ResourceLoader
from utils.tweet_generator import ( # Import the moved functions

45
poster_config.json Normal file
View File

@ -0,0 +1,45 @@
{
"variants": 1,
"model": "qwenQWQ",
"api_url": "http://localhost:8000/v1/",
"api_key": "EMPTY",
"poster_content_system_prompt": "./genPrompts/poster_content_systemPrompt.txt",
"resource_dir": [
{
"type": "Object",
"file_path": [
"./resource/Object/中山温泉宾馆.txt",
"./resource/Object/乌镇民宿.txt",
"./resource/Object/从化客天下·禧悦庄.txt"
]
},
{
"type": "Description",
"file_path": [
"./resource/Object/中山温泉宾馆.txt",
"./resource/Object/乌镇民宿.txt",
"./resource/Object/从化客天下·禧悦庄.txt"
]
},
{
"type": "Product",
"file_path": [
]
}
],
"output_dir": "./result",
"image_base_dir": "/root/autodl-tmp/TravelContentCreator/hotel_img",
"poster_assets_base_dir": "/root/autodl-tmp/poster_baseboard_0403",
"request_timeout": 120,
"max_retries": 3,
"output_collage_subdir": "collage_img",
"output_poster_subdir": "poster",
"output_poster_filename": "poster.jpg",
"poster_target_size": [
900,
1200
],
"text_possibility": 0.3,
"img_frame_possibility": 0.7,
"text_bg_possibility": 0
}

232
test_poster.py Normal file
View File

@ -0,0 +1,232 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试海报生成的脚本
"""
import os
import time
import argparse
import json
import logging
import sys
import traceback
from datetime import datetime
from core.ai_agent import AI_Agent
from utils.tweet_generator import generate_posters_for_topic
from utils.output_handler import FileSystemOutputHandler
from core.topic_parser import TopicParser
def load_config(config_path="poster_config.json"):
"""从JSON文件加载配置"""
if not os.path.exists(config_path):
print(f"错误:配置文件 '{config_path}' 未找到。")
sys.exit(1)
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
# 基本验证
required_keys = ["api_url", "model", "api_key", "resource_dir", "output_dir",
"image_base_dir", "poster_assets_base_dir",
"poster_content_system_prompt"]
if not all(key in config for key in required_keys):
missing_keys = [key for key in required_keys if key not in config]
print(f"错误:配置文件 '{config_path}' 缺少必需的键:{missing_keys}")
sys.exit(1)
return config
except json.JSONDecodeError:
print(f"错误:无法从 '{config_path}' 解码JSON。请检查文件格式。")
sys.exit(1)
except Exception as e:
print(f"'{config_path}' 加载配置时出错:{e}")
sys.exit(1)
def main():
# 设置日志记录
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 解析命令行参数
parser = argparse.ArgumentParser(description="测试海报生成")
parser.add_argument(
"--config",
type=str,
default="poster_config.json",
help="配置文件路径例如poster_config.json"
)
parser.add_argument(
"--topics_file",
type=str,
required=True,
help="必需的选题JSON文件路径用于获取海报生成的主题数据。"
)
parser.add_argument(
"--topic_index",
type=int,
default=None,
help="要生成海报的特定选题索引。如果未提供,将为所有选题生成海报。"
)
parser.add_argument(
"--run_id",
type=str,
default=None,
help="可选的指定运行ID。如果未提供将生成基于时间戳的ID。"
)
parser.add_argument(
"--debug",
action='store_true',
help="启用调试级别日志记录。"
)
args = parser.parse_args()
# 调整日志级别(如果启用了调试)
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
logging.info("已启用调试日志记录。")
logging.info("启动海报生成测试脚本...")
logging.info(f"使用配置文件:{args.config}")
logging.info(f"使用选题文件:{args.topics_file}")
if args.topic_index is not None:
logging.info(f"将仅处理选题索引:{args.topic_index}")
if args.run_id:
logging.info(f"使用指定的run_id{args.run_id}")
# 加载配置
config = load_config(args.config)
if config is None:
logging.critical("无法加载配置。退出。")
sys.exit(1)
# 初始化输出处理器
output_handler = FileSystemOutputHandler(config.get("output_dir", "result"))
logging.info(f"使用输出处理器:{output_handler.__class__.__name__}")
# 加载选题数据
logging.info(f"从以下位置加载选题:{args.topics_file}")
topics_list = TopicParser.load_topics_from_json(args.topics_file)
if not topics_list:
logging.error(f"无法从{args.topics_file}加载选题。无法继续。")
sys.exit(1)
logging.info(f"成功加载{len(topics_list)}个选题。")
# 设置run_id
run_id = args.run_id
if run_id is None:
# 尝试从文件名推断run_id
try:
base = os.path.basename(args.topics_file)
if base.startswith("tweet_topic_") and base.endswith(".json"):
run_id = base[len("tweet_topic_"): -len(".json")]
logging.info(f"从选题文件名推断的run_id{run_id}")
else:
run_id = datetime.now().strftime("%Y-%m-%d_%H-%M-%S_poster")
logging.info(f"为海报生成的run_id{run_id}")
except Exception as e:
run_id = datetime.now().strftime("%Y-%m-%d_%H-%M-%S_poster")
logging.info(f"生成的run_id{run_id}")
# 加载海报内容系统提示词
poster_content_system_prompt_path = config.get("poster_content_system_prompt")
if not os.path.exists(poster_content_system_prompt_path):
logging.error(f"海报内容系统提示词文件不存在:{poster_content_system_prompt_path}")
sys.exit(1)
with open(poster_content_system_prompt_path, "r", encoding="utf-8") as f:
poster_content_system_prompt = f.read()
# 准备海报生成参数
poster_variants = config.get("variants", 1)
poster_assets_dir = config.get("poster_assets_base_dir")
img_base_dir = config.get("image_base_dir")
res_dir_config = config.get("resource_dir", [])
poster_size = tuple(config.get("poster_target_size", [900, 1200]))
txt_possibility = config.get("text_possibility", 0.3)
img_frame_possibility = config.get("img_frame_possibility", 0.7)
text_bg_possibility = config.get("text_bg_possibility", 0)
collage_subdir = config.get("output_collage_subdir", "collage_img")
poster_subdir = config.get("output_poster_subdir", "poster")
poster_filename = config.get("output_poster_filename", "poster.jpg")
# 检查关键路径
if not poster_assets_dir or not img_base_dir:
logging.error("配置中缺少关键路径poster_assets_base_dir或image_base_dir。无法继续。")
sys.exit(1)
# 开始海报生成
pipeline_start_time = time.time()
logging.info("开始执行海报生成...")
poster_success = False
# 如果指定了topic_index只处理该选题
if args.topic_index is not None:
topics_to_process = []
for topic in topics_list:
if topic.get('index') == args.topic_index or (topic.get('index') is None and int(args.topic_index) == 1):
topics_to_process.append(topic)
break
if not topics_to_process:
logging.error(f"未找到索引为{args.topic_index}的选题。")
sys.exit(1)
else:
topics_to_process = topics_list
# 逐个处理选题
for i, topic_item in enumerate(topics_to_process):
topic_index = topic_item.get('index', i + 1)
logging.info(f"--- 处理选题 {topic_index}: {topic_item.get('object', 'N/A')} ---")
try:
posters_attempted = generate_posters_for_topic(
topic_item=topic_item,
output_dir=config["output_dir"],
run_id=run_id,
topic_index=topic_index,
output_handler=output_handler,
variants=poster_variants,
poster_assets_base_dir=poster_assets_dir,
image_base_dir=img_base_dir,
resource_dir_config=res_dir_config,
poster_target_size=poster_size,
text_possibility=txt_possibility,
img_frame_possibility=img_frame_possibility,
text_bg_possibility=text_bg_possibility,
output_collage_subdir=collage_subdir,
output_poster_subdir=poster_subdir,
output_poster_filename=poster_filename,
system_prompt=poster_content_system_prompt
)
if posters_attempted:
logging.info(f"选题{topic_index}的海报生成过程已完成。")
poster_success = True
else:
logging.warning(f"选题{topic_index}的海报生成被跳过或在早期失败。")
except Exception as e:
logging.exception(f"处理选题{topic_index}的海报生成时出错:")
logging.info(f"--- 完成选题 {topic_index} ---")
# 最终化输出
if run_id:
output_handler.finalize(run_id)
pipeline_end_time = time.time()
if poster_success:
logging.info(f"海报生成完成,耗时{pipeline_end_time - pipeline_start_time:.2f}秒。")
else:
logging.warning("海报生成完成,但可能遇到错误或未生成输出。")
logging.info(f"运行ID'{run_id}'的结果位于:{os.path.join(config.get('output_dir', 'result'), run_id)}")
if __name__ == "__main__":
main()

261
test_topic_content.py Normal file
View File

@ -0,0 +1,261 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试选题生成和文章生成的脚本
"""
import os
import time
import argparse
import json
import logging
import sys
from datetime import datetime
from core.ai_agent import AI_Agent
from utils.prompt_manager import PromptManager
from utils.tweet_generator import run_topic_generation_pipeline, generate_content_for_topic
from utils.output_handler import FileSystemOutputHandler
def load_config(config_path="topic_content_config.json"):
"""从JSON文件加载配置"""
if not os.path.exists(config_path):
print(f"错误:配置文件 '{config_path}' 未找到。")
sys.exit(1)
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
# 基本验证
required_keys = ["api_url", "model", "api_key", "resource_dir", "output_dir",
"num", "variants", "topic_system_prompt", "topic_user_prompt",
"content_system_prompt"]
if not all(key in config for key in required_keys):
missing_keys = [key for key in required_keys if key not in config]
print(f"错误:配置文件 '{config_path}' 缺少必需的键:{missing_keys}")
sys.exit(1)
# 验证prompts_dir或prompts_config至少有一个存在
if not ("prompts_dir" in config or "prompts_config" in config):
print(f"错误:配置文件 '{config_path}' 必须包含 'prompts_dir''prompts_config'")
sys.exit(1)
return config
except json.JSONDecodeError:
print(f"错误:无法从 '{config_path}' 解码JSON。请检查文件格式。")
sys.exit(1)
except Exception as e:
print(f"'{config_path}' 加载配置时出错:{e}")
sys.exit(1)
def main():
# 设置日志记录
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 解析命令行参数
parser = argparse.ArgumentParser(description="测试选题和文章生成")
parser.add_argument(
"--config",
type=str,
default="topic_content_config.json",
help="配置文件路径例如topic_content_config.json"
)
parser.add_argument(
"--run_id",
type=str,
default=None,
help="可选的指定运行ID例如'test_run_01'。如果未提供将生成基于时间戳的ID。"
)
parser.add_argument(
"--topics_file",
type=str,
default=None,
help="可选的预生成选题JSON文件路径。如果提供则跳过选题生成。"
)
parser.add_argument(
"--debug",
action='store_true',
help="启用调试级别日志记录。"
)
args = parser.parse_args()
# 调整日志级别(如果启用了调试)
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
logging.info("已启用调试日志记录。")
logging.info("启动选题和文章生成测试脚本...")
logging.info(f"使用配置文件:{args.config}")
if args.run_id:
logging.info(f"使用指定的run_id{args.run_id}")
if args.topics_file:
logging.info(f"使用现有选题文件:{args.topics_file}")
# 加载配置
config = load_config(args.config)
if config is None:
logging.critical("无法加载配置。退出。")
sys.exit(1)
# 初始化输出处理器
output_handler = FileSystemOutputHandler(config.get("output_dir", "result"))
logging.info(f"使用输出处理器:{output_handler.__class__.__name__}")
run_id = args.run_id
topics_list = None
system_prompt = None
user_prompt = None
pipeline_start_time = time.time()
# 步骤1选题生成或加载现有选题
if args.topics_file:
from core.topic_parser import TopicParser
logging.info(f"跳过选题生成步骤1- 从以下位置加载选题:{args.topics_file}")
topics_list = TopicParser.load_topics_from_json(args.topics_file)
if topics_list:
# 如果未提供run_id尝试从文件名推断
if not run_id:
try:
base = os.path.basename(args.topics_file)
# 假设格式为"tweet_topic_{run_id}.json"或"tweet_topic.json"
if base.startswith("tweet_topic_") and base.endswith(".json"):
run_id = base[len("tweet_topic_"): -len(".json")]
logging.info(f"从选题文件名推断的run_id{run_id}")
elif base == "tweet_topic.json":
logging.warning(f"从默认文件名'{base}'加载选题。未推断run_id。")
else:
logging.warning(f"无法从选题文件名推断run_id{base}")
except Exception as e:
logging.warning(f"尝试从选题文件名推断run_id时出错{e}")
# 如果尝试推断后run_id仍为None则生成一个
if run_id is None:
run_id = datetime.now().strftime("%Y-%m-%d_%H-%M-%S_loaded")
logging.info(f"为加载的选题生成的run_id{run_id}")
# 加载文件时缺少提示词
system_prompt = ""
user_prompt = ""
logging.info(f"成功加载{len(topics_list)}个选题run_id{run_id}。提示词不可用。")
else:
logging.error(f"无法从{args.topics_file}加载选题。无法继续。")
sys.exit(1)
else:
logging.info("执行选题生成步骤1...")
step1_start = time.time()
# 调用更新后的函数,接收原始数据
run_id, topics_list, system_prompt, user_prompt = run_topic_generation_pipeline(config, args.run_id)
step1_end = time.time()
if run_id is not None and topics_list is not None:
logging.info(f"步骤1成功完成耗时{step1_end - step1_start:.2f}秒。运行ID{run_id}")
# 使用输出处理器保存结果
output_handler.handle_topic_results(run_id, topics_list, system_prompt, user_prompt)
else:
logging.critical("选题生成步骤1失败。退出。")
sys.exit(1)
# 步骤2内容生成
if run_id is not None and topics_list is not None:
logging.info("执行内容生成步骤2...")
step2_start = time.time()
# 创建PromptManager实例
try:
prompt_manager = PromptManager(
topic_system_prompt_path=config.get("topic_system_prompt"),
topic_user_prompt_path=config.get("topic_user_prompt"),
content_system_prompt_path=config.get("content_system_prompt"),
prompts_config=config.get("prompts_config"),
prompts_dir=config.get("prompts_dir"),
resource_dir_config=config.get("resource_dir", []),
topic_gen_num=config.get("num", 1),
topic_gen_date=config.get("date", "")
)
logging.info("已为步骤2创建PromptManager实例。")
except KeyError as e:
logging.error(f"创建PromptManager时配置错误缺少键'{e}'。无法继续内容生成。")
sys.exit(1)
# 初始化AI Agent
ai_agent = None
content_success = False
try:
request_timeout = config.get("request_timeout", 30)
max_retries = config.get("max_retries", 3)
stream_chunk_timeout = config.get("stream_chunk_timeout", 60)
ai_agent = AI_Agent(
config["api_url"],
config["model"],
config["api_key"],
timeout=request_timeout,
max_retries=max_retries,
stream_chunk_timeout=stream_chunk_timeout
)
logging.info("已初始化用于内容生成的AI Agent。")
# 遍历选题
for i, topic_item in enumerate(topics_list):
topic_index = topic_item.get('index', i + 1)
logging.info(f"--- 处理选题 {topic_index}/{len(topics_list)}: {topic_item.get('object', 'N/A')} ---")
# 读取内容生成所需的参数
content_variants = config.get("variants", 1)
content_temp = config.get("content_temperature", 0.3)
content_top_p = config.get("content_top_p", 0.4)
content_presence_penalty = config.get("content_presence_penalty", 1.5)
# 调用generate_content_for_topic
topic_success = generate_content_for_topic(
ai_agent,
prompt_manager,
topic_item,
run_id,
topic_index,
output_handler,
variants=content_variants,
temperature=content_temp,
top_p=content_top_p,
presence_penalty=content_presence_penalty
)
if topic_success:
logging.info(f"选题{topic_index}的内容生成成功。")
content_success = True
else:
logging.warning(f"选题{topic_index}的内容生成失败或未产生有效结果。")
logging.info(f"--- 完成选题 {topic_index} ---")
except KeyError as e:
logging.error(f"内容生成过程中的配置错误:缺少键'{e}'")
traceback.print_exc()
except Exception as e:
logging.exception("内容生成过程中发生意外错误:")
finally:
# 确保AI agent已关闭
if ai_agent:
logging.info("关闭内容生成AI Agent...")
ai_agent.close()
step2_end = time.time()
if content_success:
logging.info(f"步骤2完成耗时{step2_end - step2_start:.2f}秒。")
else:
logging.warning("步骤2完成但可能遇到错误或未生成输出。")
else:
logging.error("无法进行步骤2步骤1的run_id或topics_list无效。")
# 最终化输出
if run_id:
output_handler.finalize(run_id)
pipeline_end_time = time.time()
logging.info(f"流程完成。总执行时间:{pipeline_end_time - pipeline_start_time:.2f}秒。")
logging.info(f"运行ID'{run_id}'的结果位于:{os.path.join(config.get('output_dir', 'result'), run_id)}")
if __name__ == "__main__":
main()

72
topic_content_config.json Normal file
View File

@ -0,0 +1,72 @@
{
"date": "5月15日, 5月16日, 5月17日, 6月1日",
"num": 2,
"variants": 1,
"topic_temperature": 0.2,
"topic_top_p": 0.3,
"topic_presence_penalty": 1.5,
"content_temperature": 0.3,
"content_top_p": 0.4,
"content_presence_penalty": 1.5,
"model": "qwenQWQ",
"api_url": "http://localhost:8000/v1/",
"api_key": "EMPTY",
"topic_system_prompt": "./SelectPrompt/systemPrompt.txt",
"topic_user_prompt": "./SelectPrompt/userPrompt.txt",
"content_system_prompt": "./genPrompts/systemPrompt.txt",
"prompts_config": [
{
"type": "Style",
"file_path": [
"./genPrompts/Style/攻略风文案提示词.txt",
"./genPrompts/Style/轻奢风文案提示词.txt",
"./genPrompts/Style/极力推荐风文案提示词.txt",
"./genPrompts/Style/美食风文案提示词.txt"
]
},
{
"type": "Demand",
"file_path": [
"./genPrompts/Demand/学生党文旅需求.txt",
"./genPrompts/Demand/情侣向文旅需求.txt",
"./genPrompts/Demand/职场人文旅需求.txt",
"./genPrompts/Demand/亲子向文旅需求.txt",
"./genPrompts/Demand/周边游文旅需求.txt",
"./genPrompts/Demand/夕阳红文旅需求.txt"
]
},
{
"type": "Refer",
"file_path": [
"./genPrompts/Refer/标题参考格式.txt"
]
}
],
"resource_dir": [
{
"type": "Object",
"file_path": [
"./resource/Object/中山温泉宾馆.txt",
"./resource/Object/乌镇民宿.txt",
"./resource/Object/从化客天下·禧悦庄.txt"
]
},
{
"type": "Description",
"file_path": [
"./resource/Object/中山温泉宾馆.txt",
"./resource/Object/乌镇民宿.txt",
"./resource/Object/从化客天下·禧悦庄.txt"
]
},
{
"type": "Product",
"file_path": [
]
}
],
"output_dir": "./result",
"request_timeout": 120,
"max_retries": 3,
"stream_chunk_timeout": 60
}

View File

@ -24,7 +24,7 @@ from core.ai_agent import AI_Agent
from core.topic_parser import TopicParser
from utils.prompt_manager import PromptManager # Keep this as it's importing from the same level package 'utils'
from core import contentGen as core_contentGen
from core import posterGen as core_posterGen
from core import poster_gen as core_posterGen
from core import simple_collage as core_simple_collage
from .output_handler import OutputHandler # <-- 添加导入