yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加充值审计日志

Browse Source
This commit is contained in:
yml2213
2026-08-14 11:52:30 +08:00
parent 7fefe7e9f9
commit 7fe6228901
16 changed files with 675 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/backend/main.py
+2 -1
View File
@@ -12,7 +12,7 @@ from fastapi.responses import FileResponse
from starlette.middleware.base import BaseHTTPMiddleware
from .database import init_db
from .routers import auth, users, accounts, account_check, dashboard, login, proxy, cookies, huya, douyu, yyb
from .routers import auth, users, accounts, account_check, dashboard, login, proxy, cookies, huya, douyu, yyb, audit
from .schemas import AppInfo
from .version import get_app_version
from utils import setup_logger
@@ -102,6 +102,7 @@ app.include_router(cookies.router)
app.include_router(huya.router)
app.include_router(douyu.router)
app.include_router(yyb.router)
app.include_router(audit.router)
@app.get("/api/health")
web/backend/migrations/versions/20260814_0029_audit_log_result.py
+45
View File
@@ -0,0 +1,45 @@
"""为审计日志增加操作结果标记。
Revision ID: 20260814_0029
Revises: 20260814_0028
Create Date: 2026-08-14
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "20260814_0029"
down_revision: Union[str, None] = "20260814_0028"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""保留历史日志，新增可空结果字段。"""
bind = op.get_bind()
inspector = sa.inspect(bind)
if not inspector.has_table("audit_logs"):
return
columns = {column["name"] for column in inspector.get_columns("audit_logs")}
if "success" not in columns:
op.add_column("audit_logs", sa.Column("success", sa.Boolean(), nullable=True))
indexes = {index["name"] for index in inspector.get_indexes("audit_logs")}
if "ix_audit_logs_success" not in indexes:
op.create_index("ix_audit_logs_success", "audit_logs", ["success"])
def downgrade() -> None:
"""移除新增结果字段。"""
bind = op.get_bind()
inspector = sa.inspect(bind)
if not inspector.has_table("audit_logs"):
return
indexes = {index["name"] for index in inspector.get_indexes("audit_logs")}
if "ix_audit_logs_success" in indexes:
op.drop_index("ix_audit_logs_success", table_name="audit_logs")
columns = {column["name"] for column in inspector.get_columns("audit_logs")}
if "success" in columns:
op.drop_column("audit_logs", "success")
web/backend/models.py
+2
View File
@@ -458,4 +458,6 @@ class AuditLog(Base):
action = Column(String(128), nullable=False)
target = Column(String(256), default="")
detail = Column(Text, default="")
# 旧记录未标记结果时为 NULL，避免把历史操作误判成失败。
success = Column(Boolean, nullable=True, index=True)
created_at = Column(DateTime, default=_utcnow)
web/backend/routers/audit.py
+103
View File
@@ -0,0 +1,103 @@
"""审计日志查询接口，仅超级管理员可访问。"""
from __future__ import annotations
import json
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Query
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 ""
if row.action == "recharge:douyu:create" and row.target.startswith("douyu_batch:"):
try:
detail = json.loads(detail_text) if detail_text else {}
except json.JSONDecodeError:
detail = None
if isinstance(detail, dict) and not detail.get("recharge_accounts"):
batch_id = row.target.removeprefix("douyu_batch:")
accounts = (
db.query(DouyuTask, Account)
.join(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 accounts:
detail["recharge_accounts"] = [
{
"task_id": task.id,
"username": account.username,
"douyu_uid": account.uid or "",
"douyu_nickname": account.nickname or "",
}
for task, account in accounts
]
detail["recharge_accounts_truncated"] = len(accounts) == 100
detail_text = json.dumps(detail, ensure_ascii=False, separators=(",", ":"))
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": row.created_at,
}
@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:
pattern = f"%{keyword.strip()}%"
query = query.filter((AuditLog.target.ilike(pattern)) | (AuditLog.detail.ilike(pattern)))
if success is not None:
query = query.filter(AuditLog.success == success)
if start_time:
query = query.filter(AuditLog.created_at >= start_time)
if end_time:
query = query.filter(AuditLog.created_at <= end_time)
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,
}
web/backend/routers/douyu.py
+69
View File
@@ -26,6 +26,7 @@ from ..schemas import (
DouyuXpdGoodsOut,
)
from ..services.douyu_runner import DouyuBatchRunner, douyu_batch_registry
from ..services.audit_service import record_audit
from ..services.douyu_service import (
DOUYU_CONFIG_FIELDS,
DOUYU_HANDBOOK_TASK_TYPES,
@@ -110,6 +111,19 @@ async def supplier_recharge_callback(request: Request, db: Session = Depends(get
task.status = "running"
task.message = f"供应商直充订单处理中（状态 {status if status is not None else '-'}）"
task.result = result
callback_success = status not in {3, 4}
record_audit(
db, None, action="recharge:douyu:callback", target=f"supplier_order:{out_order_id}",
detail={
"out_order_id": out_order_id,
"task_id": task.id,
"order_id": result.get("order_id"),
"supplier_order_status": status,
"task_status": task.status,
"message": task.message,
},
success=callback_success,
)
db.commit()
logger.info("[douyu] 供应商直充回调已处理: out_order_id={} status={}", out_order_id, status)
acknowledgement = {"code": 200, "message": "success"}
@@ -400,13 +414,21 @@ def update_config(
):
"""更新斗鱼活动配置。"""
config = ensure_douyu_config(db)
changed_fields = []
for field in DOUYU_CONFIG_FIELDS:
value = getattr(req, field)
if value is None:
continue
setattr(config, field, value.strip() if isinstance(value, str) else value)
changed_fields.append(field)
apply_douyu_config_defaults(config)
config.updated_at = datetime.now(timezone.utc)
recharge_fields = [field for field in changed_fields if field.startswith("gold_")]
if recharge_fields:
record_audit(
db, current, action="recharge:douyu:config", target="douyu_config",
detail={"changed_fields": recharge_fields},
)
db.commit()
db.refresh(config)
return _config_out(config)
@@ -474,6 +496,35 @@ async def create_task_batch(
if count == 0:
raise HTTPException(status_code=400, detail="没有可执行的斗鱼账号，请先登录获取 Cookie")
if req.task_type == "create_gold_qr":
recharge_accounts = (
db.query(DouyuTask, Account)
.join(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()
)
record_audit(
db, current, action="recharge:douyu:create", target=f"douyu_batch:{batch_id}",
detail={
"batch_id": batch_id, "task_type": req.task_type, "count": count,
"account_count": len(req.account_ids), "handbook_scope": req.handbook_scope,
# 仅记录充值身份，禁止在审计中写入 Cookie、密码等凭据。
"recharge_accounts": [
{
"task_id": task.id,
"username": account.username,
"douyu_uid": account.uid or "",
"douyu_nickname": account.nickname or "",
}
for task, account in recharge_accounts
],
"recharge_accounts_truncated": count > len(recharge_accounts),
},
)
db.commit()
log_queue = asyncio.Queue()
loop = asyncio.get_running_loop()
thread_db = SessionLocal()
@@ -565,6 +616,12 @@ def stop_batch(
):
"""停止正在运行的斗鱼批次。"""
_require_batch_owner(db, current, batch_id)
is_recharge_batch = (
db.query(DouyuTask.id)
.filter(DouyuTask.batch_id == batch_id, DouyuTask.task_type == "create_gold_qr")
.first()
is not None
)
batch = douyu_batch_registry.get(batch_id)
if batch:
if batch.get("finished"):
@@ -574,10 +631,22 @@ def stop_batch(
return {"message": f"批次已结束，已清理 {cleaned} 个残留任务", "success": True}
raise HTTPException(status_code=404, detail="批次已结束")
batch["runner"].stop()
if is_recharge_batch:
record_audit(
db, current, action="recharge:douyu:stop", target=f"douyu_batch:{batch_id}",
detail={"batch_id": batch_id, "mode": "running"},
)
db.commit()
return {"message": "已发送停止信号", "success": True}
cleaned = cleanup_orphan_douyu_tasks(db, batch_id=batch_id, message="任务已停止（批次不存在）")
if cleaned:
if is_recharge_batch:
record_audit(
db, current, action="recharge:douyu:stop", target=f"douyu_batch:{batch_id}",
detail={"batch_id": batch_id, "mode": "orphan_cleanup", "cleaned": cleaned},
)
db.commit()
return {"message": f"已清理 {cleaned} 个残留任务", "success": True}
raise HTTPException(status_code=404, detail="批次不存在或已结束")
web/backend/routers/huya.py
+36
View File
@@ -73,6 +73,7 @@ from ..services.huya_service import (
upsert_huya_cookie,
)
from ..services.huya_runner import HuyaBatchRunner, huya_batch_registry
from ..services.audit_service import record_audit
from ..services.huya_register_runner import (
export_success_logs_text,
huya_register_registry,
@@ -1292,12 +1293,19 @@ def update_config(
):
"""更新虎牙配置。"""
config = ensure_huya_config(db)
changed_fields = []
for field in HUYA_CONFIG_FIELDS:
value = getattr(req, field)
if value is not None:
setattr(config, field, value.strip())
changed_fields.append(field)
apply_huya_config_defaults(config)
config.updated_at = datetime.now(timezone.utc)
if changed_fields:
record_audit(
db, current, action="recharge:huya:config", target="huya_config",
detail={"changed_fields": changed_fields},
)
db.commit()
db.refresh(config)
return _config_out(config)
@@ -1356,6 +1364,16 @@ async def create_task_batch(
if count == 0:
raise HTTPException(status_code=400, detail="没有有效的虎牙账号")
if req.task_type == "create_recharge_order":
record_audit(
db, current, action="recharge:huya:create", target=f"huya_batch:{batch_id}",
detail={
"batch_id": batch_id, "task_type": req.task_type, "count": count,
"account_count": len(req.account_ids),
},
)
db.commit()
log_queue = asyncio.Queue()
loop = asyncio.get_running_loop()
thread_db = SessionLocal()
@@ -1436,6 +1454,12 @@ def stop_batch(
):
"""停止正在运行的虎牙批次。"""
_require_huya_batch_owner(db, current, batch_id)
is_recharge_batch = (
db.query(HuyaTask.id)
.filter(HuyaTask.batch_id == batch_id, HuyaTask.task_type == "create_recharge_order")
.first()
is not None
)
batch = huya_batch_registry.get(batch_id)
if batch:
if batch.get("finished"):
@@ -1449,6 +1473,12 @@ def stop_batch(
return {"message": f"批次已结束，已清理 {cleaned} 个残留任务", "success": True}
raise HTTPException(status_code=404, detail="批次已结束")
batch["runner"].stop()
if is_recharge_batch:
record_audit(
db, current, action="recharge:huya:stop", target=f"huya_batch:{batch_id}",
detail={"batch_id": batch_id, "mode": "running"},
)
db.commit()
return {"message": "已发送停止信号", "success": True}
cleaned = cleanup_orphan_huya_tasks(
@@ -1457,6 +1487,12 @@ def stop_batch(
message="任务已停止（批次不存在）",
)
if cleaned:
if is_recharge_batch:
record_audit(
db, current, action="recharge:huya:stop", target=f"huya_batch:{batch_id}",
detail={"batch_id": batch_id, "mode": "orphan_cleanup", "cleaned": cleaned},
)
db.commit()
return {"message": f"已清理 {cleaned} 个残留任务", "success": True}
raise HTTPException(status_code=404, detail="批次不存在或已结束")
web/backend/routers/yyb.py
+35
View File
@@ -14,6 +14,7 @@ from ..permissions import user_has_permission
from ..schemas import YybLoginRequest, YybSelectionRequest, YybTaskCreateRequest
from ..services.yyb_service import _utcnow, public_task, sync_task
from ..services.yyb_worker_client import YybWorkerClient, YybWorkerError
from ..services.audit_service import record_audit
router = APIRouter(prefix="/api/yyb", tags=["应用宝充值"])
@@ -58,6 +59,10 @@ def create_task(payload: YybTaskCreateRequest, db: Session = Depends(get_db), cu
worker_job_id=str(data["job_id"]), status=str(data.get("status", "created")),
phase="login", message="请选择登录方式")
db.add(task)
record_audit(
db, current, action="recharge:yyb:create", target=f"yyb_task:{task.task_id}",
detail={"task_id": task.task_id, "status": task.status},
)
db.commit()
db.refresh(task)
return public_task(task, include_qr=False, creator_username=current.username)
@@ -73,6 +78,10 @@ def login(task_id: int, payload: YybLoginRequest, db: Session = Depends(get_db),
task.message = "请扫码登录并在手机确认"
if data.get("qr_data"):
task.login_qr_data = data["qr_data"]
record_audit(
db, current, action="recharge:yyb:login", target=f"yyb_task:{task.task_id}",
detail={"task_id": task.task_id, "provider": payload.provider, "status": task.status},
)
db.commit()
return public_task(task, creator_username=_creator_username(db, task))
@@ -129,6 +138,14 @@ def selection(task_id: int, payload: YybSelectionRequest, db: Session = Depends(
setattr(task, field, selected[field])
task.price_fen = int(selected.get("price_fen") or 0)
task.phase, task.status, task.message = "payment", "ready", "选择已保存，可以生成付款码"
record_audit(
db, current, action="recharge:yyb:selection", target=f"yyb_task:{task.task_id}",
detail={
"task_id": task.task_id, "product_id": task.product_id, "points": task.points,
"zone_id": task.zone_id, "zone_name": task.zone_name, "role_id": task.role_id,
"role_name": task.role_name, "price_fen": task.price_fen,
},
)
db.commit()
return public_task(task, creator_username=_creator_username(db, task))
@@ -156,11 +173,20 @@ def payment(task_id: int, db: Session = Depends(get_db), current: User = Depends
task.phase = "payment"
task.message = "生成付款码失败，请稍后重试"
task.payment_started_at = None
record_audit(
db, current, action="recharge:yyb:payment", target=f"yyb_task:{task.task_id}",
detail="生成付款码失败", success=False,
)
db.commit()
raise
task.status = str(data.get("status", "ordering"))
task.phase = str(data.get("phase", "payment"))
task.message = str(data.get("message", task.message))
record_audit(
db, current, action="recharge:yyb:payment", target=f"yyb_task:{task.task_id}",
detail={"task_id": task.task_id, "product_id": task.product_id, "points": task.points,
"price_fen": task.price_fen, "status": task.status},
)
db.commit()
return public_task(task, include_qr=False, creator_username=_creator_username(db, task))
@@ -177,6 +203,11 @@ def payment_check(task_id: int, db: Session = Depends(get_db), current: User = D
task.payment_last_checked_at = _utcnow()
if task.status == "success":
task.phase = "completed"
record_audit(
db, current, action="recharge:yyb:payment_check", target=f"yyb_task:{task.task_id}",
detail={"task_id": task.task_id, "status": task.status, "message": task.message},
success=task.status != "failed",
)
db.commit()
return public_task(task, include_qr=False, creator_username=_creator_username(db, task))
@@ -189,5 +220,9 @@ def stop(task_id: int, db: Session = Depends(get_db), current: User = Depends(re
task.phase = str(data.get("phase", "stopped"))
task.message = str(data.get("message", "任务已停止"))
task.finished_at = _utcnow()
record_audit(
db, current, action="recharge:yyb:stop", target=f"yyb_task:{task.task_id}",
detail={"task_id": task.task_id, "status": task.status},
)
db.commit()
return public_task(task, creator_username=_creator_username(db, task))
web/backend/services/audit_service.py
+63
View File
@@ -0,0 +1,63 @@
"""操作审计记录服务，统一处理详情脱敏与序列化。"""
from __future__ import annotations
import json
from collections.abc import Mapping
from typing import Any
from sqlalchemy.orm import Session
from ..models import AuditLog, User
_SENSITIVE_KEY_PARTS = (
"cookie", "token", "password", "passwd", "secret", "signature", "sign",
"qr_data", "authorization", "credential", "private_key",
)
def _safe_value(value: Any, key: str = "") -> Any:
"""递归过滤敏感字段，避免把凭据、二维码和签名写入审计表。"""
key_lower = key.lower()
if any(part in key_lower for part in _SENSITIVE_KEY_PARTS):
return "[已脱敏]"
if isinstance(value, Mapping):
return {str(k): _safe_value(v, str(k)) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [_safe_value(item) for item in value]
if isinstance(value, (str, int, float, bool)) or value is None:
return value
return str(value)
def record_audit(
db: Session,
current: User | None,
*,
action: str,
target: str = "",
detail: Mapping[str, Any] | str | None = None,
success: bool = True,
) -> AuditLog:
"""加入一条审计记录；调用方负责与业务变更一起提交事务。"""
if isinstance(detail, Mapping):
detail_text = json.dumps(_safe_value(detail), ensure_ascii=False, separators=(",", ":"))
elif detail is None:
detail_text = ""
else:
detail_text = str(_safe_value(detail))
if not success:
detail_text = json.dumps(
{"message": detail_text}, ensure_ascii=False, separators=(",", ":")
)
entry = AuditLog(
user_id=current.id if current else None,
username=current.username if current else "supplier",
action=action,
target=target,
detail=detail_text,
success=success,
)
db.add(entry)
return entry