Merge main into dev: 登录导出/失败入流水 + AGENTS.md 修正

把 main 上的两笔带过来，dev 继续在其上开发：
- d11fa10 登录任务按状态导出 CSV + 登录失败写进账号生命周期
- 05b2bf8 AGENTS.md 分支约定修正

两条线在此之前无文件重叠（dev 38 个 / main 9 个，交集为空），所以是干净合并。
注意：main 的登录功能不含迁移，dev 的 0052 仍是唯一新增迁移。

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 </dev/null
+docker compose up -d douyu-login
+
+# ③ 部署后：版本 / CK 基线 / 新接口 / 前端产物 / 日志无 ERROR，逐条核对
+```
+
+- 部署前**必须备份**（即使这次零迁移）；**CK 检查是唯一不能省的功能性验证**
+  （CK 丢了补不回来）。
+- `docker compose build douyu-login` 内部会 `npm run build` 前端 —— 改了前端不用手工构建，
+  但部署后要**去产物里确认新代码真在里面**（`grep` 一个新增的界面文案）。
+- 只动 app 容器，**别碰 mysql**（历史上 `up -d --force-recreate` 把 mysql 一起重建，
+  每次部署 90 秒 502）。
+
+## 6. 汇报
+
+- **如实报告，包括自己的错误、误判和局限。** 不确定就说不确定，别猜；
+  报错了要主动更正（数字尤其要能复现：说「新增 N 个用例 / 基线 M」之前先跑一遍数出来）。
 - 用中文回答。
 
-## 一次踩过的坑（值得记住）
+## 7. 已踩过的坑（速查）
 
-- **别拿不同时刻的快照做对比。** 曾经把相隔两个月的两条记录对比，
-  把「时间差异」当成了「账号差异」，得出错误规则并写进设计文档。
-  对比必须在**同一时刻**。
-- **Pydantic 会静默丢字段 —— 本仓库已经踩过三次**，每次表现都不同：
-  1. 构造函数传了、模型上没声明 → `extra='ignore'` 丢掉；
-  2. 模型上声明了、但**手写的 `@model_serializer` 里没列** → 属性有值、JSON 里没有；
-  3. 字段名与 schema 声明的不一致（如 `elite_bind_status` vs `role_bind_status`）。
-  **应对**：断言要看 **`model_dump()`（序列化结果）**，只看属性会假绿；
-  改完接口先在 Python 里打一遍 `model_dump()` 确认字段真在里面。
-- **别信"测试通过" —— 用例可能是假绿的（缺 `session.commit()` 时，
-  MySQL REPEATABLE READ 会让它看不到别的会话刚提交的行）。
-  关键用例要做反向验证。
-- **删代码块时用"从 A 到 B"的区间删除很危险**，容易连带删掉中间的行
-  （曾把 `**bind_payload(acc),` 一起删掉）。改完要**读回来确认**。
-- **`docker compose run` 默认读 stdin**：脚本用 `ssh 'bash -s' < 脚本` 喂进去时，
-  它会把脚本剩下的行吃掉。要用 `docker compose run -T --rm ... </dev/null`。
+| # | 坑 | 应对 |
+| --- | --- | --- |
+| 1 | **别拿不同时刻的快照做对比** —— 曾把相隔两个月的两条记录对比，把「时间差异」当成「账号差异」，得出错误规则写进设计文档 | 对比必须在**同一时刻**；时间跨度大就重新采数 |
+| 2 | **Pydantic 静默丢字段**（本仓库踩过三次，表现各不相同）：① 构造函数传了、模型没声明 → `extra='ignore'` 丢掉；② 模型声明了、手写 `@model_serializer` 里没列 → 属性有值、JSON 没有；③ 字段名与 schema 不一致（`elite_bind_status` vs `role_bind_status`） | 断言看 **`model_dump()`**；改完接口先在 Python 里打一遍确认字段真在里面 |
+| 3 | **用例假绿**：缺 `session.commit()` 时 MySQL REPEATABLE READ 让本会话看不到别的会话刚提交的行（脚本改没改都显示没改） | 跨会话断言前先 `rollback()`/`commit()` **结束当前事务**再读；关键用例做反向验证 |
+| 4 | **用「从 A 到 B」的区间删除删代码块很危险**，会连带删掉中间的行（曾把 `**bind_payload(acc),` 一起删掉） | 改完**读回来确认**；能用精确替换就别用区间 |
+| 5 | **`docker compose run` 默认读 stdin**：用 `ssh 'bash -s' < 脚本` 喂进去时，它会把脚本剩下的行吃掉 | 用 `docker compose run -T --rm ... </dev/null` |
+| 6 | **镜像同步脚本会白等 15 分钟**：`pgrep -f 'mysqldump.*douyu_login'` 匹配到了自己所在的 shell，等待循环永不提前退出 | 判定远端进程存活别用会自匹配的 `pgrep -f` 模式 |
+| 7 | **`set -e` + 输出过滤（管道接 `grep -v`）会半途静默中止**：过滤把输出全吃掉时 grep 返回 1，被 `set -e` 当成失败（曾在「已删库、还没导入」处中止，镜像库一度是空的） | 过滤输出后别让 `set -e` 接管退出码；关键步骤显式检查成功 |
+| 8 | **`ruff check` 通过 ≠ 能提交**：钩子查的是 `ruff format` | 提交前本地跑一遍 `.githooks/pre-commit` |
 
-## 仓库事实
+## 8. 仓库事实（速查）
 
 | | |
 | --- | --- |
-| 工作分支 | `dev`（本地开发）；`main` = 已发布；远端是 `gitea` |
-| 生产 | SSH 别名 `order+douyu_huya`（47.114.114.225），部署目录 `/root/live-hub-py` |
-| 本地生产镜像 | `douyu_login_prod`（`scripts/refresh_local_prod_mirror.sh` 同步） |
-| 测试库 | `douyu_login_test`（独立库，**不要**拿镜像库跑测试套件） |
-| 迁移 | 编号按落地顺序取，**不预留**；已应用的迁移不能改名 |
-| 前端 | `tsc -b` 开了 `noUnusedLocals`，死代码/未用导入会直接报 TS6133 |
+| 分支 | `main` = 已发布（生产跑它）；`dev` = 已分叉的历史工作分支；新功能从 main 开 `feat/*` |
+| 生产 | SSH `order+douyu_huya`（47.114.114.225），目录 `/root/live-hub-py`，容器 `live-hub-py-douyu-login-1` |
+| 本地服务 | `./dev.sh`（8800 端口，`--reload`）；**启动时会自动跑迁移** |
+| 本地 MySQL | 容器 `live-hub-py-mysql-1`，端口 3307；库见 §4 |
+| 生产镜像库 | `douyu_login_prod`（`scripts/refresh_local_prod_mirror.sh` 同步，只读用） |
+| 迁移 head | `main` = `20261004_0051`；`dev` = `20261007_0052`（**两侧不一致，切分支前先对齐库**） |
+| 前端 | `tsc -b` 开了 `noUnusedLocals`，死代码/未用导入直接报 TS6133；镜像构建时会 `npm run build` |
+| 测试 | `pytest tests/ -q`（用 `douyu_login_test`）；`ruff format` + `ruff check` |

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 && (
             <>