import time import uuid import json import requests from typing import List from app.core.logger import logger from fastapi import APIRouter, Request, Response, Query, Depends, Body, UploadFile, File, HTTPException from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session 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 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.auto_booking import AutoBookingCreate, AutoBookingOut 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, VasOrderPatchUserInputs, VasOrderOut from app.schemas.payment import VasPaymentCreate, VasPaymentOut 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 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.services.configuration_service import ConfigurationService from app.services.troov_service import get_rate_by_date 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.auto_booking_service import AutoBookingService 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 PaymentProviderSerivce 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 # 公共路由 public_router = APIRouter() # 受保护路由 protected_router = APIRouter() # 管理员级别路由 admin_required_router = APIRouter() @admin_required_router.get("/ping", summary="心跳检测", tags=["测试接口"]) def ping(): return {"message": "pong"} @admin_required_router.get("/sms/upload", summary="上报短信", tags=["短信接口"], response_model=ApiResponse[ShortMessageDetail]) 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 = 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]]) def sms_download( phone: str = Query(..., description="手机号"), keyword: str = Query('', description="短信内容关键字"), sent_at: str = Query('', description="筛选时间(可选)"), redis_client: Redis = Depends(get_redis_client) ): """ 查询短信(支持关键字和时间过滤) """ obj = 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]]) def troov_rate(date: str = Query(..., description="查询的日期, 格式: YYYY-MM-DD"), redis_client: Redis = Depends(get_redis_client)): # 调用 service 层获取数据 obj = get_rate_by_date(redis_client, date) return success(data=obj) @admin_required_router.post("/dynamic-configurations", summary="创建动态配置", tags=["动态配置"], response_model=ApiResponse[ConfigurationOut]) def dynamic_config_create(config_in: ConfigurationCreate, db: Session = Depends(get_db)): obj = ConfigurationService.create(db, config_in) return success(data=obj) @admin_required_router.get("/dynamic-configurations/all", summary="读取所有动态配置", tags=["动态配置"], response_model=ApiResponse[List[ConfigurationOut]]) def dynamic_config_get_all(db: Session = Depends(get_db)): obj = ConfigurationService.get_all(db) return success(data=obj) @admin_required_router.get("/dynamic-configurations/key/{config_key}", summary="根据Key读取动态配置", tags=["动态配置"], response_model=ApiResponse[ConfigurationOut]) def dynamic_config_get_by_key(config_key: str, db: Session = Depends(get_db)): config = 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]) def dynamic_config_update_by_key(config_key: str, config_in: ConfigurationUpdate, db: Session = Depends(get_db)): config = 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]) def dynamic_config_delete_by_key(config_key: str, db: Session = Depends(get_db)): config = 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]) def http_session_create( data: HttpSessionCreate, db: Session = Depends(get_db) ): logger.info(f"[Create HttpSession] sid={data.session_id}") obj = HttpSessionService.create(db, data) return success(data=obj) @admin_required_router.delete("/http-session", summary="删除http session", tags=["会话管理"], response_model=ApiResponse) def http_session_delete_by_sid( session_id: str = Query(...), db: Session = Depends(get_db) ): logger.info(f"[Delete HttpSession] sid={session_id}") HttpSessionService.delete_by_sid(db, session_id) return success() @admin_required_router.put("/http-session", summary="更新http session", tags=["会话管理"], response_model=ApiResponse[HttpSessionOut]) def http_session_update_by_sid( session_id: str = Query(...), data: HttpSessionUpdate = Body(...), db: Session = Depends(get_db) ): logger.info(f"[Update HttpSession] sid={session_id}") obj = 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]) def http_session_get_by_sid( session_id: str = Query(...), db: Session = Depends(get_db) ): logger.info(f"[Get HttpSession] sid={session_id}") obj = HttpSessionService.get_by_sid(db, session_id) return success(data=obj) @admin_required_router.get("/email-authorizations", summary="查询所有内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[List[EmailAuthorizationOut]]) def email_authorizations_get(db: Session = Depends(get_db)): obj = EmailAuthorizationService.get_all(db) return success(data=obj) @admin_required_router.post("/email-authorizations", summary="创建内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut]) def email_authorizations_create(data: EmailAuthorizationCreate, db: Session = Depends(get_db)): obj = EmailAuthorizationService.create(db, data) return success(data=obj) @admin_required_router.get("/email-authorizations/{id}", summary="通过id查询内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut]) def email_authorizations_get_by_id(id: int, db: Session = Depends(get_db)): email_auth = 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]) def email_authorizations_update_by_id(id: int, data: EmailAuthorizationUpdate, db: Session = Depends(get_db)): updated = EmailAuthorizationService.update(db, id, data) return success(data=updated) @admin_required_router.delete("/email-authorizations/{id}", summary="通过id删除内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut]) def email_authorizations_delete_by_id(id: int, db: Session = Depends(get_db)): deleted = EmailAuthorizationService.delete(db, id) return deleted @admin_required_router.get("/email-authorizations/email/{email}", summary="通过邮箱地址查询内部邮箱", tags=["邮箱接口"], response_model=ApiResponse[EmailAuthorizationOut]) def email_authorizations_get_by_email(email: str, db: Session = Depends(get_db)): email_auth = EmailAuthorizationService.get_by_email(db, email) return email_auth @admin_required_router.post("/email-authorizations/fetch", summary="读取邮件, 仅限文本内容", tags=["邮箱接口"], response_model=ApiResponse[EmailContent]) 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: Session = Depends(get_db) ): auth = EmailAuthorizationService.get_by_email(db, email) result = 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]) 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: Session = Depends(get_db) ): auth = EmailAuthorizationService.get_by_email(db, email) result = 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]) 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: Session = Depends(get_db) ): auth = EmailAuthorizationService.get_by_email(db, emailAccount) result = 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/sendmail", summary="发送邮件", tags=["邮箱接口"], response_model=ApiResponse[EmailContent]) 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: Session = Depends(get_db) ): auth = EmailAuthorizationService.get_by_email(db, emailAccount) result = 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]) 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: Session = Depends(get_db) ): auth = EmailAuthorizationService.get_by_email(db, emailAccount) result = 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]) def resource_upload_file(file: UploadFile = File(...)): result = SeaweedFSService.upload(file) return success(data=result) @public_router.get("/resource/download_file", summary="下载文件", tags=["文件管理"]) def resource_get_file(fid: str): data = SeaweedFSService.get(fid) if not data: raise HTTPException(status_code=404, detail="文件不存在") content, mime = data return Response(content=content, media_type=mime) @admin_required_router.post("/s/generate", summary="生成短链接地址<压缩地址长度>", tags=["Short URL"], response_model=ApiResponse[ShortUrlOut]) def short_url_generate( data: ShortUrlCreate, db: Session = Depends(get_db), ): """生成短链接""" record = ShortUrlService.create_short_url(db, data.long_url) return success(data=record) @public_router.get("/s/{short_key}", summary="访问短链接地址<自动重定向到真实链接地址>", tags=["Short URL"]) def short_url_request(short_key: str, db: Session = Depends(get_db)): """访问短链接自动重定向""" long_url = 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]) def task_create(data: TaskCreate, db: Session = Depends(get_db)): """创建任务""" return TaskService.create(db, data) @admin_required_router.get("/tasks/{task_id:int}", summary="根据taskId读取任务状态", tags=["任务管理接口"], response_model=ApiResponse[TaskOut]) def task_get_by_id(task_id: int, db: Session = Depends(get_db)): """获取任务""" task = TaskService.get_by_id(db, task_id) return success(data=task) @admin_required_router.get("/tasks/pending", summary="获取等待执行的任务", tags=["任务管理接口"], response_model=ApiResponse[List[TaskOut]]) def task_get_pending( page: int = Query(0, description="第几页"), size: int = Query(10, description="分页大小"), command: str = Query(..., description="任务类型"), db: Session = Depends(get_db), ): """分页获取等待执行的任务""" obj = 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]) def task_update_by_id(task_id: int, data: TaskUpdate, db: Session = Depends(get_db)): """更新任务状态或结果""" updated = TaskService.update(db, task_id, data) return success(data=updated) @admin_required_router.post("/tg/send_message", summary="推送电报消息", tags=["消息推送接口"], response_model=ApiResponse) def tg_send_message( payload: TelegramIn ): TelegramService.push_to_telegram(payload) return success() @admin_required_router.post("/wechat/send", summary="推送微信消息", tags=["消息推送接口"], response_model=ApiResponse) def wechat_send( payload: WechatIn ): WechatService.push_to_wechat(payload) return success() @admin_required_router.post("/cards/publish", summary="创建新的消息卡片", tags=["信息卡片接口"], response_model=ApiResponse[CardOut]) def cards_publish( data: CardCreate = Body(...), db: Session = Depends(get_db) ): obj = CardService.create(db, data) return success(data=obj) @public_router.get("/cards/view2", summary="根据关键词分页查询卡片, 可选择语言", tags=["信息卡片接口"], response_model=ApiResponse[PageResponse[CardOut]]) 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: Session = Depends(get_db) ): obj = 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]]) 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]) def slots_latest_get( country: str = Query("", description="目的国家"), city: str = Query("", description="递交城市"), visa_type: str = Query("", description="签证类型"), db: Session = Depends(get_db) ): res = SlotSnapshotService.latest_for(db, country, city, visa_type) return success(data=res) @admin_required_router.get("/slots/report", summary="上报最近的slot", tags=["Slot数据"], response_model=ApiResponse[SlotSnapshotOut]) def slots_report( payload: SlotSnapshotCreate, db: Session = Depends(get_db) ): res = SlotSnapshotService.create(db, payload) return success(data=res) @public_router.post("/webhook/smshelper", summary="双开助手Webhook", tags=["webhook"], response_model=ApiResponse) def webhook_smshelper( payload: SMSHelperWebhookPayload, db: Session = 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 = 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) def webhook_stripe( request: Request, db: Session = 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 = 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]) def vas_auto_register( payload: AutoRegisterRequest, db: Session = Depends(get_db) ): res = AuthService.auto_register(db, payload) return success(data=res) @public_router.post("/auth/send-bind-code", summary="发送邮箱注册码 绑定邮箱用", tags=["用户管理"], response_model=ApiResponse) def vas_send_bind_code( payload: SendBindCodeRequest, current_user: VasUser = Depends(get_current_user), db: Session = Depends(get_db), redis_client: Redis = Depends(get_redis_client) ): 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) def vas_send_reset_code( payload: SendResetCodeRequest, db: Session = Depends(get_db), redis_client: Redis = Depends(get_redis_client) ): 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]) def vas_bind_email( payload: BindEmailRequest, current_user: VasUser = Depends(get_current_user), db: Session = Depends(get_db), redis_client: Redis = Depends(get_redis_client) ): res = AuthService.bind_email(db, payload, current_user, redis_client) return success(data=res) @public_router.post("/auth/reset-password", summary="重置密码", tags=["用户管理"], response_model=ApiResponse) def vas_reset_password( payload: ResetPasswordRequest, db: Session = Depends(get_db) ): res = AuthService.reset_password(db, payload) return success(data=res) @public_router.post("/auth/login", summary="邮箱登录", tags=["用户管理"], response_model=ApiResponse[LoginData]) def vas_login( payload: LoginRequest, db: Session = Depends(get_db) ): res = AuthService.login(db, payload) return success(data=res) @admin_required_router.get("/user/list_all", summary="获取所有用户", tags=["用户管理"], response_model=ApiResponse[PageResponse[VasUserOut]]) def vas_user_list_all( page: int = Query(0, description="第几页"), size: int = Query(10, description="分页大小"), keyword: str = Query("", description="查询条件"), db: Session = Depends(get_db) ): users = UserService.list_all(db, page, size, keyword) return success(data=users) @admin_required_router.get("/user/detail", summary="获取用户信息", tags=["用户管理"], response_model=ApiResponse[VasUserOut]) def vas_user_get_detail( user_id: str, db: Session = Depends(get_db) ): user = UserService.get(db, user_id) return success(data=user) @admin_required_router.post("/user/update", summary="更新用户信息", tags=["用户管理"], response_model=ApiResponse[VasUserOut]) def vas_user_update( uid: str, payload: VasUserUpdate, db: Session = Depends(get_db) ): updated = UserService.update(db, uid, payload) return success(data=updated) @admin_required_router.post("/user/set_profiles", summary="更新用户信息", tags=["用户管理"], response_model=ApiResponse[VasUserOut]) def vas_user_update( payload: VasUserSetProfiles, current_user: VasUser = Depends(get_current_user), db: Session = Depends(get_db) ): updated = 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]) def vas_statistics_overview( db: Session = Depends(get_db) ): overview = StatisticsService.overview(db) return success(data=overview) @admin_required_router.post("/vas/product/create", summary="创建商品", tags=["Visafly签证系统"], response_model=ApiResponse[VasProductOut]) def vas_product_create( payload: VasProductCreate, db: Session = Depends(get_db) ): created_product = ProductService.create(db, payload) return success(data=created_product) @admin_required_router.post("/vas/product/update", summary="更新商品", tags=["Visafly签证系统"], response_model=ApiResponse[VasProductOut]) def vas_product_update( id: int, payload: VasProductUpdate, db: Session = Depends(get_db) ): product = ProductService.update(db, id, payload) return success(data=product) @public_router.get("/vas/product/list", summary="获取商品列表", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasProductOut]]) 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: Session = Depends(get_db) ): products = 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]) def vas_product_get_by_id( product_id: int, db: Session = Depends(get_db) ): products = ProductService.get(db, product_id) return success(data=products) @admin_required_router.post("/vas/product_routing/create", summary="创建商品路由列表", tags=["Visafly签证系统"], response_model=ApiResponse[VasProductRoutingOut]) def vas_product_routing_create( payload: VasProductRoutingCreate, db: Session = Depends(get_db) ): payload = ProductRoutingService.create(db, payload) return success(data=payload) @admin_required_router.delete("/vas/product_routing/delete", summary="删除商品路由列表", tags=["Visafly签证系统"], response_model=ApiResponse) def vas_product_routing_date( id: int, db: Session = Depends(get_db) ): ProductRoutingService.delete(db, id) return success() @admin_required_router.get("/vas/product_routing/list", summary="获取商品路由列表", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasProductRoutingOut]]) def vas_product_routing_list_by_product( product_id: int, db: Session = Depends(get_db) ): product_routings = ProductRoutingService.list_by_product(db, product_id) return success(data=product_routings) @public_router.get("/vas/schema/detail", summary="获取schema", tags=["Visafly签证系统"], response_model=ApiResponse[VasSchemaOut]) def vas_schema_get( schema_id: int, db: Session = Depends(get_db) ): schema = SchemaService.get(db, schema_id) return success(data=schema) @admin_required_router.post("/vas/schema/create", summary="新增schema", tags=["Visafly签证系统"], response_model=ApiResponse[VasSchemaOut]) def vas_schema_create( payload: VasSchemaCreate, db: Session = Depends(get_db) ): schema = SchemaService.create(db, payload) return success(data=schema) @admin_required_router.post("/vas/schema/update", summary="更新schema", tags=["Visafly签证系统"], response_model=ApiResponse[VasSchemaOut]) def vas_schema_update( id: int, payload: VasSchemaUpdate, db: Session = Depends(get_db) ): schema = SchemaService.update(db, id, payload) return success(data=schema) @admin_required_router.delete("/vas/schema/delete", summary="删除schema", tags=["Visafly签证系统"], response_model=ApiResponse) def vas_schema_delete( id: int, db: Session = Depends(get_db) ): SchemaService.delete(db, id) return success() @admin_required_router.get("/vas/schema/list", summary="获取schema列表", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasSchemaOut]]) def vas_schema_list( db: Session = Depends(get_db) ): schemas = SchemaService.list_all(db) return success(data=schemas) @protected_router.post("/vas/order/create", summary="创建订单", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut]) def vas_order_create( payload: VasOrderCreate, current_user: VasUser = Depends(get_current_user), db: Session = Depends(get_db), redis_client: Redis = Depends(get_redis_client) ): product = ProductService.get(db, payload.product_id) # ① 获取产品绑定的 schema schema = SchemaService.get(db, product.schema_id) # ② 校验 user_inputs validate_user_inputs( schema_json=schema.schema_json, user_inputs=payload.user_inputs, ) created_order = OrderService.create(db, payload, product, current_user, redis_client) if current_user.role == "admin": OrderService.mark_as_admin_paid(db, created_order, current_user) OrderService.create_tasks_for_order(db, created_order) return success(data=created_order) @admin_required_router.post("/vas/order/patch_user_inputs", summary="更新订单的用户信息", tags=["Visafly签证系统"], response_model=ApiResponse[VasOrderOut]) def vas_order_patch_user_inputs( order_id: str, payload: VasOrderPatchUserInputs, db: Session = Depends(get_db), ): order = 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]]) 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: Session = Depends(get_db) ): orders = 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]]) def vas_order_list_all( page: int = Query(0, description="第几页"), size: int = Query(10, description="分页大小"), keyword: str = Query("", description="查询条件"), db: Session = Depends(get_db) ): orders = 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]) def vas_order_cancel( order_id: str, current_user: VasUser = Depends(get_current_user), db: Session = Depends(get_db), redis_client: Redis = Depends(get_redis_client) ): pass @protected_router.get("/vas/payment_provider/list_enabled", summary="获取支付方式", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasPaymentProviderOut]]) def vas_payment_provider_simple_get( db: Session = Depends(get_db) ): providers = PaymentProviderSerivce.list_enabled(db) return success(data=providers) @admin_required_router.get("/vas/payment_provider/list_all", summary="获取支付方式", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasPaymentProviderOut]]) def vas_payment_provider_list_all( db: Session = Depends(get_db) ): providers = PaymentProviderSerivce.list_all(db) return success(data=providers) @admin_required_router.post("/vas/payment_provider/create", summary="新增支付服务提供商", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentProviderOut]) def vas_payment_provider_create( payload: VasPaymentProviderCreate, db: Session = Depends(get_db) ): provider = PaymentProviderSerivce.create(db, payload) return success(data=provider) @admin_required_router.post("/vas/payment_provider/update", summary="更新支付服务提供商", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentProviderOut]) def vas_payment_provider_update( id: int, payload: VasPaymentProviderUpdate, db: Session = Depends(get_db) ): provider = PaymentProviderSerivce.update(db, id, payload) return success(data=provider) @admin_required_router.delete("/vas/payment_provider/delete", summary="删除支付服务提供商", tags=["Visafly签证系统"], response_model=ApiResponse) def vas_payment_provider_delete( id: int, db: Session = Depends(get_db) ): PaymentProviderSerivce.delete(db, id) return success() @protected_router.post("/vas/payment/create", summary="创建支付", tags=["Visafly签证系统"], response_model=ApiResponse[VasPaymentOut]) def vas_payment_create( payload: VasPaymentCreate, db: Session = Depends(get_db) ): rate_table = { "EUR->EUR": "1", "EUR->CNY": "8.3174", "EUR->USD": "1.0842", } res = PaymentService.create_payment(db, payload, rate_table) return success(data=res) @protected_router.get("/vas/payment/list_by_order", summary="获取某个订单下的所有payment记录", tags=["Visafly签证系统"], response_model=ApiResponse[List[VasPaymentOut]]) def vas_payment_list_by_order( order_id: str, db: Session = Depends(get_db) ): payments = 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]) def vas_payment_qr_create(payload: VasPaymentQrCreate, db: Session = Depends(get_db)): qr = 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]]) def vas_payment_qr_list_qrcode_by_provider(provider_id: int, db: Session = Depends(get_db)): qr = 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]) def vas_payment_qr_get_qrcode_by_id(id: int, db: Session = Depends(get_db)): qr = 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]) def vas_payment_qr_update( id: int, payload: VasPaymentQrSetEnableIn, db: Session = Depends(get_db) ): qr = 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) def vas_payment_qr_update( id: int, db: Session = Depends(get_db) ): PaymentQrService.delete(db, id) return success() @admin_required_router.get("/vas/task/list", summary="获取待执行的任务", tags=["Visafly签证系统"], response_model=ApiResponse[PageResponse[VasTaskOut]]) 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: Session = Depends(get_db) ): tasks = 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]) def vas_task_update( id: int, payload: VasTaskUpdate, db: Session = Depends(get_db) ): task = 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]]) def vas_task_pending( order_id: str = Query(..., description="订单编号"), script_version: str = Query("", description="脚本版本, 用来向后兼容"), db: Session = Depends(get_db) ): tasks = 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]) def vas_task_return_to_queue(task_id:int, db: Session = Depends(get_db)): obj = 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]) def vas_task_manual_confirm(task_id:int, db: Session = Depends(get_db)): obj = VasTaskService.manual_confirm(db, task_id) return success(data=obj) @protected_router.post("/vas/ticket/create", summary="创建工单", tags=["Visafly签证系统"], response_model=ApiResponse[VasTicketOut]) def vas_ticket_create( data:VasTicketCreate, current_user: VasUser = Depends(get_current_user), db: Session = Depends(get_db), redis_client: Redis = Depends(get_redis_client) ): obj = 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]]) 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: Session = Depends(get_db) ): tickets = 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]]) def vas_ticket_list_all( page: int = Query(0, description="第几页"), size: int = Query(10, description="分页大小"), keyword: str = Query("", description="查询条件"), db: Session = Depends(get_db) ): tickets = 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]) def update_ticket_status( ticket_id: int, payload: VasTicketStatusUpdate, db: Session = Depends(get_db), user=Depends(get_current_user) ): res = 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]) def create_ticket_message( ticket_id: int, payload: VasTicketMessageCreate, db: Session = Depends(get_db), user=Depends(get_current_user) ): res = 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]]) def get_ticket_messages( ticket_id: int, page: int = 1, size: int = 20, db: Session = Depends(get_db) ): msgs = TicketService.list_messages( db=db, ticket_id=ticket_id, page=page, size=size ) return success(data=msgs)