webapp/app/api/router.py

import time
import uuid
import json
import requests
import stripe
from typing import List, Dict, Any, Optional
from app.core.logger import logger
from app.core.config import settings
from fastapi import APIRouter, Request, Response, Query, Depends, Body, UploadFile, File, HTTPException
from fastapi.responses import RedirectResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.utils.redis_utils import redis_qpush
from app.utils.validation_utils import validate_user_inputs
from app.core.redis import get_redis_client
from app.core.database import get_db
from app.core.auth import get_current_user
from redis.asyncio import Redis

from app.utils.response import success, fail
from app.models.user import VasUser
from app.models.order import VasOrder
from app.models.schema import VasSchema
from app.models.product import VasProduct
from app.models.payment import VasPayment
from app.schemas.common import ApiResponse, PageResponse
from app.schemas.troov import TroovRate, TroovCheckForbiddenInput, TroovProb
from app.schemas.sms import ShortMessageDetail
from app.schemas.configuration import ConfigurationCreate, ConfigurationUpdate, ConfigurationOut
from app.schemas.email_authorizations import EmailContent, EmailAuthorizationCreate, EmailAuthorizationUpdate, EmailAuthorizationOut
from app.schemas.card import CardCreate, CardOut
from app.schemas.task import TaskCreate, TaskOut, TaskUpdate
from app.schemas.short_url import ShortUrlCreate, ShortUrlOut
from app.schemas.http_session import HttpSessionCreate, HttpSessionUpdate,HttpSessionOut
from app.schemas.fake import FakeUser
from app.schemas.auth import SendBindCodeRequest, SendResetCodeRequest, BindEmailRequest, ResetPasswordRequest, LoginRequest, LoginData, AutoRegisterRequest, AutoRegisterData
from app.schemas.user import VasUserCreate, VasUserUpdate, VasUserSetProfiles, VasUserOut
from app.schemas.product import VasProductCreate, VasProductUpdate, VasProductOut
from app.schemas.product_routing import VasProductRoutingCreate, VasProductRoutingOut
from app.schemas.schema import VasSchemaCreate, VasSchemaUpdate, VasSchemaOut
from app.schemas.order import VasOrderCreate, VasOrderAdjustPrice, VasOrderPatchUserInputs, VasOrderOut
from app.schemas.payment import VasPaymentCreate, AdminUpdateStatusPayload, VasPaymentOut
from app.schemas.payment_confirmation import VasPaymentConfirmationCreate, VasPaymentConfirmationUpdate, VasPaymentConfirmationOut
from app.schemas.payment_qr import VasPaymentQrCreate, VasPaymentQrSetEnableIn, VasPaymentQrOut
from app.schemas.payment_provider import VasPaymentProviderCreate, VasPaymentProviderUpdate, VasPaymentProviderOut
from app.schemas.webhook import SMSHelperWebhookPayload
from app.schemas.vas_task import VasTaskCreate, VasTaskUpdate, VasTaskOut
from app.schemas.ticket import VasTicketCreate, VasTicketOut, VasTicketStatusUpdate, VasTicketMessageCreate, VasTicketMessageOut
from app.schemas.slot_snapshot import SlotSnapshotCreate, SlotSnapshotOut, SlotOverviewOut
from app.schemas.slot_refresh_status import RefreshBase, RefreshFail, RefreshStatusOut
from app.schemas.telegram import TelegramIn
from app.schemas.wechat import WechatIn
from app.schemas.resource import FileUploadOut
from app.schemas.statistics import VasStatisticsOverviewOut
from app.schemas.llm import ParseUserInputsPayload, ParseUserInputsOut
from app.schemas.docker_remote import RemoteServerConfig, DockerStatusOut, DockerLogsRequest, DockerLogsOut, ConfigReadOut, ConfigReadRequest, ConfigUpdateRequest, LogReadRequest, LogReadOut, LogListOut, DockerContainerStatus, DockerActionRequest, ServerConfigItem, ServerListOut, RemoteActionRequest
from app.services.docker_remote_service import DockerRemoteService
from app.services.configuration_service import ConfigurationService
from app.services.troov_service import TroovService
from app.services.sms_service import save_short_message, query_short_message
from app.services.email_authorizations_service import EmailAuthorizationService
from app.services.short_url_service import ShortUrlService
from app.services.task_service import TaskService
from app.services.card_service import CardService
from app.services.seaweedfs_service import SeaweedFSService
from app.services.http_session_service import HttpSessionService
from app.services.fake_service import generate_fake_users
from app.services.auth_service import AuthService
from app.services.user_service import UserService
from app.services.product_service import ProductService
from app.services.product_routing_service import ProductRoutingService
from app.services.order_service import OrderService
from app.services.schema_service import SchemaService
from app.services.payment_service import PaymentService
from app.services.payment_provider_service import PaymentProviderService
from app.services.payment_qr_service import PaymentQrService
from app.services.vas_task_service import VasTaskService
from app.services.webhook_service import WebhookService
from app.services.notification_service import NotificationService
from app.services.ticket_service import TicketService
from app.services.telegram_service import TelegramService
from app.services.wechat_service import WechatService
from app.services.slot_snapshot_service import SlotSnapshotService
from app.services.statistics_service import StatisticsService
from app.services.llm_service import LlmService
from app.services.slot_refresh_status_service import SlotRefreshStatusService

# 公共路由
public_router = APIRouter()
# 受保护路由
protected_router = APIRouter()
# 管理员级别路由
admin_required_router = APIRouter()

@admin_required_router.get("/ping", summary="心跳检测", tags=["测试接口"])
async def ping():
    return {"message": "pong"}

# -----------------------
# Docker 远程控制 (预配置服务器)
# -----------------------

@admin_required_router.get("/remote/servers", summary="获取所有预配置服务器", tags=["Docker远程控制"], response_model=ApiResponse[ServerListOut])
async def list_remote_servers(db: AsyncSession = Depends(get_db)):
    from app.services.remote_server_service import RemoteServerService
    servers_db = await RemoteServerService.get_all(db)
    servers = [
        ServerConfigItem(id=s.server_id, name=s.name, host=s.host)
        for s in servers_db
    ]
    return success(data=ServerListOut(servers=servers))

@admin_required_router.post("/remote/server/docker/status", summary="获取预配置服务器容器状态", tags=["Docker远程控制"], response_model=ApiResponse[DockerStatusOut])
async def server_docker_status(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.get_container_status(config)
    return success(data=DockerStatusOut(containers=res))

@admin_required_router.post("/remote/server/docker/up", summary="启动预配置服务器Docker Compose", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def server_docker_up(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.docker_compose_up(config, request.services)
    return success(data=res)

@admin_required_router.post("/remote/server/docker/down", summary="停止预配置服务器Docker Compose", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def server_docker_down(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.docker_compose_down(config, request.services)
    return success(data=res)

@admin_required_router.post("/remote/server/docker/restart", summary="重启预配置服务器容器", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def server_docker_restart(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.docker_restart(config, request.container_name)
    return success(data=res)

@admin_required_router.post("/remote/server/docker/start", summary="启动预配置服务器容器", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def server_docker_start(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.docker_start(config, request.container_name)
    return success(data=res)

@admin_required_router.post("/remote/server/docker/stop", summary="停止预配置服务器容器", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def server_docker_stop(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.docker_stop(config, request.container_name)
    return success(data=res)

@admin_required_router.post("/remote/server/docker/logs", summary="查看预配置服务器容器日志", tags=["Docker远程控制"], response_model=ApiResponse[DockerLogsOut])
async def server_docker_logs(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    # 构造 DockerLogsRequest
    log_req = DockerLogsRequest(
        **config.model_dump(),
        container_name=request.container_name,
        lines=request.lines,
        follow=request.follow
    )
    res = await DockerRemoteService.docker_logs(config, log_req)
    return success(data=DockerLogsOut(logs=res))

@admin_required_router.post("/remote/server/config/read", summary="读取预配置服务器配置文件", tags=["Docker远程控制"], response_model=ApiResponse[Dict[str, Any]])
async def server_docker_config_read(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.read_config(config, request.config_file)
    return success(data={"config": res})

@admin_required_router.post("/remote/server/config/update", summary="更新预配置服务器配置文件", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def server_docker_config_update(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    # 构造 ConfigUpdateRequest
    update_req = ConfigUpdateRequest(
        **config.model_dump(),
        config_file=request.config_file,
        key_path=request.key_path,
        value=request.value
    )
    res = await DockerRemoteService.update_config(config, update_req)
    return success(data=res)

@admin_required_router.post("/remote/server/log/list", summary="列出预配置服务器日志文件", tags=["Docker远程控制"], response_model=ApiResponse[LogListOut])
async def server_docker_logs_list(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    res = await DockerRemoteService.list_logs(config)
    return success(data=LogListOut(log_files=res))

@admin_required_router.post("/remote/server/log/read", summary="读取预配置服务器日志文件内容", tags=["Docker远程控制"], response_model=ApiResponse[LogReadOut])
async def server_docker_logs_read(request: RemoteActionRequest, db: AsyncSession = Depends(get_db)):
    config = await DockerRemoteService.get_server_config(db, request.server_id, request.project_path)
    # 构造 LogReadRequest
    read_req = LogReadRequest(
        **config.model_dump(),
        log_file=request.log_file,
        lines=request.lines,
        from_head=request.from_head,
        full=request.full
    )
    res = await DockerRemoteService.read_log(config, read_req)
    return success(data=LogReadOut(content=res))

# -----------------------
# Docker 远程控制 (直连模式 - 仅供调试)
# -----------------------
@admin_required_router.post("/remote/docker/status", summary="获取容器状态", tags=["Docker远程控制"], response_model=ApiResponse[DockerStatusOut])
async def docker_status(config: RemoteServerConfig):
    res = await DockerRemoteService.get_container_status(config)
    return success(data=DockerStatusOut(containers=res))

@admin_required_router.post("/remote/docker/up", summary="启动Docker Compose服务", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def docker_up(config: RemoteServerConfig, services: Optional[List[str]] = None):
    res = await DockerRemoteService.docker_compose_up(config, services)
    return success(data=res)

@admin_required_router.post("/remote/docker/down", summary="停止Docker Compose服务", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def docker_down(config: RemoteServerConfig, services: Optional[List[str]] = None):
    res = await DockerRemoteService.docker_compose_down(config, services)
    return success(data=res)

@admin_required_router.post("/remote/docker/restart", summary="重启容器", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def docker_restart(request: DockerActionRequest):
    res = await DockerRemoteService.docker_restart(request, request.container_name)
    return success(data=res)

@admin_required_router.post("/remote/docker/start", summary="启动容器", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def docker_start(request: DockerActionRequest):
    res = await DockerRemoteService.docker_start(request, request.container_name)
    return success(data=res)

@admin_required_router.post("/remote/docker/stop", summary="停止容器", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def docker_stop(request: DockerActionRequest):
    res = await DockerRemoteService.docker_stop(request, request.container_name)
    return success(data=res)

@admin_required_router.post("/remote/docker/logs", summary="查看容器日志", tags=["Docker远程控制"], response_model=ApiResponse[DockerLogsOut])
async def docker_logs(request: DockerLogsRequest):
    res = await DockerRemoteService.docker_logs(request, request)
    return success(data=DockerLogsOut(logs=res))

@admin_required_router.post("/remote/config/read", summary="读取配置文件", tags=["Docker远程控制"], response_model=ApiResponse[Dict[str, Any]])
async def docker_config_read(request: ConfigReadRequest):
    res = await DockerRemoteService.read_config(request, request.config_file)
    return success(data={"config": res})

@admin_required_router.post("/remote/config/update", summary="更新配置文件", tags=["Docker远程控制"], response_model=ApiResponse[bool])
async def docker_config_update(request: ConfigUpdateRequest):
    res = await DockerRemoteService.update_config(request, request)
    return success(data=res)

@admin_required_router.post("/remote/log/list", summary="列出日志文件", tags=["Docker远程控制"], response_model=ApiResponse[LogListOut])
async def docker_logs_list(config: RemoteServerConfig):
    res = await DockerRemoteService.list_logs(config)
    return success(data=LogListOut(log_files=res))

@admin_required_router.post("/remote/log/read", summary="读取日志文件内容", tags=["Docker远程控制"], response_model=ApiResponse[LogReadOut])
async def docker_logs_read(request: LogReadRequest):
    res = await DockerRemoteService.read_log(request, request)
    return success(data=LogReadOut(content=res))

@admin_required_router.get("/sms/upload", summary="上报短信", tags=["短信接口"], response_model=ApiResponse[ShortMessageDetail])
async def sms_upload(
    phone: str = Query(..., description="手机号"),
    message: str = Query(..., description="短信内容"),
    max_ttl: int = Query(300, description="短信保存时间（秒）"),
    redis_client: Redis = Depends(get_redis_client)
):
    """
    保存短信到 Redis
    """
    received_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    msg = await save_short_message(redis_client, phone, message, received_at, max_ttl)
    return success(data=msg)

@admin_required_router.get("/sms/download", summary="读取短信", tags=["短信接口"], response_model=ApiResponse[List[ShortMessageDetail]])
async def sms_download(
    phone: str = Query(..., description="手机号"),
    keyword: str = Query('', description="短信内容关键字"),
    sent_at: str = Query('', description="筛选时间（可选）"),
    redis_client: Redis = Depends(get_redis_client)
):
    """
    查询短信（支持关键字和时间过滤）
    """
    obj = await query_short_message(redis_client, phone, keyword or None, sent_at or None)
    return success(data=obj)

@admin_required_router.get("/troov/rate", summary="TROOV 查询rate", tags=["通用接口"], response_model=ApiResponse[List[TroovRate]])
async def troov_rate(date: str = Query(..., description="查询的日期, 格式: YYYY-MM-DD"),
               redis_client: Redis = Depends(get_redis_client)):
    # 调用 service 层获取数据
    obj = await TroovService.get_rate_by_date(redis_client, date)
    return success(data=obj)

@admin_required_router.post("/troov/book", summary="TROOV 查询ForbiddenUsers", tags=["通用接口"], response_model=ApiResponse)
async def troov_check_forbiddenusers(
    payload: TroovCheckForbiddenInput,
    redis_client: Redis = Depends(get_redis_client)
):
    # 调用 service 层获取数据
    obj = await TroovService.check_for_forbiddenusers(redis_client, payload)
    return success(data=obj)

@admin_required_router.get("/troov/list-probs", summary="TROOV 查询所有概率", tags=["通用接口"], response_model=ApiResponse[List[TroovProb]])
async def troov_get_all_probs(
    redis_client: Redis = Depends(get_redis_client)
):
    obj = await TroovService.get_all_probs(redis_client)
    return success(data=obj)

@admin_required_router.post("/troov/set-prob", summary="TROOV 修改概率", tags=["通用接口"], response_model=ApiResponse[List[TroovProb]])
async def troov_add_prob(
    payload: TroovProb,
    redis_client: Redis = Depends(get_redis_client)
):
    obj = await TroovService.set_prob(redis_client, payload)
    return success(data=obj)

@admin_required_router.delete("/troov/del-prob", summary="TROOV 删除概率", tags=["通用接口"], response_model=ApiResponse[List[TroovProb]])
async def troov_del_prob(
    payload: TroovProb,
    redis_client: Redis = Depends(get_redis_client)
):
    obj = await TroovService.del_prob(redis_client, payload)
    return success(data=obj)

@admin_required_router.post("/troov/reset-probs", summary="TROOV 重置概率", tags=["通用接口"], response_model=ApiResponse[List[TroovProb]])
async def troov_del_prob(
    date: str,
    redis_client: Redis = Depends(get_redis_client)
):
    obj = await TroovService.reset_probs(redis_client, date)
    return success(data=obj)

@admin_required_router.post("/dynamic-configurations", summary="创建动态配置", tags=["动态配置"], response_model=ApiResponse[ConfigurationOut])
async def dynamic_config_create(config_in: ConfigurationCreate, db: AsyncSession = Depends(get_db)):
    obj = await ConfigurationService.create(db, config_in)
    return success(data=obj)

@admin_required_router.get("/dynamic-configurations/all", summary="读取所有动态配置", tags=["动态配置"], response_model=ApiResponse[List[ConfigurationOut]])
async def dynamic_config_get_all(db: AsyncSession = Depends(get_db)):
    obj = await ConfigurationService.get_all(db)
    return success(data=obj)
    
@admin_required_router.get("/dynamic-configurations/key/{config_key}", summary="根据Key读取动态配置", tags=["动态配置"], response_model=ApiResponse[ConfigurationOut])
async def dynamic_config_get_by_key(config_key: str, db: AsyncSession = Depends(get_db)):
    config = await ConfigurationService.get_by_key(db, config_key)
    return success(data=config)


@admin_required_router.put("/dynamic-configurations/key/{config_key}", summary="根据Key更新动态配置", tags=["动态配置"], response_model=ApiResponse[ConfigurationOut])
async def dynamic_config_update_by_key(config_key: str, config_in: ConfigurationUpdate, db: AsyncSession = Depends(get_db)):
    config = await ConfigurationService.update_by_key(db, config_key, config_in)
    return success(data=config)


@admin_required_router.delete("/dynamic-configurations/key/{config_key}", summary="根据Key删除动态配置", tags=["动态配置"], response_model=ApiResponse[ConfigurationOut])
async def dynamic_config_delete_by_key(config_key: str, db: AsyncSession = Depends(get_db)):
    config = await ConfigurationService.delete_by_key(db, config_key)
    return success(data=config)

@admin_required_router.post("/http-session", summary="创建http session", tags=["会话管理"],response_model=ApiResponse[HttpSessionOut])
async def http_session_create(
    data: HttpSessionCreate,
    db: AsyncSession = Depends(get_db)
):
    logger.info(f"[Create HttpSession] sid={data.session_id}")
    obj = await HttpSessionService.create(db, data)
    return success(data=obj)


@admin_required_router.delete("/http-session", summary="删除http session", tags=["会话管理"], response_model=ApiResponse)
async def http_session_delete_by_sid(
    session_id: str = Query(...),
    db: AsyncSession = Depends(get_db)
):
    logger.info(f"[Delete HttpSession] sid={session_id}")
    await HttpSessionService.delete_by_sid(db, session_id)
    return success()


@admin_required_router.put("/http-session", summary="更新http session", tags=["会话管理"], response_model=ApiResponse[HttpSessionOut])
async def http_session_update_by_sid(
    session_id: str = Query(...),
    data: HttpSessionUpdate = Body(...),
    db: AsyncSession = Depends(get_db)
):
    logger.info(f"[Update HttpSession] sid={session_id}")
    obj = await HttpSessionService.update_by_sid(db, session_id, data)
    return success(data=obj)

@admin_required_router.get("/http-session", summary="读取http session", tags=["会话管理"],response_model=ApiResponse[HttpSessionOut])
async def http_session_get_by_sid(
    session_id: str = Query(...),
    db: AsyncSession = Depends(get_db)
):
    logger.info(f"[Get HttpSession] sid={session_id}")
    obj = await HttpSessionService.get_by_sid(db, session_id)
    return success(data=obj)


@admin_required_router.get("/email-authorizations", summary="查询所有内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[List[EmailAuthorizationOut]])
async def email_authorizations_get(db: AsyncSession = Depends(get_db)):
    obj = await EmailAuthorizationService.get_all(db)
    return success(data=obj)


@admin_required_router.post("/email-authorizations", summary="创建内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut])
async def email_authorizations_create(data: EmailAuthorizationCreate, db: AsyncSession = Depends(get_db)):
    obj = await EmailAuthorizationService.create(db, data)
    return success(data=obj)

@admin_required_router.get("/email-authorizations/{id}", summary="通过id查询内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut])
async def email_authorizations_get_by_id(id: int, db: AsyncSession = Depends(get_db)):
    email_auth = await EmailAuthorizationService.get_by_id(db, id)
    return success(data=email_auth)


@admin_required_router.put("/email-authorizations/{id}", summary="通过id更新内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut])
async def email_authorizations_update_by_id(id: int, data: EmailAuthorizationUpdate, db: AsyncSession = Depends(get_db)):
    updated = await EmailAuthorizationService.update(db, id, data)
    return success(data=updated)


@admin_required_router.delete("/email-authorizations/{id}", summary="通过id删除内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut])
async def email_authorizations_delete_by_id(id: int, db: AsyncSession = Depends(get_db)):
    deleted = await EmailAuthorizationService.delete(db, id)
    return success(data=deleted)


@admin_required_router.get("/email-authorizations/email/{email}", summary="通过邮箱地址查询内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut])
async def email_authorizations_get_by_email(email: str, db: AsyncSession = Depends(get_db)):
    email_auth = await EmailAuthorizationService.get_by_email(db, email)
    return success(data=email_auth)

@admin_required_router.post("/email-authorizations/fetch", summary="读取邮件, 仅限文本内容", tags=["邮箱接口"], response_model=ApiResponse[EmailContent])
async def email_authorizations_fetch_email(
    email: str = Query(..., description="收件邮箱账号, 格式: xxx@xxx.xxx"),
    sender: str = Query(..., description="发件人邮箱账号或者名字"),
    recipient: str = Query(..., description="收件人账号或者名字"),
    subjectKeywords: str = Query("", description="邮件主题关键词, 支持多个关键词, 用逗号隔开"),
    bodyKeywords: str = Query("", description="邮件内容关键词, 支持多个关键词, 用逗号隔开"),
    sentDate: str = Query(..., description="发件日期, UTC时间, 格式: yyyy-mm-dd hh:mm:ss"),
    expiry: int = Query(300, description="邮件有效期, 单位秒, 从sentDate 开始算起"),
    db: AsyncSession = Depends(get_db)
):
    auth = await EmailAuthorizationService.get_by_email(db, email)
    print(auth)
    result = await EmailAuthorizationService.fetch_email_authorizations(
        auth,
        sender=sender,
        recipient=recipient,
        subject_keywords=subjectKeywords,
        body_keywords=bodyKeywords,
        sent_date=sentDate,
        expiry=expiry,
        only_text=True
    )
    return success(data={"body": result})

@admin_required_router.post("/email-authorizations/fetch-top", summary="从最近的几封邮件读取目标邮件,仅限文本内容, 性能会更好, 邮件多时有可能漏读", tags=["邮箱接口"], response_model=ApiResponse[EmailContent])
async def email_authorizations_fetch_email_from_topn(
    email: str = Query(..., description="收件邮箱账号, 格式: xxx@xxx.xxx"),
    sender: str = Query(..., description="发件人邮箱账号或者名字"),
    recipient: str = Query(..., description="收件人账号或者名字"),
    subjectKeywords: str = Query("", description="邮件主题关键词, 支持多个关键词, 用逗号隔开"),
    bodyKeywords: str = Query("", description="邮件内容关键词, 支持多个关键词, 用逗号隔开"),
    top: int = Query(10, description="指定从最近几封邮件读取"),
    db: AsyncSession = Depends(get_db)
):
    auth = await EmailAuthorizationService.get_by_email(db, email)
    result = await EmailAuthorizationService.fetch_email_authorizations_from_top_n(
        auth,
        sender=sender,
        recipient=recipient,
        subject_keywords=subjectKeywords,
        body_keywords=bodyKeywords,
        top=top,
        only_text=True
    )
    return success(data={"body": result})

@admin_required_router.post("/email-authorizations/forward", summary="转发邮件", tags=["邮箱接口"], response_model=ApiResponse[EmailContent])
async def email_authorizations_forward_email(
    emailAccount: str = Query(..., description="收件邮箱账号, 格式: xxx@xxx.xxx"),
    forwardTo: str = Query(..., description="转发到哪个邮箱地址, 格式: xxx@xxx.xxx"),
    sender: str = Query(..., description="发件人邮箱账号或者名字"),
    recipient: str = Query(..., description="收件人账号或者名字"),
    subjectKeywords: str = Query("", description="邮件主题关键词, 支持多个关键词, 用逗号隔开"),
    bodyKeywords: str = Query("", description="邮件内容关键词, 支持多个关键词, 用逗号隔开"),
    db: AsyncSession = Depends(get_db)
):
    auth = await EmailAuthorizationService.get_by_email(db, emailAccount)
    result = await EmailAuthorizationService.forward_first_matching_email(
        auth,
        forward_to = forwardTo,
        sender = sender,
        recipient = recipient,
        subject_keywords = subjectKeywords,
        body_keywords = bodyKeywords
    )
    return success(data={"body": result})

@admin_required_router.post("/email-authorizations/forward2", summary="转发邮件(新)", tags=["邮箱接口"], response_model=ApiResponse[EmailContent])
async def email_authorizations_forward_email2(
    emailAccount: str = Query(..., description="收件邮箱账号, 格式: xxx@xxx.xxx"),
    forwardTo: str = Query(..., description="转发到哪个邮箱地址, 格式: xxx@xxx.xxx"),
    sender: str = Query(..., description="发件人邮箱账号或者名字"),
    recipient: str = Query(..., description="收件人账号或者名字"),
    subjectKeywords: str = Query("", description="邮件主题关键词, 支持多个关键词, 用逗号隔开"),
    bodyKeywords: str = Query("", description="邮件内容关键词, 支持多个关键词, 用逗号隔开"),
    db: AsyncSession = Depends(get_db)
):
    auth = await EmailAuthorizationService.get_by_email(db, emailAccount)
    result = await EmailAuthorizationService.forward_first_matching_email2(
        db,
        auth,
        forward_to = forwardTo,
        sender = sender,
        recipient = recipient,
        subject_keywords = subjectKeywords,
        body_keywords = bodyKeywords
    )
    return success(data={"body": result})

@admin_required_router.post("/email-authorizations/sendmail", summary="发送邮件", tags=["邮箱接口"], response_model=ApiResponse[EmailContent])
async def email_authorizations_send_email(
    emailAccount: str = Query(..., description="收件邮箱账号, 格式: xxx@xxx.xxx"),
    sendTo: str = Query(..., description="收件人邮箱账号"),
    subject: str = Query(..., description="邮件主题"),
    contentType: str = Query("text", description="内容格式,支持 text 和 html"),
    content: EmailContent = Body("", description="邮件内容"),
    db: AsyncSession = Depends(get_db)
):
    auth = await EmailAuthorizationService.get_by_email(db, emailAccount)
    result = await EmailAuthorizationService.send_email(
        auth,
        send_to = sendTo,
        subject = subject,
        content_type = contentType,
        content = content.body
    )
    return success(data={"body": result})

@admin_required_router.post("/email-authorizations/sendmail-bulk", summary="群发送邮件", tags=["邮箱接口"], response_model=ApiResponse[EmailContent])
async def email_authorizations_send_email_bulk(
    emailAccount: str = Query(..., description="收件邮箱账号, 格式: xxx@xxx.xxx"),
    sendTo: str = Query(..., description="收件人邮箱账号,多个用逗号隔开"),
    subject: str = Query(..., description="邮件主题"),
    contentType: str = Query("text", description="内容格式,支持 text 和 html"),
    content: EmailContent = Body(..., description="邮件内容"),
    db: AsyncSession = Depends(get_db)
):
    auth = await EmailAuthorizationService.get_by_email(db, emailAccount)
    result = await EmailAuthorizationService.send_email_bulk(
        auth,
        send_to = sendTo,
        subject = subject,
        content_type = contentType,
        content = content.body
    )
    return success(data={"body": result})

@public_router.post("/resource/upload_file", summary="上传文件", tags=["文件管理"], response_model=ApiResponse[FileUploadOut])
async def resource_upload_file(file: UploadFile = File(...)):
    result = await SeaweedFSService.upload(file)
    return success(data=result)

@public_router.get("/resource/download_file", summary="下载文件", tags=["文件管理"])
async def resource_get_file(fid: str):
    data = await SeaweedFSService.get(fid)
    content, mime = data
    return Response(content=content, media_type=mime)

@admin_required_router.post("/s/generate", summary="生成短链接地址<压缩地址长度>", tags=["Short URL"], response_model=ApiResponse[ShortUrlOut])
async def short_url_generate(
    data: ShortUrlCreate,
    db: AsyncSession = Depends(get_db),
):
    """生成短链接"""
    record = await ShortUrlService.create_short_url(db, data.long_url)
    return success(data=record)

@public_router.get("/s/{short_key}", summary="访问短链接地址<自动重定向到真实链接地址>", tags=["Short URL"])
async def short_url_request(short_key: str, db: AsyncSession = Depends(get_db)):
    """访问短链接自动重定向"""
    long_url = await ShortUrlService.get_long_url(db, short_key)
    return RedirectResponse(url=long_url, status_code=302)

@admin_required_router.post("/tasks", summary="创建任务", tags=["任务管理接口"], response_model=ApiResponse[TaskOut])
async def task_create(data: TaskCreate, db: AsyncSession = Depends(get_db)):
    """创建任务"""
    task = await TaskService.create(db, data)
    return success(data=task)

@admin_required_router.get("/tasks/{task_id:int}", summary="根据taskId读取任务状态", tags=["任务管理接口"], response_model=ApiResponse[TaskOut])
async def task_get_by_id(task_id: int, db: AsyncSession = Depends(get_db)):
    """获取任务"""
    task = await TaskService.get_by_id(db, task_id)
    return success(data=task)

@admin_required_router.get("/tasks/pending", summary="获取等待执行的任务", tags=["任务管理接口"], response_model=ApiResponse[List[TaskOut]])
async def task_get_pending(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    command: str = Query(..., description="任务类型"),
    db: AsyncSession = Depends(get_db),
):
    """分页获取等待执行的任务"""
    obj = await TaskService.get_pending(db, command, page, size)
    return success(data=obj)

@admin_required_router.put("/tasks/{task_id}", summary="根据taskId更新任务状态", tags=["任务管理接口"], response_model=ApiResponse[TaskOut])
async def task_update_by_id(task_id: int, data: TaskUpdate, db: AsyncSession = Depends(get_db)):
    """更新任务状态或结果"""
    updated = await TaskService.update(db, task_id, data)
    return success(data=updated)

@admin_required_router.get("/task/pop", summary="任务出队（pop）", tags=["任务管理接口"], response_model=ApiResponse[TaskOut])
async def task_pop_task(
    queue_name: str,
    db: AsyncSession = Depends(get_db),
):
    task = await TaskService.pop_task(db, queue_name)
    return success(data=task)

@admin_required_router.post("/tg/send_message", summary="推送电报消息", tags=["消息推送接口"], response_model=ApiResponse)
async def tg_send_message(
    payload: TelegramIn
):
    await TelegramService.push_to_telegram(payload)
    return success()

@admin_required_router.post("/wechat/send", summary="推送微信消息", tags=["消息推送接口"], response_model=ApiResponse)
async def wechat_send(
    payload: WechatIn
):
    await WechatService.push_to_wechat(payload)
    return success()

@admin_required_router.post("/cards/publish", summary="创建新的消息卡片", tags=["信息卡片接口"], response_model=ApiResponse[CardOut])
async def cards_publish(
    data: CardCreate = Body(...),
    db: AsyncSession = Depends(get_db)
):
    obj = await CardService.create(db, data)
    return success(data=obj)

@public_router.get("/cards/view2", summary="根据关键词分页查询卡片, 可选择语言", tags=["信息卡片接口"], response_model=ApiResponse[PageResponse[CardOut]])
async def cards_view_paginated2(
    keyword: str = Query("", description="查询的关键词"),
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    culture: str = Query("english", description="语言, 可设置 chinese, english"),
    db: AsyncSession = Depends(get_db)
):
    obj = await CardService.list_by_keyword(db, keyword, page, size, culture)
    return success(data=obj)

@admin_required_router.get("/fake/users", summary="生成虚假的预约人信息", tags=["数据生成"], response_model=ApiResponse[List[FakeUser]])
async def fake_generate_fake_users(
    num: int = Query(1, description="生成几个数据"),
    living_country = Query("Ireland", description="居住在哪个国家, China, India, United Kingdom, Ireland"),
):
    obj = generate_fake_users(num, living_country=living_country)
    return success(data=obj)

@public_router.get("/slots/latest", summary="查询最近的slot", tags=["Slot数据"], response_model=ApiResponse[SlotSnapshotOut])
async def slots_latest_get(
    country: str = Query("", description="目的国家"),
    city: str = Query("", description="递交城市"),
    visa_type: str = Query("", description="签证类型"),
    db: AsyncSession = Depends(get_db)
):
    res = await SlotSnapshotService.latest_for(db, country, city, visa_type)
    return success(data=res)

@public_router.get("/slots/overview", summary="查询最近的slot", tags=["Slot数据"], response_model=ApiResponse[List[SlotOverviewOut]])
async def slots_latest_get(
    city: str = Query("", description="递交城市"),
    db: AsyncSession = Depends(get_db)
):
    res = await SlotSnapshotService.get_slot_overview(db, city)
    return success(data=res)

@admin_required_router.post("/slots/report", summary="上报最近的slot", tags=["Slot数据"], response_model=ApiResponse[SlotSnapshotOut])
async def slots_report(
    payload: SlotSnapshotCreate,
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    res = await SlotSnapshotService.report(db, redis_client, payload)
    return success(data=res)

@admin_required_router.post("/slot_refresh/start", summary="刷新slot开始", tags=["Slot Monitor 监控"], response_model=ApiResponse[RefreshStatusOut])
async def slot_refresh_start(data: RefreshBase, db: AsyncSession = Depends(get_db)
):
    data = await SlotRefreshStatusService.refresh_start(db, data)
    return success(data=data)

@admin_required_router.post("/slot_refresh/success", summary="刷新slot成功", tags=["Slot Monitor 监控"], response_model=ApiResponse[RefreshStatusOut])
async def  slot_refresh_success(data: RefreshBase, db: AsyncSession = Depends(get_db)
):
    data = await SlotRefreshStatusService.refresh_success(db, data)
    return success(data=data)

@admin_required_router.post("/slot_refresh/fail", summary="刷新slot失败", tags=["Slot Monitor 监控"], response_model=ApiResponse[RefreshStatusOut])
async def slot_refresh_fail(data: RefreshFail,db: AsyncSession = Depends(get_db)
):
    data = await SlotRefreshStatusService.refresh_fail(db, data)
    return success(data=data)

@admin_required_router.get("/slot_refresh/status", summary="查询刷新纪录", tags=["Slot Monitor 监控"], response_model=ApiResponse[List[RefreshStatusOut]])
async def slot_refresh_status(db: AsyncSession = Depends(get_db)):
    data = await SlotRefreshStatusService.list_all(db)
    return success(data=data)

@public_router.post("/webhook/smshelper", summary="双开助手Webhook", tags=["webhook"], response_model=ApiResponse)
async def webhook_smshelper(
    payload: SMSHelperWebhookPayload,
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    logger.info(f'smshelper webhook title={payload.title}, content={payload.content}')
    if "微信支付" in payload.title:
        res = await WebhookService.smshelper_payment_webhook(db, payload)
        if res:
            print(f"📧 send payment succeeded notification email")
    return success()

@public_router.post("/webhook/stripe", summary="Stripe Webhook", tags=["webhook"], response_model=ApiResponse)
async def webhook_stripe(
    request: Request,
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    payload = request.body()
    sig_header = request.headers.get("stripe-signature")
    event = stripe.Webhook.construct_event(
        payload=payload,
        sig_header=sig_header,
        secret=settings.STRIPE_WEBHOOK_SECRET,
    )
    res = await WebhookService.stripe_payment_webhook(db, event)
    if res:
        print(f"📧 send payment succeeded notification email")
    return success()

@public_router.post("/auth/auto-register", summary="自动注册", tags=["用户管理"], response_model=ApiResponse[AutoRegisterData])
async def vas_auto_register(
    payload: AutoRegisterRequest,
    request: Request,
    db: AsyncSession = Depends(get_db)
):
    user_agent = request.headers.get("user-agent", "unknown")
    client_host = request.client.host
    x_forwarded_for = request.headers.get("x-forwarded-for")
    if x_forwarded_for:
        client_ip = x_forwarded_for.split(",")[0].strip()
    else:
        client_ip = client_host
    res = await AuthService.auto_register(db, payload, client_ip, user_agent)
    return success(data=res)

@public_router.post("/auth/send-bind-code", summary="发送邮箱注册码 绑定邮箱用", tags=["用户管理"], response_model=ApiResponse)
async def vas_send_bind_code(
    payload: SendBindCodeRequest,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    await AuthService.send_bind_code(db, payload, current_user, redis_client)
    return success(message="verify email sent, please check your inbox")

@public_router.post("/auth/send-reset-code", summary="发送邮箱注册码 重置密码用", tags=["用户管理"], response_model=ApiResponse)
async def vas_send_reset_code(
    payload: SendResetCodeRequest,
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    await AuthService.send_reset_code(db, payload, redis_client)
    return success(message="verify email sent, please check your inbox")

@protected_router.post("/auth/bind-email", summary="绑定邮箱", tags=["用户管理"], response_model=ApiResponse[LoginData])
async def vas_bind_email(
    payload: BindEmailRequest,
    request: Request,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    user_agent = request.headers.get("user-agent", "unknown")
    client_host = request.client.host
    x_forwarded_for = request.headers.get("x-forwarded-for")
    if x_forwarded_for:
        client_ip = x_forwarded_for.split(",")[0].strip()
    else:
        client_ip = client_host
    res = await AuthService.bind_email(db, payload, current_user, redis_client, client_ip, user_agent)
    return success(data=res)

@public_router.post("/auth/reset-password", summary="重置密码", tags=["用户管理"], response_model=ApiResponse)
async def vas_reset_password(
    payload: ResetPasswordRequest,
    db: AsyncSession = Depends(get_db)
):
    res = await AuthService.reset_password(db, payload)
    return success(data=res)

@public_router.post("/auth/login", summary="邮箱登录", tags=["用户管理"], response_model=ApiResponse[LoginData])
async def vas_login(
    payload: LoginRequest,
    request: Request,
    db: AsyncSession = Depends(get_db)
):
    user_agent = request.headers.get("user-agent", "unknown")
    client_host = request.client.host
    x_forwarded_for = request.headers.get("x-forwarded-for")
    if x_forwarded_for:
        client_ip = x_forwarded_for.split(",")[0].strip()
    else:
        client_ip = client_host
    res = await AuthService.login(db, payload, client_ip, user_agent)
    return success(data=res)

@admin_required_router.get("/user/list_all", summary="获取所有用户", tags=["用户管理"], response_model=ApiResponse[PageResponse[VasUserOut]])
async def vas_user_list_all(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    db: AsyncSession = Depends(get_db)
):
    users = await UserService.list_all(db, page, size, keyword)
    return success(data=users)

@admin_required_router.get("/user/detail", summary="获取用户信息", tags=["用户管理"], response_model=ApiResponse[VasUserOut])
async def vas_user_get_detail(
    user_id: str,
    db: AsyncSession = Depends(get_db)
):
    user = await UserService.get(db, user_id)
    return success(data=user)

@admin_required_router.post("/user/update", summary="更新用户信息", tags=["用户管理"], response_model=ApiResponse[VasUserOut])
async def vas_user_update(
    uid: str,
    payload: VasUserUpdate,
    db: AsyncSession = Depends(get_db)
):
    updated = await UserService.update(db, uid, payload)
    return success(data=updated)

@protected_router.post("/user/set_profiles", summary="更新用户信息", tags=["用户管理"], response_model=ApiResponse[VasUserOut])
async def vas_user_update(
    payload: VasUserSetProfiles,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    updated = await UserService.set_profiles(db, current_user, payload)
    return success(data=updated)

@admin_required_router.get("/vas/statistics/overview", summary="系统概览", tags=["Visafly签证系统"], response_model=ApiResponse[VasStatisticsOverviewOut])
async def vas_statistics_overview(
    db: AsyncSession = Depends(get_db)
):
    overview = await StatisticsService.overview(db)
    return success(data=overview)

@admin_required_router.post("/vas/product/create", summary="创建商品", tags=["Visafly签证系统"], response_model=ApiResponse[VasProductOut])
async def vas_product_create(
    payload: VasProductCreate,
    db: AsyncSession = Depends(get_db)
):
    created_product = await ProductService.create(db, payload)
    return success(data=created_product)

@admin_required_router.post("/vas/product/update", summary="更新商品", tags=["Visafly签证系统"], response_model=ApiResponse[VasProductOut])
async def vas_product_update(
    id: int,
    payload: VasProductUpdate,
    db: AsyncSession = Depends(get_db)
):
    product = await ProductService.update(db, id, payload)
    return success(data=product)

@public_router.get("/vas/product/list", summary="获取商品列表", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasProductOut]])
async def vas_product_list(
    country: str = Query("", description="目的国家"),
    visa_type: str = Query("", description="签证类型"),
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    db: AsyncSession = Depends(get_db)
):
    products = await ProductService.list_product(db, country, visa_type, page, size, keyword)
    return success(data=products)

@public_router.get("/vas/product/detail", summary="获取商品列表", tags=["Visafly签证系统"], response_model=ApiResponse[VasProductOut])
async def vas_product_get_by_id(
    product_id: int,
    db: AsyncSession = Depends(get_db)
):
    products = await ProductService.get(db, product_id)
    return success(data=products)

@admin_required_router.post("/vas/product_routing/create", summary="创建商品路由列表", tags=["Visafly签证系统"], response_model=ApiResponse[VasProductRoutingOut])
async def vas_product_routing_create(
    payload: VasProductRoutingCreate,
    db: AsyncSession = Depends(get_db)
):
    payload = await ProductRoutingService.create(db, payload)
    return success(data=payload)

@admin_required_router.delete("/vas/product_routing/delete", summary="删除商品路由列表", tags=["Visafly签证系统"], response_model=ApiResponse)
async def vas_product_routing_date(
    id: int,
    db: AsyncSession = Depends(get_db)
):
    await ProductRoutingService.delete(db, id)
    return success()

@admin_required_router.get("/vas/product_routing/list", summary="获取商品路由列表", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasProductRoutingOut]])
async def vas_product_routing_list_by_product(
    product_id: int,
    db: AsyncSession = Depends(get_db)
):
    product_routings = await ProductRoutingService.list_by_product(db, product_id)
    return success(data=product_routings)

@admin_required_router.post("/vas/llm/data_parsing", summary="llm数据解析", tags=["Visafly签证系统"], response_model=ApiResponse[ParseUserInputsOut])
async def vas_llm_data_parsing(
    payload: ParseUserInputsPayload,
    db: AsyncSession = Depends(get_db)
):
    out = await LlmService.handle_parse(db, payload)
    return success(data=out)

@public_router.get("/vas/schema/detail", summary="获取schema", tags=["Visafly签证系统"], response_model=ApiResponse[VasSchemaOut])
async def vas_schema_get(
    schema_id: int,
    db: AsyncSession = Depends(get_db)
):
    schema = await SchemaService.get(db, schema_id)
    return success(data=schema)

@admin_required_router.post("/vas/schema/create", summary="新增schema", tags=["Visafly签证系统"], response_model=ApiResponse[VasSchemaOut])
async def vas_schema_create(
    payload: VasSchemaCreate,
    db: AsyncSession = Depends(get_db)
):
    schema = await SchemaService.create(db, payload)
    return success(data=schema)

@admin_required_router.post("/vas/schema/update", summary="更新schema", tags=["Visafly签证系统"], response_model=ApiResponse[VasSchemaOut])
async def vas_schema_update(
    id: int,
    payload: VasSchemaUpdate,
    db: AsyncSession = Depends(get_db)
):
    schema = await SchemaService.update(db, id, payload)
    return success(data=schema)

@admin_required_router.delete("/vas/schema/delete", summary="删除schema", tags=["Visafly签证系统"], response_model=ApiResponse)
async def vas_schema_delete(
    id: int,
    db: AsyncSession = Depends(get_db)
):
    await SchemaService.delete(db, id)
    return success()

@admin_required_router.get("/vas/schema/list", summary="获取schema列表", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasSchemaOut]])
async def vas_schema_list(
    db: AsyncSession = Depends(get_db)
):
    schemas = await SchemaService.list_all(db)
    return success(data=schemas)

@protected_router.post("/vas/order/create", summary="创建订单", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut])
async def vas_order_create(
    payload: VasOrderCreate,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    product = await ProductService.get(db, payload.product_id)
    # ① 获取产品绑定的 schema
    schema = await SchemaService.get(db, product.schema_id)
    # ② 校验 user_inputs
    validate_user_inputs(
        schema_json=schema.schema_json,
        user_inputs=payload.user_inputs,
    )
    created_order = await OrderService.create(db, payload, product, current_user, redis_client)
    return success(data=created_order)

@admin_required_router.post("/vas/order/create_by_admin", summary="管理员创建订单", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut])
async def vas_order_create_by_admin(
    payload: VasOrderCreate,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    product = await ProductService.get(db, payload.product_id)
    # ① 获取产品绑定的 schema
    schema = await SchemaService.get(db, product.schema_id)
    # ② 校验 user_inputs
    validate_user_inputs(
        schema_json=schema.schema_json,
        user_inputs=payload.user_inputs,
    )
    created_order = await OrderService.create_by_admin(db, payload, product, current_user, redis_client)
    return success(data=created_order)

@admin_required_router.post("/vas/order/adjust-price", summary="管理员调整订单价格", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut])
async def vas_order_adjust_price(
    order_id: str,
    payload: VasOrderAdjustPrice,
    db: AsyncSession = Depends(get_db),
    current_user: VasUser = Depends(get_current_user)
):
    order = await OrderService.adjust_order_price(db, order_id, payload)
    return success(data=order)

@admin_required_router.get("/vas/order/detail", summary="查询订单", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut])
async def vas_order_create(
    order_id: str,
    db: AsyncSession = Depends(get_db),
):
    order = await OrderService.get(db, order_id)
    return success(data=order)

@admin_required_router.post("/vas/order/patch_user_inputs", summary="更新订单的用户信息", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut])
async def vas_order_patch_user_inputs(
    order_id: str,
    payload: VasOrderPatchUserInputs,
    db: AsyncSession = Depends(get_db),
):
    order = await OrderService.patch_user_inputs(db, order_id, payload)
    return success(data=order)

@protected_router.get("/vas/order/list_by_user", summary="查看所有订单", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasOrderOut]])
async def vas_order_list_by_user(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    orders = await OrderService.list_by_user(db, current_user.id, page, size, keyword)
    return success(data=orders)

@protected_router.get("/vas/order/list_all", summary="查看所有订单", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasOrderOut]])
async def vas_order_list_all(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    db: AsyncSession = Depends(get_db)
):
    orders = await OrderService.list_all(db, page, size, keyword)
    return success(data=orders)

@admin_required_router.post("/vas/order/cancel", summary="取消订单", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut])
async def vas_order_cancel(
    order_id: str,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    cancelled_order = await OrderService.cancel(db, order_id)
    return success(data=cancelled_order)

@protected_router.get("/vas/payment_provider/list_enabled", summary="获取支付方式", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasPaymentProviderOut]])
async def vas_payment_provider_simple_get(
    db: AsyncSession = Depends(get_db)
):
    providers = await PaymentProviderService.list_enabled(db)
    return success(data=providers)

@admin_required_router.get("/vas/payment_provider/list_all", summary="获取支付方式", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasPaymentProviderOut]])
async def vas_payment_provider_list_all(
    db: AsyncSession = Depends(get_db)
):
    providers = await PaymentProviderService.list_all(db)
    return success(data=providers)

@admin_required_router.post("/vas/payment_provider/create", summary="新增支付服务提供商", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentProviderOut])
async def vas_payment_provider_create(
    payload: VasPaymentProviderCreate,
    db: AsyncSession = Depends(get_db)
):
    provider = await PaymentProviderService.create(db, payload)
    return success(data=provider)

@admin_required_router.post("/vas/payment_provider/update", summary="更新支付服务提供商", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentProviderOut])
async def vas_payment_provider_update(
    id: int,
    payload: VasPaymentProviderUpdate,
    db: AsyncSession = Depends(get_db)
):
    provider = await PaymentProviderService.update(db, id, payload)
    return success(data=provider)

@admin_required_router.delete("/vas/payment_provider/delete", summary="删除支付服务提供商", tags=["Visafly签证系统"], response_model=ApiResponse)
async def vas_payment_provider_delete(
    id: int,
    db: AsyncSession = Depends(get_db)
):
    await PaymentProviderService.delete(db, id)
    return success()

@protected_router.post("/vas/payment/create", summary="创建支付", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentOut])
async def vas_payment_create(
    payload: VasPaymentCreate,
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    res = await PaymentService.create_payment(db, payload, redis_client)
    return success(data=res)

@protected_router.get("/vas/payment/detail", summary="获取支付详情", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentOut])
async def vas_payment_create(
    payment_id: int,
    db: AsyncSession = Depends(get_db),
):
    res = await PaymentService.get_by_id(db, payment_id)
    return success(data=res)

@admin_required_router.post("/vas/payment/confirm_by_admin", summary="管理员确认支付", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentConfirmationOut])
async def vas_payment_confirm_by_admin(
    id: int,
    payload: VasPaymentConfirmationUpdate,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    res = await PaymentService.confirm_by_admin(db, id, payload, current_user)
    return success(data=res)

@admin_required_router.post("/vas/payment/admin_update_status", summary="管理员更新支付状态", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentOut])
async def vas_payment_admin_update_status(
    payment_id: int,
    payload: AdminUpdateStatusPayload,
    db: AsyncSession = Depends(get_db)
):
    res = await PaymentService.admin_update_status(db, payment_id, payload)
    return success(data=res)

@protected_router.post("/vas/payment/confirm_by_user", summary="用户确认支付", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentConfirmationOut])
async def vas_payment_confirm_by_user(
    payload: VasPaymentConfirmationCreate,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    res = await PaymentService.confirm_by_user(db, payload, current_user, redis_client)
    return success(data=res)

@admin_required_router.post("/vas/payment_confirmation/list_all", summary="查询所有待确认支付", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasPaymentConfirmationOut]])
async def vas_payment_confirm_list(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    db: AsyncSession = Depends(get_db)
):
    res = await PaymentService.list_payment_confirmation(db, keyword, page, size)
    return success(data=res)

@protected_router.get("/vas/payment/list_by_order", summary="获取某个订单下的所有payment记录", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasPaymentOut]])
async def vas_payment_list_by_order(
    order_id: str,
    db: AsyncSession = Depends(get_db)
):
    payments = await PaymentService.list_by_order(db, order_id)
    return success(data=payments)

@admin_required_router.post("/vas/payment_qr/create", summary="新增收款码", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentQrOut])
async def vas_payment_qr_create(payload: VasPaymentQrCreate, db: AsyncSession = Depends(get_db)):
    qr = await PaymentQrService.create(db, payload)
    return success(data=qr)

@protected_router.get("/vas/payment_qr/list_by_provider", summary="获取某个服务商的所有付款码", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasPaymentQrOut]])
async def vas_payment_qr_list_qrcode_by_provider(provider_id: int, db: AsyncSession = Depends(get_db)):
    qr = await PaymentQrService.list_by_provider(db, provider_id)
    return success(data=qr)
    
@protected_router.get("/vas/payment_qr/qrcode", summary="获取支付的QRCode", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentQrOut])
async def vas_payment_qr_get_qrcode_by_id(id: int, db: AsyncSession = Depends(get_db)):
    qr = await PaymentQrService.get_by_id(db, id)
    return success(data=qr)

@admin_required_router.post("/vas/payment_qr/set_enable", summary="修改QRCode", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentQrOut])
async def vas_payment_qr_update(
    id: int,
    payload: VasPaymentQrSetEnableIn,
    db: AsyncSession = Depends(get_db)
):
    qr = await PaymentQrService.set_enable(db, id, payload)
    return success(data=qr)

@admin_required_router.delete("/vas/payment_qr/delete", summary="删除QRCode", tags=["Visafly签证系统"], response_model=ApiResponse)
async def vas_payment_qr_update(
    id: int,
    db: AsyncSession = Depends(get_db)
):
    await PaymentQrService.delete(db, id)
    return success()

@admin_required_router.get("/vas/task/list", summary="获取待执行的任务", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasTaskOut]])
async def vas_task_list(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    status: str = Query("", description="task 自定义索引"),
    routing_key: str = Query("", description="task 自定义索引"),
    script_version: str = Query("", description="脚本版本, 用来向后兼容"),
    db: AsyncSession = Depends(get_db)
):
    tasks = await VasTaskService.list_task(db, status, routing_key, script_version, keyword, page, size)
    return success(data=tasks)

@admin_required_router.post("/vas/task/update", summary="更新任务数据", tags=["Visafly签证系统"], response_model=ApiResponse[VasTaskOut])
async def vas_task_update(
    id: int,
    payload: VasTaskUpdate,
    db: AsyncSession = Depends(get_db)
):
    task = await VasTaskService.update(db, id, payload)
    return success(data=task)

@admin_required_router.get("/vas/task/get_by_order", summary="根据订单查找任务", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasTaskOut]])
async def vas_task_pending(
    order_id: str = Query(..., description="订单编号"),
    script_version: str = Query("", description="脚本版本, 用来向后兼容"),
    db: AsyncSession = Depends(get_db)
):
    tasks = await VasTaskService.get_active_task_by_order_id(db, order_id)
    return success(data=tasks)

@admin_required_router.post("/vas/task/return_to_queue", summary="重新进入等候状态", tags=["Visafly签证系统"], response_model=ApiResponse[VasTaskOut])
async def vas_task_return_to_queue(task_id:int, db: AsyncSession = Depends(get_db)):
    obj = await VasTaskService.return_to_queue(db, task_id)
    return success(data=obj)

@admin_required_router.post("/vas/task/manual_confirm", summary="设置任务完成", tags=["Visafly签证系统"], response_model=ApiResponse[VasTaskOut])
async def vas_task_manual_confirm(task_id:int, db: AsyncSession = Depends(get_db)):
    obj = await VasTaskService.manual_confirm(db, task_id)
    return success(data=obj)

@admin_required_router.get("/vas/task/pop", summary="任务出队（pop）", tags=["Visafly签证系统"], response_model=ApiResponse[VasTaskOut])
async def vas_task_pop_task(
    queue_name: str,
    db: AsyncSession = Depends(get_db),
):
    task = await VasTaskService.pop_vas_task(db, queue_name, 180)
    return success(data=task)

@protected_router.post("/vas/ticket/create", summary="创建工单", tags=["Visafly签证系统"], response_model=ApiResponse[VasTicketOut])
async def vas_ticket_create(
    data:VasTicketCreate,
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client)
):
    obj = await TicketService.create(db, data, current_user, redis_client)
    return success(data=obj)

@protected_router.get("/vas/ticket/list_by_user", summary="查看工单", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasTicketOut]])
async def vas_ticket_list_by_user(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    current_user: VasUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    tickets = await TicketService.list_by_user(db, current_user.id, page, size, keyword)
    return success(data=tickets)

@admin_required_router.get("/vas/ticket/list_all", summary="查看工单", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasTicketOut]])
async def vas_ticket_list_all(
    page: int = Query(0, description="第几页"),
    size: int = Query(10, description="分页大小"),
    keyword: str = Query("", description="查询条件"),
    db: AsyncSession = Depends(get_db)
):
    tickets = await TicketService.list_all(db, page, size, keyword)
    return success(data=tickets)

@admin_required_router.post("/vas/tickets/status", summary="管理员更新工单状态", tags=["Visafly签证系统"], response_model=ApiResponse[VasTicketOut])
async def update_ticket_status(
    ticket_id: int,
    payload: VasTicketStatusUpdate,
    db: AsyncSession = Depends(get_db),
    user=Depends(get_current_user)
):
    res = await TicketService.update_status(
        db=db,
        ticket_id=ticket_id,
        status=payload.status,
        comment=payload.comment,
        admin_id=user.id
    )
    return success(data=res)
    
@protected_router.post("/vas/tickets/send_message", summary="发送工单消息", tags=["Visafly签证系统"], response_model=ApiResponse[VasTicketMessageOut])
async def create_ticket_message(
    ticket_id: int,
    payload: VasTicketMessageCreate,
    db: AsyncSession = Depends(get_db),
    user=Depends(get_current_user)
):
    res = await TicketService.add_message(
        db=db,
        ticket_id=ticket_id,
        sender_type=user.role,
        sender_id=user.id,
        content=payload.content,
        attachments=payload.attachments
    )
    return success(data=res)
    
@protected_router.get("/vas/tickets/fetch_message", summary="分页获取工单会话", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasTicketMessageOut]])
async def get_ticket_messages(
    ticket_id: int,
    page: int = 1,
    size: int = 20,
    db: AsyncSession = Depends(get_db)
):
    msgs = await TicketService.list_messages(
        db=db,
        ticket_id=ticket_id,
        page=page,
        size=size
    )
    return success(data=msgs)