Ubuntu
190 lines
5.9 KiB
Python
190 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
小红书热点趋势工作流主模块
|
|
整合热点发现、AI创作、发布全流程
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import argparse
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
# 导入子模块
|
|
from topic_discovery import TopicDiscovery
|
|
from content_generator import ContentGenerator
|
|
from publisher_adapter import XHSPublisherAdapter
|
|
|
|
class XHSTrendingWorkflow:
|
|
"""
|
|
小红书热点趋势工作流
|
|
|
|
完整流程：
|
|
1. 搜索热门话题
|
|
2. AI创作原创内容
|
|
3. 去AI化处理
|
|
4. 发布到小红书
|
|
"""
|
|
|
|
def __init__(self, config_path: str = None):
|
|
self.config = self._load_config(config_path)
|
|
|
|
# 初始化各模块
|
|
self.discovery = TopicDiscovery()
|
|
self.generator = ContentGenerator()
|
|
self.publisher = XHSPublisherAdapter()
|
|
|
|
# 日志
|
|
self.log_dir = self.config.get('paths', {}).get('log_dir', './logs')
|
|
os.makedirs(self.log_dir, exist_ok=True)
|
|
|
|
def _load_config(self, config_path: str) -> dict:
|
|
"""加载配置"""
|
|
if not config_path:
|
|
config_path = os.path.join(
|
|
os.path.dirname(__file__), '..', 'config', 'config.json'
|
|
)
|
|
|
|
try:
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
print(f"配置加载失败: {e}，使用默认配置")
|
|
return {
|
|
'workflow': {
|
|
'search_keyword': 'AI科技',
|
|
'search_limit': 5,
|
|
'content_style': 'tech'
|
|
}
|
|
}
|
|
|
|
def _log(self, message: str):
|
|
"""记录日志"""
|
|
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
log_message = f"[{timestamp}] {message}"
|
|
print(log_message)
|
|
|
|
# 写入日志文件
|
|
log_file = os.path.join(
|
|
self.log_dir,
|
|
f"workflow_{datetime.now().strftime('%Y%m%d')}.log"
|
|
)
|
|
with open(log_file, 'a', encoding='utf-8') as f:
|
|
f.write(log_message + '\n')
|
|
|
|
def run(self,
|
|
keyword: str = None,
|
|
topic: str = None,
|
|
style: str = None,
|
|
dry_run: bool = False) -> bool:
|
|
"""
|
|
执行完整工作流
|
|
|
|
Args:
|
|
keyword: 搜索关键词（自动发现模式）
|
|
topic: 直接指定话题（直接创作模式）
|
|
style: 内容风格
|
|
dry_run: 仅生成不发布
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
self._log("🚀 工作流启动")
|
|
|
|
# 模式1：自动发现热点
|
|
if keyword and not topic:
|
|
self._log(f"🔍 搜索热门话题: {keyword}")
|
|
notes = self.discovery.search_notes(keyword, limit=5)
|
|
|
|
if not notes:
|
|
self._log("❌ 未找到热门笔记")
|
|
return False
|
|
|
|
# 选择第一篇
|
|
selected_note = notes[0]
|
|
self._log(f"✅ 选中笔记: {selected_note['title'][:40]}...")
|
|
|
|
# 提取话题
|
|
topic = self.discovery.extract_topic(selected_note)
|
|
self._log(f"🏷️ 提取话题: {topic}")
|
|
|
|
# 模式2：直接创作
|
|
if not topic:
|
|
topic = self.config.get('workflow', {}).get('search_keyword', 'AI科技')
|
|
self._log(f"📝 使用默认话题: {topic}")
|
|
|
|
# AI创作内容
|
|
self._log("🤖 AI创作原创内容...")
|
|
style = style or self.config.get('workflow', {}).get('content_style', 'tech')
|
|
content = self.generator.generate(topic, style=style)
|
|
|
|
self._log(f"✅ 内容生成完成")
|
|
self._log(f" 标题: {content['title']}")
|
|
self._log(f" 正文长度: {len(content['body'])} 字符")
|
|
|
|
# 去AI化处理
|
|
self._log("🎭 去AI化处理...")
|
|
content = self.generator.humanize(content)
|
|
|
|
# 保存到文件
|
|
output_file = os.path.join(
|
|
self.config.get('paths', {}).get('output_dir', './output'),
|
|
f"content_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
)
|
|
with open(output_file, 'w', encoding='utf-8') as f:
|
|
json.dump(content, f, ensure_ascii=False, indent=2)
|
|
self._log(f"💾 内容已保存: {output_file}")
|
|
|
|
# 发布
|
|
if dry_run:
|
|
self._log("🧪 试运行模式，跳过发布")
|
|
return True
|
|
|
|
self._log("📱 发布到小红书...")
|
|
success = self.publisher.publish(
|
|
title=content['title'],
|
|
content=content['body'],
|
|
tags=content['tags'],
|
|
style=style
|
|
)
|
|
|
|
if success:
|
|
self._log("🎉 发布成功！")
|
|
else:
|
|
self._log("❌ 发布失败")
|
|
|
|
return success
|
|
|
|
|
|
def main():
|
|
"""命令行入口"""
|
|
parser = argparse.ArgumentParser(description='小红书热点趋势工作流')
|
|
parser.add_argument('--keyword', '-k', help='搜索关键词（自动发现模式）')
|
|
parser.add_argument('--topic', '-t', help='直接指定话题（直接创作模式）')
|
|
parser.add_argument('--style', '-s', default='tech',
|
|
choices=['tech', 'warm', 'business'],
|
|
help='内容风格')
|
|
parser.add_argument('--dry-run', '-d', action='store_true',
|
|
help='仅生成不发布')
|
|
parser.add_argument('--config', '-c', help='配置文件路径')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# 创建工作流
|
|
workflow = XHSTrendingWorkflow(args.config)
|
|
|
|
# 执行
|
|
success = workflow.run(
|
|
keyword=args.keyword,
|
|
topic=args.topic,
|
|
style=args.style,
|
|
dry_run=args.dry_run
|
|
)
|
|
|
|
sys.exit(0 if success else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|