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.schemas.admin import (
    CreditAdjustRequest,
    ModelConfigCreate,
    ModelConfigOut,
    SystemConfigUpdate,
    SystemConfigOut,
    AdminUserOut,
    AdminStatsOut,
    CreateUserRequest,
    UpdateMenusRequest,
    ResetPasswordRequest,
    UpdateFrontendUserKindRequest,
    OperationLogOut,
)
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.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.resource_signed_url_service import build_resource_signed_url
from app.services.payment import sync_pending_orders, process_refund

from app.services.generation_billing_service import (
    OWNER_GENERATION_RECORD,
    charge_generation_media_by_params,
    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)$"),
    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)
    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 {"items": [AdminUserOut.model_validate(u) for u in items], "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.user_type == "admin"),
        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,
    )
    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", ip=None)
    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")
    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)
    return user


@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")
    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")
    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")
    return user


@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")
    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")
    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),
    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,
        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
        )

    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()
    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()
    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),
    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).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 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,
            "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 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()
    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", "退款失败"))
    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()
    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()
    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()
    return {"message": "ok"}


# ── Video Engine ─────────────────────────────────────────

@router.get("/video-engines", response_model=list[VideoEngineOut])
async def list_video_engines(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(VideoEngine).order_by(VideoEngine.priority.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()
    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).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()
    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).limit(1)
    )
    engine = result.scalar_one_or_none()
    if not engine:
        raise HTTPException(status_code=404, detail="视频引擎不存在")
    await db.delete(engine)
    await db.flush()
    return {"message": "ok"}


# ── Image Engine ──────────────────────────────────────────

@router.get("/image-engines", response_model=list[ImageEngineOut])
async def list_image_engines(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(ImageEngine).order_by(ImageEngine.priority.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()
    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).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()
    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).limit(1)
    )
    engine = result.scalar_one_or_none()
    if not engine:
        raise HTTPException(status_code=404, detail="图片引擎不存在")
    await db.delete(engine)
    await db.flush()
    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).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))
    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()
    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()
    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()
    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(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(ModelConfig).order_by(ModelConfig.priority.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()
    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).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()
    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).limit(1))
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(status_code=404, detail="配置不存在")
    await db.delete(config)
    await db.flush()
    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.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.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().replace(hour=0, minute=0, second=0, microsecond=0)
    
    try:
        if start_date:
            date_start = datetime.strptime(start_date, "%Y-%m-%d")
        else:
            date_start = today_start
        if end_date:
            date_end = datetime.strptime(end_date, "%Y-%m-%d")
            date_end = date_end.replace(hour=23, minute=59, second=59, microsecond=999999)
        else:
            date_end = datetime.now()
    except:
        date_start = today_start
        date_end = datetime.now()

    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

    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),
    )


# ── 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)
        .join(User, GenerationRecord.user_id == User.id)
        .join(Project, GenerationRecord.project_id == Project.id)
        .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()

    items = []
    for record, username, project_name in rows:
        refs = None
        if record.media_references:
            try:
                refs = json.loads(record.media_references)
            except (json.JSONDecodeError, TypeError):
                refs = None
        items.append({
            "id": record.id,
            "user_id": record.user_id,
            "username": username,
            "project_id": record.project_id,
            "project_name": project_name,
            "original_prompt": record.original_prompt,
            "optimized_prompt": record.optimized_prompt,
            "duration": record.duration,
            "aspect_ratio": record.aspect_ratio,
            "resolution": record.resolution,
            "status": record.status,
            "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),
):
    """Admin update generation record status (e.g., confirm/reject)."""
    result = await db.execute(
        select(GenerationRecord).where(
            GenerationRecord.id == record_id,
            GenerationRecord.deleted_at.is_(None),
        )
        .with_for_update()
        .limit(1)
    )
    record = result.scalar_one_or_none()
    if not record:
        raise HTTPException(status_code=404, detail="记录不存在")

    new_status = body.get("status")
    if new_status not in ("prompt_optimized", "generating", "completed", "failed"):
        raise HTTPException(status_code=400, detail="无效状态")

    if new_status == "failed":
        await mark_generation_record_failed_and_refund_once(
            db,
            record=record,
            error_message=body.get("error_message") or record.error_message or "管理员设置为失败",
        )
    else:
        record.status = new_status

    if body.get("video_url"):
        record.video_url = body["video_url"]
    if body.get("video_cover_url"):
        record.video_cover_url = body["video_cover_url"]
    if body.get("image_url"):
        record.image_url = body["image_url"]
    if new_status == "completed":
        record.generated_at = datetime.now()
    await db.flush()
    return {"message": "ok"}


@router.post("/generation-records/{record_id}/generate")
async def admin_generate_video(
    record_id: str,
    body: dict,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Admin trigger video/image generation for a record with specified params."""
    from app.models.project import Project
    from app.services.video_queue import task_queue

    result = await db.execute(
        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()
    )
    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}")

    attempt_no = await get_next_credit_attempt_no(
        db,
        owner_type=OWNER_GENERATION_RECORD,
        owner_id=record.id,
    )

    if record.gen_type == GenerationType.video:
        # Video Generation
        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="不支持的分辨率")

        duration = record.duration or 5
        media_billing = await charge_generation_media_by_params(
            db,
            user_id=record.user_id,
            record_id=record.id,
            gen_type="video",
            duration=duration,
            resolution=resolution,
            project_name=project_name,
            description_prefix="视频生成(管理后台)",
            owner_type=OWNER_GENERATION_RECORD,
            attempt_no=attempt_no,
        )

        record.aspect_ratio = aspect_ratio
        record.resolution = resolution
        record.credits_cost = round(float(record.credits_cost or 0) + media_billing.total_charged, 2)
        record.status = "generating"
        record.error_message = None
        record.video_url = None
        record.video_cover_url = None
        record.image_url = None
        record.seedance_task_id = None
        await db.flush()

        try:
            from app.services.video_gen import get_active_engine, submit_video_task
            engine = await get_active_engine(db)
            task_id = await submit_video_task(
                db,
                engine,
                record,
                include_media_references=False,
            )
            record.seedance_task_id = task_id
            await db.flush()
            await task_queue.enqueue(record_id)
        except Exception as e:
            await mark_generation_record_failed_and_refund_once(
                db,
                record=record,
                error_message=str(e),
            )
            await db.flush()

    elif record.gen_type == GenerationType.image:
        # Image generation

        post_image_size = body.get("image_size", "")
        image_size = post_image_size or record.image_size or "2K"
        media_billing = await charge_generation_media_by_params(
            db,
            user_id=record.user_id,
            record_id=record.id,
            gen_type="image",
            image_size=image_size,
            project_name=project_name,
            description_prefix="图片生成(管理后台)",
            owner_type=OWNER_GENERATION_RECORD,
            attempt_no=attempt_no,
        )

        record.image_size = image_size
        record.credits_cost = round(float(record.credits_cost or 0) + media_billing.total_charged, 2)
        record.status = "generating"
        record.error_message = None
        record.image_url = None
        record.video_url = None
        record.video_cover_url = None
        record.seedance_task_id = None
        await db.flush()

        try:
            await task_queue.enqueue(record_id)
        except Exception as e:
            await mark_generation_record_failed_and_refund_once(
                db,
                record=record,
                error_message=str(e),
            )
            await db.flush()

    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.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.commit()

    return {"url": url}


# ── Payment Stats ────────────────────────────────────────