nanocode/it_automation_enhanced/scripts/manual_full_test.py

#!/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())