diff --git a/docs/设计-账号全生命周期.md b/docs/设计-账号全生命周期.md index 462f6c8..1c96c21 100644 --- a/docs/设计-账号全生命周期.md +++ b/docs/设计-账号全生命周期.md @@ -256,21 +256,30 @@ CREATE INDEX ix_account_events_ledger **事件类型表（收敛后的完整取值域）**： -| category | event_type | 含义 | amount | -| --- | --- | --- | --- | -| lifecycle | `imported` | 导入 | — | -| lifecycle | `status_changed` | **所有**状态变迁（detail 带 from/to/reason），停用与归档不再单列事件类型 | — | -| auth | `login_ok` | 登录成功（拿到新 CK） | — | -| auth | `ck_invalid` | 巡检判定 CK 失效 | — | -| auth | `relogin_ok` / `relogin_failed` | 自动重登结果 | — | -| auth | `ck_cleared` | CK 被清除（人工或检测判定） | — | -| business | `recharge` | 充值到账 | + ¥ | -| business | `exchange` | 兑换（含消耗积分） | − 积分 | -| business | `points_credit` | 积分到账 | + 积分 | -| business | `points_expire` | 积分到期（由 `willExpireTime` 派生） | — | -| business | `elite_opened` / `esports_opened` | 开通精英宝典 / 电竞手册 | — | -| business | `bind_ok` / `rebind` | 角色绑定 / 换绑 | — | -| ops | `assigned` / `tag_changed` / `remark_changed` | 人工操作 | — | +| category | event_type | 含义 | 发射点 | amount | +| --- | --- | --- | --- | --- | +| lifecycle | `status_changed` | **所有**状态变迁（detail 带 from/to/reason），停用与归档不再单列事件类型 | `account_lifecycle._apply`（唯一） | — | +| auth | `login_ok` | 登录成功（拿到新 CK） | `login_service._write_login_success` | — | +| auth | `ck_invalid` | CK **由好变坏**（转变才记，见下） | `account_lifecycle.on_ck_verdict` | — | +| auth | `ck_cleared` | CK 被清除（**不可逆**，必须留痕） | `account_lifecycle.on_ck_cleared` | — | +| auth | `relogin_ok` / `relogin_failed` | 自动重登结果 | `login_service` / `ck_relogin.settle_results` | — | +| ops | `assigned` | 分配 / 取消分配 | `audit_service.record_account_action`、批量走 `emit_bulk` | — | +| ops | `tag_changed` | 标签变更（detail 带 from/to） | 同上 | — | +| business | `recharge` | 充值到账 | 收集器（S5） | + ¥ | +| business | `exchange` | 兑换（含消耗积分） | 收集器（S5） | − 积分 | +| business | `points_credit` | 积分到账 | 收集器（S5） | + 积分 | +| business | `elite_opened` / `esports_opened` | 开通精英宝典 / 电竞手册 | `douyu_runner_manual` | — | +| business | `bind_ok` / `rebind` | 角色绑定 / 换绑 | 收集器（S5） | — | + +> **不落的两个**： +> - `imported`：导入时间直接读 `accounts.created_at`，不落表（设计 §3.5 起就这么定）； +> - `remark_changed`：**目前没有改备注的接口**，所以这个类型没有发射点。要么补接口， +> 要么从表里删掉 —— 留着一个永远不发的类型只会让人以为它有数据。 +> +> **`ck_invalid` 只认「转变」**（2026-10-04 S4 修正）：旧实现每轮巡检只要判定 invalid +> 就发一条，注释写着「避免每轮刷屏」而代码并没有做到 —— 6 小时一轮、失效一周 = +> 28 条一模一样的事件，把时间线淹掉。判定失效是**一次状态转变**，不是每轮读数。 +> 同理 `error`（查不到）不改变结论、也不发事件。 **「开通 / 积分 / 兑换记录」怎么呈现**：就是这张表按 `category` 或 `event_type` 过滤的视图， **不另建三张表**。理由： @@ -429,7 +438,7 @@ WHERE a.cookie <> src.cookie; | **S1** | 迁移 0042：加列 + 回填 + 自校验；`accounts` 成为状态载体 | 账号能独立描述登录态 | — | | **S2** | 读写一次切换：12 个文件、24 读点 / 10 写点全部改 `LoginTask.*` → `Account.*`；批次行复用逻辑重写；在途子查询 + 索引 | 登录记录与 CK 彻底解耦 | S1 | | **S3** ✅ | 状态收口（**已完成 2026-10-04**）：`account_lifecycle` 提供 `derive_status()` / `sync_status()` / `set_manual_state()` / `on_login_success()`；9 处判据写入点全部接线；6 状态状态机上线；自动化选行排除 `disabled`/`archived`；新增 `account:manage` 权限与停用/归档/恢复接口；前端列表加状态列与筛选、详情页加状态卡与操作入口 | `status` 可信 | S2 | -| **S4** | `account_events` 扩展（4 列 + 索引）+ 事件发射点补齐（含 `audit_logs.account_id` 打通） | 时间线完整 | S2 | +| **S4** ✅ | 事件发射点补齐（**已完成 2026-10-04**）：`assigned` / `tag_changed` 接上（此前定义了却从没发过）、新增 `ck_cleared`、`ck_invalid` 改为只在转变时发；`audit_logs.account_id` 在分配/标签/检测/清 CK 处打通；新增 `emit_bulk`（批量操作逐账号写事件）与 `record_account_action`（审计+事件一次写完）；时间线支持 `category` 筛选；前端补 `status_changed` / `ck_cleared` 样式与分类筛选 | 时间线完整 | S2 | | **S5** | 业务事实直发事件：开通 / 充值 / 积分 / 兑换 / 绑定在任务成功时写事件；`douyu_tasks` 加保留策略 | 台账可用，687MB 可控 | S4 | | **S6** | 账号页重构：列表加状态与关键列；详情页做成「生命周期主页」（状态卡 + 台账 + 时间线 + 操作）；`bind_status` 三列拆解落地 | 账号管理真正独立 | S3、S5 | | **S7** | 账号检测并入账号域（`account-check` 从「粘贴文本」改为「按账号选」，结果落事件） | 检测不再是孤岛 | S6 | diff --git a/tests/test_account_actions.py b/tests/test_account_actions.py new file mode 100644 index 0000000..feafea2 --- /dev/null +++ b/tests/test_account_actions.py @@ -0,0 +1,415 @@ +"""S4 回归：账号操作的事件发射与审计关联。 + +S4 之前的状态（都是「定义了却从没发过」）： + +- `assigned` / `tag_changed` 两个事件类型**定义了、前端也渲染了，但没有任何代码发**； +- 分配/改标签的审计**不带 `account_id`**（`target` 是自由文本，账号改名就关联不上）； +- 清 CK 不发事件 —— 而这是**不可逆**操作，时间线里却不留痕； +- `ck_invalid` **每轮巡检都发**：注释写着「避免每轮刷屏」，代码却是一个失效一周的 + 账号攒 28 条重复事件。 + +本文件守住修复后的行为。 +""" + +from datetime import UTC, datetime + +from tests.support.db import get_test_engine, make_test_session +from web.backend.models import Account, AccountEvent, AuditLog, User +from web.backend.services import account_events_service as svc + +NOW = datetime(2026, 10, 4, 12, 0, 0, tzinfo=UTC) + + +class _Base: + def setup_method(self): + self.engine = get_test_engine() + self.session = make_test_session(self.engine) + self.admin = User( + username="act-admin", password_hash="h", role="super_admin", is_active=True + ) + self.support = User( + username="act-kf", password_hash="h", role="support", is_active=True + ) + self.session.add_all([self.admin, self.support]) + self.session.commit() + + def teardown_method(self): + self.session.close() + + def make_account(self, name="acc", **kwargs) -> Account: + account = Account( + username=name, + password="p", + email=f"{name}@e.com", + email_password="ep", + cookie=kwargs.pop("cookie", "acf_uid=1"), + ck_check_status=kwargs.pop("ck_check_status", "valid"), + **kwargs, + ) + self.session.add(account) + self.session.commit() + return account + + def events(self, account_id: int, event_type: str) -> list[AccountEvent]: + return ( + self.session.query(AccountEvent) + .filter( + AccountEvent.account_id == account_id, + AccountEvent.event_type == event_type, + ) + .order_by(AccountEvent.id.asc()) + .all() + ) + + def latest_audit(self, action: str) -> AuditLog | None: + return ( + self.session.query(AuditLog) + .filter(AuditLog.action == action) + .order_by(AuditLog.id.desc()) + .first() + ) + + +class TestAssignEmitsEvent(_Base): + def test_single_assign_writes_event_and_links_audit(self): + from web.backend.routers.accounts import assign_account + from web.backend.schemas import AccountAssign + + account = self.make_account("a") + assign_account( + account.id, + AccountAssign(assigned_to=self.support.id), + db=self.session, + current=self.admin, + ) + + self.session.refresh(account) + assert account.assigned_to == self.support.id + + events = self.events(account.id, svc.EVENT_ASSIGNED) + assert len(events) == 1 + assert events[0].category == svc.CATEGORY_OPS + assert "act-kf" in events[0].title + assert events[0].actor_type == "user" + assert events[0].actor_id == self.admin.id + + # 审计也要能反查到账号（以前只写自由文本 target） + audit = self.latest_audit("account:assign") + assert audit is not None and audit.account_id == account.id + + def test_unassign_writes_event(self): + from web.backend.routers.accounts import assign_account + from web.backend.schemas import AccountAssign + + account = self.make_account("a", assigned_to=self.support.id) + assign_account( + account.id, + AccountAssign(assigned_to=None), + db=self.session, + current=self.admin, + ) + events = self.events(account.id, svc.EVENT_ASSIGNED) + assert len(events) == 1 + assert "取消分配" in events[0].title + + def test_batch_assign_emits_one_event_per_account(self): + from web.backend.routers.accounts import batch_assign_accounts + from web.backend.schemas import BatchAssign + + accounts = [self.make_account(f"a{i}") for i in range(3)] + batch_assign_accounts( + BatchAssign( + account_ids=[a.id for a in accounts], assigned_to=self.support.id + ), + db=self.session, + current=self.admin, + ) + for account in accounts: + self.session.refresh(account) + assert account.assigned_to == self.support.id + assert len(self.events(account.id, svc.EVENT_ASSIGNED)) == 1 + + +class TestTagEmitsEvent(_Base): + def test_single_tag_writes_event_with_from_to(self): + from web.backend.routers.accounts import set_account_tag + from web.backend.schemas import AccountTag + + account = self.make_account("a", tag="旧标签") + set_account_tag( + account.id, AccountTag(tag="新标签"), db=self.session, current=self.admin + ) + self.session.refresh(account) + assert account.tag == "新标签" + + events = self.events(account.id, svc.EVENT_TAG_CHANGED) + assert len(events) == 1 + assert events[0].detail == {"from": "旧标签", "to": "新标签"} + assert events[0].category == svc.CATEGORY_OPS + + audit = self.latest_audit("account:tag") + assert audit is not None and audit.account_id == account.id + + def test_batch_tag_emits_per_account_events(self): + from web.backend.routers.accounts import batch_tag + from web.backend.schemas import AccountTag + + accounts = [self.make_account(f"a{i}") for i in range(3)] + batch_tag( + AccountTag(tag="批量标签", account_ids=[a.id for a in accounts]), + db=self.session, + current=self.admin, + ) + for account in accounts: + self.session.refresh(account) + assert account.tag == "批量标签" + events = self.events(account.id, svc.EVENT_TAG_CHANGED) + assert len(events) == 1 + assert events[0].detail["to"] == "批量标签" + + +class TestCkCleared(_Base): + def test_single_clear_emits_event_and_audit_link(self): + from web.backend.routers.cookies import delete_cookie + + account = self.make_account("a") + delete_cookie(account.id, db=self.session, current=self.admin) + self.session.refresh(account) + + assert account.cookie == "" + events = self.events(account.id, svc.EVENT_CK_CLEARED) + assert len(events) == 1 + assert events[0].category == svc.CATEGORY_AUTH + audit = self.latest_audit("cookie:delete") + assert audit is not None and audit.account_id == account.id + + def test_batch_clear_emits_event_per_account(self): + from web.backend.routers.cookies import delete_cookies_batch + + accounts = [self.make_account(f"a{i}") for i in range(3)] + delete_cookies_batch( + account_ids=",".join(str(a.id) for a in accounts), + db=self.session, + current=self.admin, + ) + for account in accounts: + self.session.refresh(account) + assert account.cookie == "" + assert len(self.events(account.id, svc.EVENT_CK_CLEARED)) == 1 + + +class TestCkInvalidOnlyOnTransition(_Base): + """`ck_invalid` 是**一次状态转变**，不是每轮读数。 + + 旧实现每轮巡检只要判定 invalid 就发一条（源码注释却写着「避免每轮刷屏」）—— + 6 小时一轮、失效一周 = 28 条一模一样的事件，把时间线淹掉。 + """ + + def _job(self): + from web.backend.models import AutomationJob + from web.backend.services.automation.jobs import ck_health + + return AutomationJob( + key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200 + ) + + def _run(self, message="鱼丸接口: 账户登陆异常，请重新登陆"): + """跑一轮巡检。 + + ⚠️ 先清掉 `ck_checked_at`：候选查询会过滤「距上次检测超过作业间隔」， + interval 最小 60 秒，同一轮里刚查过的账号**根本不会被再次选中** —— + 不重置的话这个用例是假绿的（跑三轮其实只跑了一轮）。 + """ + from web.backend.services.automation.jobs import ck_health + + self.session.query(Account).update( + {Account.ck_checked_at: None}, synchronize_session=False + ) + self.session.commit() + + original = ck_health.check_douyu_cookie + ck_health.check_douyu_cookie = lambda _c: { + "valid": False, + "message": message, + "fish_ball": None, + "nickname": None, + "level": None, + "checked_at": NOW, + } + try: + return ck_health.run_ck_health(self.session, self._job()) + finally: + ck_health.check_douyu_cookie = original + + def test_repeated_invalid_rounds_emit_only_once(self): + account = self.make_account("a", ck_check_status="valid") + + self._run() + self._run() + self._run() + + assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 1 + + def test_recovering_then_failing_again_emits_a_new_event(self): + """由好变坏 → 恢复 → 再变坏，应当各记一条（是两次转变）。""" + from web.backend.services.account_lifecycle import on_ck_verdict + + account = self.make_account("a", ck_check_status="valid") + self._run() + assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 1 + + # 恢复（巡检判 valid） + from web.backend.services.automation.jobs import ck_health + + original = ck_health.check_douyu_cookie + ck_health.check_douyu_cookie = lambda _c: { + "valid": True, + "message": "ok", + "fish_ball": 1, + "nickname": "n", + "level": 1, + "checked_at": NOW, + } + self.session.query(Account).update( + {Account.ck_checked_at: None}, synchronize_session=False + ) + self.session.commit() + try: + ck_health.run_ck_health(self.session, self._job()) + finally: + ck_health.check_douyu_cookie = original + + # 再失效 + self._run() + assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 2 + assert account.id # 避免未使用告警 + + # error（查不到）不改变结论、也不发事件 + on_ck_verdict(self.session, account, "error", detail={}, occurred_at=NOW) + self.session.commit() + assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 2 + + +class TestEmitBulk(_Base): + def test_writes_many_and_returns_count(self): + accounts = [self.make_account(f"a{i}") for i in range(5)] + written = svc.emit_bulk( + self.session, + [ + { + "account_id": acc.id, + "event_type": svc.EVENT_TAG_CHANGED, + "title": f"批量 {acc.username}", + "category": svc.CATEGORY_OPS, + } + for acc in accounts + ], + ) + self.session.commit() + assert written == 5 + for acc in accounts: + assert len(self.events(acc.id, svc.EVENT_TAG_CHANGED)) == 1 + + def test_skips_invalid_rows_and_never_raises(self): + account = self.make_account("a") + written = svc.emit_bulk( + self.session, + [ + {"account_id": None, "event_type": "x", "title": "无账号"}, + {"account_id": account.id, "event_type": "", "title": "无类型"}, + { + "account_id": account.id, + "event_type": svc.EVENT_TAG_CHANGED, + "title": "有效", + }, + ], + ) + self.session.commit() + assert written == 1 + + def test_empty_input_is_noop(self): + assert svc.emit_bulk(self.session, []) == 0 + + +class TestEventsCategoryFilter(_Base): + def test_filter_by_category(self): + from web.backend.routers.accounts import account_events + + account = self.make_account("a") + svc.emit( + self.session, + account_id=account.id, + event_type=svc.EVENT_TAG_CHANGED, + title="改标签", + category=svc.CATEGORY_OPS, + ) + svc.emit( + self.session, + account_id=account.id, + event_type=svc.EVENT_CK_INVALID, + title="失效", + category=svc.CATEGORY_AUTH, + ) + self.session.commit() + + def titles(**kwargs): + page = account_events( + account.id, + page=1, + page_size=50, + db=self.session, + current=self.admin, + **kwargs, + ) + return [item["title"] for item in page["items"]] + + assert set(titles(category="")) == {"改标签", "失效"} + assert titles(category=svc.CATEGORY_OPS) == ["改标签"] + assert titles(category=svc.CATEGORY_AUTH) == ["失效"] + # 事件类型筛选仍然可用 + assert titles(event_type=svc.EVENT_CK_INVALID) == ["失效"] + + def test_response_carries_ledger_fields(self): + from web.backend.routers.accounts import account_events + + account = self.make_account("a") + svc.emit( + self.session, + account_id=account.id, + event_type=svc.EVENT_EXCHANGE, + title="兑换 兔兔白日梦", + category=svc.CATEGORY_BUSINESS, + subject_type="commodity", + subject_key="sku-1", + actor_type="user", + actor_id=self.admin.id, + ) + self.session.commit() + + page = account_events( + account.id, page=1, page_size=50, db=self.session, current=self.admin + ) + item = page["items"][0] + assert item["category"] == svc.CATEGORY_BUSINESS + assert item["subject_key"] == "sku-1" + assert item["actor_id"] == self.admin.id + + +class TestBatchDeleteRecordsIds(_Base): + def test_delete_audit_keeps_account_ids_for_forensics(self): + """删账号会**连带删掉它的事件时间线**，所以审计是唯一痕迹。""" + from web.backend.routers.accounts import batch_delete_accounts + + accounts = [self.make_account(f"a{i}") for i in range(3)] + ids = [a.id for a in accounts] + batch_delete_accounts( + account_ids=",".join(str(i) for i in ids), + db=self.session, + current=self.admin, + ) + + audit = self.latest_audit("account:delete") + assert audit is not None + assert audit.detail is not None + for account_id in ids: + assert str(account_id) in audit.detail diff --git a/tests/test_account_lifecycle.py b/tests/test_account_lifecycle.py index d3472cb..38b2d21 100644 --- a/tests/test_account_lifecycle.py +++ b/tests/test_account_lifecycle.py @@ -12,6 +12,7 @@ """ from datetime import UTC, datetime, timedelta +from typing import ClassVar import pytest @@ -655,3 +656,58 @@ class TestLifecycleApi(_Base): assert res["success"] and account.auto_managed is False # 关掉自动化不等于停用：账号状态仍可用 assert account.status == life.STATUS_ACTIVE + + +class TestStatusWriteFunnel: + """架构不变量：**`account.status` 只能由 `account_lifecycle` 写**。 + + `accounts.bind_status` 就是这么烂掉的 —— 谁都能写、语义没人管，最后 36 种取值 + 互相覆盖，连「这账号到底绑没绑」都查不出来。 + + 这条用源码扫描把它钉死：任何地方直接给 `status` 赋值都会让用例变红，逼你去用 + `sync_status()`（按判据推导）或 `set_manual_state()`（人工态）。 + 比写在文档里的约定可靠 —— 约定会被忘记，红了的用例不会。 + """ + + #: 允许直接写 status 的地方（模型定义本体、收口模块、以及迁移脚本的历史回填） + ALLOWED: ClassVar[frozenset[str]] = frozenset( + { + "web/backend/models.py", + "web/backend/services/account_lifecycle.py", + } + ) + + def test_only_lifecycle_module_writes_account_status(self): + import re + from pathlib import Path + + root = Path(__file__).resolve().parents[1] + backend = root / "web" / "backend" + patterns = ( + re.compile(r"\b(?:account|acc)\.status\s*=(?!=)"), + # 批量更新：update({Account.status: ...}) + re.compile(r"\{\s*Account\.status\s*:"), + # bulk update 的多行写法 + re.compile(r"^\s*\{\s*Account\.status\s*,"), + ) + offenders: list[str] = [] + for path in backend.rglob("*.py"): + rel = path.relative_to(root).as_posix() + # 虎牙是**另一套账号模型**（`huya_accounts.status` 自由文本， + # 20+ 种取值当状态机用，与这里的 bind_status 是同一个病）。 + # 那一摊单独治理，不在本不变量范围内。 + if rel in self.ALLOWED or "/migrations/" in rel or "huya" in rel: + continue + for lineno, line in enumerate( + path.read_text(encoding="utf-8").splitlines(), 1 + ): + if line.lstrip().startswith("#"): + continue + if any(p.search(line) for p in patterns): + offenders.append(f"{rel}:{lineno}: {line.strip()}") + + assert not offenders, ( + "account.status 只能由 account_lifecycle 写（bind_status 的教训）：\n" + "请改用 sync_status() / set_manual_state()。违规点：\n" + + "\n".join(offenders) + ) diff --git a/web/backend/routers/accounts.py b/web/backend/routers/accounts.py index 0b87c6f..bb996a1 100644 --- a/web/backend/routers/accounts.py +++ b/web/backend/routers/accounts.py @@ -30,11 +30,15 @@ from ..schemas import ( BatchAssign, ) from ..services.account_events_service import ( + CATEGORY_OPS, + EVENT_ASSIGNED, EVENT_CK_INVALID, EVENT_ELITE_OPENED, EVENT_ESPORTS_OPENED, EVENT_EXCHANGE, EVENT_RECHARGE, + EVENT_TAG_CHANGED, + emit_bulk, ) from ..services.account_lifecycle import ( STATUS_ARCHIVED, @@ -45,6 +49,7 @@ from ..services.account_service import ( cookie_account_ids_query, parse_and_build_accounts, ) +from ..services.audit_service import record_account_action, record_audit from ..services.douyu_service import cookie_holding_filter router = APIRouter(prefix="/api/accounts", tags=["账号管理"]) @@ -336,13 +341,16 @@ def assign_account( db.query(DouyuTask).filter(DouyuTask.account_id == acc.id).update( {DouyuTask.account_assigned_to: req.assigned_to}, synchronize_session=False ) - db.add( - AuditLog( - user_id=current.id, - username=current.username, - action="account:assign", - target=acc.username, - ) + target = db.query(User).filter(User.id == req.assigned_to).first() + target_name = target.username if target else "" + record_account_action( + db, + current, + account=acc, + action="account:assign", + event_type=EVENT_ASSIGNED, + title=f"分配给 {target_name}" if target_name else "取消分配", + detail={"assigned_to": req.assigned_to, "assigned_username": target_name}, ) db.commit() return {"message": "已分配", "success": True} @@ -389,6 +397,8 @@ def batch_assign_accounts( detail=f"以下账号尚未成功登录，无法分配：{name_list}{suffix}", ) + # 先取出账号对象（批量时间线要逐个写事件，光有 count 不够） + target_accounts = db.query(Account).filter(Account.id.in_(req.account_ids)).all() count = ( db.query(Account) .filter(Account.id.in_(req.account_ids)) @@ -397,13 +407,40 @@ def batch_assign_accounts( db.query(DouyuTask).filter(DouyuTask.account_id.in_(req.account_ids)).update( {DouyuTask.account_assigned_to: req.assigned_to}, synchronize_session=False ) - db.add( - AuditLog( - user_id=current.id, - username=current.username, - action="account:assign", - target=f"批量{'分配' if req.assigned_to else '取消分配'}{count}个账号", - ) + target_user = ( + db.query(User).filter(User.id == req.assigned_to).first() + if req.assigned_to is not None + else None + ) + target_name = target_user.username if target_user else "" + action_word = "分配" if req.assigned_to else "取消分配" + title = f"批量分配：{target_name}" if target_name else "批量取消分配" + record_audit( + db, + current, + action="account:assign", + target=f"批量{action_word}{count}个账号", + detail={"assigned_to": req.assigned_to, "assigned_username": target_name}, + ) + # 每个账号的时间线各留一条 —— 「这个账号什么时候被分配给谁」是账号级事实。 + emit_bulk( + db, + [ + { + "account_id": acc.id, + "event_type": EVENT_ASSIGNED, + "title": title, + "category": CATEGORY_OPS, + "detail": { + "assigned_to": req.assigned_to, + "assigned_username": target_name, + }, + "source": "audit", + "actor_type": "user", + "actor_id": current.id, + } + for acc in target_accounts + ], ) db.commit() action = "分配" if req.assigned_to else "取消分配" @@ -526,6 +563,45 @@ def set_account_automation( } +def _apply_tag_bulk(db: Session, current: User, ids: list[int], tag: str) -> None: + """批量改标签：更新 + 审计 + **逐账号写事件**。 + + 先取出对象再更新：`bulk update` 不加载实体，而每个账号的时间线都要各留一条 + 「标签从什么改成了什么」——这正是本仓库以前的缺口（`tag_changed` 事件类型 + 定义了、前端也渲染了，却从来没有代码发过）。 + """ + accounts = _visible_accounts_query(db, current).filter(Account.id.in_(ids)).all() + if not accounts: + return + previous = {acc.id: (acc.tag or "") for acc in accounts} + db.query(Account).filter(Account.id.in_([a.id for a in accounts])).update( + {Account.tag: tag}, synchronize_session=False + ) + record_audit( + db, + current, + action="account:tag", + target=f"批量设置标签「{tag}」{len(accounts)} 个账号", + detail={"to": tag}, + ) + emit_bulk( + db, + [ + { + "account_id": acc.id, + "event_type": EVENT_TAG_CHANGED, + "title": f"标签：{previous[acc.id] or '（空）'} → {tag or '（空）'}", + "category": CATEGORY_OPS, + "detail": {"from": previous[acc.id], "to": tag}, + "source": "audit", + "actor_type": "user", + "actor_id": current.id, + } + for acc in accounts + ], + ) + + @router.put("/{account_id}/tag") def set_account_tag( account_id: int, @@ -538,15 +614,16 @@ def set_account_tag( acc = _visible_accounts_query(db, current).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") + old_tag = acc.tag or "" acc.tag = (req.tag or "").strip() - db.add( - AuditLog( - user_id=current.id, - username=current.username, - action="account:tag", - target=acc.username, - detail=acc.tag, - ) + record_account_action( + db, + current, + account=acc, + action="account:tag", + event_type=EVENT_TAG_CHANGED, + title=f"标签：{old_tag or '（空）'} → {acc.tag or '（空）'}", + detail={"from": old_tag, "to": acc.tag}, ) db.commit() return {"message": "标签已更新", "success": True} @@ -563,11 +640,8 @@ def batch_tag( if not req.account_ids: raise HTTPException(status_code=400, detail="请选择账号") tag = (req.tag or "").strip() - count = ( - _visible_accounts_query(db, current) - .filter(Account.id.in_(req.account_ids)) - .update({Account.tag: tag}, synchronize_session=False) - ) + _apply_tag_bulk(db, current, req.account_ids, tag) + count = len(req.account_ids) db.commit() return {"message": f"已为 {count} 个账号设置标签", "success": True} @@ -584,11 +658,8 @@ def batch_tag_selection( if not ids: raise HTTPException(status_code=400, detail="请选择账号") tag = (req.tag_value or "").strip() - count = ( - _visible_accounts_query(db, current) - .filter(Account.id.in_(ids)) - .update({Account.tag: tag}, synchronize_session=False) - ) + _apply_tag_bulk(db, current, ids, tag) + count = len(ids) db.commit() scope = "当前筛选下" if req.all_matching else "选中的" return { @@ -644,13 +715,15 @@ def batch_delete_accounts( db.query(Account).filter(Account.id.in_(ids)).delete(synchronize_session=False) ) - db.add( - AuditLog( - user_id=current.id, - username=current.username, - action="account:delete", - target=f"批量删除{deleted}个账号", - ) + # 账号删除会**连带删掉它的事件时间线**（外键），所以留下的痕迹只有审计。 + # 批量删时 account_id 无法表达 N 个账号，改成把删掉的 id 记进 detail —— + # 出了事故才知道删了哪些，正是本项目 10-03 那次事故的教训。 + record_audit( + db, + current, + action="account:delete", + target=f"批量删除{deleted}个账号", + detail={"account_ids": ids[:200], "total_ids": len(ids), "deleted": deleted}, ) db.commit() return {"message": f"已删除 {deleted} 个账号", "deleted": deleted, "success": True} @@ -681,13 +754,12 @@ def batch_delete_accounts_selection( .filter(Account.id.in_(ids)) .delete(synchronize_session=False) ) - db.add( - AuditLog( - user_id=current.id, - username=current.username, - action="account:delete", - target=f"{'按筛选' if req.all_matching else '批量'}删除{deleted}个账号", - ) + record_audit( + db, + current, + action="account:delete", + target=f"{'按筛选' if req.all_matching else '批量'}删除{deleted}个账号", + detail={"account_ids": ids[:200], "total_ids": len(ids), "deleted": deleted}, ) db.commit() return {"message": f"已删除 {deleted} 个账号", "deleted": deleted, "success": True} @@ -809,12 +881,18 @@ def account_events( page: int = 1, page_size: int = 50, event_type: str = "", + category: str = Query( + "", description="按粗分类筛选：lifecycle / auth / business / ops" + ), db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """账号时间线（倒序分页）。 权限沿用账号可见性：客服只能看自己名下账号的事件。 + + `category` 是给页面用的「大分类」过滤（状态变迁 / 登录与 CK / 业务 / 人工操作）， + 比逐一点 `event_type` 好用：客服最常看的是「人工操作」与「状态变迁」两条线。 """ acc = _visible_accounts_query(db, current).filter(Account.id == account_id).first() if not acc: @@ -824,9 +902,14 @@ def account_events( page_size = max(1, min(int(page_size), 200)) where = "WHERE account_id = :aid" params: dict[str, Any] = {"aid": account_id} - if event_type: + sql_event_type = _query_str(event_type) + sql_category = _query_str(category) + if sql_event_type: where += " AND event_type = :etype" - params["etype"] = event_type + params["etype"] = sql_event_type + if sql_category: + where += " AND category = :category" + params["category"] = sql_category total = ( db.execute( @@ -839,8 +922,9 @@ def account_events( db.execute( text( f""" - SELECT id, event_type, occurred_at, title, amount, balance_after, - detail, source + SELECT id, event_type, category, occurred_at, title, amount, + balance_after, subject_type, subject_key, detail, source, + actor_type, actor_id FROM account_events {where} ORDER BY occurred_at DESC, id DESC LIMIT :limit OFFSET :offset diff --git a/web/backend/routers/cookies.py b/web/backend/routers/cookies.py index 0fc6696..d47cef7 100644 --- a/web/backend/routers/cookies.py +++ b/web/backend/routers/cookies.py @@ -33,7 +33,7 @@ from ..models import Account, AuditLog, LoginTask, User from ..models import ProxyConfig as ProxyConfigModel from ..permissions import get_user_permissions, user_has_permission from ..schemas import CookieReloginRequest -from ..services.account_lifecycle import sync_status +from ..services.account_lifecycle import on_ck_cleared, on_ck_verdict from ..services.cookie_check_service import check_douyu_cookie from ..services.douyu_service import ( LOGIN_ATTEMPT_IN_FLIGHT_STATUSES, @@ -513,17 +513,15 @@ def _check_cookies(ids: str, db: Session, current: User, *, detailed: bool) -> d acc = accounts_by_id.get(item["id"]) if not acc: continue - acc.ck_check_status = "valid" if item["valid"] else "invalid" - acc.ck_check_result = { - "fish_ball": item.get("fish_ball"), - "nickname": item.get("nickname"), - "level": item.get("level"), - "message": item.get("message", ""), - } - acc.ck_checked_at = checked_at - # 人工检测同样改变生命周期状态（正常 ↔ CK 失效） - sync_status( - db, acc, occurred_at=checked_at, actor_type="user", actor_id=current.id + # 人工检测结论同样走统一入口（写结论 + 由好变坏记事件 + 重算状态） + on_ck_verdict( + db, + acc, + "valid" if item["valid"] else "invalid", + detail=item, + occurred_at=checked_at, + actor_type="user", + actor_id=current.id, ) # 审计带上账号：单个检测时可直接反查（批量时只能落到 target 文本里） db.add( @@ -787,13 +785,11 @@ def delete_cookies_batch( ) now = datetime.now(UTC) for acc in accounts: - acc.cookie = "" - acc.cookie_updated_at = now - acc.ck_check_status = "" - acc.ck_check_result = None - acc.ck_checked_at = None - # 没有 CK 就跑不了业务 → 状态回到「未登录」 - sync_status(db, acc, occurred_at=now, actor_type="user", actor_id=current.id) + # 清 CK 是**不可逆**操作：清本体 + 清健康状态 + 记事件 + 重算状态， + # 全在 on_ck_cleared 一处（单个清除与批量清除共用）。 + on_ck_cleared( + db, acc, occurred_at=now, actor_id=current.id, reason="manual_clear_batch" + ) # 批量清 CK 涉及多个账号，只能落到 target 文本里；单个账号时补 account_id， # 便于审计页/账号时间线按索引反查。 db.add( @@ -835,13 +831,7 @@ def delete_cookie( account_id = acc.id username = acc.username now = datetime.now(UTC) - acc.cookie = "" - acc.cookie_updated_at = now - acc.ck_check_status = "" - acc.ck_check_result = None - acc.ck_checked_at = None - # 没有 CK 就跑不了业务 → 状态回到「未登录」 - sync_status(db, acc, occurred_at=now, actor_type="user", actor_id=current.id) + on_ck_cleared(db, acc, occurred_at=now, actor_id=current.id) db.add( AuditLog( user_id=current.id, diff --git a/web/backend/services/account_events_service.py b/web/backend/services/account_events_service.py index 6d4eedf..a305651 100644 --- a/web/backend/services/account_events_service.py +++ b/web/backend/services/account_events_service.py @@ -36,6 +36,7 @@ from sqlalchemy.orm import Session EVENT_IMPORTED = "imported" # 账号导入（读 accounts.created_at，一般不落表） EVENT_LOGIN_OK = "login_ok" # 登录成功（含新 CK） EVENT_CK_INVALID = "ck_invalid" # CK 判定失效 +EVENT_CK_CLEARED = "ck_cleared" # CK 被清除（人工清除 / 账号删除前的清理） EVENT_RELOGIN_OK = "relogin_ok" # 自动重登成功 EVENT_RELOGIN_FAILED = "relogin_failed" # 自动重登失败（转人工/退避） EVENT_BIND_OK = "bind_ok" # 绑定/换绑 @@ -84,6 +85,60 @@ def _now() -> datetime: return datetime.now(UTC) +_INSERT_SQL = text( + """ + INSERT INTO account_events ( + account_id, event_type, category, occurred_at, title, + amount, balance_after, subject_type, subject_key, + detail, source, actor_type, actor_id, source_ref, created_at + ) VALUES ( + :account_id, :event_type, :category, :occurred_at, :title, + :amount, :balance_after, :subject_type, :subject_key, + :detail, :source, :actor_type, :actor_id, :source_ref, :created_at + ) + """ +) + + +def _event_params( + *, + account_id: int, + event_type: str, + title: str, + occurred_at: datetime | None = None, + amount: int | None = None, + balance_after: int | None = None, + detail: dict[str, Any] | None = None, + source: str = "manual", + source_ref: str | None = None, + category: str = "", + subject_type: str = "", + subject_key: str = "", + actor_type: str = "", + actor_id: int | None = None, +) -> dict[str, Any]: + """把一次调用归一成落库参数（`emit` 与 `emit_bulk` 共用，避免 SQL 写两遍）。""" + return { + "account_id": account_id, + "event_type": event_type, + "category": (category or "")[:16], + "occurred_at": to_naive_utc(occurred_at or _now()), + "title": (title or "")[:256], + "amount": amount, + "balance_after": balance_after, + "subject_type": (subject_type or "")[:16], + "subject_key": (subject_key or "")[:96], + "detail": json.dumps(detail, ensure_ascii=False, default=str) + if detail + else None, + "source": source[:24], + "actor_type": (actor_type or "")[:16], + "actor_id": actor_id, + "source_ref": (source_ref or None), + "created_at": to_naive_utc(_now()), + } + + def emit( db: Session, *, @@ -117,50 +172,65 @@ def emit( if not account_id or not event_type: return False - sql = text( - """ - INSERT INTO account_events ( - account_id, event_type, category, occurred_at, title, - amount, balance_after, subject_type, subject_key, - detail, source, actor_type, actor_id, source_ref, created_at - ) VALUES ( - :account_id, :event_type, :category, :occurred_at, :title, - :amount, :balance_after, :subject_type, :subject_key, - :detail, :source, :actor_type, :actor_id, :source_ref, :created_at - ) - """ + params = _event_params( + account_id=account_id, + event_type=event_type, + title=title, + occurred_at=occurred_at, + amount=amount, + balance_after=balance_after, + detail=detail, + source=source, + source_ref=source_ref, + category=category, + subject_type=subject_type, + subject_key=subject_key, + actor_type=actor_type, + actor_id=actor_id, ) - params = { - "account_id": account_id, - "event_type": event_type, - "category": (category or "")[:16], - "occurred_at": to_naive_utc(occurred_at or _now()), - "title": (title or "")[:256], - "amount": amount, - "balance_after": balance_after, - "subject_type": (subject_type or "")[:16], - "subject_key": (subject_key or "")[:96], - "detail": json.dumps(detail, ensure_ascii=False, default=str) - if detail - else None, - "source": source[:24], - "actor_type": (actor_type or "")[:16], - "actor_id": actor_id, - "source_ref": (source_ref or None), - "created_at": to_naive_utc(_now()), - } try: # SAVEPOINT：即使这条插入失败（例如 source_ref 撞唯一键）， # 也只回滚它自己，调用方事务里已完成的写入不受影响。 with db.begin_nested(): - db.execute(sql, params) + db.execute(_INSERT_SQL, params) return True except Exception as exc: # noqa: BLE001 - 旁路，任何失败都不上抛 logger.warning(f"写入账号事件失败（account={account_id} {event_type}）：{exc}") return False +def emit_bulk(db: Session, events: list[dict[str, Any]]) -> int: + """一次写多条事件，返回实际写入条数。 + + 用于**批量账号操作**（批量分配 / 批量改标签）：一次操作可能覆盖几千个账号， + 每个账号的时间线都该各留一条「谁在什么时候改了什么」。逐条走 `emit()` 会给 + 每条开一个 SAVEPOINT，几千条时请求明显变慢 —— 这里改成一次 executemany， + 整批包在**一个** SAVEPOINT 里。 + + 与 `emit()` 一样**绝不抛异常**：整批失败就整批丢弃（记 warning）， + 不会污染调用方的事务，也不会让「批量分配」因为记不上时间线而失败。 + + 每个 dict 的键与 `emit()` 的关键字参数同名（`account_id` / `event_type` / + `title` 必填）。 + """ + rows = [ + _event_params(**{**item, "occurred_at": item.get("occurred_at")}) + for item in events + if item.get("account_id") and item.get("event_type") + ] + if not rows: + return 0 + + try: + with db.begin_nested(): + db.execute(_INSERT_SQL, rows) + return len(rows) + except Exception as exc: # noqa: BLE001 - 旁路，任何失败都不上抛 + logger.warning(f"批量写入账号事件失败（{len(rows)} 条）：{exc}") + return 0 + + # ── douyu_tasks 增量收集 ────────────────────────────────── # 只收集「业务事件」，例行轮询（query_*）一律不进时间线 —— 它们在 douyu_tasks 里 # 占绝大多数（query_limited_goods 就 6.7 万行），收进去时间线没法看。 diff --git a/web/backend/services/account_lifecycle.py b/web/backend/services/account_lifecycle.py index eef9c42..45417dc 100644 --- a/web/backend/services/account_lifecycle.py +++ b/web/backend/services/account_lifecycle.py @@ -36,7 +36,10 @@ from sqlalchemy.orm import Session from ..models import Account from .account_events_service import ( + CATEGORY_AUTH, CATEGORY_LIFECYCLE, + EVENT_CK_CLEARED, + EVENT_CK_INVALID, EVENT_STATUS_CHANGED, emit, ) @@ -248,6 +251,61 @@ def set_manual_state( ) +def on_ck_verdict( + db: Session, + account: Account, + verdict: str, + *, + detail: dict[str, Any] | None = None, + occurred_at: datetime | None = None, + actor_type: str = "system", + actor_id: int | None = None, +) -> None: + """一次 CK 检测结论落定后的统一处理（**顺序固定**，避免各处自己拼）。 + + 1. 写 `ck_check_status` / `ck_check_result` / `ck_checked_at`； + `error` = 「查不到」，**保留上一次结论** —— 查不到不等于失效； + 2. **由好变坏**的那一刻记一条 `ck_invalid` 事件； + 3. 重算生命周期状态。 + + 第 2 步只认「转变」：巡检 6 小时一轮，如果每轮都记，一个失效一周的账号会攒下 + 28 条一模一样的「CK 检测失效」把时间线淹掉。判定失效是**一次状态转变**， + 不是每轮读数。（旧实现注释里写着「避免每轮刷屏」，代码却没做到。） + """ + previous = account.ck_check_status or "" + now = occurred_at or datetime.now(UTC) + if verdict != "error": + account.ck_check_status = verdict + data = detail or {} + account.ck_check_result = { + "fish_ball": data.get("fish_ball"), + "nickname": data.get("nickname"), + "level": data.get("level"), + "message": data.get("message", ""), + "verdict": verdict, + } + account.ck_checked_at = now + + if verdict == "invalid" and previous != "invalid": + emit( + db, + account_id=account.id, + event_type=EVENT_CK_INVALID, + title="CK 检测失效", + category=CATEGORY_AUTH, + occurred_at=now, + detail={ + "message": data.get("message", ""), + "nickname": data.get("nickname"), + }, + source="login", + actor_type=actor_type, + actor_id=actor_id, + ) + + sync_status(db, account, occurred_at=now, actor_type=actor_type, actor_id=actor_id) + + def on_login_success( db: Session, account: Account, @@ -270,6 +328,49 @@ def on_login_success( sync_status(db, account, occurred_at=occurred_at) +def on_ck_cleared( + db: Session, + account: Account, + *, + occurred_at: datetime | None = None, + actor_id: int | None = None, + reason: str = "manual_clear", +) -> None: + """清空账号的 CK：**CK 本体与健康状态一起清**，记一条 `ck_cleared`，再重算状态。 + + ⚠️ 这是**不可逆**操作（重登当前成功率极低，清掉很可能补不回来），所以时间线 + 必须留痕。单个清除与批量清除共用这一条路径 —— 以前是两处各写一遍， + 改一处漏一处只是时间问题。 + """ + now = occurred_at or datetime.now(UTC) + account.cookie = "" + account.cookie_updated_at = now + account.ck_check_status = "" + account.ck_check_result = None + account.ck_checked_at = None + + emit( + db, + account_id=account.id, + event_type=EVENT_CK_CLEARED, + title="CK 被清除", + category=CATEGORY_AUTH, + occurred_at=now, + source="audit", + actor_type="user" if actor_id else "system", + actor_id=actor_id, + detail={"reason": reason}, + ) + # 没有 CK 就跑不了业务 → 状态回到「未登录」 + sync_status( + db, + account, + occurred_at=now, + actor_type="user" if actor_id else "system", + actor_id=actor_id, + ) + + def status_payload(account: Account) -> dict[str, Any]: """给接口用的状态片段（含中文标签，前端不必再维护一份映射）。""" return { diff --git a/web/backend/services/audit_service.py b/web/backend/services/audit_service.py index 6eee960..78725eb 100644 --- a/web/backend/services/audit_service.py +++ b/web/backend/services/audit_service.py @@ -4,11 +4,13 @@ from __future__ import annotations import json from collections.abc import Mapping +from datetime import datetime from typing import Any from sqlalchemy.orm import Session -from ..models import AuditLog, User +from ..models import Account, AuditLog, User +from .account_events_service import CATEGORY_OPS, emit _SENSITIVE_KEY_PARTS = ( "cookie", @@ -71,3 +73,57 @@ def record_audit( ) db.add(entry) return entry + + +def record_account_action( + db: Session, + current: User | None, + *, + account: Account, + action: str, + event_type: str, + title: str, + target: str = "", + detail: Mapping[str, Any] | str | None = None, + success: bool = True, + occurred_at: datetime | None = None, +) -> AuditLog: + """账号相关操作：**审计与账号时间线一次写完**。 + + 为什么合成一个函数：这两件事必须成对出现 —— + 只写审计，账号时间线里就缺一条「谁在什么时候改了什么」； + 只发事件，合规审计里就缺一条。 + 分两次调用写，迟早有人漏一边（S4 之前 `assigned` / `tag_changed` + 这两个事件类型定义了、前端也渲染了，但**从来没有代码发过**）。 + + 顺带把 `account_id` 落到审计上：审计页可以按账号过滤，且账号被删除后 + 仍然查得到 —— `audit_logs.account_id` 是普通整型列、**没有外键**， + 所以允许指向已删除的账号（这正是删除账号时也记它的原因）。 + """ + entry = record_audit( + db, + current, + action=action, + target=target or account.username, + detail=detail, + success=success, + ) + entry.account_id = account.id + + # 事件是旁路：emit 内部吞异常，记不上也绝不影响业务写入。 + emit( + db, + account_id=account.id, + event_type=event_type, + title=title, + category=CATEGORY_OPS, + occurred_at=occurred_at, + # 细节同时给事件一份（先脱敏，避免凭据类字段进时间线）。 + # Mapping 直接作为事件 detail，不再套一层 —— 时间线里看到的就是 + # {"from": ..., "to": ...}，前端不必再往里挖一层。 + detail=dict(_safe_value(dict(detail))) if isinstance(detail, Mapping) else None, + source="audit", + actor_type="user", + actor_id=current.id if current else None, + ) + return entry diff --git a/web/backend/services/automation/jobs/ck_health.py b/web/backend/services/automation/jobs/ck_health.py index 57d4bc4..bd00556 100644 --- a/web/backend/services/automation/jobs/ck_health.py +++ b/web/backend/services/automation/jobs/ck_health.py @@ -17,8 +17,7 @@ from sqlalchemy import or_ from sqlalchemy.orm import Session from ....models import Account, AutomationJob -from ...account_events_service import EVENT_CK_INVALID, emit -from ...account_lifecycle import MANUAL_STATES, sync_status +from ...account_lifecycle import MANUAL_STATES, on_ck_verdict from ...cookie_check_service import check_douyu_cookie from ...douyu_service import cookie_operable_filter @@ -159,40 +158,9 @@ def run_ck_health(db: Session, job: AutomationJob) -> dict[str, Any]: invalid += 1 else: errors += 1 - # error 时保留该账号原来的结论：查不到不等于失效，宁可维持上一次判断。 - if verdict != "error": - account.ck_check_status = verdict - account.ck_check_result = { - "fish_ball": result.get("fish_ball"), - "nickname": result.get("nickname"), - "level": result.get("level"), - "message": result.get("message", ""), - "verdict": verdict, - } - account.ck_checked_at = now - - # 巡检结论直接决定生命周期状态（正常 ↔ CK 失效），推导交给收口函数。 - sync_status(db, account, occurred_at=now) - - # 判定失效时记一条账号事件：`ck_check_status` / `ck_checked_at` 都是**覆盖式** - # 字段，覆盖之后就再也看不出「什么时候开始失效的」。只在真的判成 invalid 时记， - # 避免每轮巡检都给同一个账号刷屏。 - if verdict == "invalid": - emit( - db, - account_id=account.id, - event_type=EVENT_CK_INVALID, - title="CK 检测失效", - occurred_at=now, - detail={ - "message": result.get("message", ""), - "nickname": result.get("nickname"), - }, - source="login", - # 幂等键带检测时间：同一账号**再次**失效算新事件，但同一次检测 - # 重跑（例如失败重试）不会重复入库。 - source_ref=f"ck_invalid:{account.id}:{int(now.timestamp())}", - ) + # 结论落定 = 写 ck_check_* + 由好变坏时记事件 + 重算生命周期状态。 + # 三件事的**顺序与口径**收在 `on_ck_verdict` 里，四个检测入口共用。 + on_ck_verdict(db, account, verdict, detail=result, occurred_at=now) if index % FLUSH_EVERY == 0: db.commit() diff --git a/web/backend/services/douyu_runner.py b/web/backend/services/douyu_runner.py index 6d88980..47dc404 100644 --- a/web/backend/services/douyu_runner.py +++ b/web/backend/services/douyu_runner.py @@ -10,7 +10,7 @@ from core.douyu import DouyuActivityError from ..database import SessionLocal from ..models import DouyuTask -from .account_lifecycle import sync_status +from .account_lifecycle import on_ck_verdict from .cookie_check_service import check_douyu_cookie from .douyu_runner_bind import BindMixin from .douyu_runner_core import ( # noqa: F401 (douyu_batch_registry 供 routers 重导出) @@ -67,14 +67,14 @@ class DouyuBatchRunner( return cookie_check = check_douyu_cookie(cookie) - account.ck_check_status = "valid" if cookie_check["valid"] else "invalid" - account.ck_check_result = { - key: cookie_check.get(key) - for key in ("fish_ball", "nickname", "level", "message") - } - account.ck_checked_at = cookie_check["checked_at"] - # 执行前的这次检测同样会改变账号状态（正常 ↔ CK 失效）。 - sync_status(worker_db, account, occurred_at=cookie_check["checked_at"]) + # 执行前的这次检测同样要写结论、由好变坏记事件、重算状态。 + on_ck_verdict( + worker_db, + account, + "valid" if cookie_check["valid"] else "invalid", + detail=cookie_check, + occurred_at=cookie_check["checked_at"], + ) worker_db.commit() if not cookie_check["valid"]: message = f"Cookie 已失效，请重新登录：{cookie_check['message']}" diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 4d7eb2b..19fd360 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -22,7 +22,7 @@ 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_lifecycle import on_login_success +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 @@ -466,17 +466,15 @@ class LoginBatchRunner: message = result.message or "登录成功" if self.mode == "relogin" and account is not None: check_result = check_douyu_cookie(result.cookie) - # 复检结论写账号（CK 在哪，健康状态就在哪）。 - account.ck_check_status = ( - "valid" if check_result["valid"] else "invalid" + # 复检结论写账号（CK 在哪，健康状态就在哪），走统一入口： + # 写结论 + 由好变坏记事件 + 重算状态。 + on_ck_verdict( + worker_db, + account, + "valid" if check_result["valid"] else "invalid", + detail=check_result, + occurred_at=check_result["checked_at"], ) - account.ck_check_result = { - "fish_ball": check_result["fish_ball"], - "nickname": check_result["nickname"], - "level": check_result["level"], - "message": check_result["message"], - } - account.ck_checked_at = check_result["checked_at"] if check_result["valid"]: message = "重新登录成功，Cookie 有效" else: diff --git a/web/frontend/src/api/accounts.ts b/web/frontend/src/api/accounts.ts index ae5166b..f923314 100644 --- a/web/frontend/src/api/accounts.ts +++ b/web/frontend/src/api/accounts.ts @@ -49,8 +49,17 @@ export const accountApi = { // 详情页：状态卡 + 统计 overview: (id: number) => api.get(`/accounts/${id}/overview`), // 时间线（倒序分页），event_type 为空表示全部 - events: (id: number, params: { page?: number; page_size?: number; event_type?: string }) => + events: ( + id: number, + params: { page?: number; page_size?: number; event_type?: string; category?: string }, + ) => api.get(`/accounts/${id}/events`, { - params: { page: 1, page_size: 50, ...params, event_type: params.event_type || '' }, + params: { + page: 1, + page_size: 50, + ...params, + event_type: params.event_type || '', + category: params.category || '', + }, }), }; diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index a2304c3..4eeeee8 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -92,13 +92,19 @@ export type AccountEventType = export interface AccountEventItem { id: number; event_type: AccountEventType | string; + /** 粗分类：lifecycle / auth / business / ops */ + category?: string; occurred_at: string; title: string; /** 有符号：充值为正、兑换消耗为负 */ amount: number | null; balance_after: number | null; + subject_type?: string; + subject_key?: string; detail: Record | null; source: string; + actor_type?: string; + actor_id?: number | null; } export interface AccountEventPage { diff --git a/web/frontend/src/pages/AccountDetailPage.tsx b/web/frontend/src/pages/AccountDetailPage.tsx index 710a567..f6aa98f 100644 --- a/web/frontend/src/pages/AccountDetailPage.tsx +++ b/web/frontend/src/pages/AccountDetailPage.tsx @@ -22,8 +22,10 @@ const EVENT_STYLE: Record = { imported: { label: '导入', color: 'default' }, login_ok: { label: '登录', color: 'blue' }, ck_invalid: { label: 'CK 失效', color: 'red' }, + ck_cleared: { label: 'CK 已清除', color: 'red' }, relogin_ok: { label: '重登成功', color: 'green' }, relogin_failed: { label: '重登转人工', color: 'orange' }, + status_changed: { label: '状态变更', color: 'geekblue' }, bind_ok: { label: '绑定', color: 'cyan' }, recharge: { label: '充值', color: 'green' }, exchange: { label: '兑换', color: 'purple' }, @@ -34,6 +36,14 @@ const EVENT_STYLE: Record = { tag_changed: { label: '标签', color: 'default' }, }; +/** 事件粗分类（与后端 `account_events_service.CATEGORY_*` 对应）。 */ +const CATEGORY_OPTIONS = [ + { value: 'lifecycle', label: '状态变迁' }, + { value: 'auth', label: '登录与 CK' }, + { value: 'business', label: '业务流水' }, + { value: 'ops', label: '人工操作' }, +]; + /** 生命周期状态 → 展示色（与账号列表页保持一致，取值域见后端 account_lifecycle）。 */ const STATUS_TONES: Record = { imported: 'default', @@ -78,6 +88,7 @@ export default function AccountDetailPage() { const [page, setPage] = useState(1); const [pageSize, setPageSize] = useState(20); const [eventType, setEventType] = useState(''); + const [category, setCategory] = useState(''); const [eventsLoading, setEventsLoading] = useState(false); const loadOverview = useCallback(async () => { @@ -100,6 +111,7 @@ export default function AccountDetailPage() { page, page_size: pageSize, event_type: eventType, + category, }); setEvents(data.items); setEventTotal(data.total); @@ -108,7 +120,7 @@ export default function AccountDetailPage() { } finally { setEventsLoading(false); } - }, [accountId, page, pageSize, eventType]); + }, [accountId, page, pageSize, eventType, category]); useEffect(() => { void loadOverview(); @@ -303,6 +315,15 @@ export default function AccountDetailPage() { title="账号时间线" extra={ +