from datetime import datetime, timezone, timedelta
import json

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db, get_admin_user
from app.models.user import User
from app.models.project import Project
from app.models.generation_record import GenerationRecord
from app.models.chat_generation_task import ChatGenerationTask
from app.models.credit_record import CreditRecord
from app.models.model_config import ModelConfig
from app.models.system_config import SystemConfig
from app.models.payment_order import PaymentOrder
from app.models.token_usage import TokenUsage
from app.models.notification import Notification
from app.models.notification_read import NotificationRead
from app.models.industry_config import IndustryConfig
from app.models.video_engine import VideoEngine
from app.models.image_engine import ImageEngine
from app.models.credit_ratio import CreditRatio
from app.models.operation_log import OperationLog
from app.enums.user import FrontendUserKind, UserType
from app.enums.team import TEAM_UNASSIGNED_VALUE
from app.schemas.admin import (
    CreditAdjustRequest,
    ModelConfigCreate,
    ModelConfigOut,
    SystemConfigCreate,
    SystemConfigUpdate,
    SystemConfigOut,
    AdminUserOut,
    AdminStatsOut,
    DailyCreditOut,
    TeamCreditOut,
    ModelUsageOut,
    CreateUserRequest,
    UpdateMenusRequest,
    ResetPasswordRequest,
    UpdateFrontendUserKindRequest,
)
from app.schemas.team import UpdateUserTeamRequest
from app.schemas.industry import IndustryConfigCreate
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.private_portrait.reference_resolver import batch_resolve_private_portrait_reference_display_urls
from app.services.resource_signed_url_service import build_resource_signed_url
from app.services.payment import 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.utils.id_gen import generate_id


CST = timezone(timedelta(hours=8))


def _safe_json_object(value: str | None) -> dict | None:
    if not value:
        return None
    try:
        parsed = json.loads(value)
    except (TypeError, json.JSONDecodeError):
        return None
    return parsed if isinstance(parsed, dict) else None


def _iso(dt):
    """Serialize datetime as naive ISO string (UTC→CST, strip tzinfo)."""
    if dt is None:
        return None
    try:
        d = dt if isinstance(dt, datetime) else datetime.fromisoformat(str(dt))
        if d.tzinfo and d.tzinfo.utcoffset(None) == timedelta(0):
            d = d.astimezone(CST)
        return d.replace(tzinfo=None).isoformat()
    except (ValueError, TypeError):
        return str(dt)


router = APIRouter(prefix="/admin", tags=["admin"])


# ── User Management ──────────────────────────────────────

@router.get("/users")
async def list_users(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=1000),
    search: str = Query(""),
    user_type: str | None = Query(None, pattern="^(frontend|admin)$"),
    frontend_user_kind: str | None = Query(None, pattern="^(internal|external)$"),
    team_id: str | None = Query(None),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    query = select(User).order_by(User.created_at.desc())
    count_query = select(func.count(User.id))
    if search:
        like = f"%{search}%"
        search_filter = (User.username.ilike(like)) | (User.email.ilike(like)) | (User.phone.ilike(like))
        query = query.where(search_filter)
        count_query = count_query.where(search_filter)
    if user_type:
        query = query.where(User.user_type == user_type)
        count_query = count_query.where(User.user_type == user_type)
    if frontend_user_kind:
        query = query.where(User.user_type == UserType.FRONTEND.value, User.frontend_user_kind == frontend_user_kind)
        count_query = count_query.where(User.user_type == UserType.FRONTEND.value, User.frontend_user_kind == frontend_user_kind)
    if team_id:
        team_filter = User.team_id.is_(None) if team_id == TEAM_UNASSIGNED_VALUE else (User.team_id == team_id)
        query = query.where(team_filter)
        count_query = count_query.where(team_filter)
    total = (await db.execute(count_query)).scalar() or 0
    result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
    users = list(result.scalars().all())
    user_ids = [u.id for u in users]
    team_ids = [getattr(u, "team_id", None) for u in users if getattr(u, "team_id", None)]
    capacity_map = await batch_get_user_resource_capacity_usage(db, user_ids)
    team_name_map = await batch_get_team_name_map(db, team_ids)
    return {
        "items": [
            AdminUserOut.model_validate(user)
            .model_copy(
                update={
                    "resource_capacity": capacity_map.get(user.id),
                    "team_name": team_name_map.get(getattr(user, "team_id", None)),
                }
            )
            .model_dump(mode="json")
            for user in users
        ],
        "total": total,
    }


@router.post("/users", response_model=AdminUserOut)
async def create_user(
    req: CreateUserRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    import random
    # Frontend users: use phone, auto-generate username
    if req.user_type == "frontend":
        if not req.phone:
            raise HTTPException(status_code=400, detail="前台用户必须填写手机号")
        existing_phone = await db.execute(select(User).where(User.phone == req.phone).limit(1))
        if existing_phone.scalar_one_or_none():
            raise HTTPException(status_code=400, detail="该手机号已注册")
        username = f"用户{req.phone[-4:]}"
        existing_name = await db.execute(select(User).where(User.username == username).limit(1))
        if existing_name.scalar_one_or_none():
            username = f"用户{req.phone[-4:]}{random.randint(10, 99)}"
    else:
        if not req.username:
            raise HTTPException(status_code=400, detail="后台用户必须填写用户名")
        username = req.username
        existing = await db.execute(select(User).where(User.username == username).limit(1))
        if existing.scalar_one_or_none():
            raise HTTPException(status_code=400, detail="用户名已存在")

    user = User(
        id=generate_id(),
        username=username,
        hashed_password=hash_password(req.password),
        email=req.email,
        phone=req.phone,
        credits=req.credits,
        is_admin=req.is_admin if req.user_type == "admin" else False,
        user_type=req.user_type,
        frontend_user_kind=req.frontend_user_kind if req.user_type == "frontend" else FrontendUserKind.EXTERNAL.value,
        allowed_menus=req.allowed_menus,
        private_portrait_asset_limit=req.private_portrait_asset_limit,
    )
    user.credits = round(user.credits, 2)
    db.add(user)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"创建用户 {username}",
        "POST",
        "/admin/users",
        detail=json.dumps(
            {
                "user_id": user.id,
                "username": username,
                "user_type": req.user_type,
                "frontend_user_kind": user.frontend_user_kind,
                "private_portrait_asset_limit": user.private_portrait_asset_limit,
                "credits": user.credits,
                "phone": user.phone,
                "email": user.email,
            },
            ensure_ascii=False,
            default=str,
        ),
        ip=None,
    )
    await db.commit()
    return user


@router.put("/users/{user_id}/menus")
async def update_user_menus(
    user_id: str,
    req: UpdateMenusRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).where(User.id == user_id).limit(1))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.allowed_menus = req.allowed_menus
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新菜单权限",
        "PUT",
        f"/admin/users/{user_id}/menus",
        detail=json.dumps(
            {
                "user_id": user_id,
                "allowed_menus": req.allowed_menus,
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()
    return {"message": "ok"}


@router.get("/users/{user_id}", response_model=AdminUserOut)
async def get_user(
    user_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).where(User.id == user_id).limit(1))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.credits = round(user.credits, 2)
    resource_capacity = await get_user_resource_capacity_usage(db, user.id)
    team_name_map = await batch_get_team_name_map(db, [getattr(user, "team_id", None)])
    return AdminUserOut.model_validate(user).model_copy(
        update={
            "resource_capacity": resource_capacity,
            "team_name": team_name_map.get(getattr(user, "team_id", None)),
        }
    )


@router.post("/users/{user_id}/credits")
async def adjust_credits(
    user_id: str,
    req: CreditAdjustRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    if req.amount > 0:
        await add_credits(db, user_id, req.amount, f"管理员调整: {req.description}", record_meta=build_admin_adjust_meta())
    else:
        await deduct_credits(db, user_id, abs(req.amount), f"管理员调整: {req.description}", record_meta=build_admin_adjust_meta())
    await create_notification(
        db, user_id, "积分变动通知",
        f"您的积分已{'增加' if req.amount > 0 else '扣除'}{abs(req.amount)}积分。原因：{req.description}",
        "credit",
    )
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"调整积分 {'+' if req.amount > 0 else ''}{req.amount}",
        "POST",
        f"/admin/users/{user_id}/credits",
        detail=json.dumps(
            {
                "user_id": user_id,
                "amount": req.amount,
                "description": req.description,
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()
    return {"message": "ok"}


@router.put("/users/{user_id}/status")
async def update_user_status(
    user_id: str,
    body: dict,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    is_active = body.get("is_active", True)
    await db.execute(
        update(User).where(User.id == user_id).values(is_active=is_active)
    )
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"{'启用' if is_active else '禁用'}用户",
        "PUT",
        f"/admin/users/{user_id}/status",
        detail=json.dumps(
            {
                "user_id": user_id,
                "is_active": is_active,
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()
    return {"message": "ok"}


@router.put("/users/{user_id}/admin-status")
async def update_user_admin_status(
    user_id: str,
    body: dict,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    is_admin = body.get("is_admin", False)
    result = await db.execute(select(User).where(User.id == user_id).limit(1))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    if user.user_type != "admin":
        raise HTTPException(status_code=400, detail="仅后台用户支持设置超级管理员状态")
    user.is_admin = is_admin
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"{is_admin and '设为' or '取消'}超级管理员",
        "PUT",
        f"/admin/users/{user_id}/admin-status",
        detail=json.dumps(
            {
                "user_id": user_id,
                "is_admin": is_admin,
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}


@router.put("/users/{user_id}/frontend-kind", response_model=AdminUserOut)
async def update_user_frontend_kind(
    user_id: str,
    req: UpdateFrontendUserKindRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).where(User.id == user_id).limit(1))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    if user.user_type != UserType.FRONTEND.value:
        raise HTTPException(status_code=400, detail="仅前台用户支持设置内部/外部归类")
    user.frontend_user_kind = req.frontend_user_kind or FrontendUserKind.EXTERNAL.value
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"设置前台用户归类为 {user.frontend_user_kind}",
        "PUT",
        f"/admin/users/{user_id}/frontend-kind",
        detail=json.dumps(
            {
                "user_id": user_id,
                "frontend_user_kind": user.frontend_user_kind,
            },
            ensure_ascii=False,
        ),
    )
    return user


@router.put("/users/{user_id}/team", response_model=AdminUserOut)
async def update_user_team(
    user_id: str,
    req: UpdateUserTeamRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    user, before, after = await set_frontend_user_team(db, user_id=user_id, team_id=req.team_id)
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"设置用户团队 {user.username}",
        "PUT",
        f"/admin/users/{user_id}/team",
        detail=json.dumps({"before": before, "after": after}, ensure_ascii=False),
    )
    resource_capacity = await get_user_resource_capacity_usage(db, user.id)
    return AdminUserOut.model_validate(user).model_copy(
        update={"resource_capacity": resource_capacity, "team_name": after.get("team_name")}
    )


@router.put("/users/{user_id}/reset-password")
async def reset_user_password(
    user_id: str,
    req: ResetPasswordRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).where(User.id == user_id).limit(1))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.hashed_password = hash_password(req.new_password)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"重置密码",
        "PUT",
        f"/admin/users/{user_id}/reset-password",
        detail=json.dumps(
            {
                "user_id": user_id,
                "new_password": "***",
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}


@router.post("/change-password")
async def admin_change_password(
    body: dict,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    old_password = body.get("old_password", "")
    new_password = body.get("new_password", "")
    if not verify_password(old_password, admin.hashed_password):
        raise HTTPException(status_code=400, detail="原密码错误")
    if len(new_password) < 6:
        raise HTTPException(status_code=400, detail="密码至少6位")
    admin.hashed_password = hash_password(new_password)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        "修改密码",
        "POST",
        "/admin/change-password",
        detail=json.dumps(
            {
                "old_password": "***",
                "new_password": "***",
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "密码修改成功"}


# ── Credit Records ───────────────────────────────────────

@router.get("/credit-records")
async def list_credit_records(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=1000),
    user_id: str | None = Query(None),
    user_name: str | None = Query(None),
    user_type: str | None = Query(None),
    frontend_user_kind: str | None = Query(None),
    team_id: str | None = Query(None),
    record_type: str | None = Query(None),
    type: str | None = Query(None),
    credit_subject: str | None = Query(None),
    media_type: str | None = Query(None),
    charge_kind: str | None = Query(None),
    source_module: str | None = Query(None),
    source_step_code: str | None = Query(None),
    billing_scene: str | None = Query(None),
    start_date: str = Query(None),
    end_date: str = Query(None),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """List all credit transaction records with filters and full summary."""
    return await list_admin_credit_records(
        db,
        page=page,
        page_size=page_size,
        user_id=user_id,
        user_name=user_name,
        user_type=user_type,
        frontend_user_kind=frontend_user_kind,
        team_id=team_id,
        record_type=record_type or type,
        credit_subject=credit_subject,
        media_type=media_type,
        charge_kind=charge_kind,
        source_module=source_module,
        source_step_code=source_step_code,
        billing_scene=billing_scene,
        start_date=start_date,
        end_date=end_date,
    )


# ── Notification Admin ───────────────────────────────────

@router.get("/notifications")
async def list_admin_notifications(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=500),
    user_id: str | None = Query(None),
    is_read: bool = Query(None),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """List all notifications with optional filters."""
    query = select(Notification).order_by(Notification.created_at.desc())
    count_query = select(func.count(Notification.id))

    if user_id:
        query = query.where(Notification.user_id == user_id)
        count_query = count_query.where(Notification.user_id == user_id)

    if is_read is not None:
        query = query.where(Notification.is_read == is_read)
        count_query = count_query.where(Notification.is_read == is_read)

    total = (await db.execute(count_query)).scalar() or 0
    result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
    items = result.scalars().all()

    return {
        "total": total,
        "items": [
            {
                "id": item.id,
                "user_id": item.user_id,
                "title": item.title,
                "content": item.content,
                "type": item.type,
                "is_read": item.is_read,
                "related_id": item.related_id,
                "created_at": _iso(item.created_at),
            }
            for item in items
        ],
    }


@router.post("/notifications")
async def create_admin_notification(
    body: dict,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Create notification. target_user_id=None broadcasts to all users."""
    title = body.get("title", "")
    content = body.get("content", "")
    notif_type = body.get("type", "system")
    target_user_id = body.get("target_user_id")

    if not title or not content:
        raise HTTPException(status_code=400, detail="标题和内容不能为空")

    if target_user_id:
        # Send to specific user
        await create_notification(
            db, target_user_id, title, content, notif_type, push_ws=True
        )
    else:
        # Broadcast: create one notification with user_id=NULL
        await create_notification(
            db, None, title, content, notif_type, push_ws=True
        )

    await log_operation(
        db,
        admin.id,
        admin.username,
        f"{'发送通知给指定用户' if target_user_id else '广播通知'}: {title}",
        "POST",
        "/admin/notifications",
        detail=json.dumps(
            {
                "title": title,
                "target_user_id": target_user_id,
                "type": notif_type,
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}


@router.delete("/notifications/{notification_id}")
async def delete_admin_notification(
    notification_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Notification).where(Notification.id == notification_id).limit(1)
    )
    notif = result.scalar_one_or_none()
    if not notif:
        raise HTTPException(status_code=404, detail="通知不存在")
    await db.execute(
        delete(NotificationRead).where(NotificationRead.notification_id == notification_id)
    )
    await db.delete(notif)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"删除通知: {notif.title}",
        "DELETE",
        f"/admin/notifications/{notification_id}",
        detail=json.dumps(
            {
                "notification_id": notification_id,
                "title": notif.title,
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}


@router.get("/notifications/{notification_id}/read-users")
async def get_notification_read_users(
    notification_id: str,
    page: int = Query(1, ge=1),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Get list of users who have read a specific notification."""
    from app.services.notification import get_notification_read_users
    items, total = await get_notification_read_users(db, notification_id, page)
    return {"total": total, "items": items}


# ── Payment Config ───────────────────────────────────────

@router.get("/payment-configs")
async def list_payment_configs(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Return system configs that start with payment_ prefix."""
    result = await db.execute(
        select(SystemConfig).where(SystemConfig.key.like("payment_%"))
    )
    configs = result.scalars().all()
    return [
        {
            "id": c.id,
            "key": c.key,
            "value": c.value,
            "description": c.description,
        }
        for c in configs
    ]


@router.put("/payment-configs/batch")
async def batch_update_payment_configs(
    req: dict[str, str],
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Batch upsert payment configs. Creates missing keys, updates existing ones."""
    from app.utils.id_gen import generate_id

    for key, value in req.items():
        if not key.startswith("payment_"):
            continue
        result = await db.execute(
            select(SystemConfig).where(SystemConfig.key == key).limit(1)
        )
        config = result.scalar_one_or_none()
        if config:
            config.value = value
        else:
            db.add(SystemConfig(
                id=generate_id(),
                key=key,
                value=value,
            ))
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        "批量更新支付配置",
        "PUT",
        "/admin/payment-configs/batch",
        detail=json.dumps(list(req.keys()), ensure_ascii=False),
    )
    return {"ok": True}


@router.get("/payment-stats")
async def get_payment_stats(
    payment_method: str | None = Query(None),
    status: str | None = Query(None),
    start_date: str | None = Query(None),
    end_date: str | None = Query(None),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Return payment statistics for admin dashboard with filters."""
    from sqlalchemy import func

    # Ensure by_status has all expected statuses with defaults
    by_status = {
        "pending": {"count": 0, "amount": 0.0},
        "paid": {"count": 0, "amount": 0.0},
        "cancelled": {"count": 0, "amount": 0.0},
        "refunded": {"count": 0, "amount": 0.0},
    }

    # Parse dates and build base query filters
    now_cst = datetime.now(CST)
    today_start = now_cst.replace(hour=0, minute=0, second=0, microsecond=0)
    today_end = today_start + timedelta(days=1)

    # Default to today if no date range provided
    query_start = today_start
    query_end = today_end

    if start_date:
        query_start = datetime.fromisoformat(start_date).replace(tzinfo=CST)
    if end_date:
        query_end = (datetime.fromisoformat(end_date) + timedelta(days=1)).replace(tzinfo=CST)

    # Build filter list for status breakdown
    breakdown_filters = []
    if payment_method:
        breakdown_filters.append(PaymentOrder.payment_method == payment_method)
    if status:
        breakdown_filters.append(PaymentOrder.status == status)
    # Always apply date range to breakdown
    breakdown_filters.append(PaymentOrder.created_at >= query_start)
    breakdown_filters.append(PaymentOrder.created_at < query_end)

    # Status breakdown
    status_result = await db.execute(
        select(
            PaymentOrder.status,
            func.count().label("count"),
            func.coalesce(func.sum(PaymentOrder.amount), 0).label("amount"),
        )
        .where(*breakdown_filters)
        .group_by(PaymentOrder.status)
    )
    for row in status_result.all():
        if row.status in by_status:
            by_status[row.status] = {
                "count": row.count,
                "amount": round(float(row.amount), 2)
            }
        else:
            # Map any unexpected status to cancelled
            by_status["cancelled"]["count"] += row.count
            by_status["cancelled"]["amount"] += round(float(row.amount), 2)

    # Today's stats (CST time zone) - independent of filter
    today_result = await db.execute(
        select(
            func.count().label("paid_count"),
            func.coalesce(func.sum(PaymentOrder.amount), 0).label("paid_amount"),
        ).where(
            PaymentOrder.status == "paid",
            PaymentOrder.paid_at >= today_start,
            PaymentOrder.paid_at < today_end,
        )
    )
    today_row = today_result.one()

    # Monthly cumulative stats
    month_start = now_cst.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    month_end = (month_start + timedelta(days=32)).replace(day=1, hour=0, minute=0, second=0, microsecond=0)

    month_result = await db.execute(
        select(
            func.count().label("paid_count"),
            func.coalesce(func.sum(PaymentOrder.amount), 0).label("paid_amount"),
        ).where(
            PaymentOrder.status == "paid",
            PaymentOrder.paid_at >= month_start,
            PaymentOrder.paid_at < month_end,
        )
    )
    month_row = month_result.one()

    return {
        "by_status": by_status,
        "today": {
            "paid_count": today_row.paid_count,
            "paid_amount": round(float(today_row.paid_amount), 2),
        },
        "month": {
            "paid_count": month_row.paid_count,
            "paid_amount": round(float(month_row.paid_amount), 2),
        },
    }


@router.get("/payment-orders")
async def list_payment_orders(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=500),
    payment_method: str | None = Query(None),
    status: str | None = Query(None),
    phone: str | None = Query(None, description="按用户手机号模糊搜索"),
    start_date: str | None = Query(None),
    end_date: str | None = Query(None),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Return paginated payment orders for admin dashboard."""
    query = select(PaymentOrder, User.username, User.phone).join(User, PaymentOrder.user_id == User.id)
    count_query = select(func.count(PaymentOrder.id))

    filters = []
    if payment_method:
        filters.append(PaymentOrder.payment_method == payment_method)
    if status:
        filters.append(PaymentOrder.status == status)
    if phone:
        filters.append(User.phone.ilike(f"%{phone.strip()}%"))
    if start_date:
        filters.append(PaymentOrder.created_at >= datetime.fromisoformat(start_date).replace(tzinfo=CST))
    if end_date:
        filters.append(PaymentOrder.created_at < (datetime.fromisoformat(end_date) + timedelta(days=1)).replace(tzinfo=CST))

    for f in filters:
        query = query.where(f)
        count_query = count_query.where(f)

    total = (await db.execute(count_query)).scalar() or 0
    result = await db.execute(
        query.order_by(PaymentOrder.created_at.desc()).offset((page - 1) * page_size).limit(page_size)
    )
    rows = result.all()

    items = [
        {
            "id": o.id,
            "orderNo": o.order_no,
            "order_no": o.order_no,
            "userId": o.user_id,
            "user_id": o.user_id,
            "username": username,
            "phone": user_phone,
            "amount": round(float(o.amount), 2),
            "credits": round(float(o.credits), 2),
            "paymentMethod": o.payment_method,
            "payment_method": o.payment_method,
            "status": o.status,
            "tradeNo": o.trade_no,
            "trade_no": o.trade_no,
            "paidAt": o.paid_at.isoformat() if o.paid_at else None,
            "paid_at": o.paid_at.isoformat() if o.paid_at else None,
            "createdAt": o.created_at.isoformat() if o.created_at else None,
            "created_at": o.created_at.isoformat() if o.created_at else None,
        }
        for o, username, user_phone in rows
    ]

    return {"items": items, "total": total, "page": page, "page_size": page_size}

@router.put("/payment-configs/{config_id}")
async def update_payment_config(
    config_id: str,
    req: SystemConfigUpdate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Update a payment system config value."""
    result = await db.execute(
        select(SystemConfig).where(
            SystemConfig.id == config_id,
            SystemConfig.key.like("payment_%"),
        )
        .limit(1)
    )
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(status_code=404, detail="支付配置不存在")
    config.value = req.value
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新支付配置: {config.key}",
        "PUT",
        f"/admin/payment-configs/{config_id}",
        detail=json.dumps(
            {
                "config_id": config_id,
                "key": config.key,
                "value": req.value,
            },
            ensure_ascii=False,
        ),
    )
    return {
        "id": config.id,
        "key": config.key,
        "value": config.value,
        "description": config.description,
    }


@router.post("/payment-orders/{order_no}/refund")
async def refund_payment_order(
    order_no: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Refund a paid payment order."""
    result = await process_refund(db, order_no)
    if not result.get("success"):
        raise HTTPException(status_code=400, detail=result.get("message", "退款失败"))
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"订单退款: {order_no}",
        "POST",
        f"/admin/payment-orders/{order_no}/refund",
        detail=json.dumps(
            {
                "order_no": order_no,
            },
            ensure_ascii=False,
        ),
    )
    return result


# ── Industry Config ──────────────────────────────────────

def _serialize_industry(ind: IndustryConfig) -> dict:
    """Convert industry config to dict with parsed skills."""
    skills = []
    if ind.skills:
        try:
            raw = json.loads(ind.skills)
            if isinstance(raw, list):
                skills = raw if raw and isinstance(raw[0], dict) else [{"key": s, "label": s} for s in raw]
        except (json.JSONDecodeError, TypeError):
            skills = []
    return {
        "id": ind.id,
        "key": ind.key,
        "label": ind.label,
        "icon": ind.icon or "",
        "description": ind.description,
        "skills": skills,
        "is_active": ind.is_active,
        "sort_order": ind.sort_order,
        "created_at": ind.created_at,
    }


@router.get("/industry-configs")
async def list_industry_configs(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(IndustryConfig).order_by(IndustryConfig.sort_order.asc())
    )
    return [_serialize_industry(ind) for ind in result.scalars().all()]


@router.post("/industry-configs")
async def create_industry_config(
    req: IndustryConfigCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    config = IndustryConfig(
        id=generate_id(),
        key=req.key,
        label=req.label,
        icon=req.icon or "",
        description=req.description,
        skills=json.dumps(req.skills, ensure_ascii=False),
        is_active=req.is_active,
        sort_order=req.sort_order,
    )
    db.add(config)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"创建行业配置: {req.label}",
        "POST",
        "/admin/industry-configs",
        detail=json.dumps(
            {
                "config_id": config.id,
                "key": req.key,
                "label": req.label,
            },
            ensure_ascii=False,
        ),
    )
    return _serialize_industry(config)


@router.put("/industry-configs/{config_id}")
async def update_industry_config(
    config_id: str,
    req: IndustryConfigCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(IndustryConfig).where(IndustryConfig.id == config_id).limit(1)
    )
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(status_code=404, detail="行业配置不存在")
    config.key = req.key
    config.label = req.label
    config.icon = req.icon or ""
    config.description = req.description
    config.skills = json.dumps(req.skills, ensure_ascii=False)
    config.is_active = req.is_active
    config.sort_order = req.sort_order
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新行业配置: {req.label}",
        "PUT",
        f"/admin/industry-configs/{config_id}",
        detail=json.dumps(
            {
                "config_id": config_id,
                "key": req.key,
                "label": req.label,
            },
            ensure_ascii=False,
        ),
    )
    return _serialize_industry(config)


@router.delete("/industry-configs/{config_id}")
async def delete_industry_config(
    config_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(IndustryConfig).where(IndustryConfig.id == config_id).limit(1)
    )
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(status_code=404, detail="行业配置不存在")
    await db.delete(config)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"删除行业配置: {config.label}",
        "DELETE",
        f"/admin/industry-configs/{config_id}",
        detail=json.dumps(
            {
                "config_id": config_id,
                "key": config.key,
                "label": config.label,
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}


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

@router.get("/video-engines", response_model=list[VideoEngineOut])
async def list_video_engines(
    include_deleted: bool = Query(False),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    query = select(VideoEngine)
    if not include_deleted:
        query = query.where(VideoEngine.deleted_at.is_(None))
    result = await db.execute(
        query.order_by(VideoEngine.priority.desc(), VideoEngine.id.desc())
    )
    return result.scalars().all()


@router.post("/video-engines", response_model=VideoEngineOut)
async def create_video_engine(
    req: VideoEngineCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    engine = VideoEngine(id=generate_id(), **req.model_dump())
    db.add(engine)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"创建视频引擎: {req.name}",
        "POST",
        "/admin/video-engines",
        detail=json.dumps(
            {
                "engine_id": engine.id,
                "name": req.name,
            },
            ensure_ascii=False,
        ),
    )
    return engine


@router.put("/video-engines/{engine_id}", response_model=VideoEngineOut)
async def update_video_engine(
    engine_id: str,
    req: VideoEngineCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(VideoEngine).where(VideoEngine.id == engine_id, VideoEngine.deleted_at.is_(None)).limit(1)
    )
    engine = result.scalar_one_or_none()
    if not engine:
        raise HTTPException(status_code=404, detail="视频引擎不存在")
    for k, v in req.model_dump().items():
        setattr(engine, k, v)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新视频引擎: {engine.name}",
        "PUT",
        f"/admin/video-engines/{engine_id}",
        detail=json.dumps(
            {
                "engine_id": engine_id,
                "name": engine.name,
            },
            ensure_ascii=False,
        ),
    )
    return engine


@router.delete("/video-engines/{engine_id}")
async def delete_video_engine(
    engine_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(VideoEngine).where(VideoEngine.id == engine_id, VideoEngine.deleted_at.is_(None)).limit(1)
    )
    engine = result.scalar_one_or_none()
    if not engine:
        raise HTTPException(status_code=404, detail="视频引擎不存在")
    engine_name = engine.name
    engine.deleted_at = datetime.now(timezone.utc)
    engine.is_active = False
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"软删除视频引擎: {engine_name}",
        "DELETE",
        f"/admin/video-engines/{engine_id}",
        detail=json.dumps(
            {
                "engine_id": engine_id,
                "name": engine_name,
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}


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

@router.get("/image-engines", response_model=list[ImageEngineOut])
async def list_image_engines(
    include_deleted: bool = Query(False),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    query = select(ImageEngine)
    if not include_deleted:
        query = query.where(ImageEngine.deleted_at.is_(None))
    result = await db.execute(
        query.order_by(ImageEngine.priority.desc(), ImageEngine.id.desc())
    )
    return result.scalars().all()


@router.post("/image-engines", response_model=ImageEngineOut)
async def create_image_engine(
    req: ImageEngineCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    engine = ImageEngine(id=generate_id(), **req.model_dump())
    db.add(engine)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"创建图片引擎: {req.name}",
        "POST",
        "/admin/image-engines",
        detail=json.dumps(
            {
                "engine_id": engine.id,
                "name": req.name,
            },
            ensure_ascii=False,
        ),
    )
    return engine


@router.put("/image-engines/{engine_id}", response_model=ImageEngineOut)
async def update_image_engine(
    engine_id: str,
    req: ImageEngineCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(ImageEngine).where(ImageEngine.id == engine_id, ImageEngine.deleted_at.is_(None)).limit(1)
    )
    engine = result.scalar_one_or_none()
    if not engine:
        raise HTTPException(status_code=404, detail="图片引擎不存在")
    for k, v in req.model_dump().items():
        setattr(engine, k, v)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新图片引擎: {engine.name}",
        "PUT",
        f"/admin/image-engines/{engine_id}",
        detail=json.dumps(
            {
                "engine_id": engine_id,
                "name": engine.name,
            },
            ensure_ascii=False,
        ),
    )
    return engine


@router.delete("/image-engines/{engine_id}")
async def delete_image_engine(
    engine_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(ImageEngine).where(ImageEngine.id == engine_id, ImageEngine.deleted_at.is_(None)).limit(1)
    )
    engine = result.scalar_one_or_none()
    if not engine:
        raise HTTPException(status_code=404, detail="图片引擎不存在")
    engine_name = engine.name
    engine.deleted_at = datetime.now(timezone.utc)
    engine.is_active = False
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"软删除图片引擎: {engine_name}",
        "DELETE",
        f"/admin/image-engines/{engine_id}",
        detail=json.dumps(
            {
                "engine_id": engine_id,
                "name": engine_name,
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}




async def _validate_credit_ratio_engine(db: AsyncSession, req: CreditRatioCreate) -> None:
    """校验积分规则绑定的引擎是否存在。

    CreditRatio.model_config_id 为兼容旧字段名，当前实际保存引擎ID：
    - gen_type=image 时对应 image_engines.id
    - gen_type=video 时对应 video_engines.id
    """
    gen_type = (req.gen_type or "").lower().strip()
    engine_id = (req.model_config_id or "").strip()
    if gen_type not in ("image", "video"):
        raise HTTPException(status_code=400, detail="gen_type 仅支持 image 或 video")
    if not engine_id:
        raise HTTPException(status_code=400, detail="model_config_id 不能为空，当前字段用于保存图片/视频引擎ID")

    model = ImageEngine if gen_type == "image" else VideoEngine
    result = await db.execute(select(model).where(model.id == engine_id, model.deleted_at.is_(None)).limit(1))
    engine = result.scalar_one_or_none()
    if not engine:
        detail = "图片积分规则绑定的图片引擎不存在" if gen_type == "image" else "视频积分规则绑定的视频引擎不存在"
        raise HTTPException(status_code=400, detail=detail)


# ── Credit Ratio ─────────────────────────────────────────

@router.get("/credit-ratios", response_model=list[CreditRatioOut])
async def list_credit_ratios(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(CreditRatio).order_by(CreditRatio.gen_type.desc(), CreditRatio.model_config_id.desc()))
    return result.scalars().all()


@router.post("/credit-ratios", response_model=CreditRatioOut)
async def create_credit_ratio(
    req: CreditRatioCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    await _validate_credit_ratio_engine(db, req)
    data = req.model_dump()
    data["gen_type"] = data["gen_type"].lower().strip()
    data["model_config_id"] = data["model_config_id"].strip()
    ratio = CreditRatio(id=generate_id(), **data)
    db.add(ratio)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"创建积分规则: {data.get('gen_type', '')} - {data.get('model_config_id', '')}",
        "POST",
        "/admin/credit-ratios",
        detail=json.dumps(
            {
                "ratio_id": ratio.id,
                "gen_type": data.get("gen_type"),
                "model_config_id": data.get("model_config_id"),
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()
    return ratio


@router.put("/credit-ratios/{ratio_id}", response_model=CreditRatioOut)
async def update_credit_ratio(
    ratio_id: str,
    req: CreditRatioCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(CreditRatio).where(CreditRatio.id == ratio_id).limit(1)
    )
    ratio = result.scalar_one_or_none()
    if not ratio:
        raise HTTPException(status_code=404, detail="积分比例不存在")
    await _validate_credit_ratio_engine(db, req)
    data = req.model_dump()
    data["gen_type"] = data["gen_type"].lower().strip()
    data["model_config_id"] = data["model_config_id"].strip()
    for k, v in data.items():
        setattr(ratio, k, v)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新积分规则: {data.get('gen_type', '')} - {data.get('model_config_id', '')}",
        "PUT",
        f"/admin/credit-ratios/{ratio_id}",
        detail=json.dumps(
            {
                "ratio_id": ratio_id,
                "gen_type": data.get("gen_type"),
                "model_config_id": data.get("model_config_id"),
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()
    return ratio


@router.delete("/credit-ratios/{ratio_id}")
async def delete_credit_ratio(
    ratio_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(CreditRatio).where(CreditRatio.id == ratio_id).limit(1)
    )
    ratio = result.scalar_one_or_none()
    if not ratio:
        raise HTTPException(status_code=404, detail="积分比例不存在")
    await db.delete(ratio)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"删除积分规则: {ratio.gen_type} - {ratio.model_config_id}",
        "DELETE",
        f"/admin/credit-ratios/{ratio_id}",
        detail=json.dumps(
            {
                "ratio_id": ratio_id,
                "gen_type": ratio.gen_type,
                "model_config_id": ratio.model_config_id,
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()
    return {"message": "ok"}


@router.get("/credit-ratios/grouped", response_model=dict)
async def list_credit_ratios_grouped(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(CreditRatio))
    ratios = result.scalars().all()

    grouped = {}
    for ratio in ratios:
        if ratio.gen_type not in grouped:
            grouped[ratio.gen_type] = []
        grouped[ratio.gen_type].append(CreditRatioOut.model_validate(ratio))

    return grouped


# ── Model Config ─────────────────────────────────────────

@router.get("/model-configs", response_model=list[ModelConfigOut])
async def list_model_configs(
    include_deleted: bool = Query(False),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    query = select(ModelConfig)
    if not include_deleted:
        query = query.where(ModelConfig.deleted_at.is_(None))
    result = await db.execute(query.order_by(ModelConfig.priority.desc(), ModelConfig.id.desc()))
    return result.scalars().all()


@router.post("/model-configs", response_model=ModelConfigOut)
async def create_model_config(
    req: ModelConfigCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    config = ModelConfig(id=generate_id(), **req.model_dump())
    db.add(config)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"创建模型配置: {req.name}",
        "POST",
        "/admin/model-configs",
        detail=json.dumps(
            {
                "config_id": config.id,
                "name": req.name,
            },
            ensure_ascii=False,
        ),
    )
    return config


@router.put("/model-configs/{config_id}", response_model=ModelConfigOut)
async def update_model_config(
    config_id: str,
    req: ModelConfigCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(ModelConfig).where(ModelConfig.id == config_id, ModelConfig.deleted_at.is_(None)).limit(1))
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(status_code=404, detail="配置不存在")
    for k, v in req.model_dump().items():
        setattr(config, k, v)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新模型配置: {config.name}",
        "PUT",
        f"/admin/model-configs/{config_id}",
        detail=json.dumps(
            {
                "config_id": config_id,
                "name": config.name,
            },
            ensure_ascii=False,
        ),
    )
    return config


@router.delete("/model-configs/{config_id}")
async def delete_model_config(
    config_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(ModelConfig).where(ModelConfig.id == config_id, ModelConfig.deleted_at.is_(None)).limit(1))
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(status_code=404, detail="配置不存在")
    config_name = config.name
    config.deleted_at = datetime.now(timezone.utc)
    config.is_active = False
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"软删除模型配置: {config_name}",
        "DELETE",
        f"/admin/model-configs/{config_id}",
        detail=json.dumps(
            {
                "config_id": config_id,
                "name": config_name,
            },
            ensure_ascii=False,
        ),
    )
    return {"message": "ok"}


# ── System Config ────────────────────────────────────────

@router.get("/system-configs", response_model=list[SystemConfigOut])
async def list_system_configs(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(SystemConfig))
    return result.scalars().all()


@router.post("/system-configs", response_model=SystemConfigOut)
async def create_system_config(
    req: SystemConfigCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    from app.utils.id_gen import generate_id
    config = SystemConfig(
        id=generate_id(),
        key=req.key,
        value=str(req.value),
        description=req.description or "",
    )
    db.add(config)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"创建系统配置: {config.key}",
        "POST",
        "/admin/system-configs",
        detail=json.dumps({"key": req.key, "value": req.value}, ensure_ascii=False),
    )
    await db.commit()
    return config


@router.put("/system-configs/{config_id}", response_model=SystemConfigOut)
async def update_system_config(
    config_id: str,
    req: SystemConfigUpdate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(SystemConfig).where(SystemConfig.id == config_id).limit(1))
    config = result.scalar_one_or_none()
    if not config:
        raise HTTPException(status_code=404, detail="配置不存在")
    config.value = str(req.value)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"更新系统配置: {config.key}",
        "PUT",
        f"/admin/system-configs/{config_id}",
        detail=json.dumps(
            {
                "config_id": config_id,
                "key": config.key,
                "value": req.value,
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()
    return config


# ── Operation Logs ──────────────────────────────────────

@router.get("/operation-logs")
async def list_operation_logs(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=500),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    query = select(OperationLog).order_by(OperationLog.created_at.desc())
    count_query = select(func.count(OperationLog.id))
    total = (await db.execute(count_query)).scalar() or 0
    result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
    items = result.scalars().all()
    return {
        "total": total,
        "items": [
            {
                "id": item.id,
                "user_id": item.user_id,
                "username": item.username,
                "action": item.action,
                "method": item.method,
                "path": item.path,
                "detail": item.detail,
                "ip": item.ip,
                "created_at": _iso(item.created_at),
            }
            for item in items
        ],
    }


# ── Stats ────────────────────────────────────────────────

@router.get("/stats", response_model=AdminStatsOut)
async def get_stats(
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
    start_date: str = Query(None),
    end_date: str = Query(None),
):
    today_start = datetime.now(CST).replace(hour=0, minute=0, second=0, microsecond=0)

    try:
        if start_date:
            date_start = datetime.strptime(start_date, "%Y-%m-%d").replace(tzinfo=CST)
        else:
            date_start = today_start
        if end_date:
            date_end = datetime.strptime(end_date, "%Y-%m-%d").replace(tzinfo=CST)
            date_end = date_end.replace(hour=23, minute=59, second=59, microsecond=999999)
        else:
            date_end = datetime.now(CST)
    except:
        date_start = today_start
        date_end = datetime.now(CST)

    total_users = (await db.execute(
        select(func.count(User.id)).where(
            User.user_type == "frontend",
            User.created_at >= date_start,
            User.created_at <= date_end,
        )
    )).scalar() or 0

    total_projects = (await db.execute(
        select(func.count(Project.id)).where(
            Project.deleted_at.is_(None),
            Project.created_at >= date_start,
            Project.created_at <= date_end,
        )
    )).scalar() or 0

    total_generations = (await db.execute(
        select(func.count(ChatGenerationTask.id)).where(
            ChatGenerationTask.created_at >= date_start,
            ChatGenerationTask.created_at <= date_end,
        )
    )).scalar() or 0

    total_records = (await db.execute(
        select(func.count(GenerationRecord.id)).where(
            GenerationRecord.deleted_at.is_(None),
            GenerationRecord.created_at >= date_start,
            GenerationRecord.created_at <= date_end,
        )
    )).scalar() or 0

    total_revenue = (await db.execute(
        select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
            PaymentOrder.status == "paid",
            PaymentOrder.created_at >= date_start,
            PaymentOrder.created_at <= date_end,
        )
    )).scalar() or 0

    credits_consumed = (await db.execute(
        select(func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0)).where(
            CreditRecord.type == "consume",
            CreditRecord.created_at >= date_start,
            CreditRecord.created_at <= date_end,
        )
    )).scalar() or 0

    alipay_revenue = (await db.execute(
        select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
            PaymentOrder.status == "paid",
            PaymentOrder.payment_method == "alipay",
            PaymentOrder.created_at >= date_start,
            PaymentOrder.created_at <= date_end,
        )
    )).scalar() or 0

    wechat_revenue = (await db.execute(
        select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
            PaymentOrder.status == "paid",
            PaymentOrder.payment_method == "wechat",
            PaymentOrder.created_at >= date_start,
            PaymentOrder.created_at <= date_end,
        )
    )).scalar() or 0

    period_duration = date_end - date_start

    last_period_start = date_start - period_duration
    last_period_end = date_start

    last_period_users = (await db.execute(
        select(func.count(User.id)).where(
            User.user_type == "frontend",
            User.created_at >= last_period_start,
            User.created_at <= last_period_end,
        )
    )).scalar() or 0

    last_period_projects = (await db.execute(
        select(func.count(Project.id)).where(
            Project.deleted_at.is_(None),
            Project.created_at >= last_period_start,
            Project.created_at <= last_period_end,
        )
    )).scalar() or 0

    last_period_generations = (await db.execute(
        select(func.count(ChatGenerationTask.id)).where(
            ChatGenerationTask.created_at >= last_period_start,
            ChatGenerationTask.created_at <= last_period_end,
        )
    )).scalar() or 0

    last_period_records = (await db.execute(
        select(func.count(GenerationRecord.id)).where(
            GenerationRecord.deleted_at.is_(None),
            GenerationRecord.created_at >= last_period_start,
            GenerationRecord.created_at <= last_period_end,
        )
    )).scalar() or 0

    last_period_revenue = (await db.execute(
        select(func.coalesce(func.sum(PaymentOrder.amount), 0)).where(
            PaymentOrder.status == "paid",
            PaymentOrder.created_at >= last_period_start,
            PaymentOrder.created_at <= last_period_end,
        )
    )).scalar() or 0

    last_period_credits_consumed = (await db.execute(
        select(func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0)).where(
            CreditRecord.type == "consume",
            CreditRecord.created_at >= last_period_start,
            CreditRecord.created_at <= last_period_end,
        )
    )).scalar() or 0

    # ── 每日各模块积分消耗（始终返回选中日期往前7天，便于图表展示）
    from sqlalchemy import Date, cast as sa_cast
    _day_expr = sa_cast(CreditRecord.created_at, Date)
    # 图表固定展示 [date_end - 6天, date_end] 共7天
    _chart_end_dt = date_end
    _chart_start_dt = _chart_end_dt - timedelta(days=6)
    _inner = (
        select(
            _day_expr.label('date'),
            CreditRecord.source_module.label('module'),
            func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0).label('credits'),
        )
        .where(
            CreditRecord.type == "consume",
            CreditRecord.created_at >= _chart_start_dt,
            CreditRecord.created_at <= _chart_end_dt,
        )
        .group_by(_day_expr, CreditRecord.source_module)
        .subquery()
    )
    daily_credits_rows = (await db.execute(
        select(
            _inner.c.date,
            func.coalesce(_inner.c.module, 'other').label('module'),
            _inner.c.credits,
        ).order_by(_inner.c.date)
    )).all()
    daily_credits_by_module = [
        DailyCreditOut(date=str(row.date), module=row.module, credits=float(row.credits or 0))
        for row in daily_credits_rows
    ]

    # ── 选中周期内各模块积分占比（按 source_module 分组，不拆日期）
    _period_inner = (
        select(
            CreditRecord.source_module.label('module'),
            func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0).label('credits'),
        )
        .where(
            CreditRecord.type == "consume",
            CreditRecord.created_at >= date_start,
            CreditRecord.created_at <= date_end,
        )
        .group_by(CreditRecord.source_module)
        .subquery()
    )
    period_credits_rows = (await db.execute(
        select(
            func.coalesce(_period_inner.c.module, 'other').label('module'),
            _period_inner.c.credits,
        ).order_by(_period_inner.c.credits.desc())
    )).all()
    period_credits_by_module = [
        DailyCreditOut(date='', module=row.module, credits=float(row.credits or 0))
        for row in period_credits_rows
    ]

    # ── 各团队积分消耗（有团队 vs 无团队，使用流水中的团队快照）
    team_credit_rows = (await db.execute(
        select(
            func.coalesce(CreditRecord.team_name_snapshot, '未分配团队').label('team_name'),
            CreditRecord.team_id_snapshot.label('team_id'),
            func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0).label('credits'),
        )
        .where(
            CreditRecord.type == "consume",
            CreditRecord.created_at >= date_start,
            CreditRecord.created_at <= date_end,
        )
        .group_by(CreditRecord.team_id_snapshot, CreditRecord.team_name_snapshot)
        .order_by(func.coalesce(func.sum(func.abs(CreditRecord.amount)), 0).desc())
    )).all()
    credits_by_team = [
        TeamCreditOut(team_name=row.team_name, team_id=row.team_id, credits=float(row.credits or 0))
        for row in team_credit_rows
    ]

    # ── 各模型使用次数（通过 engine 快照字段统计）
    model_usage_rows = (await db.execute(
        select(
            func.coalesce(CreditRecord.engine_name, '未知').label('model_name'),
            func.coalesce(CreditRecord.engine_provider, 'unknown').label('provider'),
            func.count(CreditRecord.id).label('count'),
        )
        .where(
            CreditRecord.type == "consume",
            CreditRecord.created_at >= date_start,
            CreditRecord.created_at <= date_end,
        )
        .group_by(CreditRecord.engine_name, CreditRecord.engine_provider)
        .order_by(func.count(CreditRecord.id).desc())
    )).all()
    model_usage = [
        ModelUsageOut(model_name=row.model_name, provider=row.provider, count=int(row.count or 0))
        for row in model_usage_rows
    ]

    return AdminStatsOut(
        total_users=total_users,
        total_projects=total_projects,
        total_generations=total_generations,
        total_records=total_records,
        total_revenue=float(total_revenue),
        credits_consumed_today=float(credits_consumed),
        today_alipay_revenue=float(alipay_revenue),
        today_wechat_revenue=float(wechat_revenue),
        last_period_users=last_period_users,
        last_period_projects=last_period_projects,
        last_period_generations=last_period_generations,
        last_period_records=last_period_records,
        last_period_revenue=float(last_period_revenue),
        last_period_credits_consumed=float(last_period_credits_consumed),
        daily_credits_by_module=daily_credits_by_module,
        period_credits_by_module=period_credits_by_module,
        credits_by_team=credits_by_team,
        model_usage=model_usage,
    )


# ── Token Usage ─────────────────────────────────────────

@router.get("/token-usage")
async def list_token_usage(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=500),
    user_id: str | None = Query(None),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """List token usage records with optional user filter."""
    query = select(TokenUsage).order_by(TokenUsage.created_at.desc())
    if user_id:
        query = query.where(TokenUsage.user_id == user_id)

    count_query = select(func.count(TokenUsage.id))
    if user_id:
        count_query = count_query.where(TokenUsage.user_id == user_id)

    total = (await db.execute(count_query)).scalar() or 0
    result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
    items = result.scalars().all()

    return {
        "total": total,
        "items": [
            {
                "id": item.id,
                "model_config_id": item.model_config_id,
                "user_id": item.user_id,
                "input_tokens": item.input_tokens,
                "output_tokens": item.output_tokens,
                "total_tokens": item.total_tokens,
                "created_at": _iso(item.created_at),
            }
            for item in items
        ],
    }


# ── Generation Records (Admin) ─────────────────────────────

@router.get("/generation-records")
async def admin_list_generation_records(
    user_id: str | None = Query(None),
    status: str | None = Query(None),
    engine_id: str | None = Query(None),
    include_media_references: bool | 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)
    if engine_id:
        query = query.where(GenerationRecord.engine_id == engine_id)
    if include_media_references is not None:
        query = query.where(GenerationRecord.include_media_references.is_(include_media_references))

    # Count total
    count_query = (
        select(func.count(GenerationRecord.id))
        .join(Project, GenerationRecord.project_id == Project.id)
        .where(GenerationRecord.deleted_at.is_(None), Project.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)
    if engine_id:
        count_query = count_query.where(GenerationRecord.engine_id == engine_id)
    if include_media_references is not None:
        count_query = count_query.where(GenerationRecord.include_media_references.is_(include_media_references))
    total_result = await db.execute(count_query)
    total = total_result.scalar() or 0

    # Paginate
    offset = (page - 1) * page_size
    query = query.offset(offset).limit(page_size)
    result = await db.execute(query)
    rows = result.all()
    refs_map = await batch_resolve_private_portrait_reference_display_urls(
        db,
        {record.id: json.loads(record.media_references) if record.media_references else None for record, _username, _project_name, _industry, _industry_label in rows},
        user_id=user_id,
    )

    items = []
    for record, username, project_name, industry, industry_label in rows:
        refs = refs_map.get(record.id)
        items.append({
            "id": record.id,
            "user_id": record.user_id,
            "username": username,
            "project_id": record.project_id,
            "project_name": project_name,
            "industry": industry_label or industry,
            "original_prompt": record.original_prompt,
            "optimized_prompt": record.optimized_prompt,
            "duration": record.duration,
            "aspect_ratio": record.aspect_ratio,
            "resolution": record.resolution,
            "status": record.status,
            "pipeline_stage": record.pipeline_stage,
            "video_upscale_enabled": bool(record.video_upscale_enabled_snapshot),
            "video_url": build_resource_signed_url(record.video_url) if record.video_url else '',
            "video_cover_url": build_resource_signed_url(record.video_cover_url) if record.video_cover_url else '',
            "references": refs,
            "engine_id": record.engine_id,
            "engine_name": (_safe_json_object(record.engine_snapshot_json) or {}).get("name"),
            "engine_snapshot": _safe_json_object(record.engine_snapshot_json),
            "include_media_references": bool(record.include_media_references),
            "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}


# ── 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

    if not file.filename:
        raise HTTPException(status_code=400, detail="请选择文件")

    if not file.filename.lower().endswith(".pdf"):
        raise HTTPException(status_code=400, detail="仅支持PDF格式")

    content = await file.read()
    if len(content) > 10 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="文件大小不能超过10MB")

    safe_name = f"pdf_{generate_id()}.pdf"
    file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, safe_name)
    with open(file_path, "wb") as f:
        f.write(content)

    url = f"/uploads/{safe_name}"

    if config_key:
        result = await db.execute(
            select(SystemConfig).where(SystemConfig.key == config_key).limit(1)
        )
        config = result.scalar_one_or_none()
        if config:
            config.value = url
        else:
            db.add(SystemConfig(
                id=generate_id(),
                key=config_key,
                value=url,
            ))
        await db.flush()
        await log_operation(
            db,
            admin.id,
            admin.username,
            f"上传PDF文件: {file.filename}",
            "POST",
            "/admin/upload-pdf",
            detail=json.dumps(
                {
                    "filename": file.filename,
                    "config_key": config_key,
                    "url": url,
                },
                ensure_ascii=False,
            ),
        )
        await db.commit()

    return {"url": url}


@router.post("/upload-logo")
async def upload_logo(
    file: UploadFile = File(...),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """Upload a Logo image file and save URL to system config."""
    from app.config import settings

    if not file.filename:
        raise HTTPException(status_code=400, detail="请选择文件")

    allowed_extensions = ('.png', '.jpg', '.jpeg', '.gif', '.webp')
    if not file.filename.lower().endswith(allowed_extensions):
        raise HTTPException(status_code=400, detail="仅支持 PNG、JPG、GIF、WebP 格式图片")

    content = await file.read()
    if len(content) > 2 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="文件大小不能超过2MB")

    safe_name = "site_logo.png"
    file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, safe_name)
    with open(file_path, "wb") as f:
        f.write(content)

    url = f"/uploads/{safe_name}"

    result = await db.execute(
        select(SystemConfig).where(SystemConfig.key == "site_logo").limit(1)
    )
    config = result.scalar_one_or_none()
    if config:
        config.value = url
    else:
        db.add(SystemConfig(
            id="cfg_site_logo",
            key="site_logo",
            value=url,
            description="网站Logo图片",
        ))
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"上传Logo图片: {file.filename}",
        "POST",
        "/admin/upload-logo",
        detail=json.dumps(
            {
                "filename": file.filename,
                "url": url,
            },
            ensure_ascii=False,
        ),
    )
    await db.commit()

    return {"url": url}


@router.post("/upload-login-video")
async def upload_login_video(
    file: UploadFile = File(...),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    """上传登录页背景视频/动图，保存 URL 到 system config login_bg_video。"""
    from app.config import settings

    if not file.filename:
        raise HTTPException(status_code=400, detail="请选择文件")

    content = await file.read()
    if len(content) > 50 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="文件大小不能超过50MB")

    ext = os.path.splitext(file.filename)[1].lower()
    safe_name = f"login_bg_{generate_id()}{ext}"
    file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, safe_name)
    with open(file_path, "wb") as f:
        f.write(content)

    url = f"/uploads/{safe_name}"

    result = await db.execute(
        select(SystemConfig).where(SystemConfig.key == "login_bg_video").limit(1)
    )
    config = result.scalar_one_or_none()
    if config:
        config.value = url
    else:
        db.add(SystemConfig(
            id=generate_id(),
            key="login_bg_video",
            value=url,
            description="登录页背景视频",
        ))
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"上传登录背景视频: {file.filename}",
        "POST",
        "/admin/upload-login-video",
        detail=json.dumps({"filename": file.filename, "url": url}, ensure_ascii=False),
    )
    await db.commit()

    return {"url": url}


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