Ubuntu
159 lines
5.1 KiB
Python
159 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
小红书技术博客工作流
|
|
从国际技术博客抓取文章，改写后发布
|
|
|
|
数据来源：/home/ubuntu/.nanobot/workspace/skills/tech-blog-sources/SKILL.md
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import argparse
|
|
from datetime import datetime
|
|
from typing import Optional, Dict
|
|
|
|
# 导入模块
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
from tech_blog_fetcher import TechBlogFetcher
|
|
from tech_content_rewrite import TechContentRewriter
|
|
from publisher_adapter import XHSPublisherAdapter
|
|
|
|
|
|
class TechBlogWorkflow:
|
|
"""
|
|
技术博客工作流
|
|
|
|
完整流程：
|
|
1. 从15个可用技术博客源随机选择
|
|
2. 抓取最近3天内、字数<2000的文章
|
|
3. 使用大模型改写成小红书风格
|
|
4. 发布到小红书
|
|
"""
|
|
|
|
def __init__(self, config_path: str = None):
|
|
self.config = self._load_config(config_path)
|
|
|
|
# 初始化各模块
|
|
self.fetcher = TechBlogFetcher()
|
|
self.rewriter = TechContentRewriter()
|
|
self.publisher = XHSPublisherAdapter()
|
|
|
|
# 路径
|
|
self.log_dir = self.config.get('paths', {}).get('log_dir', './logs')
|
|
self.output_dir = self.config.get('paths', {}).get('output_dir', './output')
|
|
os.makedirs(self.log_dir, exist_ok=True)
|
|
os.makedirs(self.output_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': {'content_style': 'tech'},
|
|
'paths': {'log_dir': './logs', 'output_dir': './output'}
|
|
}
|
|
|
|
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"tech_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, dry_run: bool = False) -> bool:
|
|
"""
|
|
执行技术博客工作流
|
|
|
|
Args:
|
|
dry_run: 仅生成不发布
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
self._log("="*50)
|
|
self._log("🚀 技术博客工作流启动")
|
|
self._log("="*50)
|
|
|
|
# 步骤1: 抓取文章
|
|
self._log("\n📡 步骤1: 从技术博客源抓取文章...")
|
|
article = self.fetcher.fetch_daily_articles()
|
|
|
|
if not article:
|
|
self._log("❌ 未能获取到合适的文章")
|
|
return False
|
|
|
|
self._log(f"✅ 获取文章: {article['title'][:50]}...")
|
|
self._log(f" 来源: {article['source']}")
|
|
self._log(f" 原文链接: {article['url']}")
|
|
self._log(f" 字数: {len(article['content'])}")
|
|
|
|
# 步骤2: 改写内容
|
|
self._log("\n🤖 步骤2: 使用大模型改写成小红书风格...")
|
|
content = self.rewriter.rewrite(article)
|
|
|
|
self._log(f"✅ 改写完成")
|
|
self._log(f" 新标题: {content['title']}")
|
|
self._log(f" 正文字数: {len(content['body'])}")
|
|
self._log(f" 标签: {', '.join(content['tags'])}")
|
|
|
|
# 步骤3: 保存内容
|
|
output_file = os.path.join(
|
|
self.output_dir,
|
|
f"tech_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"\n💾 内容已保存: {output_file}")
|
|
|
|
# 步骤4: 发布
|
|
if dry_run:
|
|
self._log("\n🧪 试运行模式，跳过发布")
|
|
self._log("="*50)
|
|
return True
|
|
|
|
self._log("\n📱 步骤3: 发布到小红书...")
|
|
success = self.publisher.publish(
|
|
title=content['title'],
|
|
content=content['body'],
|
|
tags=content['tags'],
|
|
style=self.config.get('workflow', {}).get('content_style', 'tech')
|
|
)
|
|
|
|
if success:
|
|
self._log("🎉 发布成功！")
|
|
else:
|
|
self._log("❌ 发布失败")
|
|
|
|
self._log("="*50)
|
|
return success
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='小红书技术博客工作流')
|
|
parser.add_argument('--dry-run', action='store_true', help='试运行不发布')
|
|
parser.add_argument('--config', help='配置文件路径')
|
|
args = parser.parse_args()
|
|
|
|
workflow = TechBlogWorkflow(config_path=args.config)
|
|
success = workflow.run(dry_run=args.dry_run)
|
|
|
|
sys.exit(0 if success else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|