"""审计日志查询接口，仅超级管理员可访问。""" from __future__ import annotations import json from datetime import UTC, datetime from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import String, cast, or_ from sqlalchemy.orm import Session from ..database import get_db from ..deps import get_current_user from ..models import Account, AuditLog, DouyuTask, User router = APIRouter(prefix="/api/audit-logs", tags=["审计日志"]) def require_super_admin(current: User = Depends(get_current_user)) -> User: """审计日志属于高敏感运维数据，只允许超级管理员查看。""" if current.role != "super_admin": raise HTTPException(status_code=403, detail="仅超级管理员可查看审计日志") return current def _audit_log_out(db: Session, row: AuditLog) -> dict: """序列化审计记录，并为斗鱼直充批次补齐账号身份与供应商订单信息。""" detail_text = row.detail or "" try: detail = json.loads(detail_text) if detail_text else {} except json.JSONDecodeError: detail = None if ( isinstance(detail, dict) and row.action == "recharge:douyu:create" and row.target.startswith("douyu_batch:") ): batch_id = row.target.removeprefix("douyu_batch:") tasks_and_accounts = ( db.query(DouyuTask, Account) .outerjoin(Account, Account.id == DouyuTask.account_id) .filter( DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr", ) .order_by(DouyuTask.id.asc()) .limit(100) .all() ) if tasks_and_accounts: task_info_by_id: dict[int, dict] = {} supplier_orders: list[dict] = [] supplier_order_ids: list[str] = [] supplier_out_order_ids: list[str] = [] has_supplier = False for task, account in tasks_and_accounts: task_result = task.result if isinstance(task.result, dict) else {} channel = str(task_result.get("recharge_channel") or "") # 先落到局部变量再判 isinstance：直接写 # `x.get(k) if isinstance(x.get(k), dict) else {}` 是两次独立取值， # 类型检查器无法据此收窄，会报「可能为 None」。 raw_supplier_order = task_result.get("supplier_order") supplier_order = ( raw_supplier_order if isinstance(raw_supplier_order, dict) else {} ) supplier_order_id = str( supplier_order.get("order_id") or supplier_order.get("orderId") or task_result.get("order_id") or "" ).strip() out_order_id = str( task.supplier_out_order_id or task_result.get("out_order_id") or "" ).strip() if channel == "supplier_api" or supplier_order_id or out_order_id: has_supplier = True info = { "task_id": task.id, "username": (account.username if account else "") or "", "douyu_uid": (account.uid if account else "") or "", "douyu_nickname": (account.nickname if account else "") or "", "status": task.status, "supplier_out_order_id": out_order_id, "supplier_order_id": supplier_order_id, "amount": task_result.get("buy_num") or task_result.get("pay_amount"), } task_info_by_id[task.id] = info if out_order_id or supplier_order_id: supplier_orders.append( { "task_id": task.id, "supplier_out_order_id": out_order_id, "supplier_order_id": supplier_order_id, "douyu_nickname": info["douyu_nickname"] or info["username"], "douyu_uid": info["douyu_uid"], "status": task.status, "amount": info["amount"], } ) if ( supplier_order_id and supplier_order_id not in supplier_order_ids ): supplier_order_ids.append(supplier_order_id) if out_order_id and out_order_id not in supplier_out_order_ids: supplier_out_order_ids.append(out_order_id) if supplier_orders: detail["supplier_orders"] = supplier_orders detail["supplier_order_ids"] = supplier_order_ids detail["supplier_out_order_ids"] = supplier_out_order_ids if not detail.get("recharge_accounts"): detail["recharge_accounts"] = list(task_info_by_id.values()) detail["recharge_accounts_truncated"] = len(tasks_and_accounts) == 100 else: for item in detail.get("recharge_accounts", []): if isinstance(item, dict) and "task_id" in item: tid = item.get("task_id") if tid in task_info_by_id: matched = task_info_by_id[tid] if matched["supplier_out_order_id"]: item["supplier_out_order_id"] = matched[ "supplier_out_order_id" ] if matched["supplier_order_id"]: item["supplier_order_id"] = matched["supplier_order_id"] if matched["status"]: item["status"] = matched["status"] if matched["amount"] is not None: item["amount"] = matched["amount"] if not detail.get("payment_method") or not detail.get("recharge_channel"): detail["recharge_channel"] = ( "supplier_api" if has_supplier else "wechat_qr" ) detail["payment_method"] = ( "API 直充支付" if has_supplier else "微信扫码支付" ) detail_text = json.dumps(detail, ensure_ascii=False, separators=(",", ":")) created_at = row.created_at if created_at is not None: if created_at.tzinfo is None: created_at = created_at.replace(tzinfo=UTC) created_at_out = created_at.isoformat() else: created_at_out = None return { "id": row.id, "user_id": row.user_id, "username": row.username or "", "action": row.action, "target": row.target or "", "detail": detail_text, "success": row.success, "created_at": created_at_out, } @router.get("") def list_audit_logs( page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), username: str | None = Query(None, max_length=64), action: str | None = Query(None, max_length=128), keyword: str | None = Query(None, max_length=128), success: bool | None = Query(None), start_time: datetime | None = Query(None), end_time: datetime | None = Query(None), db: Session = Depends(get_db), _: User = Depends(require_super_admin), ): """分页查询审计记录，不提供修改和删除能力。""" query = db.query(AuditLog) if username: query = query.filter(AuditLog.username.ilike(f"%{username.strip()}%")) if action: query = query.filter(AuditLog.action == action.strip()) if keyword: kw = keyword.strip() pattern = f"%{kw}%" conditions = [ AuditLog.target.ilike(pattern), AuditLog.detail.ilike(pattern), AuditLog.username.ilike(pattern), ] # 关联查询 DouyuTask，支持按外部单号、供应商单号、账号昵称、UID 直接检索对应的批次审计记录 matching_tasks = ( db.query(DouyuTask.batch_id) .outerjoin(Account, Account.id == DouyuTask.account_id) .filter( or_( DouyuTask.supplier_out_order_id.ilike(pattern), cast(DouyuTask.result, String).ilike(pattern), Account.uid.ilike(pattern), Account.nickname.ilike(pattern), Account.username.ilike(pattern), ) ) .distinct() .limit(100) .all() ) batch_targets = [f"douyu_batch:{b[0]}" for b in matching_tasks if b[0]] if batch_targets: conditions.append(AuditLog.target.in_(batch_targets)) query = query.filter(or_(*conditions)) if success is not None: query = query.filter(AuditLog.success == success) if start_time: st = ( start_time.astimezone(UTC).replace(tzinfo=None) if start_time.tzinfo else start_time ) query = query.filter(AuditLog.created_at >= st) if end_time: et = ( end_time.astimezone(UTC).replace(tzinfo=None) if end_time.tzinfo else end_time ) query = query.filter(AuditLog.created_at <= et) total = query.count() rows = ( query.order_by(AuditLog.id.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return { "items": [_audit_log_out(db, row) for row in rows], "total": total, "page": page, "page_size": page_size, }