Ubuntu
332 lines
10 KiB
Python
332 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
手动完整流程测试 - 带Moonshot处理
|
|
从数据库选择一篇文章，执行完整流程（包含Moonshot翻译/总结）
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import sqlite3
|
|
import datetime
|
|
import logging
|
|
import requests
|
|
from pathlib import Path
|
|
|
|
# 配置日志
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def load_config():
|
|
"""加载配置"""
|
|
config_path = "/home/ubuntu/nanocode/it_automation_enhanced/config/config.json"
|
|
try:
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
logger.error(f"配置加载失败: {e}")
|
|
return {}
|
|
|
|
def call_moonshot_api(prompt: str, api_key: str) -> str:
|
|
"""调用Moonshot API"""
|
|
try:
|
|
headers = {
|
|
'Authorization': f'Bearer {api_key}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
data = {
|
|
'model': 'kimi-k2.5',
|
|
'messages': [
|
|
{'role': 'user', 'content': prompt}
|
|
],
|
|
'max_tokens': 2000
|
|
}
|
|
|
|
logger.info("调用Moonshot API...")
|
|
response = requests.post(
|
|
'https://api.moonshot.cn/v1/chat/completions',
|
|
headers=headers,
|
|
json=data,
|
|
timeout=60
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
content = result['choices'][0]['message']['content']
|
|
logger.info(f"Moonshot API调用成功，返回 {len(content)} 字符")
|
|
return content
|
|
else:
|
|
logger.error(f"Moonshot API调用失败: {response.status_code}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Moonshot API调用异常: {e}")
|
|
return None
|
|
|
|
def translate_and_summarize(title: str, url: str, source: str, api_key: str) -> dict:
|
|
"""翻译和总结文章"""
|
|
|
|
# 构建翻译提示
|
|
translate_prompt = f"""请将以下英文技术文章标题翻译成中文，并基于标题生成一篇详细的中文技术文章摘要：
|
|
|
|
标题: {title}
|
|
来源: {source}
|
|
原文链接: {url}
|
|
|
|
请提供：
|
|
1. 中文标题
|
|
2. 文章摘要（300-500字）
|
|
3. 关键要点（3-5条）
|
|
4. 技术亮点分析
|
|
|
|
格式要求：
|
|
【中文标题】
|
|
（翻译后的标题）
|
|
|
|
【文章摘要】
|
|
（详细的中文摘要内容）
|
|
|
|
【关键要点】
|
|
• 要点1
|
|
• 要点2
|
|
• 要点3
|
|
|
|
【技术亮点】
|
|
（技术分析）"""
|
|
|
|
result = call_moonshot_api(translate_prompt, api_key)
|
|
|
|
if result:
|
|
# 解析结果
|
|
lines = result.split('\n')
|
|
chinese_title = ""
|
|
summary = ""
|
|
key_points = []
|
|
tech_highlights = ""
|
|
|
|
current_section = ""
|
|
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
if '【中文标题】' in line:
|
|
current_section = "title"
|
|
elif '【文章摘要】' in line:
|
|
current_section = "summary"
|
|
elif '【关键要点】' in line:
|
|
current_section = "points"
|
|
elif '【技术亮点】' in line:
|
|
current_section = "highlights"
|
|
elif current_section == "title" and line and '【' not in line:
|
|
chinese_title = line
|
|
elif current_section == "summary" and line and '【' not in line:
|
|
summary += line + "\n"
|
|
elif current_section == "points" and line.startswith('•'):
|
|
key_points.append(line[1:].strip())
|
|
elif current_section == "highlights" and line and '【' not in line:
|
|
tech_highlights += line + "\n"
|
|
|
|
return {
|
|
'chinese_title': chinese_title or title,
|
|
'summary': summary.strip() or f"来自{source}的技术文章",
|
|
'key_points': key_points or ["技术发展", "行业趋势", "创新应用"],
|
|
'tech_highlights': tech_highlights.strip() or "技术亮点分析"
|
|
}
|
|
else:
|
|
# 返回默认内容
|
|
return {
|
|
'chinese_title': title,
|
|
'summary': f"这是一篇来自{source}的技术文章，讨论了当前技术领域的最新发展。",
|
|
'key_points': ["技术发展趋势", "行业应用案例", "未来发展展望"],
|
|
'tech_highlights': "文章提供了有价值的技术洞察。"
|
|
}
|
|
|
|
def select_article_from_db():
|
|
"""从数据库选择一篇文章"""
|
|
db_path = "/home/ubuntu/nanocode/it_automation_enhanced/data/articles.db"
|
|
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
# 选择一篇未处理的文章
|
|
cursor.execute('''
|
|
SELECT id, title, url, source, category, publish_date, word_count
|
|
FROM articles
|
|
WHERE processed = 0
|
|
ORDER BY word_count DESC
|
|
LIMIT 1
|
|
''')
|
|
|
|
article = cursor.fetchone()
|
|
conn.close()
|
|
|
|
if article:
|
|
article_data = {
|
|
'id': article[0],
|
|
'title': article[1],
|
|
'url': article[2],
|
|
'source': article[3],
|
|
'category': article[4],
|
|
'publish_date': article[5],
|
|
'word_count': article[6]
|
|
}
|
|
logger.info(f"选择文章: {article_data['title'][:50]}...")
|
|
return article_data
|
|
else:
|
|
logger.warning("没有未处理的文章")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"数据库查询失败: {e}")
|
|
return None
|
|
|
|
def run_xiaohongshu_with_moonshot(article_data, processed_content):
|
|
"""运行小红书测试（带Moonshot处理）"""
|
|
logger.info("运行小红书多图测试（带Moonshot处理）...")
|
|
|
|
try:
|
|
sys.path.append("/home/ubuntu/nanocode/it_automation_enhanced/scripts")
|
|
from xiaohongshu_multi_image import XiaohongshuMultiImagePublisher
|
|
|
|
publisher = XiaohongshuMultiImagePublisher()
|
|
|
|
# 使用Moonshot处理后的内容创建小红书内容
|
|
xhs_content = f"""{processed_content['summary']}
|
|
|
|
关键要点：
|
|
"""
|
|
|
|
# 添加关键要点
|
|
for point in processed_content['key_points'][:5]:
|
|
xhs_content += f"• {point}\n"
|
|
|
|
xhs_content += f"""
|
|
技术亮点：
|
|
{processed_content['tech_highlights'][:200]}
|
|
|
|
#技术 #{article_data['source']} #行业分析 #科技资讯"""
|
|
|
|
# 根据文章类别选择风格
|
|
if "tech" in article_data.get('category', '').lower() or "tech" in article_data.get('source', '').lower():
|
|
style = "tech"
|
|
else:
|
|
style = "warm"
|
|
|
|
# 使用中文标题
|
|
chinese_title = processed_content['chinese_title'][:50]
|
|
|
|
logger.info(f"小红书发布 - 中文标题: {chinese_title}")
|
|
logger.info(f"小红书发布 - 风格: {style}")
|
|
logger.info(f"小红书发布 - 内容长度: {len(xhs_content)} 字符")
|
|
|
|
# 发布到小红书
|
|
tags = ["技术", article_data['source'], "行业分析", "科技资讯", "自动化"]
|
|
success = publisher.publish_to_xiaohongshu(
|
|
title=chinese_title,
|
|
content=xhs_content,
|
|
tags=tags,
|
|
style=style
|
|
)
|
|
|
|
if success:
|
|
logger.info("✅ 小红书多图发布成功！")
|
|
return True
|
|
else:
|
|
logger.warning("小红书发布失败（可能是Cookie问题）")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"小红书测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("=" * 60)
|
|
print("🚀 手动完整流程测试 - 带Moonshot处理")
|
|
print("=" * 60)
|
|
print(f"时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print("")
|
|
|
|
# 加载配置
|
|
config = load_config()
|
|
api_key = config.get('moonshot', {}).get('api_key')
|
|
|
|
if not api_key:
|
|
print("❌ Moonshot API密钥未配置")
|
|
return 1
|
|
|
|
print(f"✅ Moonshot API已配置")
|
|
print("")
|
|
|
|
# 1. 选择文章
|
|
print("1️⃣ 选择文章...")
|
|
article_data = select_article_from_db()
|
|
if not article_data:
|
|
print("❌ 没有可用的文章，退出")
|
|
return 1
|
|
|
|
print(f" 标题: {article_data['title']}")
|
|
print(f" 来源: {article_data['source']}")
|
|
print(f" URL: {article_data['url']}")
|
|
print("")
|
|
|
|
# 2. Moonshot处理
|
|
print("2️⃣ Moonshot翻译和总结...")
|
|
processed_content = translate_and_summarize(
|
|
title=article_data['title'],
|
|
url=article_data['url'],
|
|
source=article_data['source'],
|
|
api_key=api_key
|
|
)
|
|
|
|
print(f" 中文标题: {processed_content['chinese_title'][:50]}...")
|
|
print(f" 摘要长度: {len(processed_content['summary'])} 字符")
|
|
print(f" 关键要点: {len(processed_content['key_points'])} 条")
|
|
print("")
|
|
|
|
# 3. 小红书发布
|
|
print("3️⃣ 小红书多图发布（带Moonshot内容）...")
|
|
xhs_success = run_xiaohongshu_with_moonshot(article_data, processed_content)
|
|
print("")
|
|
|
|
# 4. 更新状态
|
|
print("4️⃣ 更新文章状态...")
|
|
try:
|
|
conn = sqlite3.connect("/home/ubuntu/nanocode/it_automation_enhanced/data/articles.db")
|
|
cursor = conn.cursor()
|
|
cursor.execute('UPDATE articles SET processed = 1, xhs_status = 1 WHERE id = ?', (article_data['id'],))
|
|
conn.commit()
|
|
conn.close()
|
|
print(" ✅ 文章状态更新完成")
|
|
except Exception as e:
|
|
print(f" ⚠️ 状态更新失败: {e}")
|
|
print("")
|
|
|
|
# 5. 结果总结
|
|
print("=" * 60)
|
|
print("📋 测试结果总结")
|
|
print("=" * 60)
|
|
print(f" 文章: {article_data['title'][:50]}...")
|
|
print(f" 中文标题: {processed_content['chinese_title'][:50]}...")
|
|
print(f" 小红书发布: {'✅ 成功' if xhs_success else '❌ 失败'}")
|
|
print("")
|
|
|
|
if xhs_success:
|
|
print("🎉 完整流程测试成功！")
|
|
print("✅ 包含Moonshot翻译和总结")
|
|
return 0
|
|
else:
|
|
print("⚠️ 小红书发布失败")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |