From d11fa1032d13472d2383a56635c9bdd1008a3ac7 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Wed, 7 Oct 2026 12:54:57 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(login-tasks):=20=E6=8C=89=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E5=AF=BC=E5=87=BA=20CSV=20+=20=E7=99=BB=E5=BD=95?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E5=86=99=E8=BF=9B=E8=B4=A6=E5=8F=B7=E7=94=9F?= =?UTF-8?q?=E5=91=BD=E5=91=A8=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、失败入账号一生（此前完全缺失） 生产实测：`login_ok` 1,456 条、`relogin_failed` 73 条，但**没有 login_failed**， 而 login_tasks 里 attempt_status='failed' 有 247 条 —— 这 247 次失败在账号时间线上 一条痕迹都没有，「这账号为什么一直登录不上」在账号页上无从回答。 - 新增事件类型 login_failed，在 `_mark_attempt_failed` 里写入 - **被停止 / 被中断不写**：分型把「已停止/中断/服务重启」归到 interrupted， 那不是账号的问题 —— 否则运营按一次停止就能给几十个账号刷满假失败 - 事件带分型（captcha/credential/email_code/email_quota/network）， 同时落进 detail.kind，以后可直接按「验证码失败」筛账号 - 幂等键 login_task:{id}:failed:{ts}，与成功路径同构 二、失败分型下沉成共用 service `classify_failure` 原在 automation/jobs/ck_relogin.py，手动登录也要用。 挪到 services/login_failure.py（job 改为 import，`ck_relogin.classify_failure` 这个名字保留，旧调用方与用例不受影响），并提供型别中文标签与 `is_account_fault` 判据 —— 与仓库既有的「service 不该依赖 job」分层一致。 三、按状态导出 CSV - 新增 GET /api/login/tasks/export：不分页，导出当前筛选的**全部**匹配记录； 列：账号/状态/失败原因/失败分型/批次/最近尝试时间(UTC)/标签/归属客服 - 权限与筛选与列表**完全共用**，客服只导出自己名下账号 - 带 UTF-8 BOM（不加 Excel 打开中文表头是乱码），导出动作写审计 四、修一个口径不一致的真 bug 页面「失败 N」= failed + error 之和，但筛选只认 failed —— 所以截图里「失败 389」点进去只有 386 条（差的就是那几条 error）； 「清空失败」同样漏删 error。现在把状态组收敛成一份 `_apply_status_filter`， 列表 / 导出 / 批量删除三处共用。 测试：684 passed, 6 skipped（本次新增 13 个）；三条关键断言做了反向验证 （去掉 stopped 判断 / 失败组去掉 error / 去掉 BOM，对应用例均变红）。 --- tests/test_login_failure_events.py | 269 ++++++++++++++++++ web/backend/routers/login.py | 174 +++++++++-- .../services/account_events_service.py | 7 + .../services/automation/jobs/ck_relogin.py | 24 +- web/backend/services/login_failure.py | 68 +++++ web/backend/services/login_service.py | 44 ++- web/frontend/src/api/login.ts | 14 + web/frontend/src/pages/LoginTasksPage.tsx | 47 ++- 8 files changed, 598 insertions(+), 49 deletions(-) create mode 100644 tests/test_login_failure_events.py create mode 100644 web/backend/services/login_failure.py diff --git a/tests/test_login_failure_events.py b/tests/test_login_failure_events.py new file mode 100644 index 0000000..ed5b548 --- /dev/null +++ b/tests/test_login_failure_events.py @@ -0,0 +1,269 @@ +"""登录失败进账号时间线 + 登录任务按状态导出 CSV。 + +两条都是这次新加的，各有必须守住的东西： + +- **失败入流水**：真失败要写，被停止/被中断**不能**写 —— + 否则运营按一次「停止」就能给几十个账号刷满假失败； +- **导出**：口径必须和列表完全一致（尤其「失败 = failed + error」这个组）， + 而且要带 UTF-8 BOM，不然 Excel 打开中文乱码。 +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime + +from tests.support.db import get_test_engine, make_test_session +from web.backend.models import Account, AccountEvent, LoginTask, User +from web.backend.routers.login import LOGIN_STATUS_LABELS, export_tasks, list_tasks +from web.backend.services import account_events_service as svc +from web.backend.services.login_service import _mark_attempt_failed + +NOW = datetime(2026, 10, 7, 18, 0, 0, tzinfo=UTC) + + +async def _read_body(response) -> str: + """把 StreamingResponse 的内容读成字符串。""" + chunks: list[bytes] = [] + async for chunk in response.body_iterator: + chunks.append(chunk if isinstance(chunk, bytes) else str(chunk).encode("utf-8")) + return b"".join(chunks).decode("utf-8") + + +def _body(response) -> str: + return asyncio.run(_read_body(response)) + + +class _Base: + def setup_method(self): + self.engine = get_test_engine() + self.session = make_test_session(self.engine) + self.admin = User(username="lf-admin", password_hash="h", role="super_admin") + self.support = User(username="lf-kf", password_hash="h", role="support") + self.session.add_all([self.admin, self.support]) + self.session.flush() + self.admin_id = self.admin.id + self.support_id = self.support.id + + def teardown_method(self): + self.session.close() + + def make_account(self, name: str, *, assigned_to=None) -> Account: + account = Account( + username=name, + password="p", + email="e@e.com", + email_password="ep", + assigned_to=assigned_to if assigned_to is not None else self.support_id, + ) + self.session.add(account) + self.session.commit() + return account + + def make_task( + self, + account: Account, + *, + attempt_status: str = "failed", + status: str = "failed", + message: str = "", + batch_id: str = "b1", + ) -> LoginTask: + task = LoginTask( + batch_id=batch_id, + account_id=account.id, + status=status, + attempt_status=attempt_status, + attempt_message=message, + attempt_batch_id=batch_id, + message=message, + created_by=self.support_id, + ) + self.session.add(task) + self.session.commit() + return task + + def events(self, account_id: int) -> list[AccountEvent]: + self.session.rollback() + return ( + self.session.query(AccountEvent) + .filter( + AccountEvent.account_id == account_id, + AccountEvent.event_type == svc.EVENT_LOGIN_FAILED, + ) + .all() + ) + + +class TestLoginFailureEvents(_Base): + def test_real_failure_writes_account_event(self): + account = self.make_account("lf-1") + task = self.make_task(account, message="极验验证失败: slide") + _mark_attempt_failed(task, "极验验证失败: slide") + self.session.commit() + + events = self.events(account.id) + assert len(events) == 1 + event = events[0] + assert event.event_type == "login_failed" + assert event.source == "login" + # 分型要落进 detail —— 以后按「验证码失败」筛账号靠它 + assert event.detail["kind"] == "captcha" + assert event.detail["kind_label"] == "极验/验证码拦截" + assert "极验" in event.title + + def test_stopped_does_not_write_event(self): + """用户按「停止」不是账号的问题，不该污染账号一生。""" + account = self.make_account("lf-2") + task = self.make_task(account) + _mark_attempt_failed(task, "任务已停止，旧 Cookie 已保留") + self.session.commit() + assert self.events(account.id) == [] + + def test_restart_interrupted_does_not_write_event(self): + account = self.make_account("lf-3") + task = self.make_task(account) + _mark_attempt_failed(task, "登录已中断（服务重启），旧 Cookie 已保留") + self.session.commit() + assert self.events(account.id) == [] + + def test_credential_failure_writes_event(self): + account = self.make_account("lf-4") + task = self.make_task(account) + _mark_attempt_failed(task, "登录失败: 账号密码错误") + self.session.commit() + assert self.events(account.id)[0].detail["kind"] == "credential" + + def test_same_attempt_is_idempotent(self): + """同一次尝试重复收敛，只该留一条事件（source_ref 幂等）。""" + account = self.make_account("lf-5") + task = self.make_task(account) + _mark_attempt_failed(task, "极验验证失败: slide") + self.session.commit() + _mark_attempt_failed(task, "极验验证失败: slide") + self.session.commit() + assert len(self.events(account.id)) == 1 + + def test_event_does_not_break_when_task_has_no_session(self): + """纯内存构造的任务（没有 session）不能因为写事件而炸掉主流程。""" + account = Account( + username="lf-detached", + password="p", + email="e@e.com", + email_password="ep", + ) + task = LoginTask( + batch_id="b", + account_id=1, + status="failed", + attempt_status="failed", + created_by=1, + ) + task.account = account + _mark_attempt_failed(task, "极验验证失败: slide") # 不应抛异常 + + +class TestFailedGroupFilter(_Base): + def test_failed_filter_includes_error(self): + """页面「失败 N」= failed + error；筛选也必须一致，否则数对不上。""" + ok = self.make_account("fg-ok") + bad = self.make_account("fg-bad") + err = self.make_account("fg-err") + self.make_task(ok, attempt_status="success", status="success") + self.make_task(bad, attempt_status="failed", status="failed") + self.make_task(err, attempt_status="error", status="error") + + failed = list_tasks( + batch_id=None, + page=1, + page_size=50, + status="failed", + db=self.session, + current=self.admin, + ) + assert failed["total"] == 2, "error 必须算进「失败」这一组" + + success = list_tasks( + batch_id=None, + page=1, + page_size=50, + status="success", + db=self.session, + current=self.admin, + ) + assert success["total"] == 1 + + +class TestExportCsv(_Base): + def _export(self, *, status="", current=None, batch_id=None): + return _body( + export_tasks( + batch_id=batch_id, + status=status, + db=self.session, + current=current or self.admin, + ) + ) + + def test_has_bom_and_header(self): + self.make_account("ex-1") + body = self._export() + assert body.startswith("\ufeff"), "没有 BOM，Excel 打开中文会乱码" + header = body.lstrip("\ufeff").splitlines()[0] + assert header.startswith("账号,状态,失败原因,失败分型,批次,最近尝试时间") + assert "标签" in header and "归属客服" in header + + def test_rows_carry_account_status_reason_and_kind(self): + account = self.make_account("ex-2") + self.make_task(account, message="极验验证失败: slide") + lines = self._export(status="failed").lstrip("\ufeff").splitlines() + assert len(lines) == 2 + row = lines[1].split(",") + assert row[0] == "ex-2" + assert row[1] == "失败" + assert "极验" in row[2] + assert row[3] == "极验/验证码拦截" + assert row[7] == "lf-kf" # 归属客服：运营拿去分派 + + def test_failed_export_includes_error_rows(self): + """导出「失败」必须把 error 一起带上，否则运营照 CSV 补登会漏人。""" + bad = self.make_account("ex-3") + err = self.make_account("ex-4") + self.make_task( + bad, attempt_status="failed", status="failed", message="密码错误" + ) + self.make_task(err, attempt_status="error", status="error", message="网络超时") + lines = self._export(status="failed").lstrip("\ufeff").splitlines() + assert len(lines) == 3 + names = {line.split(",")[0] for line in lines[1:]} + assert names == {"ex-3", "ex-4"} + + def test_status_is_rendered_in_chinese(self): + account = self.make_account("ex-5") + self.make_task(account, attempt_status="success", status="success") + lines = self._export(status="success").lstrip("\ufeff").splitlines() + assert lines[1].split(",")[1] == LOGIN_STATUS_LABELS["success"] + + def test_kind_column_empty_for_non_failure(self): + account = self.make_account("ex-6") + self.make_task(account, attempt_status="success", status="success") + lines = self._export(status="success").lstrip("\ufeff").splitlines() + assert lines[1].split(",")[3] == "" + + def test_support_only_exports_own_accounts(self): + mine = self.make_account("ex-mine", assigned_to=self.support_id) + other = self.make_account("ex-other", assigned_to=self.admin_id) + self.make_task(mine) + self.make_task(other) + + body = self._export(status="failed", current=self.support) + names = {line.split(",")[0] for line in body.lstrip("\ufeff").splitlines()[1:]} + assert names == {"ex-mine"} + + all_names = { + line.split(",")[0] + for line in self._export(status="failed", current=self.admin) + .lstrip("\ufeff") + .splitlines()[1:] + } + assert all_names == {"ex-mine", "ex-other"} diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index 98cfb2b..4e5ef10 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -1,7 +1,10 @@ """登录任务路由 + WebSocket 实时日志""" import asyncio +import csv +import io import threading +from datetime import UTC, datetime from fastapi import ( APIRouter, @@ -10,6 +13,7 @@ from fastapi import ( WebSocket, WebSocketDisconnect, ) +from fastapi.responses import StreamingResponse from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session, load_only @@ -20,6 +24,7 @@ from ..models import ProxyConfig as ProxyConfigModel from ..permissions import get_user_permissions, user_has_permission from ..schemas import LoginBatchRequest, LoginTaskOut from ..services.audit_service import record_audit +from ..services.login_failure import classify_failure, failure_kind_label from ..services.login_service import LoginBatchRunner, batch_registry router = APIRouter(prefix="/api/login", tags=["登录任务"]) @@ -37,6 +42,53 @@ def _effective_attempt_status(): ) +#: 页面上「失败」是一个**组**：库里 `failed` 与 `error` 分两列存，但用户眼里都是失败。 +#: 汇总口径早就是两者相加（见 `_login_task_summary`），筛选却只认 `failed` —— +#: 于是「失败 389」点进去只剩 386 条（差的就是那几条 `error`）。 +#: 这里统一成组，列表 / 导出 / 批量删除共用一份，避免三处各写一遍再漂移。 +STATUS_GROUPS: dict[str, tuple[str, ...]] = { + "failed": ("failed", "error"), +} + +#: 导出 CSV 用的中文状态。与前端 `LoginTasksPage.STATUS_LABELS` 同源同义； +#: 后端要自己出一份是因为 CSV 在服务端生成，拿不到前端的表。 +LOGIN_STATUS_LABELS: dict[str, str] = { + "pending": "等待中", + "running": "登录中", + "success": "成功", + "failed": "失败", + "error": "异常", + "stopped": "已停止", + "account_cancelled": "账号已注销", + "password_wrong": "账号密码错误", + "account_unverified": "账号未认证", + "account_verified": "账号已认证", + "account_auth_unknown": "认证状态未知", +} + + +def _status_values(status: str) -> tuple[str, ...]: + """把页面上的状态名展开成库里真正的取值集合。""" + return STATUS_GROUPS.get(status, (status,)) + + +def _apply_status_filter(query, status: str | None): + """按「最近一次尝试的状态」筛选（与列表序列化口径一致）。 + + 优先看 `attempt_status`，为空时回退到 `status` —— 两者分开是刻意的： + `status` 只表示这条记录是否持有可用 CK（见 `_effective_attempt_status`）。 + """ + if not status: + return query + wanted = _status_values(status) + return query.filter( + or_( + and_(LoginTask.attempt_status != "", LoginTask.attempt_status.in_(wanted)), + and_(LoginTask.attempt_status == "", LoginTask.status.in_(wanted)), + ) + ) + + def _login_task_summary(query): """按状态汇总登录任务，避免概览页拉完整任务列表。""" rows = ( @@ -213,17 +265,7 @@ def list_tasks( 传 `page` 时同时返回 `total`，是为了让前端能「全选所有筛选结果」—— 否则跨页选择无从下手，而任务量早就过了单页能装下的规模。 """ - query = _visible_tasks_query(db, current, batch_id) - if status: - # 前端看到的是「最近一次尝试」的状态，与列表序列化口径保持一致。 - query = query.filter( - or_( - and_( - LoginTask.attempt_status != "", LoginTask.attempt_status == status - ), - and_(LoginTask.attempt_status == "", LoginTask.status == status), - ) - ) + query = _apply_status_filter(_visible_tasks_query(db, current, batch_id), status) ordered = query.order_by(LoginTask.id.desc()) @@ -242,6 +284,100 @@ def list_tasks( } +def _fmt_dt(value) -> str: + """导出用的时间格式。 + + 统一输出 **UTC**，并在表头写明 —— 服务端猜本地时区迟早会错， + 而错的时间在人工筛选时是看不出来的。 + """ + if value is None: + return "" + return value.strftime("%Y-%m-%d %H:%M:%S") + + +@router.get("/tasks/export") +def export_tasks( + batch_id: str | None = None, + status: str | None = None, + db: Session = Depends(get_db), + current: User = Depends(get_current_user), +): + """按当前筛选导出登录任务 CSV（账号名 / 状态 / 失败原因 / 分型 / 批次 / 时间 / 标签 / 归属客服）。 + + 几个刻意的选择： + + - **不分页**：导出的用途就是「拿去 Excel 人工筛」，只给一页没有意义； + - **筛选与权限跟列表完全共用**（`_visible_tasks_query` + `_apply_status_filter`）， + 所以「客服只能导出自己名下账号」这条不需要在这里重写一遍； + - **带 UTF-8 BOM**：不加的话 Excel 打开中文表头是乱码； + - 失败分型复用登录失败那套（`login_failure.classify_failure`）， + 与账号时间线上写的是同一份判据，不会出现「页面上算 A、导出算 B」。 + """ + query = _apply_status_filter(_visible_tasks_query(db, current, batch_id), status) + rows = query.order_by(LoginTask.id.desc()).all() + + account_ids = [t.account_id for t in rows] + info: dict[int, tuple[str, str, str]] = {} + if account_ids: + for acc_id, username, tag, owner in ( + db.query(Account.id, Account.username, Account.tag, User.username) + .outerjoin(User, Account.assigned_to == User.id) + .filter(Account.id.in_(account_ids)) + .all() + ): + info[acc_id] = (username or "", tag or "", owner or "") + + output = io.StringIO() + writer = csv.writer(output) + writer.writerow( + [ + "账号", + "状态", + "失败原因", + "失败分型", + "批次", + "最近尝试时间(UTC)", + "标签", + "归属客服", + ] + ) + for task in rows: + attempt_status = task.attempt_status or "" + effective = attempt_status or task.status or "" + message = (task.attempt_message if attempt_status else "") or task.message or "" + kind = classify_failure(message) if effective in {"failed", "error"} else "" + username, tag, owner = info.get(task.account_id, ("", "", "")) + writer.writerow( + [ + username, + LOGIN_STATUS_LABELS.get(effective, effective), + message, + failure_kind_label(kind) if kind else "", + task.attempt_batch_id or task.batch_id or "", + _fmt_dt(task.attempt_at or task.finished_at or task.created_at), + tag, + owner, + ] + ) + + content = "\ufeff" + output.getvalue() + stamp = datetime.now(UTC).strftime("%Y%m%d_%H%M") + filename = f"login_tasks_{status or 'all'}_{stamp}.csv" + record_audit( + db, + current, + action="login:tasks:export", + target=f"status={status or 'all'}", + detail={"rows": len(rows), "batch_id": batch_id or ""}, + ) + db.commit() + return StreamingResponse( + iter([content]), + media_type="text/csv; charset=utf-8", + headers={"Content-Disposition": f"attachment; filename={filename}"}, + ) + + @router.get("/tasks/summary") def tasks_summary( db: Session = Depends(get_db), @@ -336,17 +472,11 @@ def delete_tasks( elif all_matching: # 带 join 的 query 不能直接 bulk delete（SQLAlchemy 会拒绝）， # 所以先按权限与筛选条件查出 id，再按主键删。 - query = _visible_tasks_query(db, current, batch_id) - if status: - query = query.filter( - or_( - and_( - LoginTask.attempt_status != "", - LoginTask.attempt_status == status, - ), - and_(LoginTask.attempt_status == "", LoginTask.status == status), - ) - ) + # 状态筛选与列表/导出共用一份 —— 「清空失败」以前只删 `failed`、 + # 漏掉 `error`，删完与页面上那个「失败 N」的数还对不上。 + query = _apply_status_filter( + _visible_tasks_query(db, current, batch_id), status + ) ids = [row[0] for row in query.with_entities(LoginTask.id).all()] else: raise HTTPException(status_code=400, detail="请指定任务ID或 all_matching=true") diff --git a/web/backend/services/account_events_service.py b/web/backend/services/account_events_service.py index 9b74b80..0794137 100644 --- a/web/backend/services/account_events_service.py +++ b/web/backend/services/account_events_service.py @@ -43,6 +43,13 @@ from sqlalchemy.orm import Session # 命名保持「名词_状态」风格，便于前端归类与筛选。 EVENT_IMPORTED = "imported" # 账号导入（读 accounts.created_at，一般不落表） EVENT_LOGIN_OK = "login_ok" # 登录成功（含新 CK） +#: 登录/重登**失败**（手动与批量登录都走这条）。 +#: +#: 2026-10-07 补：此前只有成功会写事件，失败只改 `login_tasks` 行 —— +#: 生产实测 247 次登录失败在账号时间线上**一条痕迹都没有**（成功却有 1,456 条）， +#: 「这个账号为什么一直登录不上」在账号页上无从回答。 +#: ⚠️ **被停止 / 被中断不算失败**，那种情况不写（见 `services/login_failure`）。 +EVENT_LOGIN_FAILED = "login_failed" EVENT_CK_INVALID = "ck_invalid" # CK 判定失效 EVENT_CK_CLEARED = "ck_cleared" # CK 被清除（人工清除 / 账号删除前的清理） EVENT_RELOGIN_OK = "relogin_ok" # 自动重登成功 diff --git a/web/backend/services/automation/jobs/ck_relogin.py b/web/backend/services/automation/jobs/ck_relogin.py index f38d6cf..330bc9a 100644 --- a/web/backend/services/automation/jobs/ck_relogin.py +++ b/web/backend/services/automation/jobs/ck_relogin.py @@ -32,6 +32,7 @@ from ...account_lifecycle import ( ) from ...account_value import value_tier_expr from ...douyu_service import cookie_holding_filter, cookie_operable_filter +from ...login_failure import classify_failure from ...login_service import LoginBatchRunner, get_relogin_limits JOB_KEY = "ck_relogin" @@ -54,29 +55,6 @@ AUTO_MAX_TOTAL_TIME = 240 QUOTA_JITTER_MINUTES = 30 -def classify_failure(message: str | None) -> str: - """把重登失败消息映射成分型（见规划 §7.4.3）。 - - 顺序很重要：「邮箱验证码发送次数已达上限」既含「验证码」也含「上限」， - 必须先在 ``email_quota`` 命中，否则会被归到可重试的 ``email_code`` 上， - 导致次日才该重试的账号被按小时反复撞配额。 - """ - text = message or "" - if "上限" in text or "配额" in text: - return "email_quota" - if "极验" in text or "验证" in text and "滑块" in text: - return "captcha" - if "验证码" in text: - return "email_code" - if "中断" in text or "已停止" in text or "任务已停止" in text: - return "interrupted" - if "密码" in text or "账号不存在" in text or "缺少密码或邮箱" in text: - return "credential" - # 请求失败/超时/连接异常，以及未知文案：都按可重试处理。 - # 宁可多给一次机会（有每日上限兜底），也不要把可能只是抖动的账号推给人工。 - return "network" - - def _cutoff(now: datetime) -> datetime: return now diff --git a/web/backend/services/login_failure.py b/web/backend/services/login_failure.py new file mode 100644 index 0000000..c22a729 --- /dev/null +++ b/web/backend/services/login_failure.py @@ -0,0 +1,68 @@ +"""登录失败分型：把「失败消息」映射成稳定的型别。 + +## 为什么要单独一个模块 + +这个分类原先只活在 `automation/jobs/ck_relogin.py`（自动重登作业）里。但 +**手动/批量登录**同样需要它： + +1. 账号时间线要写「登录失败」事件，写的时候得带型别，否则以后想筛 + 「验证码失败的账号」只能去猜中文； +2. 导出 CSV 要给运营一列可直接筛选的型别。 + +放 service 而不是留在 job 里，是沿用仓库已有的分层决定 +（`MANUAL_KINDS` 当初就是从 job 挪到 `account_lifecycle.py` 的， +注释写着「service 不该反过来依赖 job」）。 + +## 一条重要推论：`interrupted` ≠ 失败 + +分型把「已停止 / 被中断 / 服务重启」都归到 `interrupted`。**这不是账号的失败**： +调用方（`_mark_attempt_failed`）据此决定**不往账号时间线写失败事件** —— +否则运营按一次停止就能把几十个账号的时间线刷满假失败。 +""" + +from __future__ import annotations + +#: 型别 → 中文标签（导出 CSV 与页面上用同一份，避免两处各写一遍漂移）。 +FAILURE_KIND_LABELS: dict[str, str] = { + "captcha": "极验/验证码拦截", + "credential": "账号凭证问题", + "email_code": "邮箱取码失败", + "email_quota": "邮箱配额已满", + "network": "网络/未知异常", + "interrupted": "被中断（停止/重启）", +} + +#: 不写进账号时间线的型别：这些是**人的动作或进程重启**，不是账号状况。 +NOT_AN_ACCOUNT_FAULT: frozenset[str] = frozenset({"interrupted"}) + + +def classify_failure(message: str | None) -> str: + """把重登/登录失败消息映射成分型（见规划 §7.4.3）。 + + 顺序很重要：「邮箱验证码发送次数已达上限」既含「验证码」也含「上限」， + 必须先在 ``email_quota`` 命中，否则会被归到可重试的 ``email_code`` 上， + 导致次日才该重试的账号被按小时反复撞配额。 + """ + text = message or "" + if "上限" in text or "配额" in text: + return "email_quota" + if "极验" in text or "验证" in text and "滑块" in text: + return "captcha" + if "验证码" in text: + return "email_code" + if "中断" in text or "已停止" in text or "任务已停止" in text: + return "interrupted" + if "密码" in text or "账号不存在" in text or "缺少密码或邮箱" in text: + return "credential" + # 请求失败/超时/连接异常，以及未知文案：都按可重试处理。 + # 宁可多给一次机会（有每日上限兜底），也不要把可能只是抖动的账号推给人工。 + return "network" + + +def is_account_fault(kind: str | None) -> bool: + """这个型别算不算「账号出了问题」——不算就不该写进账号一生。""" + return (kind or "") not in NOT_AN_ACCOUNT_FAULT + + +def failure_kind_label(kind: str | None) -> str: + return FAILURE_KIND_LABELS.get(kind or "", "") diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 780261e..2643903 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -12,7 +12,7 @@ from typing import cast from loguru import logger from sqlalchemy import and_, func, or_ -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, object_session from core.douyu import DouyuLogin, create_login_api_strategy from core.douyu.login import AccountLike @@ -21,10 +21,16 @@ from core.douyu.proxy_fetcher import ProxyFetcher from ..models import Account as AccountModel from ..models import LoginTask from ..models import ProxyConfig as ProxyConfigModel -from .account_events_service import EVENT_LOGIN_OK, EVENT_RELOGIN_OK, emit +from .account_events_service import ( + EVENT_LOGIN_FAILED, + EVENT_LOGIN_OK, + EVENT_RELOGIN_OK, + emit, +) from .account_lifecycle import on_ck_verdict, on_login_success from .cookie_check_service import check_douyu_cookie from .douyu_service import LOGIN_ATTEMPT_IN_FLIGHT_STATUSES +from .login_failure import classify_failure, failure_kind_label, is_account_fault CHECK_STATUS_MESSAGES = { "account_cancelled": "账号已注销", @@ -121,6 +127,15 @@ def _mark_attempt_failed( 这里读的是 `task.account.cookie`（CK 自 2026-10-04 起挂在账号上）， 而不是这条登录记录 —— 记录不再持有凭证。 + + 顺带往**账号时间线**写一条失败事件（这是账号级的资产变化：这个账号登录不上）。 + 两条边界： + + - **被停止 / 被中断不写**：`classify_failure` 把它们归到 `interrupted`， + 那是人的动作或进程重启，不是账号出了状况。不挡住的话， + 运营按一次「停止」就能给几十个账号的时间线刷满假失败。 + - **幂等**：一次尝试一条（`source_ref` 带 task id + 尝试时刻）， + 失败重试会新建记录或新 attempt，因此是新的一条 —— 这是对的，那是新的事实。 """ completed_at = datetime.now(UTC) task.attempt_status = status if status in {"failed", "error"} else "failed" @@ -133,6 +148,31 @@ def _mark_attempt_failed( task.message = message task.finished_at = completed_at + kind = classify_failure(message) + if not is_account_fault(kind): + return + db = object_session(task) + if db is None or account is None: + # 没有会话（纯内存构造的用例）或没有账号：跳过旁路事件，绝不影响主流程。 + return + emit( + db, + account_id=account.id, + event_type=EVENT_LOGIN_FAILED, + title=f"登录失败：{message}", + occurred_at=completed_at, + detail={ + "message": message, + "kind": kind, + "kind_label": failure_kind_label(kind), + "attempt_status": task.attempt_status, + }, + source="login", + subject_type="login", + subject_key=kind, + source_ref=f"login_task:{task.id}:failed:{int(completed_at.timestamp())}", + ) + def _write_login_success( db: Session, diff --git a/web/frontend/src/api/login.ts b/web/frontend/src/api/login.ts index d5727dd..a81c0eb 100644 --- a/web/frontend/src/api/login.ts +++ b/web/frontend/src/api/login.ts @@ -26,6 +26,20 @@ export const loginApi = { listTasksPaged: (params: { page: number; page_size: number; batch_id?: string; status?: string }) => api.get('/login/tasks', { params: { ...params, batch_id: params.batch_id ?? '' } }), tasksSummary: () => api.get('/login/tasks/summary'), + /** + * 导出登录任务 CSV（按当前筛选，不分页）。 + * + * `status` 传 'failed' 时后端会展开成 failed + error —— 页面上「失败」 + * 本来就是一个组，导出与列表口径必须一致。 + */ + exportTasks: (params: { batch_id?: string; status?: string }) => + api.get('/login/tasks/export', { + responseType: 'blob', + params: { + batch_id: params.batch_id ?? '', + status: params.status ?? '', + }, + }), stop: (batch_id: string) => api.post(`/login/stop/${batch_id}`), deleteTask: (id: number) => api.delete(`/login/tasks/${id}`), deleteTasks: (ids: number[]) => api.delete(`/login/tasks`, { params: { task_ids: ids.join(',') } }), diff --git a/web/frontend/src/pages/LoginTasksPage.tsx b/web/frontend/src/pages/LoginTasksPage.tsx index 1d97306..d3eae32 100644 --- a/web/frontend/src/pages/LoginTasksPage.tsx +++ b/web/frontend/src/pages/LoginTasksPage.tsx @@ -1,11 +1,11 @@ import { useEffect, useState, useMemo, useCallback, useRef } from 'react'; import { - Table, Button, Select, Tag, Space, InputNumber, Tooltip, Popconfirm, theme, Modal, Form, + Table, Button, Select, Tag, Space, InputNumber, Tooltip, Popconfirm, theme, Modal, Form, Dropdown, } from 'antd'; import { message } from '../utils/antdMessage'; import { PlayCircleOutlined, StopOutlined, FilterOutlined, ReloadOutlined, DeleteOutlined, - SettingOutlined, SafetyCertificateOutlined, + SettingOutlined, SafetyCertificateOutlined, DownloadOutlined, } from '@ant-design/icons'; import { accountApi, loginApi, type AccountItem, type LoginTaskItem, type TaskSummary } from '../api/modules'; import RealtimeLogPanel from '../components/RealtimeLogPanel'; @@ -391,6 +391,34 @@ export default function LoginTasksPage() { } }; + /** + * 导出登录任务 CSV。 + * + * `status` 传 undefined 表示「当前筛选」—— 与列表看到的完全是同一批； + * 传 'all' 表示不受筛选限制导出全部（后端把空串当不加条件）。 + * 导出的是**全部匹配记录**，不是当前页。 + */ + const handleExport = async (scope: 'current' | 'all' | 'success' | 'failed') => { + const status = + scope === 'current' ? taskStatus : scope === 'all' ? '' : scope; + try { + const blob = await loginApi.exportTasks({ + batch_id: batchId || undefined, + status, + }); + const url = URL.createObjectURL(blob instanceof Blob ? blob : new Blob([blob])); + const a = document.createElement('a'); + a.href = url; + const label = scope === 'current' ? (taskStatus || 'all') : scope; + a.download = `login_tasks_${label}.csv`; + a.click(); + URL.revokeObjectURL(url); + message.success('已导出，可用 Excel 打开（已带 BOM，中文不乱码）'); + } catch (e: unknown) { + message.error(getErrorMessage(e)); + } + }; + // 一键清空当前筛选下的失败任务（跨页，不用先翻完所有页去勾） const handleClearFailed = async () => { try { @@ -649,6 +677,21 @@ export default function LoginTasksPage() { { value: 'running', label: '进行中' }, ]} /> + void handleExport(key as 'current' | 'all' | 'success' | 'failed'), + }} + > + +
{canBatch && ( <> From 05b2bf8579a5e5dc2a97df7573d413b20902ee7e Mon Sep 17 00:00:00 2001 From: yml2213 Date: Wed, 7 Oct 2026 13:20:33 +0800 Subject: [PATCH 2/2] =?UTF-8?q?docs(agents):=20=E4=BF=AE=E6=AD=A3=E5=88=86?= =?UTF-8?q?=E6=94=AF=E7=BA=A6=E5=AE=9A=EF=BC=8C=E8=A1=A5=E4=B8=8A=E8=BF=99?= =?UTF-8?q?=E8=BD=AE=E5=AE=9E=E8=B7=B5=E4=B9=B0=E7=9A=84=E6=95=99=E8=AE=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - **删掉已作废的约定**「推完 main 再 `dev merge --ff-only main`」： dev 现在与 main 已分叉（上面有未上线的账号自动分配功能），ff-only 必然失败。 改为：`main` = 已发布，新功能从 main 开 `feat/*`；要同步 dev 只能真合并且需显式确认。 - 新增一批实证教训：提交/推送的钩子要求（`pre-commit` 查的是 `ruff format`， 只跑 `ruff check` 会被拦）、测试库 `create_all` 不会 ALTER 已有表（症状 `Unknown column`）、`dev.sh` 会自动应用迁移导致切分支后应用起不来、 生产部署的完整流程与三项前置检查。 - 坑速查 4 → 8 条：新增 `pgrep -f` 自匹配白等、`set -e` + 管道过滤半途静默中止 （曾让镜像库变空）、`ruff check` ≠ 能提交；并补上 REPEATABLE READ 的具体应对。 - 据实补全本地库清单：先 grep 全仓库确认无引用，再查内容，确认 `douyu_login_prodclone` / `douyu_login_rehearsal` 是 S1–S7 演练留下的快照。 - 原有教训一条未删，整体按「干活顺序」重排（60 → 116 行）。 --- AGENTS.md | 142 +++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 99 insertions(+), 43 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 76afc5b..eb578f0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,60 +1,116 @@ # 协作约定（给 AI 助手 / 协作者） -> 这些是用户明确表达过的偏好。**每次在这个仓库里干活前先读一遍。** +> 用户明确表达过的偏好，加上**已经用事故买来的教训**。每次在这个仓库里干活前先读一遍。 +> 每条都对应一个真实踩过的坑 —— 删之前先确认那个坑不会再犯。 -## 提交与推送 +## 1. 分支与提交 -- **不要改一点就提交、推送。** - 小改动先攒着，**攒够一批**再提交；只有用户明确说「提交 / 推送」， - 或者到了一个自然的里程碑（一个完整功能做完、一轮反馈全部改完）才 commit / push。 -- 一次提交要**整合**相关改动，不要拆成一堆碎提交。 -- 推送时 `main` 与 `dev` 一起同步（`dev` 是工作分支）： - `git push gitea main` → `git checkout dev && git merge --ff-only main && git push gitea dev`。 +- `main` = **已发布**（生产跑的就是它）。功能开发**从 main 开分支**： + `git checkout -b feat/<功能名> main`，做完合回 `main`。 +- `dev` 是**历史工作分支，已与 main 分叉**（dev 上有未上线的账号自动分配功能）。 + ⚠️ 旧的「推完 main 再 `dev merge --ff-only main`」约定**已作废** —— + 分叉后这么做会失败；以后要同步 dev 只能真合并，并且必须显式确认。 +- **不要改一点就提交、推送。** 小改动先攒着、攒够一批再提交；只有用户明确说 + 「提交 / 推送」，或到了一个自然的里程碑（一个完整功能做完、一轮反馈全部改完） + 才 commit / push。一次提交要**整合**相关改动，别拆成一堆碎提交。 +- 远端名不统一：本地是 `gitea`，生产服务器上的同一个仓库叫 `origin`（URL 相同）。 -## 功能改动 +## 2. 功能改动 -- **新功能是「增加」，不是「替换」。** 旧路径要完整保留（例：账号检测保留了 - 手动粘贴，新增「按账号选择」；旧的手动输入逻辑一行未动）。 +- **新功能是「增加」，不是「替换」。** 旧路径要完整保留（例：账号检测保留了手动粘贴， + 新增「按账号选择」；旧的手动输入逻辑一行未动）。做新功能时先想清楚**旧路径会不会被影响**， + 需要扩展的地方用**可选参数 / 并集**，不要改默认行为。 - 改动要**务实、克制**：不做没人要的重构，不加用不上的抽象。 -- **不要重复展示同一份数据。** 数据库里已有的字段（如 `accounts.points` / - `gold_balance` / `elite_opened`）就让已有界面显示，查询去**刷新**它们， - 不要另开一份副本 —— 同一屏出现两份会让人分不清哪份是准的。 -- **单项外部调用失败不要让整个流程失败**，但要**如实暴露**（错误写进 `errors` +- **不要重复展示同一份数据。** 数据库里已有的字段（如 `accounts.points` / `gold_balance` / + `elite_opened`）就让已有界面显示，查询去**刷新**它们，不要另开一份副本 —— + 同一屏两份会让人分不清哪份是准的。 +- **单项外部调用失败不要让整个流程失败**，但要**如实暴露**（错误写进 `errors` / 返回值 并在 UI 展示），别让「没查到」被读成「没有」。 +- 旁路动作（写事件、写审计、释放资源）**绝不能带崩主流程**：单独 try/except + 记 warning， + 但要留兜底（如定时作业）把漏掉的补上。 -## 报表 / 汇报 +## 3. 验证要求（怎么证明你做对了） -- **如实报告，包括自己的错误、误判和局限。** 不确定就说不确定，别猜。 -- **结论要有验证**：能实测就实测（本仓库有生产镜像库可用）， - 能反向验证（把 bug 注入回去看用例是否变红）就反向验证。 +- **结论要有验证**：能实测就实测（有生产镜像库 `douyu_login_prod` 可只读查）， + 关键是**反向验证** —— 把 bug 注入回去，确认用例真的会变红，再还原。 + 「跑了一下没报错」不算验证。 +- **提交/推送会跑钩子**，所以本地必须按同样标准自查： + - `pre-commit`：`ruff format --check`（**只跑 `ruff check` 不够**，格式不对会直接拦下提交） + + 前端 `tsc -b`； + - `pre-push`：上面这些 + **全量 pytest**。 +- **断言要看 `model_dump()`（序列化结果）**，只看属性会假绿（原因见坑 #2）。 +- **别信「测试通过」** —— 用例可能是假绿的（原因见坑 #3）。 + +## 4. 数据库与迁移 + +- 迁移编号**按落地顺序取，不预留**；**已应用的迁移不能改名**。 +- 加列/加表后，**测试库不会自己长出新结构** —— `tests/support/db.py` 用的是 + `Base.metadata.create_all`，它**只建缺失的表，不会 ALTER 已有表**。 + 症状是 `Unknown column`，处理办法：重建测试库（DROP/CREATE `douyu_login_test`）。 +- **本地 `dev.sh` 在跑时，改迁移文件会被它自动应用到你正在用的库**。 + 所以切分支前先确认「库的迁移版本 == 目标分支的 head」，否则应用会因为 + alembic 找不到 revision **直接起不来**（`run_migrations` 没有兜底，会带崩 lifespan）。 +- 本地库：`douyu_login`（dev）、`douyu_login_test`（测试，**不要**拿别的库跑测试套件）、 + `douyu_login_prod`（生产镜像）。 + 另有 `douyu_login_prodclone`（停在 `20261003_0040`）与 `douyu_login_rehearsal` + （停在 `20261004_0049`）：是 S1–S7 上线演练留下的**生产克隆快照**（各 15,182 账号）， + 代码里没有任何引用 —— 只读参考，别当测试库用。 + +## 5. 部署与生产 + +生产：SSH 别名 `order+douyu_huya`，目录 `/root/live-hub-py`，跑的就是 `main`。 +**完整清单见 `docs/上线清单-S1-S7部署.md`**，核心步骤： + +```bash +# ① 部署前（缺一不可）：备份 + 采基线 +docker exec live-hub-py-mysql-1 sh -c 'mysqldump ... douyu_login' | gzip > /root/backup_$(date +%m%d_%H%M).sql.gz +gzip -t /root/backup_*.sql.gz +# 采 CK 基线：样本解密长度 + 持有 CK 账号数（部署后要一字不差地对比） +# 查在途任务：正在跑的批次会被重启打断，先看能不能等 + +# ② 执行：先停 → 再迁移 → 再起 +docker compose stop douyu-login +docker compose run --rm -T douyu-login alembic upgrade head