Initial commit: 百家号文章采集系统

test_database.py

@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+"""
+测试 SQLite 数据库迁移和功能
+"""
+import os
+import sys
+from database import get_database, migrate_from_json
+from task_queue import get_task_queue, TaskStatus
+
+def test_database():
+    """测试数据库功能"""
+    print("=" * 60)
+    print("开始测试 SQLite 数据库功能")
+    print("=" * 60)
+    
+    # 1. 测试数据库初始化
+    print("\n1. 测试数据库初始化...")
+    db = get_database()
+    print(f"✓ 数据库初始化成功: {db.db_path}")
+    
+    # 2. 测试从 JSON 迁移数据
+    print("\n2. 测试数据迁移...")
+    json_file = "data/task_queue.json"
+    if os.path.exists(json_file):
+        count = migrate_from_json(json_file)
+        print(f"✓ 迁移了 {count} 个任务")
+    else:
+        print("! 未找到旧 JSON 文件，跳过迁移")
+    
+    # 3. 测试任务队列功能
+    print("\n3. 测试任务队列功能...")
+    queue = get_task_queue()
+    
+    # 3.1 添加测试任务
+    print("\n3.1 添加测试任务...")
+    task_id = queue.add_task(
+        url="https://baijiahao.baidu.com/u?app_id=test123",
+        months=3,
+        use_proxy=True,
+        username="test_user"
+    )
+    print(f"✓ 添加任务成功: {task_id}")
+    
+    # 3.2 获取任务
+    print("\n3.2 获取任务...")
+    task = queue.get_task(task_id)
+    if task:
+        print(f"✓ 获取任务成功:")
+        print(f"  - URL: {task['url']}")
+        print(f"  - 状态: {task['status']}")
+        print(f"  - 创建时间: {task['created_at']}")
+    
+    # 3.3 更新任务状态
+    print("\n3.3 更新任务状态...")
+    queue.update_task_status(task_id, TaskStatus.PROCESSING)
+    task = queue.get_task(task_id)
+    print(f"✓ 更新状态成功: {task['status']}")
+    
+    # 3.4 更新任务进度
+    print("\n3.4 更新任务进度...")
+    queue.update_task_progress(task_id, 50, "正在处理中...", 25)
+    task = queue.get_task(task_id)
+    print(f"✓ 更新进度成功: {task['progress']}%")
+    
+    # 3.5 获取队列统计
+    print("\n3.5 获取队列统计...")
+    stats = queue.get_queue_stats()
+    print(f"✓ 队列统计:")
+    print(f"  - 总任务数: {stats['total']}")
+    print(f"  - 等待中: {stats['pending']}")
+    print(f"  - 处理中: {stats['processing']}")
+    print(f"  - 已完成: {stats['completed']}")
+    print(f"  - 失败: {stats['failed']}")
+    
+    # 3.6 获取所有任务
+    print("\n3.6 获取所有任务...")
+    all_tasks = queue.get_all_tasks()
+    print(f"✓ 获取所有任务成功，共 {len(all_tasks)} 个任务")
+    
+    # 3.7 删除测试任务
+    print("\n3.7 删除测试任务...")
+    queue.delete_task(task_id)
+    print(f"✓ 删除任务成功: {task_id}")
+    
+    print("\n" + "=" * 60)
+    print("所有测试通过！SQLite 数据库运行正常")
+    print("=" * 60)
+
+if __name__ == "__main__":
+    test_database()