import csv import io from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, Response from sqlalchemy import and_, case, func, or_, select, text from sqlalchemy.orm import Session, defer, joinedload from ..database import get_db from ..deps import get_current_user, require_permission from ..models import ( Account, AccountEvent, AccountLease, AuditLog, DouyuTask, DouyuWorkbenchAccount, LoginTask, User, ) from ..permissions import user_has_permission from ..schemas import ( AccountAssign, AccountAutomationUpdate, AccountBulkSelection, AccountBulkTag, AccountExportRequest, AccountImport, AccountLifecycleUpdate, AccountOut, AccountRemark, AccountTag, BatchAssign, ) from ..services.account_binding import bind_payload from ..services.account_events_service import ( CATEGORY_OPS, EVENT_ASSIGNED, EVENT_CK_INVALID, EVENT_ELITE_OPENED, EVENT_ESPORTS_OPENED, EVENT_EXCHANGE, EVENT_RECHARGE, EVENT_REMARK_CHANGED, EVENT_TAG_CHANGED, OFFICIAL_SYNC_SOURCE, emit_bulk, ) from ..services.account_lifecycle import ( STATUS_ACTIVE, STATUS_ARCHIVED, STATUS_DISABLED, STATUS_IMPORTED, STATUS_INVALID, STATUS_LABELS, STATUS_MANUAL, set_manual_state, ) from ..services.account_service import ( parse_and_build_accounts, ) from ..services.account_value import ( SUIT_BAIRIMENG, SUIT_MENGTU, is_high_value_expr, is_suit_queried_expr, suit_owned_expr, suit_sort_rank_expr, ) from ..services.audit_service import record_account_action, record_audit from ..services.douyu_service import cookie_holding_filter from ..services.login_failure import classify_failure, failure_kind_label router = APIRouter(prefix="/api/accounts", tags=["账号管理"]) def _visible_accounts_query(db: Session, current: User): """返回当前用户可见的斗鱼账号查询。 可见 = **我名下** ∪ **我当前持有有效租约**。 后半条是账号池加的：池化后客服名下不再挂账号（归属是管理员/运营的维护责任）， 不把自己租到的号算进来，客服连刚做完的那单都看不到。 """ query = db.query(Account) if user_has_permission(current, "account:view_all"): return query if user_has_permission(current, "account:view_assigned"): my_leases = select(AccountLease.account_id).where( AccountLease.holder_user_id == current.id, AccountLease.status == "active", ) return query.filter( or_(Account.assigned_to == current.id, Account.id.in_(my_leases)) ) raise HTTPException(status_code=403, detail="无权查看账号") def _query_str(value) -> str: """把查询参数安全地取成字符串。 存在的理由：本仓库的测试**直接调用路由函数**（不经 FastAPI），此时 `Query(...)` 的默认值不会被解析，参数拿到的仍是 `Query` 对象 —— 对它调 `.strip()` 会炸。这里统一兜住，免得每加一个筛选参数就要改一遍用例。 """ return value.strip() if isinstance(value, str) else "" def _query_bool(value) -> bool: return value is True def _filter_accounts_query( db: Session, query, current: User, *, assigned_only: bool = False, assign_status: str = "", assigned_user_id: int | None = None, tag: str | None = None, has_cookie: bool = False, search: str = "", status: str = "", ck_status: str = "", failure_kind: str = "", bind_status: str = "", high_value_filter: str = "", min_points: int | None = None, max_points: int | None = None, min_gold: float | None = None, max_gold: float | None = None, elite_opened: bool | None = None, esports_opened: bool | None = None, role_bind_status: str = "", ): """复用列表筛选条件，供分页列表和全量批量操作保持一致。""" if _query_bool(has_cookie): query = query.filter(Account.cookie.isnot(None), Account.cookie != "") if _query_bool(assigned_only) and user_has_permission(current, "account:view_all"): query = query.filter(Account.assigned_to.isnot(None)) # 归属筛选：未分配 / 已分配，再叠加「指定客服」。与其它条件（标签、CK、 # 搜索…）是**同时生效**的，这正是「综合筛选」的用法。 assign_value = _query_str(assign_status) if assign_value == "unassigned": query = query.filter(Account.assigned_to.is_(None)) elif assign_value == "assigned": query = query.filter(Account.assigned_to.isnot(None)) # isinstance 兜住「测试直接调路由函数、Query 默认值未解析」的情况。 if isinstance(assigned_user_id, int) and not isinstance(assigned_user_id, bool): query = query.filter(Account.assigned_to == assigned_user_id) tag_value = _query_str(tag) if tag_value: query = query.filter(Account.tag == tag_value) status_value = _query_str(status) if status_value: query = query.filter(Account.status == status_value) else: # 归档账号默认不出现 —— 「归档」本身就是「收起来别烦我」的意思。 # 要看它们必须显式带 status=archived。 query = query.filter(Account.status != STATUS_ARCHIVED) ck_value = _query_str(ck_status) if ck_value: query = query.filter(Account.ck_check_status == ck_value) fk_value = _query_str(failure_kind) if fk_value: query = query.filter(Account.relogin_failure_kind == fk_value) bind_value = _query_str(bind_status) if bind_value: # 三个作用域任一命中即算（列表上「绑定」是一列，不分开筛） query = query.filter( or_( Account.role_bind_status == bind_value, Account.esports_bind_status == bind_value, Account.xpd_bind_status == bind_value, ) ) role_bind = _query_str(role_bind_status) if role_bind: query = query.filter(Account.role_bind_status == role_bind) # 高价值套装筛选 hv = _query_str(high_value_filter) if hv: suit_queried = is_suit_queried_expr() bairimeng_expr = suit_owned_expr(Account.high_value_owned, SUIT_BAIRIMENG) mengtu_expr = suit_owned_expr(Account.high_value_owned, SUIT_MENGTU) if hv == "both_available": query = query.filter( suit_queried, ~bairimeng_expr, ~mengtu_expr, ) elif hv == "can_bairimeng": query = query.filter( suit_queried, ~bairimeng_expr, ) elif hv == "can_mengtu": query = query.filter( suit_queried, ~mengtu_expr, ) elif hv == "owned_bairimeng": query = query.filter(bairimeng_expr) elif hv == "owned_mengtu": query = query.filter(mengtu_expr) elif hv == "both_owned": query = query.filter(bairimeng_expr, mengtu_expr) elif hv == "unqueried": query = query.filter(Account.overview_queried_at.is_(None)) elif hv == "high_value_broken": query = query.filter( is_high_value_expr(), Account.ck_check_status.in_(["invalid", "error"]), ) # 资产数值区间过滤 if isinstance(min_points, int): query = query.filter(Account.points >= min_points) if isinstance(max_points, int): query = query.filter(Account.points <= max_points) if isinstance(min_gold, (int, float)): query = query.filter(Account.gold_balance >= min_gold) if isinstance(max_gold, (int, float)): query = query.filter(Account.gold_balance <= max_gold) # 开通状态过滤 if isinstance(elite_opened, bool): query = query.filter(Account.elite_opened.is_(elite_opened)) if isinstance(esports_opened, bool): query = query.filter(Account.esports_opened.is_(esports_opened)) search_text = _query_str(search) if search_text: pattern = f"%{search_text}%" query = query.filter( or_( Account.username.ilike(pattern), Account.tag.ilike(pattern), Account.remark.ilike(pattern), ) ) return query def _selected_account_ids( db: Session, current: User, req: AccountBulkSelection ) -> list[int]: """解析批量操作目标：当前筛选全部或显式选择的 ID。""" if req.all_matching: rows = ( _filter_accounts_query( db, _visible_accounts_query(db, current), current, assigned_only=req.assigned_only, assign_status=req.assign_status, assigned_user_id=req.assigned_user_id, tag=req.tag, has_cookie=req.has_cookie, search=req.search, status=req.status, ck_status=req.ck_status, failure_kind=req.failure_kind, bind_status=req.bind_status, high_value_filter=req.high_value_filter, min_points=req.min_points, max_points=req.max_points, min_gold=req.min_gold, max_gold=req.max_gold, elite_opened=req.elite_opened, esports_opened=req.esports_opened, role_bind_status=req.role_bind_status, ) .with_entities(Account.id) .order_by(None) .all() ) return [account_id for (account_id,) in rows] seen = set() ids = [] for account_id in req.account_ids: if account_id not in seen: seen.add(account_id) ids.append(account_id) if not ids: return [] rows = ( _visible_accounts_query(db, current) .filter(Account.id.in_(ids)) .with_entities(Account.id) .order_by(None) .all() ) allowed = {account_id for (account_id,) in rows} return [account_id for account_id in ids if account_id in allowed] def _official_exchange_count_subquery(): """每个账号的官方兑换次数。只在按「兑换记录」排序时才接上，平时列表不扫这张表。""" return ( select( AccountEvent.account_id.label("account_id"), func.count(AccountEvent.id).label("exchange_count"), ) .where( AccountEvent.event_type == EVENT_EXCHANGE, AccountEvent.source == OFFICIAL_SYNC_SOURCE, ) .group_by(AccountEvent.account_id) .subquery() ) def _sort_dir(column, descending: bool): return column.desc() if descending else column.asc() def _nulls_last(column): """没查过的空值始终沉底，不跟 0 混在一起。""" return column.is_(None).asc() def _bind_rank(column): """已绑定 → 待确认 → 未绑定 → 未知。和列表上的颜色一致，不按字母排。""" return case( (column == "bound", 0), (column == "pending", 1), (column == "unbound", 2), else_=3, ) def _status_rank(): """正常 → CK 失效 → 待人工 → 未登录 → 已停用 → 已归档。""" return case( (Account.status == STATUS_ACTIVE, 0), (Account.status == STATUS_INVALID, 1), (Account.status == STATUS_MANUAL, 2), (Account.status == STATUS_IMPORTED, 3), (Account.status == STATUS_DISABLED, 4), (Account.status == STATUS_ARCHIVED, 5), else_=6, ) def _account_list_order(sort, order): """列表排序。没指定时仍按 id，别的调用方不受影响。 返回 (是否自定义排序, order_by 子句, 兑换次数子查询或 None)。 自定义排序先只取出 id，再按这个顺序装账号，避免 filesort 把 cookie 整列拖进来。 """ sort_key = _query_str(sort) descending = _query_str(order).lower() == "desc" tie = Account.id.desc() if descending else Account.id.asc() if sort_key == "username": return True, (_sort_dir(Account.username, descending), tie), None if sort_key == "status": return True, (_sort_dir(_status_rank(), descending), tie), None if sort_key == "bind": return ( True, ( _sort_dir(_bind_rank(Account.role_bind_status), descending), _sort_dir(_bind_rank(Account.esports_bind_status), descending), _sort_dir(_bind_rank(Account.xpd_bind_status), descending), tie, ), None, ) if sort_key == "points": return ( True, ( _nulls_last(Account.points), _sort_dir(Account.points, descending), _nulls_last(Account.gold_balance), _sort_dir(Account.gold_balance, descending), tie, ), None, ) if sort_key == "high_value": return True, (_sort_dir(suit_sort_rank_expr(), descending), tie), None if sort_key == "tag": return True, (_sort_dir(Account.tag, descending), tie), None if sort_key == "exchange": counts = _official_exchange_count_subquery() count_col = func.coalesce(counts.c.exchange_count, 0) return True, (_sort_dir(count_col, descending), Account.id.asc()), counts return False, (Account.id.asc(),), None def _require_account_tag_permission(current: User) -> None: """标签权限兼容历史上使用账号导入权限的运营人员。""" if not ( user_has_permission(current, "account:tag") or user_has_permission(current, "account:import") ): raise HTTPException(status_code=403, detail="无权修改账号标签") @router.get("") def list_accounts( assigned_only: bool = Query(False), assign_status: str = Query( "", description="归属筛选：unassigned=未分配 / assigned=已分配；空=全部" ), assigned_user_id: int | None = Query(None, description="只看某个客服名下的账号"), tag: str = Query(None), has_cookie: bool = Query(False), search: str = Query(""), status: str = Query("", description="按生命周期状态筛选；空=全部（不含归档）"), ck_status: str = Query( "", description="按 CK 健康结论筛选：valid / invalid / error" ), failure_kind: str = Query( "", description="按最近登录失败分型筛选：must_bind_mobile/captcha/credential/email_code/email_quota/network", ), bind_status: str = Query( "", description="按角色绑定筛选：unknown/unbound/pending/bound" ), high_value_filter: str = Query( "", description="按高价值套装状态筛选：both_available/can_bairimeng/can_mengtu/owned_bairimeng/owned_mengtu/both_owned/unqueried/high_value_broken", ), min_points: int | None = Query(None, description="最低积分"), max_points: int | None = Query(None, description="最高积分"), min_gold: float | None = Query(None, description="最低鱼翅"), max_gold: float | None = Query(None, description="最高鱼翅"), elite_opened: bool | None = Query(None, description="是否开通精英宝典"), esports_opened: bool | None = Query(None, description="是否开通电竞手册"), role_bind_status: str = Query( "", description="精英角色绑定状态: unknown/unbound/pending/bound" ), sort: str = Query( "", description="排序：username/status/bind/points/high_value/exchange/tag；空=按 id", ), order: str = Query("asc", description="asc / desc"), page: int | None = Query(None, ge=1), page_size: int = Query(20, ge=1, le=200), include_sensitive: bool = Query(False), db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """列表：按角色返回不同字段和范围。""" query = _filter_accounts_query( db, _visible_accounts_query(db, current), current, assigned_only=assigned_only, assign_status=assign_status, assigned_user_id=assigned_user_id, tag=tag, has_cookie=has_cookie, search=search, status=status, ck_status=ck_status, failure_kind=failure_kind, bind_status=bind_status, high_value_filter=high_value_filter, min_points=min_points, max_points=max_points, min_gold=min_gold, max_gold=max_gold, elite_opened=elite_opened, esports_opened=esports_opened, role_bind_status=role_bind_status, ) total = None if page is not None: total = query.order_by(None).with_entities(func.count(Account.id)).scalar() or 0 custom_sort, order_by, exchange_counts = _account_list_order(sort, order) ordered = query if exchange_counts is not None: ordered = ordered.outerjoin( exchange_counts, exchange_counts.c.account_id == Account.id ) can_include_sensitive = include_sensitive and user_has_permission( current, "account:view_full" ) options = [joinedload(Account.assigned_user)] if not can_include_sensitive: options.extend( [ defer(Account.password), defer(Account.email), defer(Account.email_password), ] ) if custom_sort: # 先按排序键取出这一页的 id。套装档位、兑换次数都不是索引列， # 全表 filesort 时不能把 cookie 一起选出来。 id_query = ordered.with_entities(Account.id).order_by(*order_by) if page is not None: id_query = id_query.offset((page - 1) * page_size).limit(page_size) page_ids = [account_id for (account_id,) in id_query.all()] if page_ids: loaded = { acc.id: acc for acc in db.query(Account) .filter(Account.id.in_(page_ids)) .options(*options) .all() } accounts = [ loaded[account_id] for account_id in page_ids if account_id in loaded ] else: accounts = [] else: ordered = ordered.order_by(*order_by) if page is not None: ordered = ordered.offset((page - 1) * page_size).limit(page_size) accounts = ordered.options(*options).all() # 「兑换记录」列：**每页只跑一条聚合查询**，不是每行一条（避免 N+1）。 # 数据源是账号自己的事件流（`account_events` 的 `exchange`）， # 而不是 `douyu_tasks` 的 JSON —— 后者是执行日志，S5 起业务事实已经在事件里了。 exchange_stats: dict[int, tuple[int, Any, str]] = {} ids = [acc.id for acc in accounts] if ids: # ① 次数 + 最近一次时间（分组聚合，走 (account_id, event_type, occurred_at) 复合索引） for account_id, count, last_at in ( db.query( AccountEvent.account_id, func.count(AccountEvent.id), func.max(AccountEvent.occurred_at), ) .filter( AccountEvent.account_id.in_(ids), AccountEvent.event_type == EVENT_EXCHANGE, # 列表的「兑换记录」只反映**斗鱼官方记录**（口径权威且完整）。 # 我们当场记的那些留给详情页时间线 —— 用户明确要求这么分。 AccountEvent.source == OFFICIAL_SYNC_SOURCE, ) .group_by(AccountEvent.account_id) .all() ): exchange_stats[int(account_id)] = (int(count or 0), last_at, "") # ② 最近一次兑换的商品名（窗口函数取每个账号的最新一条） # # ⚠️ 按 **`occurred_at`** 排，不能按 `id`： # 一键查询会把斗鱼官方的**历史**兑换记录补成事件，它们的 `id` 是刚插进去的 # （最新），但 `occurred_at` 是几年前的。按 id 排会把一条老兑换显示成"最近一次" # —— 实测就把 08-30 的兔兔白日梦错报成了 10-01（那是另一次兑换）。 ranked = ( db.query( AccountEvent.account_id.label("account_id"), AccountEvent.title.label("title"), func.row_number() .over( partition_by=AccountEvent.account_id, order_by=( AccountEvent.occurred_at.desc(), AccountEvent.id.desc(), ), ) .label("rn"), ) .filter( AccountEvent.account_id.in_(ids), AccountEvent.event_type == EVENT_EXCHANGE, AccountEvent.source == OFFICIAL_SYNC_SOURCE, ) .subquery() ) for account_id, title in ( db.query(ranked.c.account_id, ranked.c.title).filter(ranked.c.rn == 1).all() ): key = int(account_id) if key in exchange_stats: count, last_at, _ = exchange_stats[key] exchange_stats[key] = (count, last_at, str(title or "")) result = [] for acc in accounts: eff_kind = acc.relogin_failure_kind or "" if acc.relogin_last_error: refined = classify_failure(acc.relogin_last_error) if not eff_kind or (eff_kind == "network" and refined != "network"): eff_kind = refined item = AccountOut( id=acc.id, username=acc.username, remark=acc.remark or "", tag=acc.tag or "", assigned_to=acc.assigned_to, assigned_username=acc.assigned_user.username if acc.assigned_user else None, created_at=acc.created_at, status=acc.status or "", status_label=STATUS_LABELS.get(acc.status or "", acc.status or ""), status_reason=acc.status_reason or "", auto_managed=bool(acc.auto_managed), relogin_failure_kind=eff_kind, relogin_failure_label=failure_kind_label(eff_kind), relogin_last_error=acc.relogin_last_error or "", # 关键列：全部是 accounts 行上的字段，无 JOIN / 无聚合 has_cookie=bool(acc.cookie), ck_check_status=acc.ck_check_status or "", ck_checked_at=acc.ck_checked_at, points=acc.points, gold_balance=acc.gold_balance, exchange_count=exchange_stats.get(acc.id, (0, None, ""))[0], last_exchange_at=exchange_stats.get(acc.id, (0, None, ""))[1], last_exchange_name=exchange_stats.get(acc.id, (0, None, ""))[2], **bind_payload(acc), # 一键查询得到的「已兑换过的高价值套装」（不是整份快照 —— # 列表带不起 1~2KB/账号 的 JSON）。兑换条数不在这里： # 上面的 exchange_count 已经是**官方事件**聚合的结果。 high_value_owned=acc.high_value_owned, # 「有没有查过」靠它判断 —— 否则"查过但没兑换"和"没查过"都是空，分不开 overview_queried_at=acc.overview_queried_at, ) # 只有管理员可看完整字段（密码、邮箱等） if can_include_sensitive: item.password = acc.password item.email = acc.email item.email_password = acc.email_password result.append(item) if page is not None: return { "items": result, "total": total or 0, "page": page, "page_size": page_size, } return result @router.get("/summary") def accounts_summary( db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """账号管理统计，避免前端为了卡片统计拉全量账号。""" query = db.query(Account) if not user_has_permission(current, "account:view_all"): if user_has_permission(current, "account:view_assigned"): query = query.filter(Account.assigned_to == current.id) else: raise HTTPException(status_code=403, detail="无权查看账号") total = query.with_entities(func.count(Account.id)).scalar() or 0 assigned_count = ( query.filter(Account.assigned_to.isnot(None)) .with_entities(func.count(Account.id)) .scalar() or 0 ) tag_count = ( query.filter(Account.tag != "", Account.tag.isnot(None)) .with_entities(func.count(func.distinct(Account.tag))) .scalar() or 0 ) # 按**生命周期状态**分布：这才是可行动的（「有几个要处理」比「有几个标签」有用得多）。 # 归档默认不计入（与列表默认口径一致 —— 归档就是「收起来别烦我」）。 visible = query.filter(Account.status != STATUS_ARCHIVED) status_rows = ( visible.with_entities(Account.status, func.count(Account.id)) .group_by(Account.status) .all() ) by_status = {str(row[0] or ""): int(row[1] or 0) for row in status_rows} visible_total = sum(by_status.values()) # 登录失败分型统计（不含空值与归档） failure_rows = ( visible.filter( Account.relogin_failure_kind != "", Account.relogin_failure_kind.isnot(None), ) .with_entities(Account.relogin_failure_kind, func.count(Account.id)) .group_by(Account.relogin_failure_kind) .all() ) login_failures = {str(row[0] or ""): int(row[1] or 0) for row in failure_rows} # 高价值及核心资产分布（单次聚合，供前端快捷胶囊展示） bairimeng_expr = suit_owned_expr(Account.high_value_owned, SUIT_BAIRIMENG) mengtu_expr = suit_owned_expr(Account.high_value_owned, SUIT_MENGTU) high_value_broken_expr = and_( is_high_value_expr(), Account.ck_check_status.in_(["invalid", "error"]), ) suit_queried = is_suit_queried_expr() hv_row = visible.with_entities( func.count(case((and_(suit_queried, ~bairimeng_expr, ~mengtu_expr), 1))), func.count(case((and_(suit_queried, ~bairimeng_expr), 1))), func.count(case((and_(suit_queried, ~mengtu_expr), 1))), func.count(case((~suit_queried, 1))), func.count(case((Account.points >= 300, 1))), func.count(case((Account.gold_balance > 0, 1))), func.count(case((Account.elite_opened.is_(True), 1))), func.count(case((high_value_broken_expr, 1))), ).first() high_value_summary = { "both_available": int(hv_row[0] or 0) if hv_row else 0, "can_bairimeng": int(hv_row[1] or 0) if hv_row else 0, "can_mengtu": int(hv_row[2] or 0) if hv_row else 0, "unqueried": int(hv_row[3] or 0) if hv_row else 0, "high_points": int(hv_row[4] or 0) if hv_row else 0, "has_gold": int(hv_row[5] or 0) if hv_row else 0, "elite_opened": int(hv_row[6] or 0) if hv_row else 0, "high_value_broken": int(hv_row[7] or 0) if hv_row else 0, } return { "total": total, "visible_total": visible_total, # 生命周期分布（键是 6 状态的枚举值） "by_status": by_status, # 登录异常分类统计 "login_failures": login_failures, # 高价值与核心资产统计 "high_value": high_value_summary, # 需要人关注的：CK 已被判失效的账号数（含待人工） "ck_invalid_count": int( visible.filter(Account.ck_check_status == "invalid") .with_entities(func.count(Account.id)) .scalar() or 0 ), "no_cookie_count": int( visible.filter(or_(Account.cookie.is_(None), Account.cookie == "")) .with_entities(func.count(Account.id)) .scalar() or 0 ), "assigned_count": assigned_count, "unassigned_count": max(0, total - assigned_count), "tag_count": tag_count, } @router.post("/export") def export_accounts( req: AccountExportRequest, db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """导出账号：支持纯文本账号名单（txt）或完整详情（csv）。""" selected_ids = _selected_account_ids(db, current, req) if not selected_ids: if req.format == "csv": content = "\ufeff账号,生命周期状态,登录异常分类,最新报错,标签,归属客服,鱼翅,积分,精英宝典,电竞手册,CK状态,备注\n" return Response( content=content, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="accounts.csv"'}, ) return Response( content="", media_type="text/plain; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="accounts.txt"'}, ) accounts = ( db.query(Account) .options(joinedload(Account.assigned_user)) .filter(Account.id.in_(selected_ids)) .order_by(Account.id.asc()) .all() ) if req.format == "txt": # 纯账号名字，换行分隔 lines = [acc.username for acc in accounts if acc.username] text_content = "\n".join(lines) return Response( content=text_content, media_type="text/plain; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="accounts.txt"'}, ) # 导出 CSV (带 UTF-8 BOM，防止 Excel 打开乱码) output = io.StringIO() writer = csv.writer(output) writer.writerow( [ "账号", "生命周期状态", "登录异常分类", "最新报错", "标签", "归属客服", "鱼翅", "积分", "精英宝典", "电竞手册", "CK状态", "备注", ] ) for acc in accounts: eff_kind = acc.relogin_failure_kind or "" if acc.relogin_last_error: refined = classify_failure(acc.relogin_last_error) if not eff_kind or (eff_kind == "network" and refined != "network"): eff_kind = refined writer.writerow( [ acc.username or "", STATUS_LABELS.get(acc.status or "", acc.status or ""), failure_kind_label(eff_kind) if eff_kind else "", acc.relogin_last_error or "", acc.tag or "", acc.assigned_user.username if acc.assigned_user else "未分配", acc.gold_balance if acc.gold_balance is not None else "", acc.points if acc.points is not None else "", "已开通" if acc.elite_opened is True else ("未开通" if acc.elite_opened is False else "未知"), "已开通" if acc.esports_opened is True else ("未开通" if acc.esports_opened is False else "未知"), acc.ck_check_status or "", acc.remark or "", ] ) csv_content = "\ufeff" + output.getvalue() return Response( content=csv_content, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="accounts.csv"'}, ) @router.post("/import") def import_accounts( req: AccountImport, db: Session = Depends(get_db), current: User = Depends(require_permission("account:import")), ): """批量导入账号。格式：用户名|密码|邮箱|邮箱密码|标签(可选) req.tag 为统一标签兜底：某行未单独写标签时使用该值，行内标签优先。 库中已存在的用户名（大小写不敏感）与同批重复行会自动跳过。 """ accounts, skipped, duplicated = parse_and_build_accounts(db, req.text, req.tag) if accounts: db.add_all(accounts) db.add( AuditLog( user_id=current.id, username=current.username, action="account:import", target=f"导入{len(accounts)}个", ) ) db.commit() duplicated_note = f"，重复跳过 {duplicated} 个" if duplicated else "" return { "message": f"导入成功 {len(accounts)} 个，跳过 {skipped} 个{duplicated_note}", "success": True, "count": len(accounts), } @router.put("/{account_id}/assign") def assign_account( account_id: int, req: AccountAssign, db: Session = Depends(get_db), current: User = Depends(require_permission("account:assign")), ): acc = db.query(Account).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") if req.assigned_to: target = db.query(User).filter(User.id == req.assigned_to).first() if not target: raise HTTPException(status_code=404, detail="目标用户不存在") if target.role != "support": raise HTTPException(status_code=400, detail="只能分配给客服角色") # 只能分配当前仍持有 CK 的账号（CK 现在挂在账号上） has_success = ( db.query(Account) .filter( Account.id == account_id, cookie_holding_filter(), ) .first() ) if not has_success: raise HTTPException(status_code=400, detail="该账号尚未成功登录，无法分配") if acc.assigned_to != req.assigned_to: acc.assigned_to = req.assigned_to db.query(DouyuTask).filter(DouyuTask.account_id == acc.id).update( {DouyuTask.account_assigned_to: req.assigned_to}, synchronize_session=False ) 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} @router.post("/batch-assign") def batch_assign_accounts( req: BatchAssign, db: Session = Depends(get_db), current: User = Depends(require_permission("account:assign")), ): """批量分配/取消分配账号给客服。 `all_matching=True` 时按当前筛选条件解析目标（与批量打标签同一套语义）—— 「筛出未分配 → 一键分给某人」不必先手动勾选，这正是账号管理页 「归属筛选 + 快速分配」的组合用法。 """ ids = _selected_account_ids(db, current, req) if not ids: raise HTTPException(status_code=400, detail="请选择账号") # 验证目标用户 if req.assigned_to is not None: target = db.query(User).filter(User.id == req.assigned_to).first() if not target: raise HTTPException(status_code=404, detail="目标用户不存在") if target.role != "support": raise HTTPException(status_code=400, detail="只能分配给客服角色") # 只能分配当前仍持有 CK 的账号（CK 现在挂在账号上） cookie_ids_query = select(Account.id).where(cookie_holding_filter()) invalid_ids = ( db.query(Account.id) .filter( Account.id.in_(ids), Account.id.notin_(cookie_ids_query), ) .all() ) if invalid_ids: names = ( db.query(Account.username) .filter(Account.id.in_([i[0] for i in invalid_ids])) .all() ) name_list = ", ".join([n[0] for n in names[:5]]) suffix = "..." if len(invalid_ids) > 5 else "" raise HTTPException( status_code=400, detail=f"以下账号尚未成功登录，无法分配：{name_list}{suffix}", ) # 先取出账号对象（批量时间线要逐个写事件，光有 count 不够） target_accounts = db.query(Account).filter(Account.id.in_(ids)).all() count = ( db.query(Account) .filter(Account.id.in_(ids)) .update({Account.assigned_to: req.assigned_to}, synchronize_session=False) ) db.query(DouyuTask).filter(DouyuTask.account_id.in_(ids)).update( {DouyuTask.account_assigned_to: req.assigned_to}, synchronize_session=False ) 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 "取消分配" scope = "当前筛选下" if req.all_matching 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, "scope": "all_matching" if req.all_matching else "selected", }, ) # 每个账号的时间线各留一条 —— 「这个账号什么时候被分配给谁」是账号级事实。 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 "取消分配" return { "message": f"已批量{action}{scope} {count} 个账号", "success": True, "count": count, } @router.get("/assignments/summary") def assignments_summary( db: Session = Depends(get_db), current: User = Depends(require_permission("account:assign")), ): """分配概览：每个客服分配了多少账号（仅统计当前仍持有 CK 的账号）。""" results = ( db.query(User.id, User.username, func.count(Account.id).label("count")) .outerjoin( Account, and_( Account.assigned_to == User.id, Account.cookie.isnot(None), Account.cookie != "", ), ) .filter(User.role == "support") .group_by(User.id, User.username) .order_by(func.count(Account.id).desc()) .all() ) total_unassigned = ( db.query(func.count(Account.id)) .filter( Account.assigned_to.is_(None), Account.cookie.isnot(None), Account.cookie != "", ) .scalar() ) or 0 total_assigned = sum(int(cnt or 0) for _, _, cnt in results) total_accounts = total_assigned + total_unassigned return { "support_users": [ {"id": uid, "username": uname, "assigned_count": int(cnt or 0)} for uid, uname, cnt in results ], "unassigned_count": total_unassigned, "assigned_count": total_assigned, "total_count": total_accounts, } @router.put("/{account_id}/lifecycle") def set_account_lifecycle( account_id: int, req: AccountLifecycleUpdate, db: Session = Depends(get_db), current: User = Depends(require_permission("account:manage")), ): """人工停用 / 归档 / 恢复账号（生命周期状态机，见 `services/account_lifecycle`）。 - `disabled`：账号不可用（封号、退单、归还）—— 自动化**不再碰它**； - `archived`：软删除，默认列表不再展示； - `auto`：解除人工态，立刻按底层判据重新推导（例如账号又能用了）。 """ acc = db.query(Account).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") if acc.status == req.status: return {"message": "状态未变化", "status": acc.status, "success": True} set_manual_state( db, acc, req.status, reason=req.reason or "", actor_id=current.id, actor_type="user", ) db.add( AuditLog( user_id=current.id, username=current.username, action=f"account:lifecycle:{req.status}", target=acc.username, account_id=acc.id, detail=req.reason or "", ) ) db.commit() return { "message": "状态已更新", "status": acc.status, "success": True, } @router.put("/{account_id}/automation") def set_account_automation( account_id: int, req: AccountAutomationUpdate, db: Session = Depends(get_db), current: User = Depends(require_permission("account:manage")), ): """自动化开关。 与 `status=disabled` 的区别：`auto_managed=False` 的账号**仍然可用** （人照样能手动跑任务），只是巡检/自动重登绕过它 —— 用于敏感账号人工维护。 """ acc = db.query(Account).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") acc.auto_managed = bool(req.auto_managed) db.add( AuditLog( user_id=current.id, username=current.username, action="account:automation", target=acc.username, account_id=acc.id, detail="参与自动化" if req.auto_managed else "不参与自动化", ) ) db.commit() return { "message": "自动化开关已更新", "auto_managed": acc.auto_managed, "success": True, } @router.put("/{account_id}/remark") def set_account_remark( account_id: int, req: AccountRemark, db: Session = Depends(get_db), current: User = Depends(require_permission("account:remark")), ): """设置单个账号的备注。 备注是自由文本（标签是短枚举），所以走单个接口、不做批量 —— 批量给一堆账号写 同一句备注没有实际意义。与改标签一样：审计 + 账号时间线一次写完。 """ acc = db.query(Account).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") old_remark = acc.remark or "" new_remark = (req.remark or "").strip()[:256] if old_remark == new_remark: return {"message": "备注未变化", "success": True, "remark": new_remark} acc.remark = new_remark record_account_action( db, current, account=acc, action="account:remark", event_type=EVENT_REMARK_CHANGED, title=f"备注：{old_remark or '（空）'} → {new_remark or '（空）'}", detail={"from": old_remark, "to": new_remark}, ) db.commit() return {"message": "备注已更新", "success": True, "remark": new_remark} 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, req: AccountTag, db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """设置单个可见账号的标签。""" _require_account_tag_permission(current) 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() 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} @router.put("/batch-tag") def batch_tag( req: AccountTag, db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """批量设置账号标签。""" _require_account_tag_permission(current) if not req.account_ids: raise HTTPException(status_code=400, detail="请选择账号") tag = (req.tag or "").strip() _apply_tag_bulk(db, current, req.account_ids, tag) count = len(req.account_ids) db.commit() return {"message": f"已为 {count} 个账号设置标签", "success": True} @router.put("/batch-tag-selection") def batch_tag_selection( req: AccountBulkTag, db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """按显式选择或当前筛选结果批量设置账号标签。""" _require_account_tag_permission(current) ids = _selected_account_ids(db, current, req) if not ids: raise HTTPException(status_code=400, detail="请选择账号") tag = (req.tag_value or "").strip() _apply_tag_bulk(db, current, ids, tag) count = len(ids) db.commit() scope = "当前筛选下" if req.all_matching else "选中的" return { "message": f"已为{scope} {count} 个账号设置标签", "success": True, "count": count, } @router.get("/tags/list") def list_tags( db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """获取当前用户可见账号的标签列表。""" query = _visible_accounts_query(db, current) tags = ( query.filter(Account.tag != "", Account.tag.isnot(None)) .with_entities(Account.tag) .distinct() .all() ) return [t[0] for t in tags if t[0]] @router.delete("/batch/delete") def batch_delete_accounts( account_ids: str = Query(..., description="逗号分隔的账号ID"), db: Session = Depends(get_db), current: User = Depends(require_permission("account:delete")), ): """批量删除账号及其关联的登录任务。""" if not account_ids: raise HTTPException(status_code=400, detail="请指定账号ID") ids = [int(x) for x in account_ids.split(",") if x.strip().isdigit()] if not ids: raise HTTPException(status_code=400, detail="无效的账号ID") # 先删除关联的登录任务 db.query(LoginTask).filter(LoginTask.account_id.in_(ids)).delete( synchronize_session=False ) db.query(DouyuTask).filter(DouyuTask.account_id.in_(ids)).delete( synchronize_session=False ) db.query(DouyuWorkbenchAccount).filter( DouyuWorkbenchAccount.account_id.in_(ids) ).delete(synchronize_session=False) # 删除账号 deleted = ( db.query(Account).filter(Account.id.in_(ids)).delete(synchronize_session=False) ) # 账号删除会**连带删掉它的事件时间线**（外键），所以留下的痕迹只有审计。 # 批量删时 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} @router.post("/batch-delete") def batch_delete_accounts_selection( req: AccountBulkSelection, db: Session = Depends(get_db), current: User = Depends(require_permission("account:delete")), ): """按显式选择或当前筛选结果批量删除账号及其关联任务。""" ids = _selected_account_ids(db, current, req) if not ids: raise HTTPException(status_code=400, detail="请选择账号") db.query(LoginTask).filter(LoginTask.account_id.in_(ids)).delete( synchronize_session=False ) db.query(DouyuTask).filter(DouyuTask.account_id.in_(ids)).delete( synchronize_session=False ) db.query(DouyuWorkbenchAccount).filter( DouyuWorkbenchAccount.account_id.in_(ids) ).delete(synchronize_session=False) deleted = ( _visible_accounts_query(db, current) .filter(Account.id.in_(ids)) .delete(synchronize_session=False) ) 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} @router.get("/{account_id}/exchange-records") def account_exchange_records( account_id: int, db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """该账号在三个作用域下的**斗鱼官方兑换记录**（按作用域分组）。 列表的「兑换记录」格点开就用这个 —— **按需拉取**，不把记录塞进列表响应 （15000 个账号各带一堆记录会把响应撑爆）。 数据源是 `source=official_sync` 的事件，也就是一键查询从官方补录的那些， 它们带**真实兑换时间**。 """ if ( not _visible_accounts_query(db, current) .filter(Account.id == account_id) .first() ): raise HTTPException(status_code=404, detail="账号不存在或无权查看") rows = ( db.query(AccountEvent) .filter( AccountEvent.account_id == account_id, AccountEvent.event_type == EVENT_EXCHANGE, AccountEvent.source == OFFICIAL_SYNC_SOURCE, ) .order_by(AccountEvent.occurred_at.desc(), AccountEvent.id.desc()) .all() ) grouped: dict[str, list[dict[str, object]]] = { "elite": [], "esports": [], "peace": [], } for row in rows: detail = row.detail or {} scope = str(detail.get("scope") or "elite") grouped.setdefault(scope, []).append( { "title": row.title or "", "commodity_id": detail.get("commodity_id") or "", "commodity_name": detail.get("commodity_name") or "", "score": detail.get("score"), "show_name": detail.get("show_name") or "", "exchange_num": detail.get("exchange_num"), "occurred_at": row.occurred_at.isoformat() if row.occurred_at else None, } ) return { "groups": grouped, "total": sum(len(v) for v in grouped.values()), "queried_at": None, } @router.get("/{account_id}/overview") def account_overview( account_id: int, db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """账号详情页的「状态卡 + 统计」。 状态取 `accounts` 的当前值（覆盖式），统计由 `account_events` 聚合 —— 这正是 「状态与流水分离」的用法：页面上两者拼起来看。 """ acc = _visible_accounts_query(db, current).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") stats = ( db.execute( text( """ SELECT SUM(CASE WHEN event_type = :recharge THEN amount ELSE 0 END) AS recharge_total, SUM(CASE WHEN event_type = :exchange THEN 1 ELSE 0 END) AS exchange_count, SUM(CASE WHEN event_type = :exchange THEN -amount ELSE 0 END) AS points_spent, SUM(CASE WHEN event_type = :ck_invalid THEN 1 ELSE 0 END) AS ck_invalid_count, MIN(CASE WHEN event_type = :elite THEN occurred_at END) AS elite_opened_at, MIN(CASE WHEN event_type = :esports THEN occurred_at END) AS esports_opened_at, COUNT(*) AS event_count FROM account_events WHERE account_id = :aid """ ), { "aid": account_id, "recharge": EVENT_RECHARGE, "exchange": EVENT_EXCHANGE, "ck_invalid": EVENT_CK_INVALID, "elite": EVENT_ELITE_OPENED, "esports": EVENT_ESPORTS_OPENED, }, ) .mappings() .first() or {} ) # 登录次数取历史记录（事件表不回填，所以登录相关计数仍以 login_tasks 为准）； # 「最近 CK 检测」改从**账号**上读（2026-10-04 CK 解耦：ck_checked_at 已搬到 accounts）。 login_stats = ( db.execute( text( """ SELECT COUNT(*) AS login_count, MAX(finished_at) AS last_login_at FROM login_tasks WHERE account_id = :aid AND status = 'success' """ ), {"aid": account_id}, ) .mappings() .first() or {} ) return { "id": acc.id, "username": acc.username, "nickname": acc.nickname, "uid": acc.uid, "tag": acc.tag, "remark": acc.remark, "assigned_to": acc.assigned_to, "created_at": acc.created_at, # 导入时间 # 生命周期状态（账号现在到底能不能用、要不要人处理） "status": acc.status or "", "status_label": STATUS_LABELS.get(acc.status or "", acc.status or ""), "status_reason": acc.status_reason or "", "status_changed_at": acc.status_changed_at, "auto_managed": bool(acc.auto_managed), # CK 的持有与健康（详情页状态卡要一眼看出「能不能用」） "has_cookie": bool(acc.cookie), "ck_check_status": acc.ck_check_status or "", "ck_checked_at": acc.ck_checked_at, # 当前状态（覆盖式字段） "points": acc.points, "gold_balance": acc.gold_balance, "exchange_balance": acc.exchange_balance, "game_name": acc.game_name, "game_channel": acc.game_channel, **bind_payload(acc), # 最近一次「一键查询」快照（列表接口刻意**不**返回，避免体积爆炸） # 积分到期时间：账号自己的字段，账号详情里展示（一键查询会刷新它） "points_expire_at": acc.points_expire_at, "overview_snapshot": acc.overview_snapshot, "overview_queried_at": acc.overview_queried_at, "high_value_owned": acc.high_value_owned, "elite_opened": acc.elite_opened, "elite_expire_time": acc.elite_expire_time, "esports_expire_time": acc.esports_expire_time, "esports_points": acc.esports_points, "esports_bind_status": acc.esports_bind_status, "esports_game_name": acc.esports_game_name, "esports_can_change_time": acc.esports_can_change_time, # 统计 "stats": { "recharge_total": int(stats.get("recharge_total") or 0), "exchange_count": int(stats.get("exchange_count") or 0), "points_spent": int(stats.get("points_spent") or 0), "ck_invalid_count": int(stats.get("ck_invalid_count") or 0), "event_count": int(stats.get("event_count") or 0), "login_count": int(login_stats.get("login_count") or 0), "last_login_at": login_stats.get("last_login_at"), "last_ck_check_at": acc.ck_checked_at, }, # 「开通时间」由事件表提供；事件表从上线起才有，早于上线的不显示。 "opened_at": { "elite": stats.get("elite_opened_at"), "esports": stats.get("esports_opened_at"), }, } @router.get("/{account_id}/events") def account_events( account_id: int, page: int = 1, page_size: int = 50, event_type: str = "", category: str = Query( "", description="按粗分类筛选：lifecycle / auth / business / ops" ), actor_id: int | None = Query( None, description="按操作人筛选：谁做的这件事（池化后按客服查流水）" ), db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """账号时间线（倒序分页）。 权限沿用账号可见性：客服能看**自己名下 ∪ 自己租用中**的账号事件。 `category` 是给页面用的「大分类」过滤（状态变迁 / 登录与 CK / 业务 / 人工操作）， 比逐一点 `event_type` 好用：客服最常看的是「人工操作」与「状态变迁」两条线。 `actor_id` 是池化后新增的：客服不再长期持有账号，所以"这个账号被谁动过、 动了什么"必须能按人查 —— 配合 `_task_actor` 落库的 `actor_id` 形成完整流水。 """ acc = _visible_accounts_query(db, current).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") page = max(1, int(page)) page_size = max(1, min(int(page_size), 200)) where = "WHERE account_id = :aid" params: dict[str, Any] = {"aid": account_id} sql_event_type = _query_str(event_type) sql_category = _query_str(category) if sql_event_type: where += " AND event_type = :etype" params["etype"] = sql_event_type if sql_category: where += " AND category = :category" params["category"] = sql_category if isinstance(actor_id, int) and not isinstance(actor_id, bool): where += " AND actor_id = :actor_id" params["actor_id"] = int(actor_id) total = ( db.execute( text(f"SELECT COUNT(*) FROM account_events {where}"), params ).scalar() or 0 ) rows = ( db.execute( text( f""" 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 """ ), {**params, "limit": page_size, "offset": (page - 1) * page_size}, ) .mappings() .all() ) items = [dict(row) for row in rows] # 补上操作人的用户名：池化后「谁做的」是这一屏最重要的信息之一， # 前端拿到 id 还要再查一次不值当。一次 IN 查询即可。 actor_ids = {item.get("actor_id") for item in items if item.get("actor_id")} actor_names: dict[int, str] = {} if actor_ids: actor_names = { user_id: name for user_id, name in db.query(User.id, User.username) .filter(User.id.in_(actor_ids)) .all() } for item in items: actor_id = item.get("actor_id") item["actor_name"] = ( actor_names.get(actor_id, "") if isinstance(actor_id, int) else "" ) return { "items": items, "total": int(total), "page": page, "page_size": page_size, } @router.delete("/{account_id}") def delete_account( account_id: int, db: Session = Depends(get_db), current: User = Depends(require_permission("account:delete")), ): acc = db.query(Account).filter(Account.id == account_id).first() if not acc: raise HTTPException(status_code=404, detail="账号不存在") # 先删除关联的登录任务，避免外键约束失败 db.query(LoginTask).filter(LoginTask.account_id == account_id).delete( synchronize_session=False ) db.query(DouyuTask).filter(DouyuTask.account_id == account_id).delete( synchronize_session=False ) db.query(DouyuWorkbenchAccount).filter( DouyuWorkbenchAccount.account_id == account_id ).delete(synchronize_session=False) # 事件表有指向 accounts 的外键，不清会直接删不掉账号。 # 账号都没了，它的时间线也没有保留意义（审计仍在 audit_logs 里，那份不带外键）。 db.query(AccountEvent).filter(AccountEvent.account_id == account_id).delete( synchronize_session=False ) db.add( AuditLog( user_id=current.id, username=current.username, action="account:delete", target=acc.username, account_id=account_id, ) ) db.delete(acc) db.commit() return {"message": "已删除", "success": True}