from datetime import datetime, timezone, timedelta
import json

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import delete, func, select, update, desc, asc, order_by
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.schemas.admin import (
    CreditAdjustRequest,
    ModelConfigCreate,
    ModelConfigOut,
    SystemConfigUpdate,
    SystemConfigOut,
    AdminUserOut,
    AdminStatsOut,
    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.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.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_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)$"),
    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.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)
    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")
    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}/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")
    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),
    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
        )

    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)).order_by(desc(CreditRatio.gen_type), desc(CreditRatio.model_config_id)) 
    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, 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()

    items = []
    for record, username, project_name, industry, industry_label 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,
            "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,
            "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 ────────────────────────────────────────