diff --git a/tests/test_cookie_operations.py b/tests/test_cookie_operations.py index b611bca..9a443f4 100644 --- a/tests/test_cookie_operations.py +++ b/tests/test_cookie_operations.py @@ -207,8 +207,8 @@ class TestCookieOperation: ) def test_relogin_failed_record_stays_visible_and_can_be_checked(self): - self.owned_task.status = "relogin_failed" - self.owned_task.message = "重新登录失败: 密码错误（旧 Cookie 已保留）" + self.owned_task.attempt_status = "failed" + self.owned_task.attempt_message = "重新登录失败: 密码错误（旧 Cookie 已保留）" self.session.commit() result = list_cookie_operations( @@ -223,9 +223,13 @@ class TestCookieOperation: assert result["total"] == 1 assert result["items"][0]["relogin_status"] == "relogin_failed" assert "旧 Cookie 已保留" in result["items"][0]["relogin_message"] + # 失败的重登尝试不影响记录持有的 CK + assert self.owned_task.cookie == "owned-secret" + assert self.owned_task.status == "success" def test_service_restart_cleans_orphan_relogin_tasks(self): - self.owned_task.status = "relogin_running" + self.owned_task.attempt_status = "running" + self.owned_task.attempt_message = "正在重新登录，旧 Cookie 保留中" self.session.commit() cleaned = cleanup_orphan_relogin_tasks(self.session) @@ -233,13 +237,17 @@ class TestCookieOperation: assert cleaned == 1 self.session.refresh(self.owned_task) self.session.refresh(self.other_task) - assert self.owned_task.status == "relogin_failed" - assert "服务重启" in self.owned_task.message + # 只收敛尝试状态，已有 CK 与成功状态原封不动 + assert self.owned_task.attempt_status == "failed" + assert "服务重启" in self.owned_task.attempt_message + assert self.owned_task.status == "success" + assert self.owned_task.cookie == "owned-secret" + assert self.other_task.attempt_status == "" assert self.other_task.status == "success" def test_service_restart_cleans_orphan_login_tasks(self): - """普通登录批次遗留的 pending/running 必须收敛，且保留旧 CK。""" - self.owned_task.status = "pending" + """普通登录批次遗留的在途尝试必须收敛，且保留旧 CK。""" + self.owned_task.attempt_status = "pending" self.owned_task.cookie = "acf_uid=owned" self.session.commit() @@ -248,11 +256,25 @@ class TestCookieOperation: assert cleaned == 1 self.session.refresh(self.owned_task) self.session.refresh(self.other_task) - assert self.owned_task.status == "failed" + assert self.owned_task.attempt_status == "failed" assert self.owned_task.cookie == "acf_uid=owned" - assert "服务重启" in self.owned_task.message + assert self.owned_task.status == "success" + assert "服务重启" in self.owned_task.attempt_message assert self.other_task.status == "success" + def test_service_restart_cleans_legacy_in_flight_status(self): + """升级前直接把状态写在 status 上的在途记录也要能收敛。""" + self.owned_task.status = "relogin_running" + self.owned_task.attempt_status = "" + self.session.commit() + + cleaned = cleanup_orphan_login_tasks(self.session) + + assert cleaned == 1 + self.session.refresh(self.owned_task) + assert self.owned_task.attempt_status == "failed" + assert self.owned_task.cookie == "owned-secret" + def test_runner_copies_proxy_config_before_background_execution(self): proxy = ProxyConfig( enabled=True, diff --git a/tests/test_login_attempt_isolation.py b/tests/test_login_attempt_isolation.py new file mode 100644 index 0000000..284d272 --- /dev/null +++ b/tests/test_login_attempt_isolation.py @@ -0,0 +1,224 @@ +"""回归：替换 CK 必须是最后一步，只有登录成功才会写回记录。 + +事故背景：批次启动时会把复用的成功记录直接改成 pending（旧 CK 仍保留）， +批次被停止或失败后没有任何地方把它改回成功，账号于是从工作台消失。 +现在尝试状态独立存放在 attempt_*，status/cookie/message 只在登录成功时 +更新，所以停止、失败、进程重启都不可能破坏账号已有的 CK。 +""" + +import tempfile +from pathlib import Path +from types import SimpleNamespace + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +import web.backend.database as database_module +from web.backend.database import Base +from web.backend.models import Account, LoginTask, User +from web.backend.routers.login import list_tasks +from web.backend.services import login_service + + +class _FakeLogin: + """替身登录器：不发网络请求，结果由类属性控制。""" + + success = True + cookie = "new-cookie" + message = "登录成功" + code = None + calls = 0 + + def __init__(self, *_args, **_kwargs): + pass + + def login(self): + type(self).calls += 1 + return SimpleNamespace( + success=type(self).success, + cookie=type(self).cookie if type(self).success else "", + message=type(self).message, + code=type(self).code, + ) + + def check_account(self): + return self.login() + + +class TestLoginAttemptIsolation: + def setup_method(self): + self.tmpdir = tempfile.TemporaryDirectory() + url = f"sqlite:///{Path(self.tmpdir.name) / 'attempt.db'}" + self.engine = create_engine(url, connect_args={"check_same_thread": False}) + Base.metadata.create_all(self.engine) + # 批次执行器在后台线程里用 SessionLocal，测试期间指向同一个文件库。 + database_module.SessionLocal.configure(bind=self.engine) + self.session = sessionmaker(bind=self.engine)() + + self.user = User(username="admin", password_hash="hash", role="super_admin") + self.session.add(self.user) + self.session.flush() + self.account = Account( + username="远行者103", + password="password", + email="yuanxingzhe@example.com", + email_password="mail-password", + ) + self.session.add(self.account) + self.session.flush() + self.task = LoginTask( + batch_id="old-batch", + account_id=self.account.id, + created_by=self.user.id, + status="success", + cookie="old-cookie", + message="登录成功", + ) + self.session.add(self.task) + self.session.commit() + self.task_id = self.task.id + + _FakeLogin.success = True + _FakeLogin.cookie = "new-cookie" + _FakeLogin.message = "登录成功" + _FakeLogin.calls = 0 + + def teardown_method(self): + database_module.SessionLocal.configure(bind=database_module.engine) + self.session.close() + Base.metadata.drop_all(self.engine) + self.engine.dispose() + self.tmpdir.cleanup() + + def _run_batch(self, monkeypatch, *, mode="login", account_ids=None, stop=False): + monkeypatch.setattr(login_service, "DouyuLogin", _FakeLogin) + # 批次结束时 runner 会关掉自己的会话，因此单独给它一个。 + runner_db = sessionmaker(bind=self.engine)() + runner = login_service.LoginBatchRunner( + db=runner_db, + account_ids=account_ids if account_ids is not None else [self.account.id], + created_by=self.user.id, + creator_permissions=["login:view_all"], + proxy_config=None, + concurrency=1, + mode=mode, + ) + if stop: + runner.stop() + runner.run() + return runner + + def _fresh(self, task_id: int) -> LoginTask: + session = sessionmaker(bind=self.engine)() + try: + return session.query(LoginTask).filter(LoginTask.id == task_id).one() + finally: + session.close() + + def _fresh_user(self, session) -> User: + return session.query(User).filter(User.id == self.user.id).one() + + def test_failed_attempt_keeps_existing_cookie(self, monkeypatch): + _FakeLogin.success = False + _FakeLogin.message = "账号密码错误" + + self._run_batch(monkeypatch) + + task = self._fresh(self.task_id) + # 记录持有 CK 的字段完全没被碰过 + assert task.status == "success" + assert task.cookie == "old-cookie" + assert task.message == "登录成功" + assert task.batch_id == "old-batch" + # 失败只体现在尝试字段上 + assert task.attempt_status == "failed" + assert task.attempt_message == "账号密码错误" + assert task.attempt_at is not None + + def test_failed_attempt_reuses_the_cookie_row(self, monkeypatch): + _FakeLogin.success = False + _FakeLogin.message = "账号密码错误" + + self._run_batch(monkeypatch) + + session = sessionmaker(bind=self.engine)() + try: + rows = ( + session.query(LoginTask) + .filter(LoginTask.account_id == self.account.id) + .all() + ) + finally: + session.close() + assert [row.id for row in rows] == [self.task_id] + + def test_successful_attempt_replaces_cookie(self, monkeypatch): + self._run_batch(monkeypatch) + + task = self._fresh(self.task_id) + assert task.status == "success" + assert task.cookie == "new-cookie" + assert task.message == "登录成功" + assert task.attempt_status == "success" + assert task.attempt_at is not None + assert task.batch_id != "old-batch" + + def test_stopped_batch_keeps_existing_cookie(self, monkeypatch): + self._run_batch(monkeypatch, stop=True) + + task = self._fresh(self.task_id) + assert _FakeLogin.calls == 0 + assert task.cookie == "old-cookie" + assert task.status == "success" + assert task.attempt_status == "failed" + assert "旧 Cookie 已保留" in task.attempt_message + + def test_list_tasks_reports_attempt_state_without_losing_ck_row(self, monkeypatch): + _FakeLogin.success = False + _FakeLogin.message = "账号密码错误" + + runner = self._run_batch(monkeypatch) + + read_db = sessionmaker(bind=self.engine)() + try: + rows = list_tasks( + batch_id=runner.batch_id, db=read_db, current=self._fresh_user(read_db) + ) + finally: + read_db.close() + assert len(rows) == 1 + row = rows[0] + # 页面看到的是「这次尝试怎么样了」，而记录仍然持有 CK + assert row.status == "failed" + assert row.message == "账号密码错误" + assert row.batch_id == runner.batch_id + assert self._fresh(self.task_id).cookie == "old-cookie" + + def test_account_without_cookie_still_reports_failure(self, monkeypatch): + """首次登录失败的账号没有 CK 需要保护，保持原来的失败语义。""" + fresh_account = Account( + username="no-cookie", + password="password", + email="nocookie@example.com", + email_password="mail-password", + ) + self.session.add(fresh_account) + self.session.commit() + _FakeLogin.success = False + _FakeLogin.message = "账号密码错误" + + self._run_batch(monkeypatch, account_ids=[fresh_account.id]) + + session = sessionmaker(bind=self.engine)() + try: + task = ( + session.query(LoginTask) + .filter(LoginTask.account_id == fresh_account.id) + .one() + ) + assert task.status == "failed" + assert task.message == "账号密码错误" + assert task.attempt_status == "failed" + assert task.cookie == "" + finally: + session.close() diff --git a/tests/test_login_attempt_migration.py b/tests/test_login_attempt_migration.py new file mode 100644 index 0000000..f078dcf --- /dev/null +++ b/tests/test_login_attempt_migration.py @@ -0,0 +1,152 @@ +"""迁移 20261001_0035 的存量数据归一验证。 + +线上升级时要保证： +- 持有 CK 的记录 status 归一到 success，历史批次状态搬进 attempt_*； +- 没有 CK 且卡在 pending/running 的记录收敛为 failed； +- 任何情况下都不丢 CK。 +""" + +import tempfile +from pathlib import Path + +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, text + +PROJECT_ROOT = Path(__file__).resolve().parents[1] + + +def _migration_config(url: str) -> Config: + config = Config(str(PROJECT_ROOT / "alembic.ini")) + config.set_main_option( + "script_location", str(PROJECT_ROOT / "web" / "backend" / "migrations") + ) + config.set_main_option("sqlalchemy.url", url) + return config + + +def _insert_task(conn, *, status: str, cookie: str, message: str = "") -> None: + conn.execute( + text( + """ + INSERT INTO login_tasks + (batch_id, account_id, status, cookie, message, created_by, + created_at, finished_at) + VALUES + ('batch-1', 1, :status, :cookie, :message, 1, + '2026-09-30 00:00:00', '2026-09-30 01:00:00') + """ + ), + {"status": status, "cookie": cookie, "message": message}, + ) + + +class TestLoginAttemptMigration: + def setup_method(self): + import web.backend.database as database_module + + self.database_module = database_module + self.original_url = database_module.DATABASE_URL + self.tmpdir = tempfile.TemporaryDirectory() + self.url = f"sqlite:///{Path(self.tmpdir.name) / 'upgrade.db'}" + # env.py 用模块级 DATABASE_URL 覆盖配置里的连接串。 + database_module.DATABASE_URL = self.url + self.config = _migration_config(self.url) + command.upgrade(self.config, "20260901_0034") + + self.engine = create_engine(self.url) + with self.engine.begin() as conn: + conn.execute( + text( + """ + INSERT INTO accounts + (id, username, password, email, email_password, + email_imap_server, email_imap_port, email_imap_ssl, + tag, remark, uid, nickname, created_at, updated_at) + VALUES + (1, '远行者103', 'p', 'e@example.com', 'm', + '', 993, 1, '', '', '', '', '2026-09-30 00:00:00', + '2026-09-30 00:00:00') + """ + ) + ) + conn.execute( + text( + """ + INSERT INTO users (id, username, password_hash, role, is_active) + VALUES (1, 'admin', 'hash', 'super_admin', 1) + """ + ) + ) + _insert_task(conn, status="success", cookie="ck-success") + _insert_task( + conn, + status="relogin_failed", + cookie="ck-relogin", + message="重新登录失败: 密码错误（旧 Cookie 已保留）", + ) + _insert_task(conn, status="pending", cookie="ck-pending") + _insert_task(conn, status="pending", cookie="") + _insert_task(conn, status="running", cookie="") + _insert_task(conn, status="failed", cookie="ck-failed", message="登录失败") + + def teardown_method(self): + self.database_module.DATABASE_URL = self.original_url + self.engine.dispose() + self.tmpdir.cleanup() + + def _rows(self) -> list[dict]: + with self.engine.connect() as conn: + rows = conn.execute( + text( + """ + SELECT status, cookie, message, attempt_status, + attempt_message, attempt_batch_id, attempt_at + FROM login_tasks + ORDER BY id + """ + ) + ).mappings() + return [dict(row) for row in rows] + + def test_upgrade_normalizes_existing_rows(self): + command.upgrade(self.config, "head") + + rows = self._rows() + ( + success, + relogin_failed, + pending_with_ck, + pending_without_ck, + running_without_ck, + failed_with_ck, + ) = rows + + # 已经成功的记录：status 不变，也没有历史尝试需要搬迁 + assert success["status"] == "success" + assert success["cookie"] == "ck-success" + assert success["attempt_status"] in (None, "") + + # 重登失败：CK 保留，status 归一为 success，失败信息进 attempt_* + assert relogin_failed["cookie"] == "ck-relogin" + assert relogin_failed["status"] == "success" + assert relogin_failed["attempt_status"] == "failed" + assert "旧 Cookie 已保留" in relogin_failed["attempt_message"] + assert relogin_failed["attempt_batch_id"] == "batch-1" + + # 卡在 pending 但持有 CK：同样归一为 success + 中断说明 + assert pending_with_ck["cookie"] == "ck-pending" + assert pending_with_ck["status"] == "success" + assert pending_with_ck["attempt_status"] == "failed" + assert "旧 Cookie 已保留" in pending_with_ck["attempt_message"] + + # 没有 CK 的在途记录：收敛为失败，不再永久排队 + for row in (pending_without_ck, running_without_ck): + assert row["status"] == "failed" + assert row["attempt_status"] == "failed" + assert "服务重启" in row["message"] + + # 复用记录上的旧 CK 不能被清掉 + assert failed_with_ck["status"] == "success" + assert failed_with_ck["cookie"] == "ck-failed" + assert failed_with_ck["attempt_status"] == "failed" diff --git a/tests/test_migrations.py b/tests/test_migrations.py index cdeb8da..9bd0c04 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -22,7 +22,7 @@ TESTS_DIR = Path(__file__).resolve().parent PROJECT_ROOT = TESTS_DIR.parent VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions" -HEAD_REVISION = "20260901_0034" +HEAD_REVISION = "20261001_0035" # 迁移新增、但不声明在模型里的复合查询索引。 EXTRA_INDEXES = ( diff --git a/web/backend/migrations/versions/20261001_0035_login_attempt_fields.py b/web/backend/migrations/versions/20261001_0035_login_attempt_fields.py new file mode 100644 index 0000000..0e45370 --- /dev/null +++ b/web/backend/migrations/versions/20261001_0035_login_attempt_fields.py @@ -0,0 +1,139 @@ +"""login_tasks 增加 attempt_* 字段：登录尝试与 CK 记录分离。 + +事故背景：批次启动时会把复用的成功记录直接改成 pending（旧 CK 仍保留）， +批次被停止或失败后没有任何地方把它改回成功，账号于是从工作台消失。 +按「替换 CK 是最后一步、只有成功后才会替换旧的 CK」的原则，把尝试状态独立 +出来：批次只写 attempt_*，只有登录成功才会把新 CK 写回 status/cookie/message， +因此停止、失败、进程重启都不可能破坏已有 CK。 + +存量数据归一： +- 持有 CK 但状态不是 success 的记录（relogin_failed/failed/pending/running…） + 把状态、消息、批次、时间搬到 attempt_*，status 归一到 success，让 status + 重新只表示「这条记录持有可用 CK」； +- 没有 CK 且卡在 pending/running 的记录收敛为 failed，避免永久显示排队中。 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "20261001_0035" +down_revision: str | None = "20260901_0034" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +# 升级前可能停留在这些状态、但旧 CK 已经保留在记录里的批次状态。 +_INTERRUPTED_STATUSES = ("pending", "running", "relogin_pending", "relogin_running") +_INTERRUPTED_SQL = "(" + ",".join(f"'{item}'" for item in _INTERRUPTED_STATUSES) + ")" +# 历史重登状态统一归一为 failed，避免 attempt_status 里混入页面不认识的值。 +_LEGACY_ATTEMPT_STATUSES = _INTERRUPTED_STATUSES + ("relogin_failed",) +_LEGACY_ATTEMPT_SQL = ( + "(" + ",".join(f"'{item}'" for item in _LEGACY_ATTEMPT_STATUSES) + ")" +) + + +def upgrade() -> None: + bind = op.get_bind() + columns = {item["name"] for item in sa.inspect(bind).get_columns("login_tasks")} + if "attempt_status" not in columns: + # NOT NULL + server_default 会让存量行直接填上空串，避免出现 NULL + # 导致 NOT IN 查询把历史记录整批排除。 + op.add_column( + "login_tasks", + sa.Column( + "attempt_status", + sa.String(length=32), + nullable=False, + server_default="", + ), + ) + if "attempt_message" not in columns: + op.add_column( + "login_tasks", + sa.Column( + "attempt_message", + sa.String(length=512), + nullable=False, + server_default="", + ), + ) + if "attempt_batch_id" not in columns: + op.add_column( + "login_tasks", + sa.Column( + "attempt_batch_id", + sa.String(length=64), + nullable=False, + server_default="", + ), + ) + if "attempt_at" not in columns: + op.add_column( + "login_tasks", + sa.Column("attempt_at", sa.DateTime(), nullable=True), + ) + + # 1) 持有 CK 的记录：历史批次状态搬进 attempt_*（status 稍后再归一， + # 避免同一条 UPDATE 里读到被覆盖后的值）。 + bind.execute( + sa.text( + f""" + UPDATE login_tasks + SET attempt_status = CASE + WHEN status IN {_LEGACY_ATTEMPT_SQL} THEN 'failed' + ELSE status + END, + attempt_message = CASE + WHEN status IN {_INTERRUPTED_SQL} + THEN '登录已中断（旧 Cookie 已保留）' + ELSE COALESCE(message, '') + END, + attempt_batch_id = COALESCE(batch_id, ''), + attempt_at = COALESCE(finished_at, created_at) + WHERE cookie IS NOT NULL + AND cookie <> '' + AND status <> 'success' + AND (attempt_status IS NULL OR attempt_status = '') + """ + ) + ) + + # 2) 持有 CK 的记录：status 只表示「持有可用 CK」。 + bind.execute( + sa.text( + """ + UPDATE login_tasks + SET status = 'success' + WHERE cookie IS NOT NULL + AND cookie <> '' + AND status <> 'success' + """ + ) + ) + + # 3) 没有 CK 且卡在批次中的记录：收敛为失败，不再永久排队中。 + bind.execute( + sa.text( + f""" + UPDATE login_tasks + SET status = 'failed', + message = '登录已中断（服务重启），无 Cookie', + finished_at = COALESCE(finished_at, created_at), + attempt_status = 'failed', + attempt_message = '登录已中断（服务重启），无 Cookie', + attempt_at = COALESCE(attempt_at, created_at) + WHERE status IN {_INTERRUPTED_SQL} + AND (cookie IS NULL OR cookie = '') + AND (attempt_status IS NULL OR attempt_status = '') + """ + ) + ) + + +def downgrade() -> None: + bind = op.get_bind() + columns = {item["name"] for item in sa.inspect(bind).get_columns("login_tasks")} + for name in ("attempt_at", "attempt_batch_id", "attempt_message", "attempt_status"): + if name in columns: + op.drop_column("login_tasks", name) diff --git a/web/backend/models.py b/web/backend/models.py index d127c40..26eba62 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -155,6 +155,18 @@ class LoginTask(Base): JSON, nullable=True ) # {fish_ball, nickname, level, message} ck_checked_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + # 最近一次登录尝试的状态。批次只写 attempt_*，只有登录成功才会把新 CK + # 写回 status/cookie/message，因此停止、失败、进程重启都不会破坏已有 CK。 + attempt_status: Mapped[str] = mapped_column( + String(32), default="", server_default="" + ) # pending / running / success / failed / stopped / 空=没有进行过的尝试 + attempt_message: Mapped[str] = mapped_column( + String(512), default="", server_default="" + ) + attempt_batch_id: Mapped[str] = mapped_column( + String(64), default="", server_default="" + ) + attempt_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) account: Mapped[Account] = relationship("Account", back_populates="login_tasks") diff --git a/web/backend/routers/cookies.py b/web/backend/routers/cookies.py index d1335c5..f8dead8 100644 --- a/web/backend/routers/cookies.py +++ b/web/backend/routers/cookies.py @@ -19,7 +19,7 @@ from ..permissions import get_user_permissions, user_has_permission from ..schemas import CookieReloginRequest from ..services.cookie_check_service import check_douyu_cookie from ..services.douyu_service import ( - LOGIN_IN_FLIGHT_STATUSES, + LOGIN_ATTEMPT_IN_FLIGHT_STATUSES, cookie_holding_filter, cookie_operable_filter, ) @@ -76,6 +76,34 @@ def _order_cookie_tasks(query, selected_names: list[str]): return query.order_by(LoginTask.finished_at.desc(), LoginTask.id.desc()) +# CK 操作页沿用的重登状态词表；attempt_* 是唯一来源，历史值一并兼容。 +_RELOGIN_STATUS_BY_ATTEMPT = { + "pending": "relogin_pending", + "running": "relogin_running", + "failed": "relogin_failed", + "error": "relogin_failed", + "relogin_pending": "relogin_pending", + "relogin_running": "relogin_running", + "relogin_failed": "relogin_failed", +} + + +def _relogin_display(task: LoginTask) -> dict: + """把最近一次登录尝试映射成 CK 操作页展示用的重登状态。 + + 账号持有的 CK 与「这次尝试的结果」分开存储，页面只关心后者；尝试结束 + 后旧 CK 仍在记录里，因此停止或失败都不会让这一行消失。 + """ + attempt_status = task.attempt_status or "" + return { + "relogin_status": _RELOGIN_STATUS_BY_ATTEMPT.get(attempt_status, ""), + "relogin_message": task.attempt_message or "", + "relogin_batch_id": task.attempt_batch_id + if attempt_status in LOGIN_ATTEMPT_IN_FLIGHT_STATUSES + else "", + } + + def _check_one_cookie(task: LoginTask) -> dict: """将共享检测结果补充为 Cookie 记录的接口返回格式。""" result = check_douyu_cookie(task.cookie or "") @@ -291,11 +319,7 @@ def list_cookie_operations( "ck_check_status": task.ck_check_status or "", "ck_checked_at": _fmt_dt(task.ck_checked_at), "created_at": _fmt_dt(task.finished_at), - "relogin_status": task.status if task.status != "success" else "", - "relogin_message": task.message or "", - "relogin_batch_id": task.batch_id - if task.status in {"relogin_pending", "relogin_running"} - else "", + **_relogin_display(task), } for task in tasks ], @@ -575,17 +599,16 @@ def _start_relogin_tasks( skipped.append(task.id) continue if not acc.password or not acc.email: - task.status = "relogin_failed" - task.message = "重新登录失败: 账号缺少密码或邮箱，旧 Cookie 已保留" - task.finished_at = datetime.now(UTC) + # 只记录这次尝试失败，账号已有的 CK 原封不动。 + task.attempt_status = "failed" + task.attempt_message = "重新登录失败: 账号缺少密码或邮箱，旧 Cookie 已保留" + task.attempt_at = datetime.now(UTC) skipped.append(task.id) continue - if task.status in LOGIN_IN_FLIGHT_STATUSES: + if task.attempt_status in LOGIN_ATTEMPT_IN_FLIGHT_STATUSES: skipped.append(task.id) continue - task.status = "relogin_pending" - task.message = "等待重新登录，旧 Cookie 已保留" - task.finished_at = None + # 排队状态由 LoginBatchRunner 在批次启动时写入 attempt_*（含批次号）。 task_ids.append(task.id) if not task_ids: diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index 74c0530..c19b4df 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -4,7 +4,7 @@ import asyncio import threading from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect -from sqlalchemy import func +from sqlalchemy import func, or_ from sqlalchemy.orm import Session, defer, load_only from ..database import SessionLocal, get_db @@ -18,12 +18,24 @@ from ..services.login_service import LoginBatchRunner, batch_registry router = APIRouter(prefix="/api/login", tags=["登录任务"]) +def _effective_attempt_status(): + """页面展示用的状态：优先最近一次登录尝试，没有尝试时回退到 CK 状态。 + + ``attempt_*`` 才是「这次登录怎么样了」，``status`` 只表示这条记录是否 + 持有可用 CK —— 两者分开，停止/失败/重启都不会让账号从列表里消失。 + """ + return func.coalesce( + func.nullif(LoginTask.attempt_status, ""), + LoginTask.status, + ) + + def _login_task_summary(query): """按状态汇总登录任务，避免概览页拉完整任务列表。""" rows = ( query.order_by(None) - .with_entities(LoginTask.status, func.count(LoginTask.id)) - .group_by(LoginTask.status) + .with_entities(_effective_attempt_status(), func.count(LoginTask.id)) + .group_by(_effective_attempt_status()) .all() ) status_counts = {status or "": count for status, count in rows} @@ -120,7 +132,10 @@ def list_tasks( query = query.filter(Account.assigned_to == current.id) if batch_id: - query = query.filter(LoginTask.batch_id == batch_id) + # 记录持有的 CK 属于产出它的批次，最近一次尝试属于新批次，两者都要算。 + query = query.filter( + or_(LoginTask.batch_id == batch_id, LoginTask.attempt_batch_id == batch_id) + ) rows = query.order_by(LoginTask.id.desc()).limit(200).all() # 批量收集 account_id，一次性查出 username @@ -137,18 +152,21 @@ def list_tasks( result = [] for t in rows: + attempt_status = t.attempt_status or "" result.append( LoginTaskOut( id=t.id, - batch_id=t.batch_id, + batch_id=t.attempt_batch_id or t.batch_id, account_id=t.account_id, account_username=accounts_map.get(t.account_id, ""), - status=t.status, + status=attempt_status or t.status, cookie="", - message=t.message or "", + message=(t.attempt_message if attempt_status else "") + or t.message + or "", created_by=t.created_by, created_at=t.created_at, - finished_at=t.finished_at, + finished_at=t.attempt_at or t.finished_at, ) ) return result diff --git a/web/backend/services/douyu_service.py b/web/backend/services/douyu_service.py index ec40cfb..1888702 100644 --- a/web/backend/services/douyu_service.py +++ b/web/backend/services/douyu_service.py @@ -5,7 +5,7 @@ from __future__ import annotations import uuid from datetime import UTC, datetime -from sqlalchemy import and_, select +from sqlalchemy import and_, func, select from sqlalchemy.orm import Session from core.douyu.activity_client import DouyuActivityClient @@ -181,18 +181,16 @@ def ensure_douyu_config(db: Session) -> DouyuConfig: return config -# 登录/重登批次正在处理的状态：这些记录已被批次接管，不允许并发操作。 -LOGIN_IN_FLIGHT_STATUSES = ("pending", "running", "relogin_pending", "relogin_running") +# 登录/重登尝试的在途状态：判断只看 attempt_*，与记录持有的 CK 无关。 +LOGIN_ATTEMPT_IN_FLIGHT_STATUSES = ("pending", "running") def cookie_holding_filter(): """判定登录记录当前是否仍持有 Cookie。 - 账号是否有 CK 只取决于记录里是否还留着 Cookie，而不是取决于批次状态： - 普通登录在排队时会把复用的成功记录临时置为 ``pending``/``running``， - CK 重登失败会置为 ``relogin_failed``，这些状态都按设计保留旧 Cookie - （见 ``login_service`` 的「旧 Cookie 已保留」分支），账号必须继续可见。 - 只有 Cookie 真正被清空（账号检测成功）时才不再算持有 CK。 + 账号是否有 CK 只取决于记录里是否还留着 Cookie。登录尝试（attempt_*） + 在成功前不会碰 status/cookie/message，所以停止、失败、进程重启都不会 + 让账号失去 CK；只有 Cookie 真正被清空（账号检测成功）时才不再持有。 """ return and_( LoginTask.cookie.isnot(None), @@ -216,12 +214,14 @@ def latest_cookie_login_task(db: Session, account_id: int) -> LoginTask | None: def cookie_operable_filter(): """判定登录记录能否基于现有 Cookie 再做检测或重登。 - 正在被登录/重登批次处理的行（pending/running/relogin_*）已由批次接管， - 并发再点检测或重登只会互相覆盖状态，因此排除在外。 + 正在被批次处理的记录（attempt_* 处于在途状态）已由批次接管，并发再点 + 检测或重登只会互相覆盖，因此排除在外；CK 与 status 不参与判断。 """ return and_( cookie_holding_filter(), - LoginTask.status.notin_(LOGIN_IN_FLIGHT_STATUSES), + func.coalesce(LoginTask.attempt_status, "").notin_( + LOGIN_ATTEMPT_IN_FLIGHT_STATUSES + ), ) diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 3df9732..775acab 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -11,6 +11,7 @@ from types import SimpleNamespace from typing import cast from loguru import logger +from sqlalchemy import or_ from sqlalchemy.orm import Session from core.douyu import DouyuLogin, IframeLoginAPI, WgapiLoginAPI @@ -21,6 +22,7 @@ from ..models import Account as AccountModel from ..models import LoginTask from ..models import ProxyConfig as ProxyConfigModel from .cookie_check_service import check_douyu_cookie +from .douyu_service import cookie_holding_filter def _create_api_strategy(strategy_name: str): @@ -89,6 +91,66 @@ def _snapshot_proxy_config( ) +ATTEMPT_IN_FLIGHT_STATUSES = ("pending", "running") +# 升级前把尝试状态写进 status 的历史取值；判断与清理时一并兼容。 +_LEGACY_IN_FLIGHT_STATUSES = ( + "pending", + "running", + "relogin_pending", + "relogin_running", +) + + +def _attempt_in_flight(task: LoginTask) -> bool: + """判断这条记录是否正被某个批次处理（只看尝试字段，不看 CK 字段）。""" + if task.attempt_status in ATTEMPT_IN_FLIGHT_STATUSES: + return True + return not task.attempt_status and task.status in _LEGACY_IN_FLIGHT_STATUSES + + +def _begin_attempt(task: LoginTask, batch_id: str, message: str) -> None: + """登记一次登录尝试。 + + 只写 attempt_*：账号当前持有的 CK（status/cookie/message）在登录成功前 + 一律不动，所以批次被停止、失败或进程重启都不会让账号失去 CK。 + """ + task.attempt_status = "pending" + task.attempt_message = message + task.attempt_batch_id = batch_id + task.attempt_at = None + + +def _mark_attempt_failed( + task: LoginTask, message: str, *, status: str = "failed" +) -> None: + """记录一次失败/被中断的尝试；已经持有 CK 的记录保持原样。""" + completed_at = datetime.now(UTC) + task.attempt_status = status if status in {"failed", "error"} else "failed" + task.attempt_message = message + task.attempt_at = completed_at + if not task.cookie: + # 还没有任何 CK 需要保护：保持旧的失败语义，便于复用记录与页面统计。 + task.status = task.attempt_status + task.message = message + task.finished_at = completed_at + + +def _write_login_success( + task: LoginTask, *, cookie: str, message: str, batch_id: str +) -> None: + """登录成功后把新 CK 写回记录：这是唯一替换旧 CK 的入口。""" + completed_at = datetime.now(UTC) + task.status = "success" + task.cookie = cookie + task.message = message + task.batch_id = batch_id + task.finished_at = completed_at + task.attempt_status = "success" + task.attempt_message = message + task.attempt_batch_id = batch_id + task.attempt_at = completed_at + + class LoginBatchRunner: """批量登录执行器，在线程中运行，通过 ThreadPoolExecutor 并发登录多个账号。""" @@ -175,14 +237,12 @@ class LoginBatchRunner: task_id: int, message: str = "重新登录已停止，旧 Cookie 已保留", ) -> None: - """将未开始或已中断的重登任务收敛为保留旧 CK 的终态。""" + """把未开始或已中断的重登尝试收敛为终态，旧 CK 原封不动。""" worker_db = SessionLocal() try: task = worker_db.query(LoginTask).filter(LoginTask.id == task_id).first() - if task and task.status in ("relogin_pending", "relogin_running"): - task.status = "relogin_failed" - task.message = message - task.finished_at = datetime.now(UTC) + if task and _attempt_in_flight(task): + _mark_attempt_failed(task, message) worker_db.commit() finally: worker_db.close() @@ -219,8 +279,7 @@ class LoginBatchRunner: """在独立线程中执行单个账号登录，使用独立的 DB 会话。""" if self._stop.is_set(): self._push_log("warning", f"任务已停止，跳过: {acc_info['username']}") - if self.mode == "relogin": - self._mark_relogin_stopped(task_id) + self._mark_relogin_stopped(task_id, "任务已停止，旧 Cookie 已保留") return worker_db = SessionLocal() @@ -229,11 +288,15 @@ class LoginBatchRunner: if not task: return - if self.mode == "relogin": - task.status = "relogin_running" - task.message = "正在重新登录，旧 Cookie 保留中" - else: - task.status = "running" + # 只推进尝试状态；记录持有的旧 CK 在成功前保持不变。 + task.attempt_status = "running" + task.attempt_message = ( + "正在重新登录，旧 Cookie 保留中" + if self.mode == "relogin" + else "正在检测" + if self.mode == "check" + else "正在登录" + ) worker_db.commit() with self._counter_lock: @@ -266,14 +329,14 @@ class LoginBatchRunner: and not proxy_dict ): if self.mode == "relogin": - task.status = "relogin_failed" - task.message = ( - "重新登录失败: 代理不可用: 未配置代理（旧 Cookie 已保留）" + _mark_attempt_failed( + task, + "重新登录失败: 代理不可用: 未配置代理（旧 Cookie 已保留）", ) else: - task.status = "error" - task.message = "代理不可用: 未配置代理" - task.finished_at = datetime.now(UTC) + _mark_attempt_failed( + task, "代理不可用: 未配置代理", status="error" + ) worker_db.commit() self._push_log( "error", f"[{current}] {acc_info['username']} 代理不可用" @@ -315,15 +378,18 @@ class LoginBatchRunner: task.status = status task.cookie = "" task.message = result.message or CHECK_STATUS_MESSAGES[status] + # 检测结果本身就是这次尝试的结论，页面按 attempt_* 展示。 + task.attempt_status = status + task.attempt_message = task.message + task.attempt_at = datetime.now(UTC) + task.finished_at = datetime.now(UTC) level = CHECK_STATUS_LOG_LEVELS.get(status, "info") self._push_log( level, f"[{current}] {acc_info['username']} 检测结果: {task.message}", ) elif result.success: - task.status = "success" - task.cookie = result.cookie - task.message = result.message or "登录成功" + message = result.message or "登录成功" if self.mode == "relogin": check_result = check_douyu_cookie(result.cookie) task.ck_check_status = ( @@ -337,16 +403,26 @@ class LoginBatchRunner: } task.ck_checked_at = check_result["checked_at"] if check_result["valid"]: - task.message = "重新登录成功，Cookie 有效" + message = "重新登录成功，Cookie 有效" + else: + message = f"重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}" + # 唯一替换旧 CK 的入口：登录成功后写回新 CK。 + _write_login_success( + task, + cookie=result.cookie, + message=message, + batch_id=self.batch_id, + ) + if self.mode == "relogin": + if task.ck_check_status == "valid": self._push_log( "success", f"[{current}] {acc_info['username']} 重新登录成功，Cookie 已替换并验证有效", ) else: - task.message = f"重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}" self._push_log( "warning", - f"[{current}] {acc_info['username']} 重新登录成功，但 Cookie 有效性检测失败: {check_result['message']}", + f"[{current}] {acc_info['username']} 重新登录成功，但 Cookie 有效性检测失败: {task.ck_check_result.get('message') if task.ck_check_result else ''}", ) else: self._push_log( @@ -355,18 +431,16 @@ class LoginBatchRunner: ) else: if self.mode == "relogin": - # 重新登录失败时保留旧 Cookie 与成功状态，仅记录失败原因，行不消失 - task.status = "relogin_failed" - task.message = ( - f"重新登录失败: {result.message}（旧 Cookie 已保留）" + # 重登失败不动旧 CK 与成功状态，只记录这次尝试失败 + _mark_attempt_failed( + task, f"重新登录失败: {result.message}（旧 Cookie 已保留）" ) self._push_log( "error", f"[{current}] {acc_info['username']} 重新登录失败: {result.message}", ) else: - task.status = "failed" - task.message = result.message + _mark_attempt_failed(task, result.message or "登录失败") self._push_log( "error", f"[{current}] {acc_info['username']} {action_name}失败: {result.message}", @@ -374,20 +448,17 @@ class LoginBatchRunner: except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 if self.mode == "relogin": - task.status = "relogin_failed" - task.message = f"重新登录异常: {e}（旧 Cookie 已保留）" + _mark_attempt_failed(task, f"重新登录异常: {e}（旧 Cookie 已保留）") self._push_log( "error", f"[{current}] {acc_info['username']} 重新登录异常: {e}" ) else: - task.status = "error" - task.message = str(e) + _mark_attempt_failed(task, str(e), status="error") self._push_log( "error", f"[{current}] {acc_info['username']} {action_name}异常: {e}", ) - task.finished_at = datetime.now(UTC) worker_db.commit() finally: @@ -416,15 +487,17 @@ class LoginBatchRunner: self._push_log("info" if ok else "warning", f"白名单预热: {msg}") def _append_task_info(task: LoginTask, acc: AccountModel): - task.batch_id = batch_id - if self.mode == "relogin": - # 旧 Cookie 仅在 result.success 后由 _execute_one 原子替换。 - task.status = "relogin_pending" - task.message = "等待重新登录，旧 Cookie 已保留" - else: - task.status = "pending" - task.message = "" - task.finished_at = None + # 只登记这次尝试：账号当前持有的 CK（status/cookie/message）在 + # 登录成功前一律不动，批次被停止或失败都不会让账号失去 CK。 + _begin_attempt( + task, + batch_id, + "等待重新登录，旧 Cookie 已保留" + if self.mode == "relogin" + else "等待检测" + if self.mode == "check" + else "等待登录", + ) self.db.flush() task_infos.append( { @@ -491,50 +564,40 @@ class LoginBatchRunner: self._push_log("warning", f"跳过无权账号: {acc.username}") continue - # 一个斗鱼账号只保留一条成功 CK：再次普通登录时更新最新成功记录。 - latest_success_task = ( + # 一个斗鱼账号只保留一条持有 CK 的记录：优先复用它， + # 登录成功后才把新 CK 写回同一条记录。 + task = ( self.db.query(LoginTask) .filter( - LoginTask.account_id == aid, LoginTask.status == "success" + LoginTask.account_id == aid, + cookie_holding_filter(), ) - .order_by(LoginTask.finished_at.desc(), LoginTask.id.desc()) + .order_by(LoginTask.id.desc()) .first() ) - 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() - ) - for duplicate_task in duplicate_success_tasks: - self.db.delete(duplicate_task) - task = latest_success_task - else: - # 复用该账号最近一条失败任务记录，避免重复产生多条失败历史。 - existing_task = ( + if task is None: + # 复用最近一条失败记录，避免重复产生多条失败历史。 + task = ( self.db.query(LoginTask) .filter( LoginTask.account_id == aid, LoginTask.status.in_(["failed", "error"]), + or_( + LoginTask.cookie.is_(None), + LoginTask.cookie == "", + ), ) .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) + if task is None: + 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) @@ -551,9 +614,12 @@ class LoginBatchRunner: for item_index, item in enumerate(task_infos): if self._stop.is_set(): self._push_log("warning", "任务已停止，跳过剩余账号") - if self.mode == "relogin": - for pending_item in task_infos[item_index:]: - self._mark_relogin_stopped(pending_item["task_id"]) + # 所有模式都要把没轮到的尝试收敛掉，否则会一直停在 + # 「等待中」；已有 CK 的记录本身不受影响。 + for pending_item in task_infos[item_index:]: + self._mark_relogin_stopped( + pending_item["task_id"], "任务已停止，旧 Cookie 已保留" + ) break future = executor.submit( self._execute_one, @@ -624,51 +690,50 @@ class BatchRegistry: batch_registry = BatchRegistry() +def _converge_interrupted_attempts(db: Session, message: str) -> int: + """把遗留的在途尝试收敛为失败。 + + 只写 attempt_*；对已经没有 CK 的记录才顺带落一个失败终态。返回收敛条数。 + """ + tasks = ( + db.query(LoginTask) + .filter( + or_( + LoginTask.attempt_status.in_(ATTEMPT_IN_FLIGHT_STATUSES), + LoginTask.status.in_(_LEGACY_IN_FLIGHT_STATUSES), + ) + ) + .all() + ) + cleaned = 0 + for task in tasks: + if not _attempt_in_flight(task): + continue + _mark_attempt_failed(task, message) + cleaned += 1 + if cleaned: + db.commit() + return cleaned + + def cleanup_orphan_relogin_tasks( db: Session, message: str = "重新登录已中断（服务重启），旧 Cookie 已保留", ) -> int: - """服务重启后收敛遗留重登状态，避免页面永久显示重登中。""" - tasks = ( - db.query(LoginTask) - .filter(LoginTask.status.in_(("relogin_pending", "relogin_running"))) - .all() - ) - if not tasks: - return 0 - - finished_at = datetime.now(UTC) - for task in tasks: - task.status = "relogin_failed" - task.message = message - task.finished_at = finished_at - db.commit() - return len(tasks) + """服务重启后收敛遗留的重登尝试，旧 CK 原封不动。""" + return _converge_interrupted_attempts(db, message) def cleanup_orphan_login_tasks( db: Session, message: str = "登录已中断（服务重启），旧 Cookie 已保留", ) -> int: - """服务重启后收敛遗留的普通登录状态。 + """服务重启后收敛遗留的登录尝试。 - 普通登录批次停止或进程重启时，未轮到的记录会永远停在 pending/running， - 页面上一直显示「排队中」，旧 CK 也失去终态说明。统一收敛为 failed 并 - 保留 Cookie，账号仍然可见可重登。 + 在途尝试只体现在 attempt_* 上，收敛它不会碰到账号已有的 CK；只有本来 + 就没有 CK 的记录才会顺带标记失败，避免页面永久显示「排队中」。 """ - tasks = ( - db.query(LoginTask).filter(LoginTask.status.in_(("pending", "running"))).all() - ) - if not tasks: - return 0 - - finished_at = datetime.now(UTC) - for task in tasks: - task.status = "failed" - task.message = message - task.finished_at = finished_at - db.commit() - return len(tasks) + return _converge_interrupted_attempts(db, message) # 在模块末尾导入 SessionLocal（避免循环导入）