fix: keep one douyu cookie per account

web/backend/migrations/versions/20260808_0019_dedupe_douyu_success_cookies.py

@@ -0,0 +1,66 @@
+"""清理斗鱼账号重复成功 Cookie 记录
+
+Revision ID: 20260808_0019
+Revises: 20260807_0018
+Create Date: 2026-08-08
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "20260808_0019"
+down_revision: Union[str, None] = "20260807_0018"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if not inspector.has_table("login_tasks"):
+        return
+
+    task_table = sa.table(
+        "login_tasks",
+        sa.column("id", sa.Integer),
+        sa.column("account_id", sa.Integer),
+        sa.column("status", sa.String),
+        sa.column("finished_at", sa.DateTime),
+    )
+    rows = bind.execute(
+        sa.select(
+            task_table.c.id,
+            task_table.c.account_id,
+            task_table.c.finished_at,
+        )
+        .where(task_table.c.status == "success")
+        .order_by(
+            task_table.c.account_id.asc(),
+            task_table.c.finished_at.desc(),
+            task_table.c.id.desc(),
+        )
+    ).all()
+
+    delete_ids: list[int] = []
+    seen_accounts: set[int] = set()
+    for row in rows:
+        if row.account_id in seen_accounts:
+            delete_ids.append(row.id)
+            continue
+        seen_accounts.add(row.account_id)
+
+    if not delete_ids:
+        return
+
+    for index in range(0, len(delete_ids), 500):
+        chunk = delete_ids[index:index + 500]
+        bind.execute(
+            task_table.delete().where(task_table.c.id.in_(chunk))
+        )
+
+
+def downgrade() -> None:
+    pass

web/backend/routers/login.py

@@ -49,7 +49,11 @@ async def create_batch(
 
     # 权限过滤账号
     valid_ids = []
+    seen_ids = set()
     for aid in req.account_ids:
+        if aid in seen_ids:
+            continue
+        seen_ids.add(aid)
         acc = db.query(Account).filter(Account.id == aid).first()
         if not acc:
             continue

web/backend/services/login_service.py

@@ -300,8 +300,17 @@ class LoginBatchRunner:
                     task.ck_checked_at = None
                     _append_task_info(task, acc)
             else:
+                seen_account_ids = set()
                 for aid in self.account_ids:
-                    acc = self.db.query(AccountModel).filter(AccountModel.id == aid).first()
+                    if aid in seen_account_ids:
+                        continue
+                    seen_account_ids.add(aid)
+                    acc = (
+                        self.db.query(AccountModel)
+                        .filter(AccountModel.id == aid)
+                        .with_for_update()
+                        .first()
+                    )
                     if not acc:
                         continue
                     # 权限检查：客服只能跑分配给自己的
@@ -310,24 +319,45 @@ class LoginBatchRunner:
                             self._push_log("warning", f"跳过无权账号: {acc.username}")
                             continue
 
-                    # 复用该账号最近一条失败任务记录，避免重复产生多条
+                    # 一个斗鱼账号只保留一条成功 CK：再次普通登录时更新最新成功记录。
-                    existing_task = (
+                    latest_success_task = (
                         self.db.query(LoginTask)
-                        .filter(LoginTask.account_id == aid, LoginTask.status.in_(["failed", "error"]))
+                        .filter(LoginTask.account_id == aid, LoginTask.status == "success")
-                        .order_by(LoginTask.id.desc())
+                        .order_by(LoginTask.finished_at.desc(), LoginTask.id.desc())
                         .first()
                     )
-                    if existing_task:
+                    if latest_success_task:
-                        existing_task.cookie = ""
+                        duplicate_success_tasks = (
-                        task = existing_task
+                            self.db.query(LoginTask)
-                    else:
+                            .filter(
-                        task = LoginTask(
+                                LoginTask.account_id == aid,
-                            batch_id=batch_id,
+                                LoginTask.status == "success",
-                            account_id=aid,
+                                LoginTask.id != latest_success_task.id,
-                            status="pending",
+                            )
-                            created_by=self.created_by,
+                            .all()
                         )
-                        self.db.add(task)
+                        for duplicate_task in duplicate_success_tasks:
+                            self.db.delete(duplicate_task)
+                        task = latest_success_task
+                    else:
+                        # 复用该账号最近一条失败任务记录，避免重复产生多条失败历史。
+                        existing_task = (
+                            self.db.query(LoginTask)
+                            .filter(LoginTask.account_id == aid, LoginTask.status.in_(["failed", "error"]))
+                            .order_by(LoginTask.id.desc())
+                            .first()
+                        )
+                        if existing_task:
+                            existing_task.cookie = ""
+                            task = existing_task
+                        else:
+                            task = LoginTask(
+                                batch_id=batch_id,
+                                account_id=aid,
+                                status="pending",
+                                created_by=self.created_by,
+                            )
+                            self.db.add(task)
 
                     _append_task_info(task, acc)