feat: 新增后台任务系统，支持抓取任务在后台持续运行

- 新增后台任务API (/api/tasks) - 抓取任务在后台独立运行，不受页面刷新影响 - 支持任务状态查询和手动停止 - 新增后台任务管理界面 - 数据库新增 background_tasks 表 - 前端使用轮询方式更新任务进度
2026-07-14 10:59:18 +08:00
parent 4f47bf2b91
commit 6534c3315b
8 changed files with 973 additions and 43 deletions
@@ -142,6 +142,25 @@ class Database:
                )
            ''')
            
+            # 后台任务表
+            cursor.execute('''
+                CREATE TABLE IF NOT EXISTS background_tasks (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    task_id TEXT NOT NULL UNIQUE,
+                    task_type TEXT NOT NULL,
+                    status TEXT DEFAULT 'pending',
+                    params TEXT,
+                    progress INTEGER DEFAULT 0,
+                    total INTEGER DEFAULT 0,
+                    current_item TEXT,
+                    result TEXT,
+                    error_message TEXT,
+                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    started_at DATETIME,
+                    finished_at DATETIME
+                )
+            ''')
+            
            conn.commit()

 # ========== 内容库操作 ==========
@@ -472,5 +491,112 @@ class Database:
            conn.commit()
            return cursor.rowcount

+# ========== 后台任务操作 ==========
+    def create_task(self, task_id, task_type, params=None):
+        """创建后台任务"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('''
+                INSERT INTO background_tasks (task_id, task_type, params, status)
+                VALUES (?, ?, ?, 'pending')
+            ''', (task_id, task_type, json.dumps(params, ensure_ascii=False) if params else None))
+            conn.commit()
+            return task_id
+    
+    def get_task(self, task_id):
+        """获取任务详情"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('SELECT * FROM background_tasks WHERE task_id = ?', (task_id,))
+            row = cursor.fetchone()
+            if row:
+                result = dict(row)
+                if result.get('params'):
+                    result['params'] = json.loads(result['params'])
+                if result.get('result'):
+                    result['result'] = json.loads(result['result'])
+                return result
+            return None
+    
+    def get_active_tasks(self, task_type=None):
+        """获取活动任务（running 或 pending）"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            if task_type:
+                cursor.execute('''
+                    SELECT * FROM background_tasks 
+                    WHERE status IN ('pending', 'running') AND task_type = ?
+                    ORDER BY created_at DESC
+                ''', (task_type,))
+            else:
+                cursor.execute('''
+                    SELECT * FROM background_tasks 
+                    WHERE status IN ('pending', 'running')
+                    ORDER BY created_at DESC
+                ''')
+            return [dict(row) for row in cursor.fetchall()]
+    
+    def update_task_status(self, task_id, status, **kwargs):
+        """更新任务状态"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            
+            updates = ['status = ?']
+            values = [status]
+            
+            if status == 'running' and 'started_at' not in kwargs:
+                updates.append('started_at = CURRENT_TIMESTAMP')
+            elif status in ('completed', 'failed', 'stopped'):
+                updates.append('finished_at = CURRENT_TIMESTAMP')
+            
+            for key in ['progress', 'total', 'current_item', 'error_message']:
+                if key in kwargs:
+                    updates.append(f'{key} = ?')
+                    values.append(kwargs[key])
+            
+            if 'result' in kwargs:
+                updates.append('result = ?')
+                values.append(json.dumps(kwargs['result'], ensure_ascii=False))
+            
+            values.append(task_id)
+            
+            cursor.execute(
+                f'UPDATE background_tasks SET {" , ".join(updates)} WHERE task_id = ?',
+                values
+            )
+            conn.commit()
+            return cursor.rowcount > 0
+    
+    def stop_task(self, task_id):
+        """停止任务"""
+        return self.update_task_status(task_id, 'stopped')
+    
+    def get_recent_tasks(self, limit=20):
+        """获取最近的任务"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('''
+                SELECT * FROM background_tasks 
+                ORDER BY created_at DESC 
+                LIMIT ?
+            ''', (limit,))
+            return [dict(row) for row in cursor.fetchall()]
+    
+    def delete_task(self, task_id):
+        """删除任务记录"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('DELETE FROM background_tasks WHERE task_id = ?', (task_id,))
+            conn.commit()
+            return cursor.rowcount > 0
+    
+    def clear_completed_tasks(self):
+        """清理已完成的任务"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('DELETE FROM background_tasks WHERE status IN ("completed", "failed", "stopped")')
+            conn.commit()
+            return cursor.rowcount
+
 # 全局数据库实例
 db = Database()