#!/usr/bin/env python3 """ 手动完整流程测试 从数据库选择一篇文章，执行完整流程 """ import os import sys import json import sqlite3 import datetime import logging import subprocess from pathlib import Path # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) 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 create_test_markdown(article_data): """创建测试Markdown内容""" content = f"""# {article_data['title']} **来源**: {article_data['source']} **日期**: {article_data['publish_date']} **字数**: {article_data['word_count']:,} **原文链接**: [{article_data['url']}]({article_data['url']}) --- ## 文章摘要 这是一篇来自 {article_data['source']} 的技术文章，标题为 "{article_data['title']}"。 文章讨论了当前技术领域的最新发展和趋势，对于了解行业动态非常有帮助。 ## 主要内容 1. **技术趋势分析** 2. **行业应用案例** 3. **未来发展展望** ## 关键要点 - 技术正在快速发展 - 创新应用不断涌现 - 行业面临新的机遇和挑战 > 本文由IT资讯自动化系统处理生成。 ![技术示意图](../images/tech_diagram.webp) *图：技术发展示意图* ## 总结 {article_data['title']} 提供了有价值的技术洞察，值得深入阅读和研究。 如需查看完整内容，请访问原文链接。""" return content def run_email_attachment_test(article_data, markdown_content): """运行邮件附件测试""" logger.info("运行邮件附件测试...") try: sys.path.append("/home/ubuntu/nanocode/it_automation_enhanced/scripts") from email_with_attachments import EmailWithAttachments sender = EmailWithAttachments() # 创建ZIP包 zip_path = sender.create_zip_package(article_data, markdown_content) if not zip_path: logger.error("ZIP包创建失败") return False logger.info(f"ZIP包创建成功: {zip_path}") # 发送邮件 success = sender.send_article_email(article_data, markdown_content) if success: logger.info("✅ 邮件附件发送成功！") return True else: logger.error("邮件附件发送失败") return False except Exception as e: logger.error(f"邮件附件测试失败: {e}") return False def run_xiaohongshu_test(article_data): """运行小红书测试""" logger.info("运行小红书多图测试...") try: sys.path.append("/home/ubuntu/nanocode/it_automation_enhanced/scripts") from xiaohongshu_multi_image import XiaohongshuMultiImagePublisher publisher = XiaohongshuMultiImagePublisher() # 创建小红书内容（去掉原文链接） xhs_content = f"""文章来自 {article_data['source']}，发布于 {article_data['publish_date']}。 这是一篇关于技术发展的深度文章，标题为 "{article_data['title']}"。 文章要点： • 技术发展趋势分析 • 行业应用案例分享 • 未来发展展望 #技术 #{article_data['source']} #行业分析 #科技资讯""" # 根据文章类别选择风格 if "tech" in article_data.get('category', '').lower() or "tech" in article_data.get('source', '').lower(): style = "tech" else: style = "warm" logger.info(f"小红书发布 - 标题: {article_data['title']}") logger.info(f"小红书发布 - 风格: {style}") # 发布到小红书 tags = ["技术", article_data['source'], "行业分析", "科技资讯", "自动化"] success = publisher.publish_to_xiaohongshu( title=article_data['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}") return False def update_article_status(article_id): """更新文章状态""" db_path = "/home/ubuntu/nanocode/it_automation_enhanced/data/articles.db" try: conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(''' UPDATE articles SET processed = 1, xhs_status = 1 WHERE id = ? ''', (article_id,)) conn.commit() conn.close() logger.info(f"文章状态更新完成 (ID: {article_id})") return True except Exception as e: logger.error(f"文章状态更新失败: {e}") return False def main(): """主函数""" print("=" * 60) print("🚀 手动完整流程测试") print("=" * 60) print(f"时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") 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" 字数: {article_data['word_count']:,}") print(f" URL: {article_data['url']}") print("") # 2. 创建Markdown内容 print("2️⃣ 创建Markdown内容...") markdown_content = create_test_markdown(article_data) print(f" Markdown长度: {len(markdown_content)} 字符") print("") # 3. 邮件附件测试 print("3️⃣ 邮件附件测试...") email_success = run_email_attachment_test(article_data, markdown_content) print("") # 4. 小红书测试 print("4️⃣ 小红书多图测试...") xhs_success = run_xiaohongshu_test(article_data) print("") # 5. 更新状态 print("5️⃣ 更新文章状态...") if email_success or xhs_success: update_success = update_article_status(article_data['id']) if update_success: print(" ✅ 文章状态更新完成") else: print(" ⚠️ 文章状态更新失败") else: print(" ⏸️ 跳过状态更新（邮件和小红书都失败）") print("") # 6. 结果总结 print("=" * 60) print("📋 测试结果总结") print("=" * 60) print(f" 文章: {article_data['title'][:50]}...") print(f" 邮件附件: {'✅ 成功' if email_success else '❌ 失败'}") print(f" 小红书多图: {'✅ 成功' if xhs_success else '❌ 失败'}") print(f" 状态更新: {'✅ 完成' if (email_success or xhs_success) else '⏸️ 跳过'}") print("") if email_success and xhs_success: print("🎉 完整流程测试成功！") return 0 elif email_success or xhs_success: print("⚠️ 部分功能测试成功") return 1 else: print("❌ 测试失败") return 1 if __name__ == "__main__": sys.exit(main())