feat(login-tasks): 按状态导出 CSV + 登录失败写进账号生命周期

一、失败入账号一生(此前完全缺失)
生产实测:`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,对应用例均变红)。
This commit is contained in:
yml2213
2026-10-07 12:54:57 +08:00
parent 281769b440
commit d11fa1032d
8 changed files with 598 additions and 49 deletions
+269
View File
@@ -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"}
+152 -22
View File
@@ -1,7 +1,10 @@
"""登录任务路由 + WebSocket 实时日志""" """登录任务路由 + WebSocket 实时日志"""
import asyncio import asyncio
import csv
import io
import threading import threading
from datetime import UTC, datetime
from fastapi import ( from fastapi import (
APIRouter, APIRouter,
@@ -10,6 +13,7 @@ from fastapi import (
WebSocket, WebSocket,
WebSocketDisconnect, WebSocketDisconnect,
) )
from fastapi.responses import StreamingResponse
from sqlalchemy import and_, func, or_ from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session, load_only 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 ..permissions import get_user_permissions, user_has_permission
from ..schemas import LoginBatchRequest, LoginTaskOut from ..schemas import LoginBatchRequest, LoginTaskOut
from ..services.audit_service import record_audit 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 from ..services.login_service import LoginBatchRunner, batch_registry
router = APIRouter(prefix="/api/login", tags=["登录任务"]) 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): def _login_task_summary(query):
"""按状态汇总登录任务,避免概览页拉完整任务列表。""" """按状态汇总登录任务,避免概览页拉完整任务列表。"""
rows = ( rows = (
@@ -213,17 +265,7 @@ def list_tasks(
传 `page` 时同时返回 `total`,是为了让前端能「全选所有筛选结果」—— 传 `page` 时同时返回 `total`,是为了让前端能「全选所有筛选结果」——
否则跨页选择无从下手,而任务量早就过了单页能装下的规模。 否则跨页选择无从下手,而任务量早就过了单页能装下的规模。
""" """
query = _visible_tasks_query(db, current, batch_id) query = _apply_status_filter(_visible_tasks_query(db, current, batch_id), status)
if status:
# 前端看到的是「最近一次尝试」的状态,与列表序列化口径保持一致。
query = query.filter(
or_(
and_(
LoginTask.attempt_status != "", LoginTask.attempt_status == status
),
and_(LoginTask.attempt_status == "", LoginTask.status == status),
)
)
ordered = query.order_by(LoginTask.id.desc()) 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") @router.get("/tasks/summary")
def tasks_summary( def tasks_summary(
db: Session = Depends(get_db), db: Session = Depends(get_db),
@@ -336,17 +472,11 @@ def delete_tasks(
elif all_matching: elif all_matching:
# 带 join 的 query 不能直接 bulk delete(SQLAlchemy 会拒绝), # 带 join 的 query 不能直接 bulk delete(SQLAlchemy 会拒绝),
# 所以先按权限与筛选条件查出 id,再按主键删。 # 所以先按权限与筛选条件查出 id,再按主键删。
query = _visible_tasks_query(db, current, batch_id) # 状态筛选与列表/导出共用一份 —— 「清空失败」以前只删 `failed`、
if status: # 漏掉 `error`,删完与页面上那个「失败 N」的数还对不上。
query = query.filter( query = _apply_status_filter(
or_( _visible_tasks_query(db, current, batch_id), status
and_( )
LoginTask.attempt_status != "",
LoginTask.attempt_status == status,
),
and_(LoginTask.attempt_status == "", LoginTask.status == status),
)
)
ids = [row[0] for row in query.with_entities(LoginTask.id).all()] ids = [row[0] for row in query.with_entities(LoginTask.id).all()]
else: else:
raise HTTPException(status_code=400, detail="请指定任务ID或 all_matching=true") raise HTTPException(status_code=400, detail="请指定任务ID或 all_matching=true")
@@ -43,6 +43,13 @@ from sqlalchemy.orm import Session
# 命名保持「名词_状态」风格,便于前端归类与筛选。 # 命名保持「名词_状态」风格,便于前端归类与筛选。
EVENT_IMPORTED = "imported" # 账号导入(读 accounts.created_at,一般不落表) EVENT_IMPORTED = "imported" # 账号导入(读 accounts.created_at,一般不落表)
EVENT_LOGIN_OK = "login_ok" # 登录成功(含新 CK) 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_INVALID = "ck_invalid" # CK 判定失效
EVENT_CK_CLEARED = "ck_cleared" # CK 被清除(人工清除 / 账号删除前的清理) EVENT_CK_CLEARED = "ck_cleared" # CK 被清除(人工清除 / 账号删除前的清理)
EVENT_RELOGIN_OK = "relogin_ok" # 自动重登成功 EVENT_RELOGIN_OK = "relogin_ok" # 自动重登成功
@@ -32,6 +32,7 @@ from ...account_lifecycle import (
) )
from ...account_value import value_tier_expr from ...account_value import value_tier_expr
from ...douyu_service import cookie_holding_filter, cookie_operable_filter from ...douyu_service import cookie_holding_filter, cookie_operable_filter
from ...login_failure import classify_failure
from ...login_service import LoginBatchRunner, get_relogin_limits from ...login_service import LoginBatchRunner, get_relogin_limits
JOB_KEY = "ck_relogin" JOB_KEY = "ck_relogin"
@@ -54,29 +55,6 @@ AUTO_MAX_TOTAL_TIME = 240
QUOTA_JITTER_MINUTES = 30 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: def _cutoff(now: datetime) -> datetime:
return now return now
+68
View File
@@ -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 "", "")
+42 -2
View File
@@ -12,7 +12,7 @@ from typing import cast
from loguru import logger from loguru import logger
from sqlalchemy import and_, func, or_ 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 import DouyuLogin, create_login_api_strategy
from core.douyu.login import AccountLike 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 Account as AccountModel
from ..models import LoginTask from ..models import LoginTask
from ..models import ProxyConfig as ProxyConfigModel 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 .account_lifecycle import on_ck_verdict, on_login_success
from .cookie_check_service import check_douyu_cookie from .cookie_check_service import check_douyu_cookie
from .douyu_service import LOGIN_ATTEMPT_IN_FLIGHT_STATUSES from .douyu_service import LOGIN_ATTEMPT_IN_FLIGHT_STATUSES
from .login_failure import classify_failure, failure_kind_label, is_account_fault
CHECK_STATUS_MESSAGES = { CHECK_STATUS_MESSAGES = {
"account_cancelled": "账号已注销", "account_cancelled": "账号已注销",
@@ -121,6 +127,15 @@ def _mark_attempt_failed(
这里读的是 `task.account.cookie`(CK 自 2026-10-04 起挂在账号上), 这里读的是 `task.account.cookie`(CK 自 2026-10-04 起挂在账号上),
而不是这条登录记录 —— 记录不再持有凭证。 而不是这条登录记录 —— 记录不再持有凭证。
顺带往**账号时间线**写一条失败事件(这是账号级的资产变化:这个账号登录不上)。
两条边界:
- **被停止 / 被中断不写**:`classify_failure` 把它们归到 `interrupted`,
那是人的动作或进程重启,不是账号出了状况。不挡住的话,
运营按一次「停止」就能给几十个账号的时间线刷满假失败。
- **幂等**:一次尝试一条(`source_ref` 带 task id + 尝试时刻),
失败重试会新建记录或新 attempt,因此是新的一条 —— 这是对的,那是新的事实。
""" """
completed_at = datetime.now(UTC) completed_at = datetime.now(UTC)
task.attempt_status = status if status in {"failed", "error"} else "failed" task.attempt_status = status if status in {"failed", "error"} else "failed"
@@ -133,6 +148,31 @@ def _mark_attempt_failed(
task.message = message task.message = message
task.finished_at = completed_at 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( def _write_login_success(
db: Session, db: Session,
+14
View File
@@ -26,6 +26,20 @@ export const loginApi = {
listTasksPaged: (params: { page: number; page_size: number; batch_id?: string; status?: string }) => listTasksPaged: (params: { page: number; page_size: number; batch_id?: string; status?: string }) =>
api.get<TaskPage, TaskPage>('/login/tasks', { params: { ...params, batch_id: params.batch_id ?? '' } }), api.get<TaskPage, TaskPage>('/login/tasks', { params: { ...params, batch_id: params.batch_id ?? '' } }),
tasksSummary: () => api.get<TaskSummary, TaskSummary>('/login/tasks/summary'), tasksSummary: () => api.get<TaskSummary, TaskSummary>('/login/tasks/summary'),
/**
* 导出登录任务 CSV(按当前筛选,不分页)。
*
* `status` 传 'failed' 时后端会展开成 failed + error —— 页面上「失败」
* 本来就是一个组,导出与列表口径必须一致。
*/
exportTasks: (params: { batch_id?: string; status?: string }) =>
api.get<Blob, Blob>('/login/tasks/export', {
responseType: 'blob',
params: {
batch_id: params.batch_id ?? '',
status: params.status ?? '',
},
}),
stop: (batch_id: string) => api.post<MessageResponse, MessageResponse>(`/login/stop/${batch_id}`), stop: (batch_id: string) => api.post<MessageResponse, MessageResponse>(`/login/stop/${batch_id}`),
deleteTask: (id: number) => api.delete<MessageResponse, MessageResponse>(`/login/tasks/${id}`), deleteTask: (id: number) => api.delete<MessageResponse, MessageResponse>(`/login/tasks/${id}`),
deleteTasks: (ids: number[]) => api.delete<MessageDeletedResponse, MessageDeletedResponse>(`/login/tasks`, { params: { task_ids: ids.join(',') } }), deleteTasks: (ids: number[]) => api.delete<MessageDeletedResponse, MessageDeletedResponse>(`/login/tasks`, { params: { task_ids: ids.join(',') } }),
+45 -2
View File
@@ -1,11 +1,11 @@
import { useEffect, useState, useMemo, useCallback, useRef } from 'react'; import { useEffect, useState, useMemo, useCallback, useRef } from 'react';
import { 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'; } from 'antd';
import { message } from '../utils/antdMessage'; import { message } from '../utils/antdMessage';
import { import {
PlayCircleOutlined, StopOutlined, FilterOutlined, ReloadOutlined, DeleteOutlined, PlayCircleOutlined, StopOutlined, FilterOutlined, ReloadOutlined, DeleteOutlined,
SettingOutlined, SafetyCertificateOutlined, SettingOutlined, SafetyCertificateOutlined, DownloadOutlined,
} from '@ant-design/icons'; } from '@ant-design/icons';
import { accountApi, loginApi, type AccountItem, type LoginTaskItem, type TaskSummary } from '../api/modules'; import { accountApi, loginApi, type AccountItem, type LoginTaskItem, type TaskSummary } from '../api/modules';
import RealtimeLogPanel from '../components/RealtimeLogPanel'; 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 () => { const handleClearFailed = async () => {
try { try {
@@ -649,6 +677,21 @@ export default function LoginTasksPage() {
{ value: 'running', label: '进行中' }, { value: 'running', label: '进行中' },
]} ]}
/> />
<Dropdown
menu={{
items: [
{ key: 'current', label: taskStatus ? '导出当前筛选' : '导出全部' },
{ key: 'all', label: '导出全部(忽略筛选)' },
{ key: 'success', label: '只导出成功' },
{ key: 'failed', label: '只导出失败(含异常)' },
],
onClick: ({ key }) => void handleExport(key as 'current' | 'all' | 'success' | 'failed'),
}}
>
<Button size="small" icon={<DownloadOutlined />}>
导出 CSV
</Button>
</Dropdown>
<div style={{ flex: 1 }} /> <div style={{ flex: 1 }} />
{canBatch && ( {canBatch && (
<> <>