2503 lines
84 KiB
Python
2503 lines
84 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
import json
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import delete, func, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.dependencies import get_db, get_admin_user
|
|
from app.models.user import User
|
|
from app.models.project import Project
|
|
from app.models.generation_record import GenerationRecord
|
|
from app.models.chat_generation_task import ChatGenerationTask
|
|
from app.models.credit_record import CreditRecord
|
|
from app.models.model_config import ModelConfig
|
|
from app.models.system_config import SystemConfig
|
|
from app.models.payment_order import PaymentOrder
|
|
from app.models.token_usage import TokenUsage
|
|
from app.models.notification import Notification
|
|
from app.models.notification_read import NotificationRead
|
|
from app.models.industry_config import IndustryConfig
|
|
from app.models.video_engine import VideoEngine
|
|
from app.models.image_engine import ImageEngine
|
|
from app.models.credit_ratio import CreditRatio
|
|
from app.models.operation_log import OperationLog
|
|
from app.enums.user import FrontendUserKind, UserType
|
|
from app.enums.team import TEAM_UNASSIGNED_VALUE
|
|
from app.enums.generation_status import GenerationRecordPipelineStage
|
|
from app.schemas.admin import (
|
|
CreditAdjustRequest,
|
|
ModelConfigCreate,
|
|
ModelConfigOut,
|
|
SystemConfigCreate,
|
|
SystemConfigUpdate,
|
|
SystemConfigOut,
|
|
AdminUserOut,
|
|
AdminStatsOut,
|
|
DailyCreditOut,
|
|
TeamCreditOut,
|
|
ModelUsageOut,
|
|
CreateUserRequest,
|
|
UpdateMenusRequest,
|
|
ResetPasswordRequest,
|
|
UpdateFrontendUserKindRequest,
|
|
OperationLogOut,
|
|
)
|
|
from app.schemas.team import UpdateUserTeamRequest
|
|
from app.schemas.industry import IndustryConfigCreate, IndustryConfigOut
|
|
from app.schemas.video_engine import VideoEngineCreate, VideoEngineOut
|
|
from app.schemas.image_engine import ImageEngineCreate, ImageEngineOut
|
|
from app.schemas.credit_ratio import CreditRatioCreate, CreditRatioOut
|
|
from app.services.generation.pipeline.db_lock_service import (
|
|
DatabaseRowLockBusy,
|
|
execute_with_lock_timeout,
|
|
)
|
|
from app.services.credits import add_credits, deduct_credits
|
|
from app.services.credit_record_meta_service import build_admin_adjust_meta
|
|
from app.services.admin_credit_record_service import list_admin_credit_records
|
|
from app.services.notification import create_notification
|
|
from app.services.auth import hash_password, verify_password
|
|
from app.services.operation_log import log_operation
|
|
from app.services.private_portrait.reference_resolver import batch_resolve_private_portrait_reference_display_urls
|
|
from app.services.resource_signed_url_service import build_resource_signed_url
|
|
from app.services.payment import sync_pending_orders, process_refund
|
|
from app.services.resource_capacity_service import batch_get_user_resource_capacity_usage, get_user_resource_capacity_usage
|
|
from app.services.team_service import batch_get_team_name_map, set_frontend_user_team
|
|
|
|
from app.services.generation.billing_service import (
|
|
OWNER_GENERATION_RECORD,
|
|
charge_generation_media_for_record,
|
|
get_next_credit_attempt_no,
|
|
)
|
|
from app.services.generation.refund_service import mark_generation_record_failed_and_refund_once
|
|
from app.utils.id_gen import generate_id
|
|
from app.schemas.generation import GenerationType, ASPECT_RATIOS, RESOLUTIONS
|
|
|
|
|
|
CST = timezone(timedelta(hours=8))
|
|
|
|
|
|
def _iso(dt):
|
|
"""Serialize datetime as naive ISO string (UTC→CST, strip tzinfo)."""
|
|
if dt is None:
|
|
return None
|
|
try:
|
|
d = dt if isinstance(dt, datetime) else datetime.fromisoformat(str(dt))
|
|
if d.tzinfo and d.tzinfo.utcoffset(None) == timedelta(0):
|
|
d = d.astimezone(CST)
|
|
return d.replace(tzinfo=None).isoformat()
|
|
except (ValueError, TypeError):
|
|
return str(dt)
|
|
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
# ── User Management ──────────────────────────────────────
|
|
|
|
@router.get("/users")
|
|
async def list_users(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=1000),
|
|
search: str = Query(""),
|
|
user_type: str | None = Query(None, pattern="^(frontend|admin)$"),
|
|
frontend_user_kind: str | None = Query(None, pattern="^(internal|external)$"),
|
|
team_id: str | None = Query(None),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
query = select(User).order_by(User.created_at.desc())
|
|
count_query = select(func.count(User.id))
|
|
if search:
|
|
like = f"%{search}%"
|
|
search_filter = (User.username.ilike(like)) | (User.email.ilike(like)) | (User.phone.ilike(like))
|
|
query = query.where(search_filter)
|
|
count_query = count_query.where(search_filter)
|
|
if user_type:
|
|
query = query.where(User.user_type == user_type)
|
|
count_query = count_query.where(User.user_type == user_type)
|
|
if frontend_user_kind:
|
|
query = query.where(User.user_type == UserType.FRONTEND.value, User.frontend_user_kind == frontend_user_kind)
|
|
count_query = count_query.where(User.user_type == UserType.FRONTEND.value, User.frontend_user_kind == frontend_user_kind)
|
|
if team_id:
|
|
team_filter = User.team_id.is_(None) if team_id == TEAM_UNASSIGNED_VALUE else (User.team_id == team_id)
|
|
query = query.where(team_filter)
|
|
count_query = count_query.where(team_filter)
|
|
total = (await db.execute(count_query)).scalar() or 0
|
|
result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
|
|
users = list(result.scalars().all())
|
|
user_ids = [u.id for u in users]
|
|
team_ids = [getattr(u, "team_id", None) for u in users if getattr(u, "team_id", None)]
|
|
capacity_map = await batch_get_user_resource_capacity_usage(db, user_ids)
|
|
team_name_map = await batch_get_team_name_map(db, team_ids)
|
|
return {
|
|
"items": [
|
|
AdminUserOut.model_validate(user)
|
|
.model_copy(
|
|
update={
|
|
"resource_capacity": capacity_map.get(user.id),
|
|
"team_name": team_name_map.get(getattr(user, "team_id", None)),
|
|
}
|
|
)
|
|
.model_dump(mode="json")
|
|
for user in users
|
|
],
|
|
"total": total,
|
|
}
|
|
|
|
|
|
@router.post("/users", response_model=AdminUserOut)
|
|
async def create_user(
|
|
req: CreateUserRequest,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
import random
|
|
# Frontend users: use phone, auto-generate username
|
|
if req.user_type == "frontend":
|
|
if not req.phone:
|
|
raise HTTPException(status_code=400, detail="前台用户必须填写手机号")
|
|
existing_phone = await db.execute(select(User).where(User.phone == req.phone).limit(1))
|
|
if existing_phone.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="该手机号已注册")
|
|
username = f"用户{req.phone[-4:]}"
|
|
existing_name = await db.execute(select(User).where(User.username == username).limit(1))
|
|
if existing_name.scalar_one_or_none():
|
|
username = f"用户{req.phone[-4:]}{random.randint(10, 99)}"
|
|
else:
|
|
if not req.username:
|
|
raise HTTPException(status_code=400, detail="后台用户必须填写用户名")
|
|
username = req.username
|
|
existing = await db.execute(select(User).where(User.username == username).limit(1))
|
|
if existing.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="用户名已存在")
|
|
|
|
user = User(
|
|
id=generate_id(),
|
|
username=username,
|
|
hashed_password=hash_password(req.password),
|
|
email=req.email,
|
|
phone=req.phone,
|
|
credits=req.credits,
|
|
is_admin=req.is_admin if req.user_type == "admin" else False,
|
|
user_type=req.user_type,
|
|
frontend_user_kind=req.frontend_user_kind if req.user_type == "frontend" else FrontendUserKind.EXTERNAL.value,
|
|
allowed_menus=req.allowed_menus,
|
|
private_portrait_asset_limit=req.private_portrait_asset_limit,
|
|
)
|
|
user.credits = round(user.credits, 2)
|
|
db.add(user)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"创建用户 {username}",
|
|
"POST",
|
|
"/admin/users",
|
|
detail=json.dumps(
|
|
{
|
|
"user_id": user.id,
|
|
"username": username,
|
|
"user_type": req.user_type,
|
|
"frontend_user_kind": user.frontend_user_kind,
|
|
"private_portrait_asset_limit": user.private_portrait_asset_limit,
|
|
"credits": user.credits,
|
|
"phone": user.phone,
|
|
"email": user.email,
|
|
},
|
|
ensure_ascii=False,
|
|
default=str,
|
|
),
|
|
ip=None,
|
|
)
|
|
await db.commit()
|
|
return user
|
|
|
|
|
|
@router.put("/users/{user_id}/menus")
|
|
async def update_user_menus(
|
|
user_id: str,
|
|
req: UpdateMenusRequest,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(User).where(User.id == user_id).limit(1))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
user.allowed_menus = req.allowed_menus
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新菜单权限",
|
|
"PUT",
|
|
f"/admin/users/{user_id}/menus",
|
|
detail=json.dumps(
|
|
{
|
|
"user_id": user_id,
|
|
"allowed_menus": req.allowed_menus,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.get("/users/{user_id}", response_model=AdminUserOut)
|
|
async def get_user(
|
|
user_id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(User).where(User.id == user_id).limit(1))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
user.credits = round(user.credits, 2)
|
|
resource_capacity = await get_user_resource_capacity_usage(db, user.id)
|
|
team_name_map = await batch_get_team_name_map(db, [getattr(user, "team_id", None)])
|
|
return AdminUserOut.model_validate(user).model_copy(
|
|
update={
|
|
"resource_capacity": resource_capacity,
|
|
"team_name": team_name_map.get(getattr(user, "team_id", None)),
|
|
}
|
|
)
|
|
|
|
|
|
@router.post("/users/{user_id}/credits")
|
|
async def adjust_credits(
|
|
user_id: str,
|
|
req: CreditAdjustRequest,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if req.amount > 0:
|
|
await add_credits(db, user_id, req.amount, f"管理员调整: {req.description}", record_meta=build_admin_adjust_meta())
|
|
else:
|
|
await deduct_credits(db, user_id, abs(req.amount), f"管理员调整: {req.description}", record_meta=build_admin_adjust_meta())
|
|
await create_notification(
|
|
db, user_id, "积分变动通知",
|
|
f"您的积分已{'增加' if req.amount > 0 else '扣除'}{abs(req.amount)}积分。原因：{req.description}",
|
|
"credit",
|
|
)
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"调整积分 {'+' if req.amount > 0 else ''}{req.amount}",
|
|
"POST",
|
|
f"/admin/users/{user_id}/credits",
|
|
detail=json.dumps(
|
|
{
|
|
"user_id": user_id,
|
|
"amount": req.amount,
|
|
"description": req.description,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.put("/users/{user_id}/status")
|
|
async def update_user_status(
|
|
user_id: str,
|
|
body: dict,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
is_active = body.get("is_active", True)
|
|
await db.execute(
|
|
update(User).where(User.id == user_id).values(is_active=is_active)
|
|
)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"{'启用' if is_active else '禁用'}用户",
|
|
"PUT",
|
|
f"/admin/users/{user_id}/status",
|
|
detail=json.dumps(
|
|
{
|
|
"user_id": user_id,
|
|
"is_active": is_active,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.put("/users/{user_id}/admin-status")
|
|
async def update_user_admin_status(
|
|
user_id: str,
|
|
body: dict,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
is_admin = body.get("is_admin", False)
|
|
result = await db.execute(select(User).where(User.id == user_id).limit(1))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
if user.user_type != "admin":
|
|
raise HTTPException(status_code=400, detail="仅后台用户支持设置超级管理员状态")
|
|
user.is_admin = is_admin
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"{is_admin and '设为' or '取消'}超级管理员",
|
|
"PUT",
|
|
f"/admin/users/{user_id}/admin-status",
|
|
detail=json.dumps(
|
|
{
|
|
"user_id": user_id,
|
|
"is_admin": is_admin,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.put("/users/{user_id}/frontend-kind", response_model=AdminUserOut)
|
|
async def update_user_frontend_kind(
|
|
user_id: str,
|
|
req: UpdateFrontendUserKindRequest,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(User).where(User.id == user_id).limit(1))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
if user.user_type != UserType.FRONTEND.value:
|
|
raise HTTPException(status_code=400, detail="仅前台用户支持设置内部/外部归类")
|
|
user.frontend_user_kind = req.frontend_user_kind or FrontendUserKind.EXTERNAL.value
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"设置前台用户归类为 {user.frontend_user_kind}",
|
|
"PUT",
|
|
f"/admin/users/{user_id}/frontend-kind",
|
|
detail=json.dumps(
|
|
{
|
|
"user_id": user_id,
|
|
"frontend_user_kind": user.frontend_user_kind,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return user
|
|
|
|
|
|
@router.put("/users/{user_id}/team", response_model=AdminUserOut)
|
|
async def update_user_team(
|
|
user_id: str,
|
|
req: UpdateUserTeamRequest,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
user, before, after = await set_frontend_user_team(db, user_id=user_id, team_id=req.team_id)
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"设置用户团队 {user.username}",
|
|
"PUT",
|
|
f"/admin/users/{user_id}/team",
|
|
detail=json.dumps({"before": before, "after": after}, ensure_ascii=False),
|
|
)
|
|
resource_capacity = await get_user_resource_capacity_usage(db, user.id)
|
|
return AdminUserOut.model_validate(user).model_copy(
|
|
update={"resource_capacity": resource_capacity, "team_name": after.get("team_name")}
|
|
)
|
|
|
|
|
|
@router.put("/users/{user_id}/reset-password")
|
|
async def reset_user_password(
|
|
user_id: str,
|
|
req: ResetPasswordRequest,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(User).where(User.id == user_id).limit(1))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
user.hashed_password = hash_password(req.new_password)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"重置密码",
|
|
"PUT",
|
|
f"/admin/users/{user_id}/reset-password",
|
|
detail=json.dumps(
|
|
{
|
|
"user_id": user_id,
|
|
"new_password": "***",
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.post("/change-password")
|
|
async def admin_change_password(
|
|
body: dict,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
old_password = body.get("old_password", "")
|
|
new_password = body.get("new_password", "")
|
|
if not verify_password(old_password, admin.hashed_password):
|
|
raise HTTPException(status_code=400, detail="原密码错误")
|
|
if len(new_password) < 6:
|
|
raise HTTPException(status_code=400, detail="密码至少6位")
|
|
admin.hashed_password = hash_password(new_password)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
"修改密码",
|
|
"POST",
|
|
"/admin/change-password",
|
|
detail=json.dumps(
|
|
{
|
|
"old_password": "***",
|
|
"new_password": "***",
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "密码修改成功"}
|
|
|
|
|
|
# ── Credit Records ───────────────────────────────────────
|
|
|
|
@router.get("/credit-records")
|
|
async def list_credit_records(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=1000),
|
|
user_id: str | None = Query(None),
|
|
user_name: str | None = Query(None),
|
|
user_type: str | None = Query(None),
|
|
frontend_user_kind: str | None = Query(None),
|
|
team_id: str | None = Query(None),
|
|
record_type: str | None = Query(None),
|
|
type: str | None = Query(None),
|
|
credit_subject: str | None = Query(None),
|
|
media_type: str | None = Query(None),
|
|
charge_kind: str | None = Query(None),
|
|
source_module: str | None = Query(None),
|
|
source_step_code: str | None = Query(None),
|
|
billing_scene: str | None = Query(None),
|
|
start_date: str = Query(None),
|
|
end_date: str = Query(None),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List all credit transaction records with filters and full summary."""
|
|
return await list_admin_credit_records(
|
|
db,
|
|
page=page,
|
|
page_size=page_size,
|
|
user_id=user_id,
|
|
user_name=user_name,
|
|
user_type=user_type,
|
|
frontend_user_kind=frontend_user_kind,
|
|
team_id=team_id,
|
|
record_type=record_type or type,
|
|
credit_subject=credit_subject,
|
|
media_type=media_type,
|
|
charge_kind=charge_kind,
|
|
source_module=source_module,
|
|
source_step_code=source_step_code,
|
|
billing_scene=billing_scene,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
)
|
|
|
|
|
|
# ── Notification Admin ───────────────────────────────────
|
|
|
|
@router.get("/notifications")
|
|
async def list_admin_notifications(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=500),
|
|
user_id: str | None = Query(None),
|
|
is_read: bool = Query(None),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List all notifications with optional filters."""
|
|
query = select(Notification).order_by(Notification.created_at.desc())
|
|
count_query = select(func.count(Notification.id))
|
|
|
|
if user_id:
|
|
query = query.where(Notification.user_id == user_id)
|
|
count_query = count_query.where(Notification.user_id == user_id)
|
|
|
|
if is_read is not None:
|
|
query = query.where(Notification.is_read == is_read)
|
|
count_query = count_query.where(Notification.is_read == is_read)
|
|
|
|
total = (await db.execute(count_query)).scalar() or 0
|
|
result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
|
|
items = result.scalars().all()
|
|
|
|
return {
|
|
"total": total,
|
|
"items": [
|
|
{
|
|
"id": item.id,
|
|
"user_id": item.user_id,
|
|
"title": item.title,
|
|
"content": item.content,
|
|
"type": item.type,
|
|
"is_read": item.is_read,
|
|
"related_id": item.related_id,
|
|
"created_at": _iso(item.created_at),
|
|
}
|
|
for item in items
|
|
],
|
|
}
|
|
|
|
|
|
@router.post("/notifications")
|
|
async def create_admin_notification(
|
|
body: dict,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Create notification. target_user_id=None broadcasts to all users."""
|
|
title = body.get("title", "")
|
|
content = body.get("content", "")
|
|
notif_type = body.get("type", "system")
|
|
target_user_id = body.get("target_user_id")
|
|
|
|
if not title or not content:
|
|
raise HTTPException(status_code=400, detail="标题和内容不能为空")
|
|
|
|
if target_user_id:
|
|
# Send to specific user
|
|
await create_notification(
|
|
db, target_user_id, title, content, notif_type, push_ws=True
|
|
)
|
|
else:
|
|
# Broadcast: create one notification with user_id=NULL
|
|
await create_notification(
|
|
db, None, title, content, notif_type, push_ws=True
|
|
)
|
|
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"{'发送通知给指定用户' if target_user_id else '广播通知'}: {title}",
|
|
"POST",
|
|
"/admin/notifications",
|
|
detail=json.dumps(
|
|
{
|
|
"title": title,
|
|
"target_user_id": target_user_id,
|
|
"type": notif_type,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.delete("/notifications/{notification_id}")
|
|
async def delete_admin_notification(
|
|
notification_id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(Notification).where(Notification.id == notification_id).limit(1)
|
|
)
|
|
notif = result.scalar_one_or_none()
|
|
if not notif:
|
|
raise HTTPException(status_code=404, detail="通知不存在")
|
|
await db.execute(
|
|
delete(NotificationRead).where(NotificationRead.notification_id == notification_id)
|
|
)
|
|
await db.delete(notif)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"删除通知: {notif.title}",
|
|
"DELETE",
|
|
f"/admin/notifications/{notification_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"notification_id": notification_id,
|
|
"title": notif.title,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.get("/notifications/{notification_id}/read-users")
|
|
async def get_notification_read_users(
|
|
notification_id: str,
|
|
page: int = Query(1, ge=1),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Get list of users who have read a specific notification."""
|
|
from app.services.notification import get_notification_read_users
|
|
items, total = await get_notification_read_users(db, notification_id, page)
|
|
return {"total": total, "items": items}
|
|
|
|
|
|
# ── Payment Config ───────────────────────────────────────
|
|
|
|
@router.get("/payment-configs")
|
|
async def list_payment_configs(
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Return system configs that start with payment_ prefix."""
|
|
result = await db.execute(
|
|
select(SystemConfig).where(SystemConfig.key.like("payment_%"))
|
|
)
|
|
configs = result.scalars().all()
|
|
return [
|
|
{
|
|
"id": c.id,
|
|
"key": c.key,
|
|
"value": c.value,
|
|
"description": c.description,
|
|
}
|
|
for c in configs
|
|
]
|
|
|
|
|
|
@router.put("/payment-configs/batch")
|
|
async def batch_update_payment_configs(
|
|
req: dict[str, str],
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Batch upsert payment configs. Creates missing keys, updates existing ones."""
|
|
from app.utils.id_gen import generate_id
|
|
|
|
for key, value in req.items():
|
|
if not key.startswith("payment_"):
|
|
continue
|
|
result = await db.execute(
|
|
select(SystemConfig).where(SystemConfig.key == key).limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if config:
|
|
config.value = value
|
|
else:
|
|
db.add(SystemConfig(
|
|
id=generate_id(),
|
|
key=key,
|
|
value=value,
|
|
))
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
"批量更新支付配置",
|
|
"PUT",
|
|
"/admin/payment-configs/batch",
|
|
detail=json.dumps(list(req.keys()), ensure_ascii=False),
|
|
)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/payment-stats")
|
|
async def get_payment_stats(
|
|
payment_method: str | None = Query(None),
|
|
status: str | None = Query(None),
|
|
start_date: str | None = Query(None),
|
|
end_date: str | None = Query(None),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Return payment statistics for admin dashboard with filters."""
|
|
from sqlalchemy import func
|
|
|
|
# Ensure by_status has all expected statuses with defaults
|
|
by_status = {
|
|
"pending": {"count": 0, "amount": 0.0},
|
|
"paid": {"count": 0, "amount": 0.0},
|
|
"cancelled": {"count": 0, "amount": 0.0},
|
|
"refunded": {"count": 0, "amount": 0.0},
|
|
}
|
|
|
|
# Parse dates and build base query filters
|
|
now_cst = datetime.now(CST)
|
|
today_start = now_cst.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
today_end = today_start + timedelta(days=1)
|
|
|
|
# Default to today if no date range provided
|
|
query_start = today_start
|
|
query_end = today_end
|
|
|
|
if start_date:
|
|
query_start = datetime.fromisoformat(start_date).replace(tzinfo=CST)
|
|
if end_date:
|
|
query_end = (datetime.fromisoformat(end_date) + timedelta(days=1)).replace(tzinfo=CST)
|
|
|
|
# Build filter list for status breakdown
|
|
breakdown_filters = []
|
|
if payment_method:
|
|
breakdown_filters.append(PaymentOrder.payment_method == payment_method)
|
|
if status:
|
|
breakdown_filters.append(PaymentOrder.status == status)
|
|
# Always apply date range to breakdown
|
|
breakdown_filters.append(PaymentOrder.created_at >= query_start)
|
|
breakdown_filters.append(PaymentOrder.created_at < query_end)
|
|
|
|
# Status breakdown
|
|
status_result = await db.execute(
|
|
select(
|
|
PaymentOrder.status,
|
|
func.count().label("count"),
|
|
func.coalesce(func.sum(PaymentOrder.amount), 0).label("amount"),
|
|
)
|
|
.where(*breakdown_filters)
|
|
.group_by(PaymentOrder.status)
|
|
)
|
|
for row in status_result.all():
|
|
if row.status in by_status:
|
|
by_status[row.status] = {
|
|
"count": row.count,
|
|
"amount": round(float(row.amount), 2)
|
|
}
|
|
else:
|
|
# Map any unexpected status to cancelled
|
|
by_status["cancelled"]["count"] += row.count
|
|
by_status["cancelled"]["amount"] += round(float(row.amount), 2)
|
|
|
|
# Today's stats (CST time zone) - independent of filter
|
|
today_result = await db.execute(
|
|
select(
|
|
func.count().label("paid_count"),
|
|
func.coalesce(func.sum(PaymentOrder.amount), 0).label("paid_amount"),
|
|
).where(
|
|
PaymentOrder.status == "paid",
|
|
PaymentOrder.paid_at >= today_start,
|
|
PaymentOrder.paid_at < today_end,
|
|
)
|
|
)
|
|
today_row = today_result.one()
|
|
|
|
# Monthly cumulative stats
|
|
month_start = now_cst.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
month_end = (month_start + timedelta(days=32)).replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
month_result = await db.execute(
|
|
select(
|
|
func.count().label("paid_count"),
|
|
func.coalesce(func.sum(PaymentOrder.amount), 0).label("paid_amount"),
|
|
).where(
|
|
PaymentOrder.status == "paid",
|
|
PaymentOrder.paid_at >= month_start,
|
|
PaymentOrder.paid_at < month_end,
|
|
)
|
|
)
|
|
month_row = month_result.one()
|
|
|
|
return {
|
|
"by_status": by_status,
|
|
"today": {
|
|
"paid_count": today_row.paid_count,
|
|
"paid_amount": round(float(today_row.paid_amount), 2),
|
|
},
|
|
"month": {
|
|
"paid_count": month_row.paid_count,
|
|
"paid_amount": round(float(month_row.paid_amount), 2),
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/payment-orders")
|
|
async def list_payment_orders(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=500),
|
|
payment_method: str | None = Query(None),
|
|
status: str | None = Query(None),
|
|
phone: str | None = Query(None, description="按用户手机号模糊搜索"),
|
|
start_date: str | None = Query(None),
|
|
end_date: str | None = Query(None),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Return paginated payment orders for admin dashboard."""
|
|
query = select(PaymentOrder, User.username, User.phone).join(User, PaymentOrder.user_id == User.id)
|
|
count_query = select(func.count(PaymentOrder.id))
|
|
|
|
filters = []
|
|
if payment_method:
|
|
filters.append(PaymentOrder.payment_method == payment_method)
|
|
if status:
|
|
filters.append(PaymentOrder.status == status)
|
|
if phone:
|
|
filters.append(User.phone.ilike(f"%{phone.strip()}%"))
|
|
if start_date:
|
|
filters.append(PaymentOrder.created_at >= datetime.fromisoformat(start_date).replace(tzinfo=CST))
|
|
if end_date:
|
|
filters.append(PaymentOrder.created_at < (datetime.fromisoformat(end_date) + timedelta(days=1)).replace(tzinfo=CST))
|
|
|
|
for f in filters:
|
|
query = query.where(f)
|
|
count_query = count_query.where(f)
|
|
|
|
total = (await db.execute(count_query)).scalar() or 0
|
|
result = await db.execute(
|
|
query.order_by(PaymentOrder.created_at.desc()).offset((page - 1) * page_size).limit(page_size)
|
|
)
|
|
rows = result.all()
|
|
|
|
items = [
|
|
{
|
|
"id": o.id,
|
|
"orderNo": o.order_no,
|
|
"order_no": o.order_no,
|
|
"userId": o.user_id,
|
|
"user_id": o.user_id,
|
|
"username": username,
|
|
"phone": user_phone,
|
|
"amount": round(float(o.amount), 2),
|
|
"credits": round(float(o.credits), 2),
|
|
"paymentMethod": o.payment_method,
|
|
"payment_method": o.payment_method,
|
|
"status": o.status,
|
|
"tradeNo": o.trade_no,
|
|
"trade_no": o.trade_no,
|
|
"paidAt": o.paid_at.isoformat() if o.paid_at else None,
|
|
"paid_at": o.paid_at.isoformat() if o.paid_at else None,
|
|
"createdAt": o.created_at.isoformat() if o.created_at else None,
|
|
"created_at": o.created_at.isoformat() if o.created_at else None,
|
|
}
|
|
for o, username, user_phone in rows
|
|
]
|
|
|
|
return {"items": items, "total": total, "page": page, "page_size": page_size}
|
|
|
|
@router.put("/payment-configs/{config_id}")
|
|
async def update_payment_config(
|
|
config_id: str,
|
|
req: SystemConfigUpdate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Update a payment system config value."""
|
|
result = await db.execute(
|
|
select(SystemConfig).where(
|
|
SystemConfig.id == config_id,
|
|
SystemConfig.key.like("payment_%"),
|
|
)
|
|
.limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail="支付配置不存在")
|
|
config.value = req.value
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新支付配置: {config.key}",
|
|
"PUT",
|
|
f"/admin/payment-configs/{config_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config_id,
|
|
"key": config.key,
|
|
"value": req.value,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {
|
|
"id": config.id,
|
|
"key": config.key,
|
|
"value": config.value,
|
|
"description": config.description,
|
|
}
|
|
|
|
|
|
@router.post("/payment-orders/{order_no}/refund")
|
|
async def refund_payment_order(
|
|
order_no: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Refund a paid payment order."""
|
|
result = await process_refund(db, order_no)
|
|
if not result.get("success"):
|
|
raise HTTPException(status_code=400, detail=result.get("message", "退款失败"))
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"订单退款: {order_no}",
|
|
"POST",
|
|
f"/admin/payment-orders/{order_no}/refund",
|
|
detail=json.dumps(
|
|
{
|
|
"order_no": order_no,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return result
|
|
|
|
|
|
# ── Industry Config ──────────────────────────────────────
|
|
|
|
def _serialize_industry(ind: IndustryConfig) -> dict:
|
|
"""Convert industry config to dict with parsed skills."""
|
|
skills = []
|
|
if ind.skills:
|
|
try:
|
|
raw = json.loads(ind.skills)
|
|
if isinstance(raw, list):
|
|
skills = raw if raw and isinstance(raw[0], dict) else [{"key": s, "label": s} for s in raw]
|
|
except (json.JSONDecodeError, TypeError):
|
|
skills = []
|
|
return {
|
|
"id": ind.id,
|
|
"key": ind.key,
|
|
"label": ind.label,
|
|
"icon": ind.icon or "",
|
|
"description": ind.description,
|
|
"skills": skills,
|
|
"is_active": ind.is_active,
|
|
"sort_order": ind.sort_order,
|
|
"created_at": ind.created_at,
|
|
}
|
|
|
|
|
|
@router.get("/industry-configs")
|
|
async def list_industry_configs(
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(IndustryConfig).order_by(IndustryConfig.sort_order.asc())
|
|
)
|
|
return [_serialize_industry(ind) for ind in result.scalars().all()]
|
|
|
|
|
|
@router.post("/industry-configs")
|
|
async def create_industry_config(
|
|
req: IndustryConfigCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
config = IndustryConfig(
|
|
id=generate_id(),
|
|
key=req.key,
|
|
label=req.label,
|
|
icon=req.icon or "",
|
|
description=req.description,
|
|
skills=json.dumps(req.skills, ensure_ascii=False),
|
|
is_active=req.is_active,
|
|
sort_order=req.sort_order,
|
|
)
|
|
db.add(config)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"创建行业配置: {req.label}",
|
|
"POST",
|
|
"/admin/industry-configs",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config.id,
|
|
"key": req.key,
|
|
"label": req.label,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return _serialize_industry(config)
|
|
|
|
|
|
@router.put("/industry-configs/{config_id}")
|
|
async def update_industry_config(
|
|
config_id: str,
|
|
req: IndustryConfigCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(IndustryConfig).where(IndustryConfig.id == config_id).limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail="行业配置不存在")
|
|
config.key = req.key
|
|
config.label = req.label
|
|
config.icon = req.icon or ""
|
|
config.description = req.description
|
|
config.skills = json.dumps(req.skills, ensure_ascii=False)
|
|
config.is_active = req.is_active
|
|
config.sort_order = req.sort_order
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新行业配置: {req.label}",
|
|
"PUT",
|
|
f"/admin/industry-configs/{config_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config_id,
|
|
"key": req.key,
|
|
"label": req.label,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return _serialize_industry(config)
|
|
|
|
|
|
@router.delete("/industry-configs/{config_id}")
|
|
async def delete_industry_config(
|
|
config_id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(IndustryConfig).where(IndustryConfig.id == config_id).limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail="行业配置不存在")
|
|
await db.delete(config)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"删除行业配置: {config.label}",
|
|
"DELETE",
|
|
f"/admin/industry-configs/{config_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config_id,
|
|
"key": config.key,
|
|
"label": config.label,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
# ── Video Engine ─────────────────────────────────────────
|
|
|
|
@router.get("/video-engines", response_model=list[VideoEngineOut])
|
|
async def list_video_engines(
|
|
include_deleted: bool = Query(False),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
query = select(VideoEngine)
|
|
if not include_deleted:
|
|
query = query.where(VideoEngine.deleted_at.is_(None))
|
|
result = await db.execute(
|
|
query.order_by(VideoEngine.priority.desc(), VideoEngine.id.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/video-engines", response_model=VideoEngineOut)
|
|
async def create_video_engine(
|
|
req: VideoEngineCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
engine = VideoEngine(id=generate_id(), **req.model_dump())
|
|
db.add(engine)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"创建视频引擎: {req.name}",
|
|
"POST",
|
|
"/admin/video-engines",
|
|
detail=json.dumps(
|
|
{
|
|
"engine_id": engine.id,
|
|
"name": req.name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return engine
|
|
|
|
|
|
@router.put("/video-engines/{engine_id}", response_model=VideoEngineOut)
|
|
async def update_video_engine(
|
|
engine_id: str,
|
|
req: VideoEngineCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(VideoEngine).where(VideoEngine.id == engine_id, VideoEngine.deleted_at.is_(None)).limit(1)
|
|
)
|
|
engine = result.scalar_one_or_none()
|
|
if not engine:
|
|
raise HTTPException(status_code=404, detail="视频引擎不存在")
|
|
for k, v in req.model_dump().items():
|
|
setattr(engine, k, v)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新视频引擎: {engine.name}",
|
|
"PUT",
|
|
f"/admin/video-engines/{engine_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"engine_id": engine_id,
|
|
"name": engine.name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return engine
|
|
|
|
|
|
@router.delete("/video-engines/{engine_id}")
|
|
async def delete_video_engine(
|
|
engine_id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(VideoEngine).where(VideoEngine.id == engine_id, VideoEngine.deleted_at.is_(None)).limit(1)
|
|
)
|
|
engine = result.scalar_one_or_none()
|
|
if not engine:
|
|
raise HTTPException(status_code=404, detail="视频引擎不存在")
|
|
engine_name = engine.name
|
|
engine.deleted_at = datetime.now(timezone.utc)
|
|
engine.is_active = False
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"软删除视频引擎: {engine_name}",
|
|
"DELETE",
|
|
f"/admin/video-engines/{engine_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"engine_id": engine_id,
|
|
"name": engine_name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
# ── Image Engine ──────────────────────────────────────────
|
|
|
|
@router.get("/image-engines", response_model=list[ImageEngineOut])
|
|
async def list_image_engines(
|
|
include_deleted: bool = Query(False),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
query = select(ImageEngine)
|
|
if not include_deleted:
|
|
query = query.where(ImageEngine.deleted_at.is_(None))
|
|
result = await db.execute(
|
|
query.order_by(ImageEngine.priority.desc(), ImageEngine.id.desc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/image-engines", response_model=ImageEngineOut)
|
|
async def create_image_engine(
|
|
req: ImageEngineCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
engine = ImageEngine(id=generate_id(), **req.model_dump())
|
|
db.add(engine)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"创建图片引擎: {req.name}",
|
|
"POST",
|
|
"/admin/image-engines",
|
|
detail=json.dumps(
|
|
{
|
|
"engine_id": engine.id,
|
|
"name": req.name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return engine
|
|
|
|
|
|
@router.put("/image-engines/{engine_id}", response_model=ImageEngineOut)
|
|
async def update_image_engine(
|
|
engine_id: str,
|
|
req: ImageEngineCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(ImageEngine).where(ImageEngine.id == engine_id, ImageEngine.deleted_at.is_(None)).limit(1)
|
|
)
|
|
engine = result.scalar_one_or_none()
|
|
if not engine:
|
|
raise HTTPException(status_code=404, detail="图片引擎不存在")
|
|
for k, v in req.model_dump().items():
|
|
setattr(engine, k, v)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新图片引擎: {engine.name}",
|
|
"PUT",
|
|
f"/admin/image-engines/{engine_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"engine_id": engine_id,
|
|
"name": engine.name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return engine
|
|
|
|
|
|
@router.delete("/image-engines/{engine_id}")
|
|
async def delete_image_engine(
|
|
engine_id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(ImageEngine).where(ImageEngine.id == engine_id, ImageEngine.deleted_at.is_(None)).limit(1)
|
|
)
|
|
engine = result.scalar_one_or_none()
|
|
if not engine:
|
|
raise HTTPException(status_code=404, detail="图片引擎不存在")
|
|
engine_name = engine.name
|
|
engine.deleted_at = datetime.now(timezone.utc)
|
|
engine.is_active = False
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"软删除图片引擎: {engine_name}",
|
|
"DELETE",
|
|
f"/admin/image-engines/{engine_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"engine_id": engine_id,
|
|
"name": engine_name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
|
|
|
|
async def _validate_credit_ratio_engine(db: AsyncSession, req: CreditRatioCreate) -> None:
|
|
"""校验积分规则绑定的引擎是否存在。
|
|
|
|
CreditRatio.model_config_id 为兼容旧字段名，当前实际保存引擎ID：
|
|
- gen_type=image 时对应 image_engines.id
|
|
- gen_type=video 时对应 video_engines.id
|
|
"""
|
|
gen_type = (req.gen_type or "").lower().strip()
|
|
engine_id = (req.model_config_id or "").strip()
|
|
if gen_type not in ("image", "video"):
|
|
raise HTTPException(status_code=400, detail="gen_type 仅支持 image 或 video")
|
|
if not engine_id:
|
|
raise HTTPException(status_code=400, detail="model_config_id 不能为空，当前字段用于保存图片/视频引擎ID")
|
|
|
|
model = ImageEngine if gen_type == "image" else VideoEngine
|
|
result = await db.execute(select(model).where(model.id == engine_id, model.deleted_at.is_(None)).limit(1))
|
|
engine = result.scalar_one_or_none()
|
|
if not engine:
|
|
detail = "图片积分规则绑定的图片引擎不存在" if gen_type == "image" else "视频积分规则绑定的视频引擎不存在"
|
|
raise HTTPException(status_code=400, detail=detail)
|
|
|
|
|
|
# ── Credit Ratio ─────────────────────────────────────────
|
|
|
|
@router.get("/credit-ratios", response_model=list[CreditRatioOut])
|
|
async def list_credit_ratios(
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(CreditRatio).order_by(CreditRatio.gen_type.desc(), CreditRatio.model_config_id.desc()))
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/credit-ratios", response_model=CreditRatioOut)
|
|
async def create_credit_ratio(
|
|
req: CreditRatioCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
await _validate_credit_ratio_engine(db, req)
|
|
data = req.model_dump()
|
|
data["gen_type"] = data["gen_type"].lower().strip()
|
|
data["model_config_id"] = data["model_config_id"].strip()
|
|
ratio = CreditRatio(id=generate_id(), **data)
|
|
db.add(ratio)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"创建积分规则: {data.get('gen_type', '')} - {data.get('model_config_id', '')}",
|
|
"POST",
|
|
"/admin/credit-ratios",
|
|
detail=json.dumps(
|
|
{
|
|
"ratio_id": ratio.id,
|
|
"gen_type": data.get("gen_type"),
|
|
"model_config_id": data.get("model_config_id"),
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
return ratio
|
|
|
|
|
|
@router.put("/credit-ratios/{ratio_id}", response_model=CreditRatioOut)
|
|
async def update_credit_ratio(
|
|
ratio_id: str,
|
|
req: CreditRatioCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(CreditRatio).where(CreditRatio.id == ratio_id).limit(1)
|
|
)
|
|
ratio = result.scalar_one_or_none()
|
|
if not ratio:
|
|
raise HTTPException(status_code=404, detail="积分比例不存在")
|
|
await _validate_credit_ratio_engine(db, req)
|
|
data = req.model_dump()
|
|
data["gen_type"] = data["gen_type"].lower().strip()
|
|
data["model_config_id"] = data["model_config_id"].strip()
|
|
for k, v in data.items():
|
|
setattr(ratio, k, v)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新积分规则: {data.get('gen_type', '')} - {data.get('model_config_id', '')}",
|
|
"PUT",
|
|
f"/admin/credit-ratios/{ratio_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"ratio_id": ratio_id,
|
|
"gen_type": data.get("gen_type"),
|
|
"model_config_id": data.get("model_config_id"),
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
return ratio
|
|
|
|
|
|
@router.delete("/credit-ratios/{ratio_id}")
|
|
async def delete_credit_ratio(
|
|
ratio_id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(CreditRatio).where(CreditRatio.id == ratio_id).limit(1)
|
|
)
|
|
ratio = result.scalar_one_or_none()
|
|
if not ratio:
|
|
raise HTTPException(status_code=404, detail="积分比例不存在")
|
|
await db.delete(ratio)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"删除积分规则: {ratio.gen_type} - {ratio.model_config_id}",
|
|
"DELETE",
|
|
f"/admin/credit-ratios/{ratio_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"ratio_id": ratio_id,
|
|
"gen_type": ratio.gen_type,
|
|
"model_config_id": ratio.model_config_id,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.get("/credit-ratios/grouped", response_model=dict)
|
|
async def list_credit_ratios_grouped(
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(CreditRatio))
|
|
ratios = result.scalars().all()
|
|
|
|
grouped = {}
|
|
for ratio in ratios:
|
|
if ratio.gen_type not in grouped:
|
|
grouped[ratio.gen_type] = []
|
|
grouped[ratio.gen_type].append(CreditRatioOut.model_validate(ratio))
|
|
|
|
return grouped
|
|
|
|
|
|
# ── Model Config ─────────────────────────────────────────
|
|
|
|
@router.get("/model-configs", response_model=list[ModelConfigOut])
|
|
async def list_model_configs(
|
|
include_deleted: bool = Query(False),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
query = select(ModelConfig)
|
|
if not include_deleted:
|
|
query = query.where(ModelConfig.deleted_at.is_(None))
|
|
result = await db.execute(query.order_by(ModelConfig.priority.desc(), ModelConfig.id.desc()))
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/model-configs", response_model=ModelConfigOut)
|
|
async def create_model_config(
|
|
req: ModelConfigCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
config = ModelConfig(id=generate_id(), **req.model_dump())
|
|
db.add(config)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"创建模型配置: {req.name}",
|
|
"POST",
|
|
"/admin/model-configs",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config.id,
|
|
"name": req.name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return config
|
|
|
|
|
|
@router.put("/model-configs/{config_id}", response_model=ModelConfigOut)
|
|
async def update_model_config(
|
|
config_id: str,
|
|
req: ModelConfigCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(ModelConfig).where(ModelConfig.id == config_id, ModelConfig.deleted_at.is_(None)).limit(1))
|
|
config = result.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail="配置不存在")
|
|
for k, v in req.model_dump().items():
|
|
setattr(config, k, v)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新模型配置: {config.name}",
|
|
"PUT",
|
|
f"/admin/model-configs/{config_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config_id,
|
|
"name": config.name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return config
|
|
|
|
|
|
@router.delete("/model-configs/{config_id}")
|
|
async def delete_model_config(
|
|
config_id: str,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(ModelConfig).where(ModelConfig.id == config_id, ModelConfig.deleted_at.is_(None)).limit(1))
|
|
config = result.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail="配置不存在")
|
|
config_name = config.name
|
|
config.deleted_at = datetime.now(timezone.utc)
|
|
config.is_active = False
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"软删除模型配置: {config_name}",
|
|
"DELETE",
|
|
f"/admin/model-configs/{config_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config_id,
|
|
"name": config_name,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok"}
|
|
|
|
|
|
# ── System Config ────────────────────────────────────────
|
|
|
|
@router.get("/system-configs", response_model=list[SystemConfigOut])
|
|
async def list_system_configs(
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(SystemConfig))
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/system-configs", response_model=SystemConfigOut)
|
|
async def create_system_config(
|
|
req: SystemConfigCreate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
from app.utils.id_gen import generate_id
|
|
config = SystemConfig(
|
|
id=generate_id(),
|
|
key=req.key,
|
|
value=str(req.value),
|
|
description=req.description or "",
|
|
)
|
|
db.add(config)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"创建系统配置: {config.key}",
|
|
"POST",
|
|
"/admin/system-configs",
|
|
detail=json.dumps({"key": req.key, "value": req.value}, ensure_ascii=False),
|
|
)
|
|
await db.commit()
|
|
return config
|
|
|
|
|
|
@router.put("/system-configs/{config_id}", response_model=SystemConfigOut)
|
|
async def update_system_config(
|
|
config_id: str,
|
|
req: SystemConfigUpdate,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(select(SystemConfig).where(SystemConfig.id == config_id).limit(1))
|
|
config = result.scalar_one_or_none()
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail="配置不存在")
|
|
config.value = str(req.value)
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"更新系统配置: {config.key}",
|
|
"PUT",
|
|
f"/admin/system-configs/{config_id}",
|
|
detail=json.dumps(
|
|
{
|
|
"config_id": config_id,
|
|
"key": config.key,
|
|
"value": req.value,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
return config
|
|
|
|
|
|
# ── Operation Logs ──────────────────────────────────────
|
|
|
|
@router.get("/operation-logs")
|
|
async def list_operation_logs(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=500),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
query = select(OperationLog).order_by(OperationLog.created_at.desc())
|
|
count_query = select(func.count(OperationLog.id))
|
|
total = (await db.execute(count_query)).scalar() or 0
|
|
result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
|
|
items = result.scalars().all()
|
|
return {
|
|
"total": total,
|
|
"items": [
|
|
{
|
|
"id": item.id,
|
|
"user_id": item.user_id,
|
|
"username": item.username,
|
|
"action": item.action,
|
|
"method": item.method,
|
|
"path": item.path,
|
|
"detail": item.detail,
|
|
"ip": item.ip,
|
|
"created_at": _iso(item.created_at),
|
|
}
|
|
for item in items
|
|
],
|
|
}
|
|
|
|
|
|
# ── Stats ────────────────────────────────────────────────
|
|
|
|
@router.get("/stats", response_model=AdminStatsOut)
|
|
async def get_stats(
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
start_date: str = Query(None),
|
|
end_date: str = Query(None),
|
|
):
|
|
today_start = datetime.now(CST).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
try:
|
|
if start_date:
|
|
date_start = datetime.strptime(start_date, "%Y-%m-%d").replace(tzinfo=CST)
|
|
else:
|
|
date_start = today_start
|
|
if end_date:
|
|
date_end = datetime.strptime(end_date, "%Y-%m-%d").replace(tzinfo=CST)
|
|
date_end = date_end.replace(hour=23, minute=59, second=59, microsecond=999999)
|
|
else:
|
|
date_end = datetime.now(CST)
|
|
except:
|
|
date_start = today_start
|
|
date_end = datetime.now(CST)
|
|
|
|
total_users = (await db.execute(
|
|
select(func.count(User.id)).where(
|
|
User.user_type == "frontend",
|
|
User.created_at >= date_start,
|
|
User.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
total_projects = (await db.execute(
|
|
select(func.count(Project.id)).where(
|
|
Project.deleted_at.is_(None),
|
|
Project.created_at >= date_start,
|
|
Project.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
total_generations = (await db.execute(
|
|
select(func.count(ChatGenerationTask.id)).where(
|
|
ChatGenerationTask.created_at >= date_start,
|
|
ChatGenerationTask.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
total_records = (await db.execute(
|
|
select(func.count(GenerationRecord.id)).where(
|
|
GenerationRecord.deleted_at.is_(None),
|
|
GenerationRecord.created_at >= date_start,
|
|
GenerationRecord.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
total_revenue = (await db.execute(
|
|
select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
|
|
PaymentOrder.status == "paid",
|
|
PaymentOrder.created_at >= date_start,
|
|
PaymentOrder.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
credits_consumed = (await db.execute(
|
|
select(func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0)).where(
|
|
CreditRecord.type == "consume",
|
|
CreditRecord.created_at >= date_start,
|
|
CreditRecord.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
alipay_revenue = (await db.execute(
|
|
select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
|
|
PaymentOrder.status == "paid",
|
|
PaymentOrder.payment_method == "alipay",
|
|
PaymentOrder.created_at >= date_start,
|
|
PaymentOrder.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
wechat_revenue = (await db.execute(
|
|
select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
|
|
PaymentOrder.status == "paid",
|
|
PaymentOrder.payment_method == "wechat",
|
|
PaymentOrder.created_at >= date_start,
|
|
PaymentOrder.created_at <= date_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
period_duration = date_end - date_start
|
|
|
|
last_period_start = date_start - period_duration
|
|
last_period_end = date_start
|
|
|
|
last_period_users = (await db.execute(
|
|
select(func.count(User.id)).where(
|
|
User.user_type == "frontend",
|
|
User.created_at >= last_period_start,
|
|
User.created_at <= last_period_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
last_period_projects = (await db.execute(
|
|
select(func.count(Project.id)).where(
|
|
Project.deleted_at.is_(None),
|
|
Project.created_at >= last_period_start,
|
|
Project.created_at <= last_period_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
last_period_generations = (await db.execute(
|
|
select(func.count(ChatGenerationTask.id)).where(
|
|
ChatGenerationTask.created_at >= last_period_start,
|
|
ChatGenerationTask.created_at <= last_period_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
last_period_records = (await db.execute(
|
|
select(func.count(GenerationRecord.id)).where(
|
|
GenerationRecord.deleted_at.is_(None),
|
|
GenerationRecord.created_at >= last_period_start,
|
|
GenerationRecord.created_at <= last_period_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
last_period_revenue = (await db.execute(
|
|
select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
|
|
PaymentOrder.status == "paid",
|
|
PaymentOrder.created_at >= last_period_start,
|
|
PaymentOrder.created_at <= last_period_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
last_period_credits_consumed = (await db.execute(
|
|
select(func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0)).where(
|
|
CreditRecord.type == "consume",
|
|
CreditRecord.created_at >= last_period_start,
|
|
CreditRecord.created_at <= last_period_end,
|
|
)
|
|
)).scalar() or 0
|
|
|
|
# ── 每日各模块积分消耗（始终返回选中日期往前7天，便于图表展示）
|
|
from sqlalchemy import Date, cast as sa_cast
|
|
_day_expr = sa_cast(CreditRecord.created_at, Date)
|
|
# 图表固定展示 [date_end - 6天, date_end] 共7天
|
|
_chart_end_dt = date_end
|
|
_chart_start_dt = _chart_end_dt - timedelta(days=6)
|
|
_inner = (
|
|
select(
|
|
_day_expr.label('date'),
|
|
CreditRecord.source_module.label('module'),
|
|
func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0).label('credits'),
|
|
)
|
|
.where(
|
|
CreditRecord.type == "consume",
|
|
CreditRecord.created_at >= _chart_start_dt,
|
|
CreditRecord.created_at <= _chart_end_dt,
|
|
)
|
|
.group_by(_day_expr, CreditRecord.source_module)
|
|
.subquery()
|
|
)
|
|
daily_credits_rows = (await db.execute(
|
|
select(
|
|
_inner.c.date,
|
|
func.coalesce(_inner.c.module, 'other').label('module'),
|
|
_inner.c.credits,
|
|
).order_by(_inner.c.date)
|
|
)).all()
|
|
daily_credits_by_module = [
|
|
DailyCreditOut(date=str(row.date), module=row.module, credits=float(row.credits or 0))
|
|
for row in daily_credits_rows
|
|
]
|
|
|
|
# ── 各团队积分消耗（有团队 vs 无团队，使用流水中的团队快照）
|
|
team_credit_rows = (await db.execute(
|
|
select(
|
|
func.coalesce(CreditRecord.team_name_snapshot, '未分配团队').label('team_name'),
|
|
CreditRecord.team_id_snapshot.label('team_id'),
|
|
func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0).label('credits'),
|
|
)
|
|
.where(
|
|
CreditRecord.type == "consume",
|
|
CreditRecord.created_at >= date_start,
|
|
CreditRecord.created_at <= date_end,
|
|
)
|
|
.group_by(CreditRecord.team_id_snapshot, CreditRecord.team_name_snapshot)
|
|
.order_by(func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0).desc())
|
|
)).all()
|
|
credits_by_team = [
|
|
TeamCreditOut(team_name=row.team_name, team_id=row.team_id, credits=float(row.credits or 0))
|
|
for row in team_credit_rows
|
|
]
|
|
|
|
# ── 各模型使用次数（通过 engine 快照字段统计）
|
|
model_usage_rows = (await db.execute(
|
|
select(
|
|
func.coalesce(CreditRecord.engine_name, '未知').label('model_name'),
|
|
func.coalesce(CreditRecord.engine_provider, 'unknown').label('provider'),
|
|
func.count(CreditRecord.id).label('count'),
|
|
)
|
|
.where(
|
|
CreditRecord.type == "consume",
|
|
CreditRecord.created_at >= date_start,
|
|
CreditRecord.created_at <= date_end,
|
|
)
|
|
.group_by(CreditRecord.engine_name, CreditRecord.engine_provider)
|
|
.order_by(func.count(CreditRecord.id).desc())
|
|
)).all()
|
|
model_usage = [
|
|
ModelUsageOut(model_name=row.model_name, provider=row.provider, count=int(row.count or 0))
|
|
for row in model_usage_rows
|
|
]
|
|
|
|
return AdminStatsOut(
|
|
total_users=total_users,
|
|
total_projects=total_projects,
|
|
total_generations=total_generations,
|
|
total_records=total_records,
|
|
total_revenue=float(total_revenue),
|
|
credits_consumed_today=float(credits_consumed),
|
|
today_alipay_revenue=float(alipay_revenue),
|
|
today_wechat_revenue=float(wechat_revenue),
|
|
last_period_users=last_period_users,
|
|
last_period_projects=last_period_projects,
|
|
last_period_generations=last_period_generations,
|
|
last_period_records=last_period_records,
|
|
last_period_revenue=float(last_period_revenue),
|
|
last_period_credits_consumed=float(last_period_credits_consumed),
|
|
daily_credits_by_module=daily_credits_by_module,
|
|
credits_by_team=credits_by_team,
|
|
model_usage=model_usage,
|
|
)
|
|
|
|
|
|
# ── Token Usage ─────────────────────────────────────────
|
|
|
|
@router.get("/token-usage")
|
|
async def list_token_usage(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=500),
|
|
user_id: str | None = Query(None),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List token usage records with optional user filter."""
|
|
query = select(TokenUsage).order_by(TokenUsage.created_at.desc())
|
|
if user_id:
|
|
query = query.where(TokenUsage.user_id == user_id)
|
|
|
|
count_query = select(func.count(TokenUsage.id))
|
|
if user_id:
|
|
count_query = count_query.where(TokenUsage.user_id == user_id)
|
|
|
|
total = (await db.execute(count_query)).scalar() or 0
|
|
result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
|
|
items = result.scalars().all()
|
|
|
|
return {
|
|
"total": total,
|
|
"items": [
|
|
{
|
|
"id": item.id,
|
|
"model_config_id": item.model_config_id,
|
|
"user_id": item.user_id,
|
|
"input_tokens": item.input_tokens,
|
|
"output_tokens": item.output_tokens,
|
|
"total_tokens": item.total_tokens,
|
|
"created_at": _iso(item.created_at),
|
|
}
|
|
for item in items
|
|
],
|
|
}
|
|
|
|
|
|
# ── Generation Records (Admin) ─────────────────────────────
|
|
|
|
@router.get("/generation-records")
|
|
async def admin_list_generation_records(
|
|
user_id: str | None = Query(None),
|
|
status: str | None = Query(None),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=500),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List all generation records across all users, with optional filters."""
|
|
query = (
|
|
select(GenerationRecord, User.username, Project.name, Project.industry, IndustryConfig.label)
|
|
.join(User, GenerationRecord.user_id == User.id)
|
|
.join(Project, GenerationRecord.project_id == Project.id)
|
|
.outerjoin(IndustryConfig, Project.industry == IndustryConfig.key)
|
|
.where(GenerationRecord.deleted_at.is_(None), Project.deleted_at.is_(None))
|
|
.order_by(GenerationRecord.created_at.desc())
|
|
)
|
|
if user_id:
|
|
query = query.where(GenerationRecord.user_id == user_id)
|
|
if status:
|
|
query = query.where(GenerationRecord.status == status)
|
|
|
|
# Count total
|
|
count_query = select(func.count(GenerationRecord.id)).where(GenerationRecord.deleted_at.is_(None))
|
|
if user_id:
|
|
count_query = count_query.where(GenerationRecord.user_id == user_id)
|
|
if status:
|
|
count_query = count_query.where(GenerationRecord.status == status)
|
|
total_result = await db.execute(count_query)
|
|
total = total_result.scalar() or 0
|
|
|
|
# Paginate
|
|
offset = (page - 1) * page_size
|
|
query = query.offset(offset).limit(page_size)
|
|
result = await db.execute(query)
|
|
rows = result.all()
|
|
refs_map = await batch_resolve_private_portrait_reference_display_urls(
|
|
db,
|
|
{record.id: json.loads(record.media_references) if record.media_references else None for record, _username, _project_name, _industry, _industry_label in rows},
|
|
user_id=user_id,
|
|
)
|
|
|
|
items = []
|
|
for record, username, project_name, industry, industry_label in rows:
|
|
refs = refs_map.get(record.id)
|
|
items.append({
|
|
"id": record.id,
|
|
"user_id": record.user_id,
|
|
"username": username,
|
|
"project_id": record.project_id,
|
|
"project_name": project_name,
|
|
"industry": industry_label or industry,
|
|
"original_prompt": record.original_prompt,
|
|
"optimized_prompt": record.optimized_prompt,
|
|
"duration": record.duration,
|
|
"aspect_ratio": record.aspect_ratio,
|
|
"resolution": record.resolution,
|
|
"status": record.status,
|
|
"pipeline_stage": record.pipeline_stage,
|
|
"video_upscale_enabled": bool(record.video_upscale_enabled_snapshot),
|
|
"video_url": build_resource_signed_url(record.video_url) if record.video_url else '',
|
|
"video_cover_url": build_resource_signed_url(record.video_cover_url) if record.video_cover_url else '',
|
|
"references": refs,
|
|
"credits_cost": record.credits_cost or 0,
|
|
"text_credits_cost": record.text_credits_cost or 0,
|
|
"text_tokens_used": record.text_tokens_used or 0,
|
|
"video_tokens_used": record.video_tokens_used or 0,
|
|
"error_message": record.error_message,
|
|
"created_at": _iso(record.created_at),
|
|
"generated_at": _iso(record.generated_at),
|
|
|
|
# append img param
|
|
"gen_type": record.gen_type,
|
|
"image_size": record.image_size or '',
|
|
"image_url": build_resource_signed_url(record.image_url) if record.image_url else '',
|
|
"image_tokens_used": record.image_tokens_used or 0,
|
|
"image_proportion": record.image_proportion or '',
|
|
"image_px": record.image_px or '',
|
|
})
|
|
|
|
return {"total": total, "items": items}
|
|
|
|
|
|
@router.put("/generation-records/{record_id}/status")
|
|
async def admin_update_generation_status(
|
|
record_id: str,
|
|
body: dict,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""管理员只能终止正在执行或待生成的记录，禁止绕过流水线裸改生成/完成状态。"""
|
|
try:
|
|
result = await execute_with_lock_timeout(
|
|
db,
|
|
select(GenerationRecord).where(
|
|
GenerationRecord.id == record_id,
|
|
GenerationRecord.deleted_at.is_(None),
|
|
)
|
|
.with_for_update()
|
|
.limit(1),
|
|
)
|
|
except DatabaseRowLockBusy as exc:
|
|
raise HTTPException(status_code=409, detail=exc.detail) from exc
|
|
record = result.scalar_one_or_none()
|
|
if not record:
|
|
raise HTTPException(status_code=404, detail="记录不存在")
|
|
|
|
new_status = str(body.get("status") or "").strip()
|
|
if new_status in {"generating", "completed", "prompt_optimized"}:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="禁止直接修改为该状态；生成请调用生成接口，完成必须由下载/超分流水线落库",
|
|
)
|
|
if new_status != "failed":
|
|
raise HTTPException(status_code=400, detail="该接口仅允许管理员终止任务")
|
|
if record.status == "completed":
|
|
raise HTTPException(status_code=409, detail="已完成记录不能直接改为失败")
|
|
|
|
error_message = body.get("error_message") or record.error_message or "管理员终止生成任务"
|
|
await mark_generation_record_failed_and_refund_once(
|
|
db,
|
|
record=record,
|
|
error_message=error_message,
|
|
generation_attempt_no=int(record.generation_attempt_no or 1),
|
|
)
|
|
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
|
|
record.provider_create_claim_token = None
|
|
record.provider_create_lease_until = None
|
|
record.poll_claim_token = None
|
|
record.poll_lease_until = None
|
|
record.next_poll_at = None
|
|
record.download_claim_token = None
|
|
record.download_lease_until = None
|
|
record.download_next_retry_at = None
|
|
|
|
# 若任务已进入超分，必须同时撤销超分数据库租约；执行中的超分 Worker
|
|
# 在回填前校验 lease_token，发现 token 被清除后会中止，不得覆盖管理员终止状态。
|
|
from app.enums.video_upscale import VideoUpscaleStage, VideoUpscaleTaskStatus
|
|
from app.models.video_upscale_task import VideoUpscaleTask
|
|
|
|
try:
|
|
upscale_result = await execute_with_lock_timeout(
|
|
db,
|
|
select(VideoUpscaleTask)
|
|
.where(VideoUpscaleTask.generation_record_id == record.id)
|
|
.with_for_update()
|
|
.limit(1),
|
|
)
|
|
except DatabaseRowLockBusy as exc:
|
|
raise HTTPException(status_code=409, detail=exc.detail) from exc
|
|
upscale = upscale_result.scalar_one_or_none()
|
|
if upscale and upscale.status not in {
|
|
VideoUpscaleTaskStatus.COMPLETED.value,
|
|
VideoUpscaleTaskStatus.FAILED.value,
|
|
}:
|
|
upscale.status = VideoUpscaleTaskStatus.FAILED.value
|
|
upscale.stage = VideoUpscaleStage.FAILED.value
|
|
upscale.last_error = error_message
|
|
upscale.failed_at = datetime.now(CST)
|
|
upscale.next_retry_at = None
|
|
upscale.lease_token = None
|
|
upscale.lease_until = None
|
|
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
"管理员终止生成记录",
|
|
"PUT",
|
|
f"/admin/generation-records/{record_id}/status",
|
|
detail=json.dumps(
|
|
{
|
|
"record_id": record_id,
|
|
"new_status": new_status,
|
|
"generation_attempt_no": int(record.generation_attempt_no or 1),
|
|
"error_message": error_message,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
|
|
# Redis 注册表只做调度加速；删除失败不回滚已提交的业务终止状态。
|
|
try:
|
|
from app.services.celery_download_recovery_service import remove_download_active
|
|
from app.services.generation.pipeline.owner_service import redis_owner_item_id
|
|
from app.services.redis_registry_service import redis_remove_registry_item
|
|
from app.config import settings
|
|
|
|
registry_id = redis_owner_item_id(
|
|
"generation_record",
|
|
record_id,
|
|
int(record.generation_attempt_no or 1),
|
|
)
|
|
await remove_download_active(registry_id)
|
|
await redis_remove_registry_item(
|
|
hash_key=settings.POLL_ACTIVE_REDIS_HASH_KEY,
|
|
zset_key=settings.POLL_ACTIVE_REDIS_ZSET_KEY,
|
|
item_id=registry_id,
|
|
log_context="admin_generation_record_terminate",
|
|
)
|
|
except Exception:
|
|
pass
|
|
return {"message": "ok"}
|
|
|
|
|
|
@router.post("/generation-records/{record_id}/generate")
|
|
async def admin_generate_record_resource(
|
|
record_id: str,
|
|
body: dict,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""管理员触发 GenerationRecord 图片或视频资源生成。"""
|
|
from app.services.generation.pipeline.generation_record_service import (
|
|
commit_and_enqueue_generation_record,
|
|
prepare_generation_record_execution,
|
|
)
|
|
|
|
try:
|
|
result = await execute_with_lock_timeout(
|
|
db,
|
|
select(GenerationRecord, Project.name)
|
|
.join(Project, GenerationRecord.project_id == Project.id)
|
|
.where(
|
|
GenerationRecord.id == record_id,
|
|
GenerationRecord.deleted_at.is_(None),
|
|
Project.deleted_at.is_(None),
|
|
)
|
|
.with_for_update(),
|
|
)
|
|
except DatabaseRowLockBusy as exc:
|
|
raise HTTPException(status_code=409, detail=exc.detail) from exc
|
|
row = result.first()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="记录不存在")
|
|
|
|
record, project_name = row
|
|
type_str = "视频" if record.gen_type == GenerationType.video else "图片"
|
|
if record.status not in ("prompt_optimized", "failed"):
|
|
raise HTTPException(status_code=400, detail=f"当前状态不允许生成{type_str}")
|
|
if record.pipeline_stage == GenerationRecordPipelineStage.UPSCALE_FAILED.value:
|
|
raise HTTPException(status_code=409, detail="该任务为画质增强失败，请使用超分恢复命令处理")
|
|
|
|
attempt_no = await get_next_credit_attempt_no(
|
|
db,
|
|
owner_type=OWNER_GENERATION_RECORD,
|
|
owner_id=record.id,
|
|
)
|
|
|
|
if record.gen_type == GenerationType.video:
|
|
aspect_ratio = body.get("aspect_ratio", "16:9")
|
|
resolution = body.get("resolution", "720p")
|
|
if aspect_ratio not in ASPECT_RATIOS:
|
|
raise HTTPException(status_code=400, detail="不支持的画面比例")
|
|
if resolution not in RESOLUTIONS:
|
|
raise HTTPException(status_code=400, detail="不支持的分辨率")
|
|
|
|
from app.services.video_gen import get_active_engine
|
|
from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
|
|
|
|
engine = await get_active_engine(db)
|
|
try:
|
|
supported_provider_resolutions = json.loads(engine.supported_resolutions or "[]")
|
|
except (TypeError, json.JSONDecodeError):
|
|
supported_provider_resolutions = []
|
|
provider_resolution, upscale_enabled, upscale_snapshot_json = await build_video_upscale_snapshot(
|
|
db,
|
|
target_resolution=resolution,
|
|
aspect_ratio=aspect_ratio,
|
|
supported_provider_resolutions=supported_provider_resolutions,
|
|
)
|
|
record.aspect_ratio = aspect_ratio
|
|
record.resolution = resolution
|
|
record.provider_generation_resolution = provider_resolution
|
|
record.video_upscale_enabled_snapshot = upscale_enabled
|
|
record.video_upscale_snapshot_json = upscale_snapshot_json
|
|
elif record.gen_type == GenerationType.image:
|
|
from app.services.image_gen import get_active_image_engine
|
|
|
|
engine = await get_active_image_engine(db)
|
|
record.image_size = body.get("image_size") or record.image_size or "2K"
|
|
record.provider_generation_resolution = None
|
|
record.video_upscale_enabled_snapshot = False
|
|
record.video_upscale_snapshot_json = None
|
|
else:
|
|
raise HTTPException(status_code=400, detail="不支持的生成类型")
|
|
|
|
media_billing = await charge_generation_media_for_record(
|
|
db,
|
|
record=record,
|
|
project_name=project_name,
|
|
description_prefix=f"{type_str}生成(管理后台)-",
|
|
attempt_no=attempt_no,
|
|
engine_id=engine.id,
|
|
)
|
|
record.credits_cost = round(float(record.credits_cost or 0) + float(media_billing.total_charged or 0), 2)
|
|
prepare_generation_record_execution(record, engine=engine, attempt_no=attempt_no)
|
|
await db.flush()
|
|
await commit_and_enqueue_generation_record(db, record, reason="generation_record_admin_generate")
|
|
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"管理员触发生成{type_str}: {record_id}",
|
|
"POST",
|
|
f"/admin/generation-records/{record_id}/generate",
|
|
detail=json.dumps(
|
|
{
|
|
"record_id": record_id,
|
|
"gen_type": record.gen_type,
|
|
"project_name": project_name,
|
|
"generation_attempt_no": record.generation_attempt_no,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
return {"message": "ok", "record_id": record_id}
|
|
|
|
|
|
# ── File Uploads ─────────────────────────────────────────
|
|
|
|
import os
|
|
from fastapi import File, UploadFile
|
|
|
|
|
|
@router.post("/upload-pdf")
|
|
async def upload_pdf(
|
|
file: UploadFile = File(...),
|
|
config_key: str | None = None,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Upload a PDF file and save URL to system config."""
|
|
from app.config import settings
|
|
from app.utils.id_gen import generate_id
|
|
|
|
if not file.filename:
|
|
raise HTTPException(status_code=400, detail="请选择文件")
|
|
|
|
if not file.filename.lower().endswith(".pdf"):
|
|
raise HTTPException(status_code=400, detail="仅支持PDF格式")
|
|
|
|
content = await file.read()
|
|
if len(content) > 10 * 1024 * 1024:
|
|
raise HTTPException(status_code=400, detail="文件大小不能超过10MB")
|
|
|
|
safe_name = f"pdf_{generate_id()}.pdf"
|
|
file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, safe_name)
|
|
with open(file_path, "wb") as f:
|
|
f.write(content)
|
|
|
|
url = f"/uploads/{safe_name}"
|
|
|
|
if config_key:
|
|
result = await db.execute(
|
|
select(SystemConfig).where(SystemConfig.key == config_key).limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if config:
|
|
config.value = url
|
|
else:
|
|
db.add(SystemConfig(
|
|
id=generate_id(),
|
|
key=config_key,
|
|
value=url,
|
|
))
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"上传PDF文件: {file.filename}",
|
|
"POST",
|
|
"/admin/upload-pdf",
|
|
detail=json.dumps(
|
|
{
|
|
"filename": file.filename,
|
|
"config_key": config_key,
|
|
"url": url,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
|
|
return {"url": url}
|
|
|
|
|
|
@router.post("/upload-logo")
|
|
async def upload_logo(
|
|
file: UploadFile = File(...),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Upload a Logo image file and save URL to system config."""
|
|
from app.config import settings
|
|
from app.utils.id_gen import generate_id
|
|
|
|
if not file.filename:
|
|
raise HTTPException(status_code=400, detail="请选择文件")
|
|
|
|
allowed_extensions = ('.png', '.jpg', '.jpeg', '.gif', '.webp')
|
|
if not file.filename.lower().endswith(allowed_extensions):
|
|
raise HTTPException(status_code=400, detail="仅支持 PNG、JPG、GIF、WebP 格式图片")
|
|
|
|
content = await file.read()
|
|
if len(content) > 2 * 1024 * 1024:
|
|
raise HTTPException(status_code=400, detail="文件大小不能超过2MB")
|
|
|
|
safe_name = "site_logo.png"
|
|
file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, safe_name)
|
|
with open(file_path, "wb") as f:
|
|
f.write(content)
|
|
|
|
url = f"/uploads/{safe_name}"
|
|
|
|
result = await db.execute(
|
|
select(SystemConfig).where(SystemConfig.key == "site_logo").limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if config:
|
|
config.value = url
|
|
else:
|
|
db.add(SystemConfig(
|
|
id="cfg_site_logo",
|
|
key="site_logo",
|
|
value=url,
|
|
description="网站Logo图片",
|
|
))
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"上传Logo图片: {file.filename}",
|
|
"POST",
|
|
"/admin/upload-logo",
|
|
detail=json.dumps(
|
|
{
|
|
"filename": file.filename,
|
|
"url": url,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
)
|
|
await db.commit()
|
|
|
|
return {"url": url}
|
|
|
|
|
|
@router.post("/upload-login-video")
|
|
async def upload_login_video(
|
|
file: UploadFile = File(...),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""上传登录页背景视频/动图，保存 URL 到 system config login_bg_video。"""
|
|
from app.config import settings
|
|
|
|
if not file.filename:
|
|
raise HTTPException(status_code=400, detail="请选择文件")
|
|
|
|
content = await file.read()
|
|
if len(content) > 50 * 1024 * 1024:
|
|
raise HTTPException(status_code=400, detail="文件大小不能超过50MB")
|
|
|
|
ext = os.path.splitext(file.filename)[1].lower()
|
|
safe_name = f"login_bg_{generate_id()}{ext}"
|
|
file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, safe_name)
|
|
with open(file_path, "wb") as f:
|
|
f.write(content)
|
|
|
|
url = f"/uploads/{safe_name}"
|
|
|
|
result = await db.execute(
|
|
select(SystemConfig).where(SystemConfig.key == "login_bg_video").limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if config:
|
|
config.value = url
|
|
else:
|
|
db.add(SystemConfig(
|
|
id=generate_id(),
|
|
key="login_bg_video",
|
|
value=url,
|
|
description="登录页背景视频",
|
|
))
|
|
await db.flush()
|
|
await log_operation(
|
|
db,
|
|
admin.id,
|
|
admin.username,
|
|
f"上传登录背景视频: {file.filename}",
|
|
"POST",
|
|
"/admin/upload-login-video",
|
|
detail=json.dumps({"filename": file.filename, "url": url}, ensure_ascii=False),
|
|
)
|
|
await db.commit()
|
|
|
|
return {"url": url}
|
|
|
|
|
|
# ── Payment Stats ────────────────────────────────────────
|
|
|
|
|