From 948e7d7246e493195a021613b92b69412bd0abec Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sat, 8 Aug 2026 21:26:32 +0800 Subject: [PATCH] fix: keep one douyu cookie per account --- ...60808_0019_dedupe_douyu_success_cookies.py | 66 +++++++++++++++++++ web/backend/routers/login.py | 4 ++ web/backend/services/login_service.py | 60 ++++++++++++----- 3 files changed, 115 insertions(+), 15 deletions(-) create mode 100644 web/backend/migrations/versions/20260808_0019_dedupe_douyu_success_cookies.py diff --git a/web/backend/migrations/versions/20260808_0019_dedupe_douyu_success_cookies.py b/web/backend/migrations/versions/20260808_0019_dedupe_douyu_success_cookies.py new file mode 100644 index 0000000..086dbd6 --- /dev/null +++ b/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 diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index ab276e6..f35883c 100644 --- a/web/backend/routers/login.py +++ b/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 diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 6d9d8c5..f4c943a 100644 --- a/web/backend/services/login_service.py +++ b/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 - # 复用该账号最近一条失败任务记录，避免重复产生多条 - existing_task = ( + # 一个斗鱼账号只保留一条成功 CK：再次普通登录时更新最新成功记录。 + latest_success_task = ( self.db.query(LoginTask) - .filter(LoginTask.account_id == aid, LoginTask.status.in_(["failed", "error"])) - .order_by(LoginTask.id.desc()) + .filter(LoginTask.account_id == aid, LoginTask.status == "success") + .order_by(LoginTask.finished_at.desc(), 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, + if latest_success_task: + duplicate_success_tasks = ( + self.db.query(LoginTask) + .filter( + LoginTask.account_id == aid, + LoginTask.status == "success", + LoginTask.id != latest_success_task.id, + ) + .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)