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，对应用例均变红）。

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"}

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")

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"  # 自动重登成功

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
 

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 "", "")

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,

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<TaskPage, TaskPage>('/login/tasks', { params: { ...params, batch_id: params.batch_id ?? '' } }),
   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}`),
   deleteTask: (id: number) => api.delete<MessageResponse, MessageResponse>(`/login/tasks/${id}`),
   deleteTasks: (ids: number[]) => api.delete<MessageDeletedResponse, MessageDeletedResponse>(`/login/tasks`, { params: { task_ids: ids.join(',') } }),

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: '进行中' },
             ]}
           />
+          <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 }} />
           {canBatch && (
             <>