fix: 账号页闪烁/浏览类型/截图复制/时区统一

task_checkpoint.py

@@ -7,11 +7,18 @@
 4. 智能重试机制
 """
 
-import time
-import json
-from datetime import datetime
-from enum import Enum
-import db_pool
+import time
+import json
+from datetime import datetime
+from enum import Enum
+import db_pool
+import pytz
+
+# 北京时区（统一）
+CST_TZ = pytz.timezone("Asia/Shanghai")
+
+def get_cst_now_str():
+    return datetime.now(CST_TZ).strftime('%Y-%m-%d %H:%M:%S')
 
 class TaskStage(Enum):
     """任务执行阶段"""
@@ -103,27 +110,28 @@ class TaskCheckpoint:
 
             conn.commit()
 
-    def create_checkpoint(self, user_id, account_id, username, browse_type):
-        """创建新的任务断点"""
-        task_id = f"{user_id}:{account_id}:{int(time.time())}"
-        with db_pool.get_db() as conn:
-            cursor = conn.cursor()
-            cursor.execute("""
-                INSERT INTO task_checkpoints
-                (task_id, user_id, account_id, username, browse_type, stage, status)
-                VALUES (?, ?, ?, ?, ?, ?, ?)
-            """, (task_id, user_id, account_id, username, browse_type,
-                  TaskStage.QUEUED.value, 'running'))
-            conn.commit()
-            return task_id
+    def create_checkpoint(self, user_id, account_id, username, browse_type):
+        """创建新的任务断点"""
+        task_id = f"{user_id}:{account_id}:{int(time.time())}"
+        with db_pool.get_db() as conn:
+            cursor = conn.cursor()
+            cst_time = get_cst_now_str()
+            cursor.execute("""
+                INSERT INTO task_checkpoints
+                (task_id, user_id, account_id, username, browse_type, stage, status, created_at, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """, (task_id, user_id, account_id, username, browse_type,
+                  TaskStage.QUEUED.value, 'running', cst_time, cst_time))
+            conn.commit()
+            return task_id
 
     def update_stage(self, task_id, stage, progress_percent=None, checkpoint_data=None):
         """更新任务阶段"""
         with db_pool.get_db() as conn:
             cursor = conn.cursor()
 
-            updates = ['stage = ?', 'updated_at = CURRENT_TIMESTAMP']
-            params = [stage.value if isinstance(stage, TaskStage) else stage]
+            updates = ['stage = ?', 'updated_at = ?']
+            params = [stage.value if isinstance(stage, TaskStage) else stage, get_cst_now_str()]
 
             if progress_percent is not None:
                 updates.append('progress_percent = ?')
@@ -155,8 +163,8 @@ class TaskCheckpoint:
         with db_pool.get_db() as conn:
             cursor = conn.cursor()
 
-            updates = ['updated_at = CURRENT_TIMESTAMP']
-            params = []
+            updates = ['updated_at = ?']
+            params = [get_cst_now_str()]
 
             for key in ['current_page', 'total_pages', 'processed_items', 'downloaded_files']:
                 if key in kwargs:
@@ -178,10 +186,11 @@ class TaskCheckpoint:
             """, params)
             conn.commit()
 
-    def record_error(self, task_id, error_message, pause=False):
-        """记录错误并决定是否暂停任务"""
-        with db_pool.get_db() as conn:
-            cursor = conn.cursor()
+    def record_error(self, task_id, error_message, pause=False):
+        """记录错误并决定是否暂停任务"""
+        with db_pool.get_db() as conn:
+            cursor = conn.cursor()
+            cst_time = get_cst_now_str()
 
             # 获取当前重试次数和最大重试次数
             cursor.execute("""
@@ -199,50 +208,51 @@ class TaskCheckpoint:
                 # 判断是否超过最大重试次数
                 if retry_count >= max_retries or pause:
                     # 超过重试次数,暂停任务等待人工处理
-                    cursor.execute("""
-                        UPDATE task_checkpoints
-                        SET status = 'paused',
-                            stage = ?,
-                            retry_count = ?,
-                            error_count = ?,
-                            last_error = ?,
-                            updated_at = CURRENT_TIMESTAMP
-                        WHERE task_id = ?
-                    """, (TaskStage.PAUSED.value, retry_count, error_count,
-                          error_message, task_id))
-                    conn.commit()
-                    return 'paused'
-                else:
-                    # 还可以重试
-                    cursor.execute("""
-                        UPDATE task_checkpoints
-                        SET retry_count = ?,
-                            error_count = ?,
-                            last_error = ?,
-                            updated_at = CURRENT_TIMESTAMP
-                        WHERE task_id = ?
-                    """, (retry_count, error_count, error_message, task_id))
-                    conn.commit()
-                    return 'retry'
+                    cursor.execute("""
+                        UPDATE task_checkpoints
+                        SET status = 'paused',
+                            stage = ?,
+                            retry_count = ?,
+                            error_count = ?,
+                            last_error = ?,
+                            updated_at = ?
+                        WHERE task_id = ?
+                    """, (TaskStage.PAUSED.value, retry_count, error_count,
+                          error_message, cst_time, task_id))
+                    conn.commit()
+                    return 'paused'
+                else:
+                    # 还可以重试
+                    cursor.execute("""
+                        UPDATE task_checkpoints
+                        SET retry_count = ?,
+                            error_count = ?,
+                            last_error = ?,
+                            updated_at = ?
+                        WHERE task_id = ?
+                    """, (retry_count, error_count, error_message, cst_time, task_id))
+                    conn.commit()
+                    return 'retry'
 
             return 'unknown'
 
-    def complete_task(self, task_id, success=True):
-        """完成任务"""
-        with db_pool.get_db() as conn:
-            cursor = conn.cursor()
-            cursor.execute("""
-                UPDATE task_checkpoints
-                SET status = ?,
-                    stage = ?,
-                    progress_percent = 100,
-                    completed_at = CURRENT_TIMESTAMP,
-                    updated_at = CURRENT_TIMESTAMP
-                WHERE task_id = ?
-            """, ('completed' if success else 'failed',
-                  TaskStage.COMPLETED.value if success else TaskStage.FAILED.value,
-                  task_id))
-            conn.commit()
+    def complete_task(self, task_id, success=True):
+        """完成任务"""
+        with db_pool.get_db() as conn:
+            cursor = conn.cursor()
+            cst_time = get_cst_now_str()
+            cursor.execute("""
+                UPDATE task_checkpoints
+                SET status = ?,
+                    stage = ?,
+                    progress_percent = 100,
+                    completed_at = ?,
+                    updated_at = ?
+                WHERE task_id = ?
+            """, ('completed' if success else 'failed',
+                  TaskStage.COMPLETED.value if success else TaskStage.FAILED.value,
+                  cst_time, cst_time, task_id))
+            conn.commit()
 
     def get_checkpoint(self, task_id):
         """获取任务断点信息"""
@@ -323,47 +333,49 @@ class TaskCheckpoint:
                 })
             return tasks
 
-    def resume_task(self, task_id):
-        """恢复暂停的任务"""
-        with db_pool.get_db() as conn:
-            cursor = conn.cursor()
-            cursor.execute("""
-                UPDATE task_checkpoints
-                SET status = 'running',
-                    retry_count = 0,
-                    updated_at = CURRENT_TIMESTAMP
-                WHERE task_id = ? AND status = 'paused'
-            """, (task_id,))
-            conn.commit()
-            return cursor.rowcount > 0
+    def resume_task(self, task_id):
+        """恢复暂停的任务"""
+        with db_pool.get_db() as conn:
+            cursor = conn.cursor()
+            cst_time = get_cst_now_str()
+            cursor.execute("""
+                UPDATE task_checkpoints
+                SET status = 'running',
+                    retry_count = 0,
+                    updated_at = ?
+                WHERE task_id = ? AND status = 'paused'
+            """, (cst_time, task_id))
+            conn.commit()
+            return cursor.rowcount > 0
 
-    def abandon_task(self, task_id):
-        """放弃暂停的任务"""
-        with db_pool.get_db() as conn:
-            cursor = conn.cursor()
-            cursor.execute("""
-                UPDATE task_checkpoints
-                SET status = 'failed',
-                    stage = ?,
-                    completed_at = CURRENT_TIMESTAMP,
-                    updated_at = CURRENT_TIMESTAMP
-                WHERE task_id = ? AND status = 'paused'
-            """, (TaskStage.FAILED.value, task_id))
-            conn.commit()
-            return cursor.rowcount > 0
+    def abandon_task(self, task_id):
+        """放弃暂停的任务"""
+        with db_pool.get_db() as conn:
+            cursor = conn.cursor()
+            cst_time = get_cst_now_str()
+            cursor.execute("""
+                UPDATE task_checkpoints
+                SET status = 'failed',
+                    stage = ?,
+                    completed_at = ?,
+                    updated_at = ?
+                WHERE task_id = ? AND status = 'paused'
+            """, (TaskStage.FAILED.value, cst_time, cst_time, task_id))
+            conn.commit()
+            return cursor.rowcount > 0
 
-    def cleanup_old_checkpoints(self, days=7):
-        """清理旧的断点数据(保留最近N天)"""
-        with db_pool.get_db() as conn:
-            cursor = conn.cursor()
-            cursor.execute("""
-                DELETE FROM task_checkpoints
-                WHERE status IN ('completed', 'failed')
-                AND datetime(completed_at) < datetime('now', '-' || ? || ' days')
-            """, (days,))
-            deleted = cursor.rowcount
-            conn.commit()
-            return deleted
+    def cleanup_old_checkpoints(self, days=7):
+        """清理旧的断点数据(保留最近N天)"""
+        with db_pool.get_db() as conn:
+            cursor = conn.cursor()
+            cursor.execute("""
+                DELETE FROM task_checkpoints
+                WHERE status IN ('completed', 'failed')
+                AND datetime(completed_at) < datetime('now', 'localtime', '-' || ? || ' days')
+            """, (days,))
+            deleted = cursor.rowcount
+            conn.commit()
+            return deleted
 
 
 # 全局单例