Files
live-hub-py/web/backend/routers/accounts.py
T
yml2213 e88c41385b Merge branch 'main' into dev
# Conflicts:
#	tests/test_migrations.py
2026-10-04 13:17:43 +08:00

785 lines
25 KiB
Python

"""账号管理路由"""
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import 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,
AuditLog,
DouyuTask,
DouyuWorkbenchAccount,
LoginTask,
User,
)
from ..permissions import user_has_permission
from ..schemas import (
AccountAssign,
AccountBulkSelection,
AccountBulkTag,
AccountImport,
AccountOut,
AccountTag,
BatchAssign,
)
from ..services.account_events_service import (
EVENT_CK_INVALID,
EVENT_ELITE_OPENED,
EVENT_ESPORTS_OPENED,
EVENT_EXCHANGE,
EVENT_RECHARGE,
)
from ..services.account_service import (
cookie_account_ids_query,
parse_and_build_accounts,
)
from ..services.douyu_service import cookie_holding_filter
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"):
return query.filter(Account.assigned_to == current.id)
raise HTTPException(status_code=403, detail="无权查看账号")
def _filter_accounts_query(
db: Session,
query,
current: User,
*,
assigned_only: bool = False,
tag: str | None = None,
has_cookie: bool = False,
search: str = "",
):
"""复用列表筛选条件,供分页列表和全量批量操作保持一致。"""
if has_cookie:
query = query.filter(Account.id.in_(cookie_account_ids_query(db)))
if assigned_only and user_has_permission(current, "account:view_all"):
query = query.filter(Account.assigned_to.isnot(None))
if tag:
query = query.filter(Account.tag == tag)
search_text = (search or "").strip()
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,
tag=req.tag,
has_cookie=req.has_cookie,
search=req.search,
)
.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 _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),
tag: str = Query(None),
has_cookie: bool = Query(False),
search: str = Query(""),
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,
tag=tag,
has_cookie=has_cookie,
search=search,
)
total = None
if page is not None:
total = query.order_by(None).with_entities(func.count(Account.id)).scalar() or 0
query = query.order_by(Account.id)
if page is not None:
query = query.offset((page - 1) * page_size).limit(page_size)
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),
]
)
accounts = query.options(*options).all()
result = []
for acc in accounts:
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,
)
# 只有管理员可看完整字段(密码、邮箱等)
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
)
return {
"total": total,
"assigned_count": assigned_count,
"unassigned_count": max(0, total - assigned_count),
"tag_count": tag_count,
}
@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(LoginTask)
.filter(
LoginTask.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
)
db.add(
AuditLog(
user_id=current.id,
username=current.username,
action="account:assign",
target=acc.username,
)
)
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")),
):
"""批量分配/取消分配账号给客服。"""
if not req.account_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(LoginTask.account_id).where(cookie_holding_filter())
invalid_ids = (
db.query(Account.id)
.filter(
Account.id.in_(req.account_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 = (
db.query(Account)
.filter(Account.id.in_(req.account_ids))
.update({Account.assigned_to: req.assigned_to}, synchronize_session=False)
)
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}个账号",
)
)
db.commit()
action = "分配" if req.assigned_to else "取消分配"
return {
"message": f"已批量{action} {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 的账号)。"""
cookie_subq = select(LoginTask.account_id).where(cookie_holding_filter())
cookie_accounts = db.query(Account).filter(Account.id.in_(cookie_subq)).subquery()
results = (
db.query(
User.id, User.username, func.count(cookie_accounts.c.id).label("count")
)
.outerjoin(cookie_accounts, cookie_accounts.c.assigned_to == User.id)
.filter(User.role == "support")
.group_by(User.id, User.username)
.order_by(func.count(cookie_accounts.c.id).desc())
.all()
)
total_unassigned = (
db.query(func.count(cookie_accounts.c.id))
.filter(cookie_accounts.c.assigned_to.is_(None))
.scalar()
) or 0
return {
"support_users": [
{"id": uid, "username": uname, "assigned_count": cnt}
for uid, uname, cnt in results
],
"unassigned_count": total_unassigned,
}
@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="账号不存在")
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,
)
)
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()
count = (
_visible_accounts_query(db, current)
.filter(Account.id.in_(req.account_ids))
.update({Account.tag: tag}, synchronize_session=False)
)
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()
count = (
_visible_accounts_query(db, current)
.filter(Account.id.in_(ids))
.update({Account.tag: tag}, synchronize_session=False)
)
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)
)
db.add(
AuditLog(
user_id=current.id,
username=current.username,
action="account:delete",
target=f"批量删除{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)
)
db.add(
AuditLog(
user_id=current.id,
username=current.username,
action="account:delete",
target=f"{'按筛选' if req.all_matching else '批量'}删除{deleted}个账号",
)
)
db.commit()
return {"message": f"已删除 {deleted} 个账号", "deleted": deleted, "success": True}
@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 {}
)
# 登录次数取历史记录(事件表不回填,所以 CK 相关计数仍以 login_tasks 为准)。
login_stats = (
db.execute(
text(
"""
SELECT COUNT(*) AS login_count,
MAX(finished_at) AS last_login_at,
MAX(ck_checked_at) AS last_ck_check_at
FROM login_tasks
WHERE account_id = :aid AND cookie IS NOT NULL AND cookie <> ''
"""
),
{"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, # 导入时间
# 当前状态(覆盖式字段)
"points": acc.points,
"gold_balance": acc.gold_balance,
"exchange_balance": acc.exchange_balance,
"game_name": acc.game_name,
"game_channel": acc.game_channel,
"bind_status": acc.bind_status,
"elite_opened": acc.elite_opened,
"elite_expire_time": acc.elite_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,
"auto_managed": acc.auto_managed,
# 统计
"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": login_stats.get("last_ck_check_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 = "",
db: Session = Depends(get_db),
current: User = Depends(get_current_user),
):
"""账号时间线(倒序分页)。
权限沿用账号可见性:客服只能看自己名下账号的事件。
"""
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}
if event_type:
where += " AND event_type = :etype"
params["etype"] = event_type
total = (
db.execute(
text(f"SELECT COUNT(*) FROM account_events {where}"), params
).scalar()
or 0
)
rows = (
db.execute(
text(
f"""
SELECT id, event_type, occurred_at, title, amount, balance_after,
detail, source
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()
)
return {
"items": [dict(row) for row in rows],
"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}