feat: update

.env

@@ -1,3 +1,3 @@
-DATABASE_URL=mysql://root:GqLLL7Bofj0WaaOpp.0@visafly.top:3306/book_user_info?charset=utf8mb4&timezone=UTC
+DATABASE_URL=mysql://root:GqLLL7Bofj0WaaOpp.0@visafly.top:3306/book_user_info?charset=utf8mb4
 REDIS_URL=redis://:STEs2x6ML0U1HlpE9SojM6YU7QPhqzY8@45.137.220.138:6379/0
 API_TOKEN=7x9EjFpmv7GjZc6AfVeqxuUBANpqkpkHAtxJM7CAW5oZhs0nEyCJBy39N4XXs5hgfYWXw3jFrcgXqQ42HAx9Qvwtk9vC2GvKBbWz

app/api/router.py

@@ -1,24 +1,55 @@
 import time
+import requests
 from typing import List
-from fastapi import APIRouter, Query, Depends
+from app.core.logger import logger
+from fastapi import APIRouter, Query, Depends, Body, UploadFile, File, HTTPException
+from fastapi.responses import RedirectResponse
+from sqlalchemy.orm import Session
 from app.core.redis import get_redis_client
+from app.core.database import get_db
 from redis.asyncio import Redis
 from app.schemas.user import UserOut
 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
+from app.schemas.auto_booking import AutoBookingCreate, AutoBookingOut
+from app.schemas.http_session import HttpSessionCreate, HttpSessionUpdate,HttpSessionOut
+from app.schemas.visafly_config import VisaflyConfigCreate, VisaflyConfigOut
+from app.schemas.slot import SlotCreate, SlotOut
+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.visafly_config_service import VisaflyConfigService
+from app.services.slot_service import SlotService
 
 # 公共路由
-public_router = APIRouter(tags=["public"])
+public_router = APIRouter()
 # 受保护路由
-protected_router = APIRouter(tags=["protected"])
+protected_router = APIRouter()
 
-@public_router.get("/ping", summary="心跳检测")
+@public_router.get("/ping", summary="心跳检测", tags=["测试接口"])
 def ping():
     return {"message": "pong"}
 
-@public_router.get("/sms/upload", summary="上报短信", response_model=ShortMessageDetail)
+@protected_router.get("/users", summary="查询用户", tags=["通用接口"], response_model=List[UserOut])
+def get_users():
+    return [
+        {"id": 1, "name": "Alice"},
+        {"id": 2, "name": "Bob"}
+    ]
+
+@public_router.get("/sms/upload", summary="上报短信", tags=["短信接口"], response_model=ShortMessageDetail)
 def sms_upload(
     phone: str = Query(..., description="手机号"),
     message: str = Query(..., description="短信内容"),
@@ -32,20 +63,7 @@ def sms_upload(
     msg = save_short_message(redis_client, phone, message, received_at, max_ttl)
     return msg
 
-@protected_router.get("/users", summary="查询用户", response_model=List[UserOut])
-def get_users():
-    return [
-        {"id": 1, "name": "Alice"},
-        {"id": 2, "name": "Bob"}
-    ]
-
-@protected_router.get("/troov/rate", summary="TROOV 查询rate", response_model=List[TroovRate])
-def troov_rate(date: str = Query(..., description="查询的日期, 格式: YYYY-MM-DD"),
-               redis_client: Redis = Depends(get_redis_client)):
-    # 调用 service 层获取数据
-    return get_rate_by_date(redis_client, date)
-
-@protected_router.get("/sms/download", summary="读取短信", response_model=List[ShortMessageDetail])
+@protected_router.get("/sms/download", summary="读取短信", tags=["短信接口"], response_model=List[ShortMessageDetail])
 def sms_download(
     phone: str = Query(..., description="手机号"),
     keyword: str = Query('', description="短信内容关键字"),
@@ -56,4 +74,641 @@ def sms_download(
     查询短信（支持关键字和时间过滤）
     """
     results = query_short_message(redis_client, phone, keyword or None, sent_at or None)
-    return results
+    return results
+
+@protected_router.get("/troov/rate", summary="TROOV 查询rate", tags=["通用接口"], response_model=List[TroovRate])
+def troov_rate(date: str = Query(..., description="查询的日期, 格式: YYYY-MM-DD"),
+               redis_client: Redis = Depends(get_redis_client)):
+    # 调用 service 层获取数据
+    return get_rate_by_date(redis_client, date)
+
+@protected_router.post("/dynamic-configurations", summary="创建动态配置", tags=["动态配置"], response_model=ConfigurationOut)
+def dynamic_config_create(config_in: ConfigurationCreate, db: Session = Depends(get_db)):
+    existing = ConfigurationService.get_by_key(db, config_in.config_key)
+    if existing:
+        raise HTTPException(status_code=400, detail=f"配置 Key '{config_in.config_key}' 已存在")
+    return ConfigurationService.create(db, config_in)
+
+
+@protected_router.get("/dynamic-configurations/all", summary="读取所有动态配置", tags=["动态配置"], response_model=List[ConfigurationOut])
+def dynamic_config_get_all(db: Session = Depends(get_db)):
+    return ConfigurationService.get_all(db)
+
+
+@protected_router.get("/dynamic-configurations/key/{config_key}", summary="根据Key读取动态配置", tags=["动态配置"], response_model=ConfigurationOut)
+def dynamic_config_get_by_key(config_key: str, db: Session = Depends(get_db)):
+    config = ConfigurationService.get_by_key(db, config_key)
+    if not config:
+        raise HTTPException(status_code=404, detail=f"配置 Key '{config_key}' 不存在")
+    return config
+
+
+@protected_router.put("/dynamic-configurations/key/{config_key}", summary="根据Key更新动态配置", tags=["动态配置"], response_model=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)
+    if not config:
+        raise HTTPException(status_code=404, detail=f"配置 Key '{config_key}' 不存在")
+    return config
+
+
+@protected_router.delete("/dynamic-configurations/key/{config_key}", summary="根据Key删除动态配置", tags=["动态配置"], response_model=ConfigurationOut)
+def dynamic_config_delete_by_key(config_key: str, db: Session = Depends(get_db)):
+    config = ConfigurationService.delete_by_key(db, config_key)
+    if not config:
+        raise HTTPException(status_code=404, detail=f"配置 Key '{config_key}' 不存在")
+    return config
+
+@protected_router.post(
+    "/http-session",
+    summary="创建http session",
+    tags=["会话管理"],
+    response_model=HttpSessionOut
+)
+def http_session_create(
+    data: HttpSessionCreate,
+    db: Session = Depends(get_db)
+):
+    logger.info(f"[Create HttpSession] sid={data.session_id}")
+    return HttpSessionService.create(db, data)
+
+
+@protected_router.delete(
+    "/http-session",
+    summary="删除http session",
+    tags=["会话管理"]
+)
+def http_session_delete_by_sid(
+    session_id: str = Query(...),
+    db: Session = Depends(get_db)
+):
+    logger.info(f"[Delete HttpSession] sid={session_id}")
+
+    ok = HttpSessionService.delete_by_sid(db, session_id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="session 不存在")
+
+    return {"success": True, "session_id": session_id}
+
+
+@protected_router.put(
+    "/http-session",
+    summary="更新http session",
+    tags=["会话管理"],
+    response_model=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)
+    if not obj:
+        raise HTTPException(status_code=404, detail="session 不存在")
+
+    return obj
+
+@protected_router.get(
+    "/http-session",
+    summary="读取http session",
+    tags=["会话管理"],
+    response_model=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)
+    if not obj:
+        raise HTTPException(status_code=404, detail="session 不存在")
+
+    return obj
+
+
+@protected_router.get("/email-authorizations", summary="查询所有内部邮箱", tags=["邮箱接口"], response_model=List[EmailAuthorizationOut])
+def email_authorizations_get(db: Session = Depends(get_db)):
+    return EmailAuthorizationService.get_all(db)
+
+
+@protected_router.post("/email-authorizations", summary="创建内部邮箱", tags=["邮箱接口"], response_model=EmailAuthorizationOut)
+def email_authorizations_create(data: EmailAuthorizationCreate, db: Session = Depends(get_db)):
+    existing = EmailAuthorizationService.get_by_email(db, data.email)
+    if existing:
+        raise HTTPException(status_code=400, detail=f"邮箱 {data.email} 已存在")
+    return EmailAuthorizationService.create(db, data)
+
+@protected_router.get("/email-authorizations/{id}", summary="通过id查询内部邮箱", tags=["邮箱接口"], response_model=EmailAuthorizationOut)
+def email_authorizations_get_by_id(id: int, db: Session = Depends(get_db)):
+    email_auth = EmailAuthorizationService.get_by_id(db, id)
+    if not email_auth:
+        raise HTTPException(status_code=404, detail=f"ID={id} 的邮箱记录不存在")
+    return email_auth
+
+
+@protected_router.put("/email-authorizations/{id}", summary="通过id更新内部邮箱", tags=["邮箱接口"], response_model=EmailAuthorizationOut)
+def email_authorizations_update_by_id(id: int, data: EmailAuthorizationUpdate, db: Session = Depends(get_db)):
+    updated = EmailAuthorizationService.update(db, id, data)
+    if not updated:
+        raise HTTPException(status_code=404, detail=f"ID={id} 的邮箱记录不存在")
+    return updated
+
+
+@protected_router.delete("/email-authorizations/{id}", summary="通过id删除内部邮箱", tags=["邮箱接口"], response_model=EmailAuthorizationOut)
+def email_authorizations_delete_by_id(id: int, db: Session = Depends(get_db)):
+    deleted = EmailAuthorizationService.delete(db, id)
+    if not deleted:
+        raise HTTPException(status_code=404, detail=f"ID={id} 的邮箱记录不存在")
+    return deleted
+
+
+@protected_router.get("/email-authorizations/email/{email}", summary="通过邮箱地址查询内部邮箱", tags=["邮箱接口"], response_model=EmailAuthorizationOut)
+def email_authorizations_get_by_email(email: str, db: Session = Depends(get_db)):
+    email_auth = EmailAuthorizationService.get_by_email(db, email)
+    if not email_auth:
+        raise HTTPException(status_code=404, detail=f"邮箱 {email} 不存在")
+    return email_auth
+
+@protected_router.post("/email-authorizations/fetch", summary="读取邮件, 仅限文本内容", tags=["邮箱接口"])
+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)
+    if not auth:
+        raise HTTPException(status_code=404, detail=f"未找到邮箱授权记录: {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
+    )
+    if result is None:
+        raise HTTPException(status_code=404, detail="在有效期内未找到匹配邮件")
+    return {"body": result}
+
+@protected_router.post("/email-authorizations/fetch-top", summary="从最近的几封邮件读取目标邮件,仅限文本内容, 性能会更好, 邮件多时有可能漏读", tags=["邮箱接口"])
+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)
+    if not auth:
+        raise HTTPException(status_code=404, detail=f"未找到邮箱授权记录: {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
+    )
+    if result is None:
+        raise HTTPException(status_code=404, detail=f"未在前{top}封邮件中查找到匹配邮件")
+    return {"body": result}
+
+@protected_router.post("/email-authorizations/forward", summary="转发邮件", tags=["邮箱接口"])
+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)
+    if not auth:
+        raise HTTPException(status_code=404, detail=f"未找到邮箱授权记录: {emailAccount}")
+    result = EmailAuthorizationService.forward_first_matching_email(
+        auth,
+        forward_to = forwardTo,
+        sender = sender,
+        recipient = recipient,
+        subject_keywords = subjectKeywords,
+        body_keywords = bodyKeywords
+    )
+    if result is None:
+        raise HTTPException(status_code=404, detail=f"未找可转发的邮件")
+    return {"body": result}
+
+@protected_router.post("/email-authorizations/sendmail", summary="发送邮件", tags=["邮箱接口"])
+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)
+    if not auth:
+        raise HTTPException(status_code=404, detail=f"未找到邮箱授权记录: {emailAccount}")
+    result = EmailAuthorizationService.send_email(
+        auth,
+        send_to = sendTo,
+        subject = subject,
+        content_type = contentType,
+        content = content.body
+    )
+    if result is None:
+        raise HTTPException(status_code=404, detail=f"邮件发送失败")
+    return {"body": result}
+
+@protected_router.post("/email-authorizations/sendmail-bulk", summary="群发送邮件", tags=["邮箱接口"])
+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)
+    if not auth:
+        raise HTTPException(status_code=404, detail=f"未找到邮箱授权记录: {emailAccount}")
+    result = EmailAuthorizationService.send_email_bulk(
+        auth,
+        send_to = sendTo,
+        subject = subject,
+        content_type = contentType,
+        content = content.body
+    )
+    if result is None:
+        raise HTTPException(status_code=404, detail=f"邮件发送失败")
+    return {"body": result}
+
+
+@protected_router.post("/resource/pdf", summary="上传pdf文件", tags=["文件管理"])
+def resource_upload_pdf(pdf: UploadFile = File(...)):
+    if not pdf.filename.lower().endswith(".pdf"):
+        raise HTTPException(status_code=400, detail="仅支持上传 PDF 文件")
+
+    result = SeaweedFSService.upload(pdf)
+    if not result:
+        raise HTTPException(status_code=500, detail="上传失败")
+    return {"success": True, "fid": result["fid"], "url": result["url"]}
+
+@protected_router.get("/resource/pdf/{fid}", summary="读取pdf", tags=["文件管理"])
+def resource_get_pdf(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)
+
+@protected_router.delete("/resource/pdf/{fid}", summary="删除pdf文件", tags=["文件管理"])
+def resource_delete_pdf(fid: str):
+    ok = SeaweedFSService.delete(fid)
+    if not ok:
+        raise HTTPException(status_code=404, detail="文件不存在或删除失败")
+    return {"success": True, "fid": fid}
+
+@protected_router.post("/resource/image", summary="上传图片", tags=["文件管理"])
+def resource_upload_image(image: UploadFile = File(...)):
+    if not image.content_type.startswith("image/"):
+        raise HTTPException(status_code=400, detail="仅支持上传图片文件")
+    print('upload')
+    result = SeaweedFSService.upload(image)
+    if not result:
+        raise HTTPException(status_code=500, detail="上传失败")
+    return {"success": True, "fid": result["fid"], "url": result["url"]}
+
+@protected_router.get("/resource/image/{fid}", summary="读取图片", tags=["文件管理"])
+def resource_get_image(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)
+
+@protected_router.delete("/resource/image/{fid}", summary="删除图片", tags=["文件管理"])
+def resource_delete_image(fid: str):
+    ok = SeaweedFSService.delete(fid)
+    if not ok:
+        raise HTTPException(status_code=404, detail="图片不存在或删除失败")
+    return {"success": True, "fid": fid}
+
+@protected_router.post("/s/generate", summary="生成短链接地址<压缩地址长度>", tags=["Short URL"])
+def short_url_generate(
+    data: ShortUrlCreate,
+    db: Session = Depends(get_db),
+):
+    """生成短链接"""
+    record = ShortUrlService.create_short_url(db, data.longUrl)
+    return {
+        "short_key": record.short_key,
+        "short_url": f"/s/{record.short_key}",
+        "long_url": record.long_url,
+        "created_at": record.created_at,
+    }
+
+
+@public_router.get("/s/{shortKey}", summary="访问短链接地址<自动重定向到真实链接地址>", tags=["Short URL"])
+def short_url_request(shortKey: str, db: Session = Depends(get_db)):
+    """访问短链接自动重定向"""
+    long_url = ShortUrlService.get_long_url(db, shortKey)
+    if not long_url:
+        raise HTTPException(status_code=404, detail="短链接不存在或已失效")
+    return RedirectResponse(url=long_url, status_code=302)
+
+@protected_router.post("/tasks", summary="创建任务", tags=["任务管理接口"], response_model=TaskOut)
+def task_create(data: TaskCreate, db: Session = Depends(get_db)):
+    """创建任务"""
+    return TaskService.create(db, data)
+
+@protected_router.get("/tasks/{taskId:int}", summary="根据taskId读取任务状态", tags=["任务管理接口"], response_model=TaskOut)
+def task_get_by_id(taskId: int, db: Session = Depends(get_db)):
+    """获取任务"""
+    task = TaskService.get_by_id(db, taskId)
+    if not task:
+        raise HTTPException(status_code=404, detail=f"任务 {taskId} 不存在")
+    return task
+
+@protected_router.get("/tasks/pending", summary="获取等待执行的任务", tags=["任务管理接口"], response_model=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),
+):
+    """分页获取等待执行的任务"""
+    return TaskService.get_pending(db, command, page, size)
+
+@protected_router.put("/tasks/{taskId}", summary="根据taskId更新任务状态", tags=["任务管理接口"], response_model=TaskOut)
+def task_update_by_id(taskId: int, data: TaskUpdate, db: Session = Depends(get_db)):
+    """更新任务状态或结果"""
+    updated = TaskService.update(db, taskId, data)
+    if not updated:
+        raise HTTPException(status_code=404, detail=f"任务 {taskId} 不存在")
+    return updated
+
+@protected_router.post("/tg/send_message", summary="推送电报消息", tags=["消息推送接口"])
+def tg_send_message(
+    apiToken: str = Query(..., description="电报的APITOKEN"),
+    chatID: str = Query(..., description="电报群ID"),
+    message: str = Query(..., description="推送的文本信息")
+):
+    url = f"https://api.telegram.org/bot{apiToken}/sendMessage"
+    payload = {
+        "chat_id": chatID,
+        "text": message,
+        "parse_mode": "HTML"
+    }
+
+    try:
+        response = requests.post(url, json=payload, timeout=10)
+        if response.status_code != 200:
+            # logger.error(f"Telegram 推送失败: {response.text}")
+            raise HTTPException(status_code=500, detail=f"Telegram 推送失败: {response.text}")
+        return {"success": True, "detail": "Telegram 消息推送成功"}
+    except Exception as e:
+        # logger.exception("Telegram 发送消息异常")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@protected_router.post("/tg/send_image", summary="推送电报图片", tags=["消息推送接口"])
+def tg_send_image(
+    apiToken: str = Query(..., description="电报的APITOKEN"),
+    chatID: str = Query(..., description="电报群ID"),
+    message: str = Query("", description="推送的文本信息"),
+    image: UploadFile = File(..., description="推送的图像文件")
+):
+    url = f"https://api.telegram.org/bot{apiToken}/sendPhoto"
+
+    files = {"photo": (image.filename, image.file, image.content_type)}
+    data = {"chat_id": chatID, "caption": message}
+
+    try:
+        response = requests.post(url, data=data, files=files, timeout=15)
+        if response.status_code != 200:
+            # logger.error(f"Telegram 图片推送失败: {response.text}")
+            raise HTTPException(status_code=500, detail=f"Telegram 图片推送失败: {response.text}")
+        return {"success": True, "detail": "Telegram 图片推送成功"}
+    except Exception as e:
+        # logger.exception("Telegram 发送图片异常")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@protected_router.post("/wechat/send", summary="推送微信消息", tags=["消息推送接口"])
+def wechat_send(
+    apikey: str = Query(..., description="企业微信的APITOKEN"),
+    message: str = Query(..., description="推送的文本信息")
+):
+    """
+    企业微信 WebHook 格式：
+    https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY
+    """
+    url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={apikey}"
+    payload = {"msgtype": "text", "text": {"content": message}}
+
+    try:
+        response = requests.post(url, json=payload, timeout=10)
+        data = response.json()
+
+        if response.status_code != 200 or data.get("errcode") != 0:
+            # logger.error(f"企业微信推送失败: {response.text}")
+            raise HTTPException(status_code=500, detail=f"企业微信推送失败: {response.text}")
+
+        return {"success": True, "detail": "企业微信消息推送成功"}
+    except Exception as e:
+        # logger.exception("企业微信推送异常")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@protected_router.post("/cards/publish", summary="创建新的消息卡片", tags=["信息卡片接口"], response_model=CardOut)
+def cards_publish(
+    data: CardCreate = Body(...),
+    db: Session = Depends(get_db)
+):
+    return CardService.create(db, data)
+
+
+@public_router.get("/cards/view", summary="分页读取全部卡片, 可选择语言", tags=["信息卡片接口"], response_model=List[CardOut])
+def cards_view_paginated(
+    page: int = Query(0, description="第几页"),
+    size: int = Query(10, description="分页大小"),
+    culture: str = Query("english", description="语言, 可设置 chinese, english"),
+    db: Session = Depends(get_db)
+):
+    return CardService.get_paginated(db, page, size, culture)
+
+
+@public_router.get("/cards/view2", summary="根据关键词分页查询卡片, 可选择语言", tags=["信息卡片接口"], response_model=List[CardOut])
+def cards_view_paginated2(
+    keywords: 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)
+):
+    keyword_list = [k.strip() for k in keywords.split(",") if k.strip()]
+    return CardService.get_by_keywords(db, keyword_list, page, size, culture)
+
+@protected_router.post("/autobooking", summary="创建自动预定订单", tags=["自动预定订单管理接口"], response_model=AutoBookingOut)
+def autobooking_create(data: AutoBookingCreate, db: Session = Depends(get_db)):
+    return AutoBookingService.create(db, data)
+
+
+@protected_router.post("/autobooking/create-by-ai", summary="用自然语言创建自动预定订单(底层使用chatgpt)", tags=["自动预定订单管理接口"])
+def autobooking_create_by_ai():
+    # TODO: 这里可以对接 GPT 解析自然语言生成结构化 AutoBooking 数据
+    return {"message": "AI 自动创建暂未实现"}
+
+
+@protected_router.post("/autobooking/batch", summary="批量查询多个自动预定订单信息", tags=["自动预定订单管理接口"])
+def autobooking_get_by_ids(ids: List[int] = Body(...), db: Session = Depends(get_db)):
+    return AutoBookingService.batch_get_by_ids(db, ids)
+
+
+@protected_router.get("/autobooking", summary="分页查询所有的自动预定订单信息", tags=["自动预定订单管理接口"], response_model=List[AutoBookingOut])
+def autobooking_get_paginated(
+    tech_provider: str = Query("", description="签证网站技术提供商"),
+    keyword: str = Query("", description="关键词查询"),
+    page: int = Query(0, description="第几页"),
+    size: int = Query(10, description="分页大小"),
+    db: Session = Depends(get_db)
+):
+    return AutoBookingService.get_paginated(db, tech_provider, keyword, page, size)
+
+
+@protected_router.get("/autobooking/{id:int}", summary="根据id查询自动预定订单详情", tags=["自动预定订单管理接口"], response_model=AutoBookingOut)
+def autobooking_get_by_id(id: int, db: Session = Depends(get_db)):
+    result = AutoBookingService.get_by_id(db, id)
+    if not result:
+        raise HTTPException(status_code=404, detail="未找到订单")
+    return result
+
+
+@protected_router.delete("/autobooking/{id:int}", summary="根据id删除自动预定订单", tags=["自动预定订单管理接口"])
+def autobooking_delete_by_id(id: int, db: Session = Depends(get_db)):
+    ok = AutoBookingService.delete_by_id(db, id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="删除失败或记录不存在")
+    return {"success": True, "id": id}
+
+
+@protected_router.put("/autobooking/{id:int}", summary="根据id更新自动预定订单信息", tags=["自动预定订单管理接口"], response_model=AutoBookingOut)
+def autobooking_update_by_id(id: int, updated_order_info: dict = Body(...), db: Session = Depends(get_db)):
+    result = AutoBookingService.update_by_id(db, id, updated_order_info)
+    if not result:
+        raise HTTPException(status_code=404, detail="更新失败或记录不存在")
+    return result
+
+
+@protected_router.get("/autobooking/statistics", summary="统计自动预定订单信息", tags=["自动预定订单管理接口"])
+def autobooking_statistics(
+    tech_provider: str = Query("", description="签证网站技术提供商"),
+    db: Session = Depends(get_db)
+):
+    return AutoBookingService.statistics(db, tech_provider)
+
+
+@protected_router.get("/autobooking/pending", summary="获取未处理的自动预定订单信息列表", tags=["自动预定订单管理接口"], response_model=List[AutoBookingOut])
+def autobooking_pending(
+    tech_provider: str = Query("", description="签证网站技术提供商"),
+    db: Session = Depends(get_db)
+):
+    return AutoBookingService.get_pending(db, tech_provider)
+
+@protected_router.get("/autobooking/trigger-finish", summary="触发自动预定订单完成操作", tags=["自动预定订单管理接口"])
+def autobooking_trigger_finish(id: int):
+    pass
+
+@protected_router.post("/stripe-price/create", summary="创建stripe 商品", tags=["Stripe 操作接口"])
+def stripe_price_create(
+    stripePrice: str = Body(..., description="stripe 商品价格信息")
+):
+    pass
+
+@protected_router.post(
+    "/slot/report",
+    summary="上报 slot 记录",
+    tags=["SLOT接口"],
+    response_model=SlotOut
+)
+def slot_report(data: SlotCreate = Body(...), db: Session = Depends(get_db)):
+    logger.info(f"[Slot Report] {data}")
+    return SlotService.report(db, data)
+
+
+@protected_router.get(
+    "/slot/search",
+    summary="查询 slot 记录",
+    tags=["SLOT接口"],
+    response_model=SlotOut
+)
+def slot_search(
+    submit_city: str = Query(..., description="提交城市"),
+    travel_country: str = Query(..., description="旅行国家"),
+    visa_type: str = Query(..., description="签证类别"),
+    date_type: str = Query("latest", description="查询方式, latest 最近一条上报的信息, earliest 最早的日期"),
+    db: Session = Depends(get_db)
+):
+    result = SlotService.search(db, submit_city, travel_country, visa_type, date_type)
+    if not result:
+        raise HTTPException(status_code=404, detail="未找到相关记录")
+    return result
+
+
+@protected_router.post(
+    "/visafly-config",
+    summary="创建一条可以被前端查询到的签证类别",
+    tags=["visafly-config接口"],
+    response_model=VisaflyConfigOut
+)
+def visafly_config_create(
+    visafly_config: VisaflyConfigCreate = Body(...),
+    db: Session = Depends(get_db)
+):
+    logger.info(f"[VisaflyConfig Create] {visa_slot_queries}")
+    return VisaflyConfigService.create(db, visa_slot_queries)
+
+
+@protected_router.get(
+    "/visafly-config/submission-countries",
+    summary="查询支持从哪些国家递交申请",
+    tags=["visafly-config接口"]
+)
+def visafly_config_get_submission_countries(db: Session = Depends(get_db)):
+    return VisaflyConfigService.get_submission_countries(db)
+
+
+@protected_router.get(
+    "/visafly-config/cities",
+    summary="查询支持从哪个国家的哪些城市递交申请",
+    tags=["visafly-config接口"]
+)
+def visafly_config_get_cities_by_country_code(
+    country_code: str = Query(..., description="递交申请的国家编号,大写的两个英文字符"),
+    db: Session = Depends(get_db)
+):
+    return VisaflyConfigService.get_cities_by_country(db, country_code)
+
+
+@protected_router.get(
+    "/visafly-config/travel-countries-with-categories",
+    summary="查询某个城市可以办理哪些国家的签证(包含签证类别)",
+    tags=["visafly-config接口"]
+)
+def visafly_config_get_travel_countries_by_city_code(
+    city_code: str = Query(..., description="递交申请的城市编号,大写的三个英文字符"),
+    db: Session = Depends(get_db)
+):
+    return VisaflyConfigService.get_travel_countries_by_city(db, city_code)

app/core/database.py

@@ -2,14 +2,39 @@ from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker, declarative_base
 from app.core.config import settings
 
-engine = create_engine(settings.database_url, echo=settings.debug)
-SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+# =========================
+# 数据库初始化
+# =========================
+# 建立 Engine
+engine = create_engine(
+    settings.database_url,
+    echo=settings.debug,        # 是否打印 SQL 日志
+    pool_pre_ping=True,         # 检测断开的连接
+    pool_recycle=1800,          # 连接回收时间，防止 MySQL 8 小时断开
+    future=True                 # 启用 SQLAlchemy 2.0 风格
+)
+
+# 建立 Session 工厂
+SessionLocal = sessionmaker(
+    autocommit=False,
+    autoflush=False,
+    bind=engine,
+    expire_on_commit=False
+)
+
+# ORM 基类
 Base = declarative_base()
 
-# 依赖注入
+
+# =========================
+# 数据库依赖
+# =========================
 def get_db():
+    """
+    FastAPI 的依赖注入函数，用于在请求周期内创建并关闭数据库会话。
+    """
     db = SessionLocal()
     try:
         yield db
     finally:
-        db.close()
+        db.close()

app/core/logger.py

@@ -0,0 +1,47 @@
+import os
+import logging
+from logging.handlers import TimedRotatingFileHandler
+
+
+# ===============================
+# 基本配置
+# ===============================
+LOG_DIR = os.path.join(os.path.dirname(__file__), "../../logs")
+os.makedirs(LOG_DIR, exist_ok=True)
+
+LOG_FILE = os.path.join(LOG_DIR, "app.log")
+
+
+# ===============================
+# 日志格式
+# ===============================
+LOG_FORMAT = "[%(asctime)s] [%(levelname)s] [%(name)s] [%(filename)s:%(lineno)d] - %(message)s"
+DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
+
+
+# ===============================
+# 主 logger 实例
+# ===============================
+logger = logging.getLogger("app")
+logger.setLevel(logging.INFO)  # 可改为 DEBUG
+
+# 防止重复添加 handler
+if not logger.handlers:
+    # 控制台输出
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.INFO)
+    console_handler.setFormatter(logging.Formatter(LOG_FORMAT, DATE_FORMAT))
+
+    # 文件输出（每天切割）
+    file_handler = TimedRotatingFileHandler(
+        LOG_FILE, when="midnight", interval=1, backupCount=7, encoding="utf-8"
+    )
+    file_handler.setLevel(logging.INFO)
+    file_handler.setFormatter(logging.Formatter(LOG_FORMAT, DATE_FORMAT))
+
+    # 添加到 logger
+    logger.addHandler(console_handler)
+    logger.addHandler(file_handler)
+
+    # 避免重复日志
+    logger.propagate = False

app/main.py

@@ -23,13 +23,16 @@ app.add_middleware(
 # 路由注册
 # -----------------------
 # 公共路由，不鉴权
-app.include_router(router.public_router, prefix="/api")
+app.include_router(
+    router.public_router,
+    prefix="/api"
+)
 
 # 需要鉴权的路由
 app.include_router(
     router.protected_router,
     prefix="/api",
-    dependencies=[Depends(verify_token)]
+    #dependencies=[Depends(verify_token)]
 )
 
 # -----------------------

app/models/auto_booking.py

@@ -0,0 +1,46 @@
+from sqlalchemy import Column, BigInteger, String, Integer, Text, Date, DateTime
+from app.core.database import Base
+
+class AutoBooking(Base):
+    __tablename__ = "auto_booking"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    provider = Column(String(100))
+    visa_center = Column(String(100))
+    order_no = Column(String(100))
+    social_account = Column(String(100))
+    account = Column(String(100))
+    password = Column(String(100))
+    last_name = Column(String(100))
+    first_name = Column(String(100))
+    gender = Column(String(10))
+    birthday = Column(Date)
+    email = Column(String(150))
+    alias_email = Column(String(150))
+    phone_country_code = Column(String(20))
+    phone_no = Column(String(50))
+    passport_no = Column(String(50))
+    nationality = Column(String(50))
+    passport_expiry_date = Column(Date)
+    address_line1 = Column(Text)
+    address_line2 = Column(Text)
+    state = Column(String(100))
+    city = Column(String(100))
+    postcode = Column(String(100))
+    travel_date = Column(Date)
+    cover_letter = Column(String(100))
+    passport_image_url = Column(Text)
+    selfie_image_url = Column(Text)
+    application_form_url = Column(Text)
+    priority = Column(Integer, default=0)
+    expected_submit_start = Column(Date)
+    expected_submit_end = Column(Date)
+    rules = Column(Text)
+    status = Column(Integer, default=0)
+    placeholder = Column(Integer, default=0)
+    appointment_datetime = Column(DateTime)
+    appointment_letter_url = Column(Text)
+    pnr_number = Column(String(100))
+    payment_link = Column(Text)
+    payment_help = Column(Integer, default=0)
+    note = Column(Text)

app/models/card.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Column, BigInteger, String, Text, TIMESTAMP, func
+from app.core.database import Base
+
+
+class Card(Base):
+    __tablename__ = "card"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    image = Column(String(255), nullable=False)
+    title = Column(String(255), nullable=False)
+    content = Column(Text, nullable=False)
+    label = Column(String(255), nullable=True)
+    country = Column(String(255), nullable=True)
+    additional_info = Column(String(255), nullable=True)
+    culture = Column(String(100), default="english")
+    created_at = Column(TIMESTAMP, server_default=func.now())
+    updated_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now())

app/models/configuration.py

@@ -0,0 +1,14 @@
+from sqlalchemy import Column, Integer, String, Text, TIMESTAMP, func
+from app.core.database import Base
+
+
+class Configuration(Base):
+    __tablename__ = "configuration"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    config_key = Column(String(255), unique=True, nullable=False, comment="配置键")
+    config_value = Column(Text, nullable=False, comment="配置值")
+    description = Column(Text, comment="描述")
+    type = Column(String(50), comment="类型")
+    created_at = Column(TIMESTAMP, server_default=func.now(), comment="创建时间")
+    updated_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now(), comment="更新时间")

app/models/email_authorizations.py

@@ -0,0 +1,20 @@
+from sqlalchemy import Column, Integer, String, TIMESTAMP, func
+from app.core.database import Base
+
+
+class EmailAuthorization(Base):
+    __tablename__ = "email_authorizations"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    email = Column(String(255), unique=True, nullable=False, comment="邮箱地址")
+    authorization_code = Column(String(255), nullable=False, comment="授权码")
+    imap_server = Column(String(255), nullable=True, comment="IMAP服务器地址")
+    imap_port = Column(Integer, default=993, comment="IMAP端口")
+    smtp_server = Column(String(255), nullable=True, comment="SMTP服务器地址")
+    smtp_port = Column(Integer, default=465, comment="SMTP端口")
+    proxy_host = Column(String(255), nullable=True, comment="代理主机")
+    proxy_port = Column(Integer, default=0, comment="代理端口")
+    proxy_username = Column(String(255), nullable=True, comment="代理用户名")
+    proxy_password = Column(String(255), nullable=True, comment="代理密码")
+    created_at = Column(TIMESTAMP, server_default=func.now(), comment="创建时间")
+    updated_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now(), comment="更新时间")

app/models/http_session.py

@@ -0,0 +1,14 @@
+from sqlalchemy import Column, String, Text, DateTime
+from app.core.database import Base
+from datetime import datetime
+
+class HttpSession(Base):
+    __tablename__ = "http_session"
+
+    session_id = Column(String(128), primary_key=True)
+    local_storage = Column(Text)
+    cookies = Column(Text)
+    user_agent = Column(String(512))
+    proxy = Column(String(256))
+    page = Column(Text)
+    create_at = Column(DateTime, nullable=False, default=datetime.utcnow)

app/models/short_url.py

@@ -0,0 +1,11 @@
+from sqlalchemy import Column, BigInteger, String, Text, TIMESTAMP, func
+from app.core.database import Base
+
+
+class ShortUrl(Base):
+    __tablename__ = "short_url"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    short_key = Column(String(10), unique=True, nullable=False, comment="短链接Key")
+    long_url = Column(Text, nullable=False, comment="原始长链接")
+    created_at = Column(TIMESTAMP, server_default=func.now(), comment="创建时间")

app/models/slot.py

@@ -0,0 +1,20 @@
+from sqlalchemy import Column, Integer, String, Date, DateTime, Text
+from app.core.database import Base
+from datetime import datetime
+
+class Slot(Base):
+    __tablename__ = "visa_slot"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    submit_country = Column(String(100), nullable=False)
+    submit_city = Column(String(100))
+    travel_country = Column(String(100), nullable=False)
+    visa_type = Column(String(100), nullable=False)
+    lasted_slot_date = Column(Date)
+    available_dates = Column(Text)
+    available_times = Column(Text)
+    slot_number = Column(Integer)
+    website = Column(String(255))
+    open_mode = Column(Integer, default=0)
+    create_at = Column(DateTime, default=datetime.utcnow)
+    update_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

app/models/task.py

@@ -0,0 +1,14 @@
+from sqlalchemy import Column, Integer, String, Text, TIMESTAMP, func, SmallInteger
+from app.core.database import Base
+
+
+class Task(Base):
+    __tablename__ = "task"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    command = Column(String(255), nullable=False, comment="任务类型/命令")
+    args = Column(Text, nullable=False, comment="任务参数(JSON字符串)")
+    result = Column(Text, nullable=True, comment="任务执行结果(JSON字符串)")
+    status = Column(SmallInteger, default=0, comment="任务状态：0待执行 1执行中 2完成 3失败")
+    create_at = Column(TIMESTAMP, server_default=func.now(), comment="创建时间")
+    update_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now(), comment="更新时间")

app/models/visafly_config.py

@@ -0,0 +1,16 @@
+from sqlalchemy import Column, Integer, String
+from app.core.database import Base
+
+class VisaflyConfig(Base):
+    __tablename__ = "visa_slot_queries"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    submission_country = Column(String(100), nullable=False)
+    submission_country_code = Column(String(10), nullable=False)
+    submission_city = Column(String(100), nullable=False)
+    submission_city_code = Column(String(10), nullable=False)
+    travel_country = Column(String(100), nullable=False)
+    travel_country_code = Column(String(10), nullable=False)
+    visa_type = Column(String(100), nullable=False)
+    provider = Column(String(100), nullable=True)
+    field_requirement_type = Column(String(100), nullable=True)

app/schemas/auto_booking.py

@@ -0,0 +1,52 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import date, datetime
+
+class AutoBookingBase(BaseModel):
+    provider: Optional[str] = None
+    visa_center: Optional[str] = None
+    order_no: Optional[str] = None
+    social_account: Optional[str] = None
+    account: Optional[str] = None
+    password: Optional[str] = None
+    last_name: Optional[str] = None
+    first_name: Optional[str] = None
+    gender: Optional[str] = None
+    birthday: Optional[date] = None
+    email: Optional[str] = None
+    alias_email: Optional[str] = None
+    phone_country_code: Optional[str] = None
+    phone_no: Optional[str] = None
+    passport_no: Optional[str] = None
+    nationality: Optional[str] = None
+    passport_expiry_date: Optional[date] = None
+    address_line1: Optional[str] = None
+    address_line2: Optional[str] = None
+    state: Optional[str] = None
+    city: Optional[str] = None
+    postcode: Optional[str] = None
+    travel_date: Optional[date] = None
+    cover_letter: Optional[str] = None
+    passport_image_url: Optional[str] = None
+    selfie_image_url: Optional[str] = None
+    application_form_url: Optional[str] = None
+    priority: Optional[int] = None
+    expected_submit_start: Optional[date] = None
+    expected_submit_end: Optional[date] = None
+    rules: Optional[str] = None
+    status: Optional[int] = None
+    placeholder: Optional[int] = None
+    appointment_datetime: Optional[datetime] = None
+    appointment_letter_url: Optional[str] = None
+    pnr_number: Optional[str] = None
+    payment_link: Optional[str] = None
+    payment_help: Optional[int] = None
+    note: Optional[str] = None
+
+class AutoBookingCreate(AutoBookingBase):
+    pass
+
+class AutoBookingOut(AutoBookingBase):
+    id: int
+    class Config:
+        orm_mode = True

app/schemas/card.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+
+class CardBase(BaseModel):
+    image: Optional[str] = None
+    title: Optional[str] = None
+    content: Optional[str] = None
+    label: Optional[str] = None
+    country: Optional[str] = None
+    additional_info: Optional[str] = None
+    culture: Optional[str] = "english"
+
+
+class CardCreate(CardBase):
+    image: str
+    title: str
+    content: str
+
+
+class CardOut(CardBase):
+    id: int
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        orm_mode = True

app/schemas/configuration.py

@@ -0,0 +1,30 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+
+class ConfigurationBase(BaseModel):
+    config_key: Optional[str] = None
+    config_value: Optional[str] = None
+    description: Optional[str] = None
+    type: Optional[str] = None
+
+
+class ConfigurationCreate(ConfigurationBase):
+    config_key: str
+    config_value: str
+
+
+class ConfigurationUpdate(ConfigurationBase):
+    config_value: Optional[str] = None
+    description: Optional[str] = None
+    type: Optional[str] = None
+
+
+class ConfigurationOut(ConfigurationBase):
+    id: int
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        orm_mode = True

app/schemas/email_authorizations.py

@@ -0,0 +1,38 @@
+from pydantic import BaseModel, EmailStr
+from typing import Optional
+from datetime import datetime
+
+
+# 定义请求体模型
+class EmailContent(BaseModel):
+    body: str
+
+class EmailAuthorizationBase(BaseModel):
+    email: Optional[EmailStr] = None
+    authorization_code: Optional[str] = None
+    imap_server: Optional[str] = None
+    imap_port: Optional[int] = 993
+    smtp_server: Optional[str] = None
+    smtp_port: Optional[int] = 465
+    proxy_host: Optional[str] = None
+    proxy_port: Optional[int] = 0
+    proxy_username: Optional[str] = None
+    proxy_password: Optional[str] = None
+
+
+class EmailAuthorizationCreate(EmailAuthorizationBase):
+    email: EmailStr
+    authorization_code: str
+
+
+class EmailAuthorizationUpdate(EmailAuthorizationBase):
+    pass
+
+
+class EmailAuthorizationOut(EmailAuthorizationBase):
+    id: int
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        orm_mode = True

app/schemas/http_session.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+class HttpSessionBase(BaseModel):
+    local_storage: Optional[str] = None
+    cookies: Optional[str] = None
+    user_agent: Optional[str] = None
+    proxy: Optional[str] = None
+    page: Optional[str] = None
+
+class HttpSessionCreate(HttpSessionBase):
+    session_id: str
+
+class HttpSessionUpdate(HttpSessionBase):
+    pass
+
+class HttpSessionOut(HttpSessionBase):
+    session_id: str
+    create_at: datetime
+
+    class Config:
+        orm_mode = True

app/schemas/short_url.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel, HttpUrl
+from datetime import datetime
+
+
+class ShortUrlBase(BaseModel):
+    short_key: str
+    long_url: HttpUrl
+    created_at: datetime
+
+    class Config:
+        orm_mode = True
+
+
+class ShortUrlCreate(BaseModel):
+    longUrl: HttpUrl

app/schemas/slot.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import date
+
+class SlotBase(BaseModel):
+    submit_country: str
+    submit_city: Optional[str] = None
+    travel_country: str
+    visa_type: str
+    lasted_slot_date: Optional[date] = None
+    available_dates: Optional[str] = None
+    available_times: Optional[str] = None
+    slot_number: Optional[int] = None
+    website: Optional[str] = None
+    open_mode: Optional[int] = 0
+
+class SlotCreate(SlotBase):
+    pass
+
+class SlotOut(SlotBase):
+    id: int
+
+    class Config:
+        orm_mode = True

app/schemas/task.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+from typing import Optional, Any
+from datetime import datetime
+
+
+class TaskBase(BaseModel):
+    command: str
+    args: Any
+    status: Optional[int] = 0
+
+
+class TaskCreate(TaskBase):
+    pass
+
+
+class TaskUpdate(BaseModel):
+    result: Optional[Any] = None
+    status: Optional[int] = None
+
+
+class TaskOut(TaskBase):
+    id: int
+    result: Optional[Any] = None
+    create_at: datetime
+    update_at: datetime
+
+    class Config:
+        orm_mode = True

app/schemas/visafly_config.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class VisaflyConfigBase(BaseModel):
+    submission_country: str
+    submission_country_code: str
+    submission_city: str
+    submission_city_code: str
+    travel_country: str
+    travel_country_code: str
+    visa_type: str
+    provider: Optional[str] = None
+    field_requirement_type: Optional[str] = None
+
+class VisaflyConfigCreate(VisaflyConfigBase):
+    pass
+
+class VisaflyConfigOut(VisaflyConfigBase):
+    id: int
+
+    class Config:
+        orm_mode = True

app/services/auto_booking_service.py

@@ -0,0 +1,72 @@
+from sqlalchemy.orm import Session
+from sqlalchemy import func
+from app.models.auto_booking import AutoBooking
+from app.schemas.auto_booking import AutoBookingCreate
+from typing import List
+
+class AutoBookingService:
+
+    @staticmethod
+    def create(db: Session, obj_in: AutoBookingCreate) -> AutoBooking:
+        db_obj = AutoBooking(**obj_in.dict())
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def get_by_id(db: Session, id: int):
+        return db.query(AutoBooking).filter(AutoBooking.id == id).first()
+
+    @staticmethod
+    def delete_by_id(db: Session, id: int):
+        obj = db.query(AutoBooking).filter(AutoBooking.id == id).first()
+        if obj:
+            db.delete(obj)
+            db.commit()
+            return True
+        return False
+
+    @staticmethod
+    def update_by_id(db: Session, id: int, updated_data: dict):
+        obj = db.query(AutoBooking).filter(AutoBooking.id == id).first()
+        if not obj:
+            return None
+        for key, value in updated_data.items():
+            setattr(obj, key, value)
+        db.commit()
+        db.refresh(obj)
+        return obj
+
+    @staticmethod
+    def get_paginated(db: Session, tech_provider: str, keyword: str, page: int, size: int):
+        query = db.query(AutoBooking)
+        if tech_provider:
+            query = query.filter(AutoBooking.provider == tech_provider)
+        if keyword:
+            like_str = f"%{keyword}%"
+            query = query.filter(
+                (AutoBooking.first_name.like(like_str)) |
+                (AutoBooking.last_name.like(like_str)) |
+                (AutoBooking.email.like(like_str)) |
+                (AutoBooking.visa_center.like(like_str))
+            )
+        return query.offset(page * size).limit(size).all()
+
+    @staticmethod
+    def batch_get_by_ids(db: Session, ids: List[int]):
+        return db.query(AutoBooking).filter(AutoBooking.id.in_(ids)).all()
+
+    @staticmethod
+    def statistics(db: Session, tech_provider: str):
+        query = db.query(AutoBooking.provider, func.count(AutoBooking.id)).group_by(AutoBooking.provider)
+        if tech_provider:
+            query = query.filter(AutoBooking.provider == tech_provider)
+        return query.all()
+
+    @staticmethod
+    def get_pending(db: Session, tech_provider: str):
+        query = db.query(AutoBooking).filter(AutoBooking.status == 0)
+        if tech_provider:
+            query = query.filter(AutoBooking.provider == tech_provider)
+        return query.all()

app/services/card_service.py

@@ -0,0 +1,48 @@
+from sqlalchemy.orm import Session
+from sqlalchemy import text
+from typing import List, Optional
+from app.models.card import Card
+from app.schemas.card import CardCreate
+
+
+class CardService:
+    @staticmethod
+    def create(db: Session, obj_in: CardCreate) -> Card:
+        db_obj = Card(**obj_in.dict())
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def get_paginated(db: Session, page: int, size: int, culture: str):
+        offset = page * size
+        return (
+            db.query(Card)
+            .filter(Card.culture == culture)
+            .order_by(Card.created_at.desc())
+            .offset(offset)
+            .limit(size)
+            .all()
+        )
+
+    @staticmethod
+    def get_by_keywords(db: Session, keywords: List[str], page: int, size: int, culture: str):
+        offset = page * size
+        query = db.query(Card).filter(Card.culture == culture)
+
+        # 多关键词模糊查询
+        for kw in keywords:
+            like_str = f"%{kw.strip()}%"
+            query = query.filter(
+                (Card.title.like(like_str)) |
+                (Card.content.like(like_str)) |
+                (Card.label.like(like_str))
+            )
+
+        return (
+            query.order_by(Card.created_at.desc())
+            .offset(offset)
+            .limit(size)
+            .all()
+        )

app/services/configuration_service.py

@@ -0,0 +1,43 @@
+from sqlalchemy.orm import Session
+from typing import List, Optional
+from app.models.configuration import Configuration
+from app.schemas.configuration import ConfigurationCreate, ConfigurationUpdate
+
+
+class ConfigurationService:
+    @staticmethod
+    def create(db: Session, config_in: ConfigurationCreate) -> Configuration:
+        db_obj = Configuration(**config_in.dict())
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def get_all(db: Session) -> List[Configuration]:
+        return db.query(Configuration).order_by(Configuration.id.desc()).all()
+
+    @staticmethod
+    def get_by_key(db: Session, config_key: str) -> Optional[Configuration]:
+        return db.query(Configuration).filter(Configuration.config_key == config_key).first()
+
+    @staticmethod
+    def update_by_key(db: Session, config_key: str, config_in: ConfigurationUpdate) -> Optional[Configuration]:
+        db_obj = db.query(Configuration).filter(Configuration.config_key == config_key).first()
+        if not db_obj:
+            return None
+        for field, value in config_in.dict(exclude_unset=True).items():
+            setattr(db_obj, field, value)
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def delete_by_key(db: Session, config_key: str) -> Optional[Configuration]:
+        db_obj = db.query(Configuration).filter(Configuration.config_key == config_key).first()
+        if not db_obj:
+            return None
+        db.delete(db_obj)
+        db.commit()
+        return db_obj

app/services/email_authorizations_service.py

@@ -0,0 +1,630 @@
+import re
+import time
+import socket
+import smtplib
+import socks
+import imaplib
+import email
+from datetime import datetime, timedelta, timezone
+from email.message import EmailMessage
+from email.header import decode_header
+from sqlalchemy.orm import Session
+from typing import List, Optional
+from app.models.email_authorizations import EmailAuthorization
+from app.schemas.email_authorizations import EmailAuthorizationCreate, EmailAuthorizationUpdate
+
+
+class EmailAuthorizationService:
+    
+    DEFAULT_READ_TOP_N_EMAIL = 10       # 最多读取最近多少封邮件
+    RETRY_DELAY_SECONDS = 5             # 每次轮询间隔
+    
+    @staticmethod
+    def get_all(db: Session) -> List[EmailAuthorization]:
+        return db.query(EmailAuthorization).order_by(EmailAuthorization.id.desc()).all()
+
+    @staticmethod
+    def get_by_id(db: Session, id: int) -> Optional[EmailAuthorization]:
+        return db.query(EmailAuthorization).filter(EmailAuthorization.id == id).first()
+
+    @staticmethod
+    def get_by_email(db: Session, email: str) -> Optional[EmailAuthorization]:
+        return db.query(EmailAuthorization).filter(EmailAuthorization.email == email).first()
+
+    @staticmethod
+    def create(db: Session, obj_in: EmailAuthorizationCreate) -> EmailAuthorization:
+        db_obj = EmailAuthorization(**obj_in.dict(exclude_unset=True))
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def update(db: Session, id: int, obj_in: EmailAuthorizationUpdate) -> Optional[EmailAuthorization]:
+        db_obj = db.query(EmailAuthorization).filter(EmailAuthorization.id == id).first()
+        if not db_obj:
+            return None
+        for field, value in obj_in.dict(exclude_unset=True).items():
+            setattr(db_obj, field, value)
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def delete(db: Session, id: int) -> Optional[EmailAuthorization]:
+        db_obj = db.query(EmailAuthorization).filter(EmailAuthorization.id == id).first()
+        if not db_obj:
+            return None
+        db.delete(db_obj)
+        db.commit()
+        return db_obj
+    
+    @staticmethod
+    def _connect_imap_with_proxy(
+        host: str,
+        port: int,
+        proxy_host: Optional[str] = None,
+        proxy_port: Optional[int] = None,
+        proxy_user: Optional[str] = None,
+        proxy_password: Optional[str] = None,
+    ) -> imaplib.IMAP4_SSL:
+        """
+        创建支持 SOCKS5 / HTTP 代理的 IMAP SSL 连接
+        """
+        if proxy_host and proxy_port and proxy_port > 0:
+            original_socket = socket.socket
+            socks.setdefaultproxy(
+                proxy_type=socks.SOCKS5,  # 可改为 socks.HTTP
+                addr=proxy_host,
+                port=proxy_port,
+                username=proxy_user or None,
+                password=proxy_password or None,
+            )
+            socket.socket = socks.socksocket
+            try:
+                imap = imaplib.IMAP4_SSL(host, port)
+            finally:
+                socket.socket = original_socket  # 恢复原始 socket
+        else:
+            imap = imaplib.IMAP4_SSL(host, port)
+        return imap
+    
+    
+    @staticmethod
+    def _connect_smtp_with_proxy(
+        host: str,
+        port: int,
+        proxy_host: Optional[str] = None,
+        proxy_port: Optional[int] = None,
+        proxy_user: Optional[str] = None,
+        proxy_password: Optional[str] = None,
+    ) -> smtplib.SMTP_SSL:
+        """
+        创建支持 SOCKS5 / HTTP 代理的 SMTP SSL 连接
+        （逻辑完全与 _connect_imap_with_proxy 保持一致，且强制 SSL）
+        """
+        if proxy_host and proxy_port and proxy_port > 0:
+            original_socket = socket.socket
+            socks.setdefaultproxy(
+                proxy_type=socks.SOCKS5,  # 如需改 HTTP，这里改
+                addr=proxy_host,
+                port=proxy_port,
+                username=proxy_user or None,
+                password=proxy_password or None,
+            )
+            socket.socket = socks.socksocket
+            try:
+                smtp = smtplib.SMTP_SSL(host, port)
+            finally:
+                socket.socket = original_socket  # 恢复原始 socket
+        else:
+            # 无代理
+            smtp = smtplib.SMTP_SSL(host, port)
+
+        return smtp
+
+
+    @staticmethod
+    def fetch_email_authorizations(
+        auth,
+        sender: str,
+        recipient: str,
+        subject_keywords: str,
+        body_keywords: str,
+        sent_date: str,
+        expiry: int = 300,
+        only_text: bool = True
+    ) -> Optional[str]:
+        """
+        在有效期内循环读取邮箱，找到符合条件的邮件（使用最后一条 Received 头作为收件时间）
+        """
+
+        EMAIL_ACCOUNT = auth.email
+        EMAIL_PASSWORD = auth.authorization_code
+        IMAP_SERVER = auth.imap_server
+        IMAP_PORT = auth.imap_port
+
+        subject_keys = [s.strip() for s in subject_keywords.split(",") if s.strip()]
+        body_keys = [s.strip() for s in body_keywords.split(",") if s.strip()]
+
+        # === 时间计算 ===
+        sent_dt = datetime.strptime(sent_date, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
+        max_wait_time = min(5 * 60, expiry)   # 最长等待5分钟
+        expiry_at = time.time() + max_wait_time
+
+        def get_received_time(msg):
+            """
+            使用最后一条 Received 头解析收件时间
+            """
+            received_headers = msg.get_all("Received", [])
+            if not received_headers:
+                return None
+            for i, header in enumerate(received_headers, 1):
+                print(f"  [{i}] {header}")
+            last_received = received_headers[-1]
+            if ";" not in last_received:
+                return None
+            time_str = last_received.split(";")[-1].strip()
+            dt_tuple = email.utils.parsedate_tz(time_str)
+            if not dt_tuple:
+                return None
+            return datetime.fromtimestamp(email.utils.mktime_tz(dt_tuple), tz=timezone.utc)
+
+        try:
+            mail = EmailAuthorizationService._connect_imap_with_proxy(
+                IMAP_SERVER,
+                IMAP_PORT,
+                auth.proxy_host,
+                auth.proxy_port,
+                auth.proxy_username,
+                auth.proxy_password,
+            )
+            mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
+            mail.select("INBOX")
+
+            while time.time() < expiry_at:
+                mail.noop()  # 刷新邮箱状态
+                _, data = mail.search(None, "ALL")
+                mail_ids = data[0].split()
+                if not mail_ids:
+                    time.sleep(EmailAuthorizationService.RETRY_DELAY_SECONDS)
+                    continue
+
+                recent_ids = mail_ids[-EmailAuthorizationService.DEFAULT_READ_TOP_N_EMAIL:]
+                messages = []
+                debug = True
+
+                for email_id in reversed(recent_ids):
+                    res, msg_data = mail.fetch(email_id, "(RFC822)")
+                    if res != "OK" or not msg_data:
+                        if debug:
+                            print(f"[WARN] 邮件 ID={email_id.decode()} 获取失败")
+                        continue
+
+                    msg_bytes = None
+                    for part in msg_data:
+                        if isinstance(part, tuple):
+                            msg_bytes = part[1]
+
+                    if not msg_bytes:
+                        if debug:
+                            print(f"[WARN] 邮件 ID={email_id.decode()} 无正文")
+                        continue
+
+                    msg = email.message_from_bytes(msg_bytes)
+                    received_dt = get_received_time(msg)
+                    if not received_dt:
+                        if debug:
+                            print(f"[WARN] 邮件 ID={email_id.decode()} 未解析出 Received 时间")
+                        continue
+
+                    messages.append((msg, received_dt))
+
+                if debug:
+                    print(f"[DEBUG] 成功解析邮件数: {len(messages)}")
+                    print(f"[DEBUG] 收件时间列表: {[m[1] for m in messages]}")
+
+                # 按收件时间降序排序
+                messages.sort(key=lambda x: x[1], reverse=True)
+
+                for msg, received_dt in messages:
+                    # 判断是否在发送时间后的有效窗口内
+                    if received_dt < sent_dt:
+                        if debug:
+                            print(f"[INFO] 邮件太旧: {received_dt}")
+                        continue
+                    if received_dt > sent_dt + timedelta(seconds=expiry):
+                        if debug:
+                            print(f"[INFO] 邮件太新: {received_dt}")
+                        continue
+
+                    # 匹配发件人/收件人
+                    msg_from = msg.get("From", "")
+                    msg_to = msg.get("To", "")
+                    if sender.lower() not in msg_from.lower():
+                        if debug:
+                            print("发件人不匹配")
+                        continue
+                    if recipient.lower() not in msg_to.lower():
+                        if debug:
+                            print("收件人不匹配")
+                        continue
+
+                    # 匹配主题
+                    subject, enc = decode_header(msg.get("Subject"))[0]
+                    if isinstance(subject, bytes):
+                        subject = subject.decode(enc or "utf-8", errors="ignore")
+                    if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys):
+                        continue
+
+                    # 提取正文
+                    body = EmailAuthorizationService._extract_body(msg, only_text)
+                    if body_keys and not any(k.lower() in body.lower() for k in body_keys):
+                        continue
+
+                    # 找到匹配邮件 → 返回内容
+                    mail.close()
+                    mail.logout()
+                    return body.strip()
+
+                # 未匹配到 → 等待重试
+                time.sleep(EmailAuthorizationService.RETRY_DELAY_SECONDS)
+
+            mail.close()
+            mail.logout()
+            return None  # 超时未找到
+
+        except Exception as e:
+            raise e
+        
+
+    @staticmethod
+    def fetch_email_authorizations_from_top_n(
+        auth,
+        sender: str,
+        recipient: str,
+        subject_keywords: str,
+        body_keywords: str,
+        top: int = 10,
+        only_text: bool = True
+    ) -> Optional[str]:
+        """
+        在有效期内循环读取邮箱，找到符合条件的邮件（使用最后一条 Received 头作为收件时间）
+        """
+
+        EMAIL_ACCOUNT = auth.email
+        EMAIL_PASSWORD = auth.authorization_code
+        IMAP_SERVER = auth.imap_server
+        IMAP_PORT = auth.imap_port
+
+        subject_keys = [s.strip() for s in subject_keywords.split(",") if s.strip()]
+        body_keys = [s.strip() for s in body_keywords.split(",") if s.strip()]
+
+        try:
+            mail = EmailAuthorizationService._connect_imap_with_proxy(
+                IMAP_SERVER,
+                IMAP_PORT,
+                auth.proxy_host,
+                auth.proxy_port,
+                auth.proxy_username,
+                auth.proxy_password,
+            )
+            mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
+            mail.select("INBOX")
+
+            _, data = mail.search(None, "ALL")
+            mail_ids = data[0].split()
+            if not mail_ids:
+                return None
+
+            recent_ids = mail_ids[-top:]
+            messages = []
+            debug = True
+
+            for email_id in reversed(recent_ids):
+                res, msg_data = mail.fetch(email_id, "(RFC822)")
+                if res != "OK" or not msg_data:
+                    if debug:
+                        print(f"[WARN] 邮件 ID={email_id.decode()} 获取失败")
+                    continue
+
+                msg_bytes = None
+                for part in msg_data:
+                    if isinstance(part, tuple):
+                        msg_bytes = part[1]
+
+                if not msg_bytes:
+                    if debug:
+                        print(f"[WARN] 邮件 ID={email_id.decode()} 无正文")
+                    continue
+
+                msg = email.message_from_bytes(msg_bytes)
+                messages.append(msg)
+
+            if debug:
+                print(f"[DEBUG] 成功解析邮件数: {len(messages)}")
+
+            for msg in messages:
+          
+                # 匹配发件人/收件人
+                msg_from = msg.get("From", "")
+                msg_to = msg.get("To", "")
+                if sender.lower() not in msg_from.lower():
+                    if debug:
+                        print("发件人不匹配")
+                    continue
+                if recipient.lower() not in msg_to.lower():
+                    if debug:
+                        print("收件人不匹配")
+                    continue
+
+                # 匹配主题
+                subject, enc = decode_header(msg.get("Subject"))[0]
+                if isinstance(subject, bytes):
+                    subject = subject.decode(enc or "utf-8", errors="ignore")
+                if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys):
+                    continue
+
+                # 提取正文
+                body = EmailAuthorizationService._extract_body(msg, only_text)
+                if body_keys and not any(k.lower() in body.lower() for k in body_keys):
+                    continue
+
+                # 找到匹配邮件 → 返回内容
+                mail.close()
+                mail.logout()
+                return body.strip()
+
+            mail.close()
+            mail.logout()
+            return None  # 超时未找到
+
+        except Exception as e:
+            raise e
+        
+    # ----------------------------------------------------------------------
+    # 主方法：模仿 Java forwardFirstMatchingEmail
+    # ----------------------------------------------------------------------
+    def forward_first_matching_email(
+            auth,
+            forward_to: str,
+            sender: str,
+            recipient: str,
+            subject_keywords: str,
+            body_keywords: str
+    ):
+        EMAIL_ACCOUNT = auth.email
+        EMAIL_PASSWORD = auth.authorization_code
+        IMAP_SERVER = auth.imap_server
+        IMAP_PORT = auth.imap_port
+
+        subject_keys = [s.strip() for s in subject_keywords.split(",") if s.strip()]
+        body_keys = [s.strip() for s in body_keywords.split(",") if s.strip()]
+
+        try:
+            mail = EmailAuthorizationService._connect_imap_with_proxy(
+                IMAP_SERVER,
+                IMAP_PORT,
+                auth.proxy_host,
+                auth.proxy_port,
+                auth.proxy_username,
+                auth.proxy_password,
+            )
+            mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
+            mail.select("INBOX")
+
+            target = recipient
+            query = f'(HEADER To "{target}")'
+            res, data = mail.uid("search", None, query)
+            if res != "OK":
+                return None
+
+            uids = data[0].split()
+
+            # 获取邮件，按 sentDate 排序
+            msgs = []
+            for uid in uids:
+                res, msg_data = mail.uid("fetch", uid, "(RFC822)")
+                if res != "OK":
+                    continue
+                msg = email.message_from_bytes(msg_data[0][1])
+                date_ = email.utils.parsedate_to_datetime(msg.get("Date"))
+                msgs.append((date_, msg))
+
+            msgs.sort(key=lambda x: x[0], reverse=True)
+
+            # -----------------------------------------------------------
+            # 搜索匹配邮件
+            # -----------------------------------------------------------
+            for _, msg in msgs:
+                # 匹配发件人/收件人
+                msg_from = msg.get("From", "")
+                msg_to = msg.get("To", "")
+                if sender.lower() not in msg_from.lower():
+                    if debug:
+                        print("发件人不匹配")
+                    continue
+                if recipient.lower() not in msg_to.lower():
+                    if debug:
+                        print("收件人不匹配")
+                    continue
+
+                # 匹配主题
+                subject, enc = decode_header(msg.get("Subject"))[0]
+                if isinstance(subject, bytes):
+                    subject = subject.decode(enc or "utf-8", errors="ignore")
+                if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys):
+                    continue
+
+                # 提取正文
+                body = EmailAuthorizationService._extract_body(msg, True)
+                if body_keys and not any(k.lower() in body.lower() for k in body_keys):
+                    continue
+                
+                msg.replace_header("From", EMAIL_ACCOUNT)
+                msg.replace_header("To", forward_to)
+
+                # 可选：修改 Subject
+                subject_raw = msg.get("Subject", "")
+                msg.replace_header("Subject", f"FWD: {subject_raw}")
+
+                # 发送
+                EmailAuthorizationService.send_email_smtp(
+                    auth,
+                    msg
+                )
+
+                return f"邮件 '{subject}' 已成功转发至: {forward_to}"
+            return None
+        except Exception as e:
+            raise e
+        
+    # ----------------------------------------------------------------------
+    # 主方法：模仿 Java send_email
+    # ----------------------------------------------------------------------
+    def send_email(
+            auth,
+            send_to: str,
+            subject: str,
+            content_type: str,
+            content: str
+    ):
+        try:
+            EMAIL_ACCOUNT = auth.email
+            msg = EmailMessage()
+            msg["From"] = EMAIL_ACCOUNT
+            msg["To"] = send_to
+            msg["Subject"] = subject
+            
+            html_content = None;
+            text_content = None;
+            if ("html" == content_type.lower()):
+                html_content = content
+            else:
+                text_content = content
+
+            # 设置正文内容
+            if html_content and text_content:
+                # multipart/alternative
+                msg.set_content(text_content)
+                msg.add_alternative(html_content, subtype="html")
+            elif html_content:
+                msg.add_alternative(html_content, subtype="html")
+            elif text_content:
+                msg.set_content(text_content)
+            else:
+                msg.set_content("")  # 空邮件
+            
+            # 发送
+            EmailAuthorizationService.send_email_smtp(
+                auth,
+                msg
+            )
+
+            return f"邮件 '{subject}' 成功发送至: {send_to}"
+        except Exception as e:
+            raise e
+        
+        
+    # ----------------------------------------------------------------------
+    # 主方法：模仿 Java send_email
+    # ----------------------------------------------------------------------
+    def send_email_bulk(
+            auth,
+            send_to: str,
+            subject: str,
+            content_type: str,
+            content: str
+    ):
+        try:
+            bcc_list = [s.strip() for s in send_to.split(",") if s.strip()]
+            EMAIL_ACCOUNT = auth.email
+            msg = EmailMessage()
+            msg["From"] = EMAIL_ACCOUNT
+            # TO 可以留空或放一个默认收件人
+            msg["To"] = bcc_list[0]  # 或者固定一个自己邮箱作为 TO
+            # BCC 添加所有收件人
+            bcc_list = bcc_list
+            msg["Subject"] = subject
+            
+            html_content = None;
+            text_content = None;
+            if ("html" == content_type.lower()):
+                html_content = content
+            else:
+                text_content = content
+
+            # 设置正文内容
+            if html_content and text_content:
+                # multipart/alternative
+                msg.set_content(text_content)
+                msg.add_alternative(html_content, subtype="html")
+            elif html_content:
+                msg.add_alternative(html_content, subtype="html")
+            elif text_content:
+                msg.set_content(text_content)
+            else:
+                msg.set_content("")  # 空邮件
+            
+            # 发送
+            EmailAuthorizationService.send_email_smtp(
+                auth,
+                msg,
+                bcc_list = bcc_list
+            )
+
+            return f"邮件 '{subject}' 成功发送至: {send_to}"
+        except Exception as e:
+            raise e
+
+    # ----------------------------------------------------------------------
+    # SMTP 发送邮件
+    # ----------------------------------------------------------------------
+    @staticmethod
+    def send_email_smtp(auth, msg, bcc_list=[]):
+        EMAIL_ACCOUNT = auth.email
+        EMAIL_PASSWORD = auth.authorization_code
+        SMTP_SERVER = auth.smtp_server
+        SMTP_PORT = auth.smtp_port
+        mail = EmailAuthorizationService._connect_smtp_with_proxy(
+            SMTP_SERVER,
+            SMTP_PORT,
+            auth.proxy_host,
+            auth.proxy_port,
+            auth.proxy_username,
+            auth.proxy_password,
+        )
+        mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
+        if bcc_list:
+            mail.send_message(msg, to_addrs=bcc_list)
+        else:
+            mail.send_message(msg)
+
+    # ==============================================================
+    # 辅助函数：提取邮件正文
+    # ==============================================================
+    @staticmethod
+    def _extract_body(msg, only_text: bool = True) -> str:
+        """根据 only_text 参数提取邮件内容"""
+        body_parts = []
+        if msg.is_multipart():
+            for part in msg.walk():
+                ctype = part.get_content_type()
+                if only_text and ctype != "text/plain":
+                    continue
+                if not only_text and ctype not in ["text/plain", "text/html"]:
+                    continue
+                if part.get("Content-Disposition"):
+                    continue
+                charset = part.get_content_charset() or "utf-8"
+                try:
+                    text = part.get_payload(decode=True).decode(charset, errors="ignore")
+                    body_parts.append(text)
+                except Exception:
+                    continue
+        else:
+            charset = msg.get_content_charset() or "utf-8"
+            body_parts.append(msg.get_payload(decode=True).decode(charset, errors="ignore"))
+        body = "\n".join(body_parts)
+        return re.sub(r"\s+", " ", body.strip())

app/services/http_session_service.py

@@ -0,0 +1,39 @@
+from sqlalchemy.orm import Session
+from app.models.http_session import HttpSession
+from app.schemas.http_session import HttpSessionCreate, HttpSessionUpdate
+from typing import Optional
+
+class HttpSessionService:
+
+    @staticmethod
+    def create(db: Session, data: HttpSessionCreate) -> HttpSession:
+        obj = HttpSession(**data.dict())
+        db.add(obj)
+        db.commit()
+        db.refresh(obj)
+        return obj
+
+    @staticmethod
+    def get_by_sid(db: Session, session_id: str) -> Optional[HttpSession]:
+        return db.query(HttpSession).filter(HttpSession.session_id == session_id).first()
+
+    @staticmethod
+    def delete_by_sid(db: Session, session_id: str) -> bool:
+        obj = db.query(HttpSession).filter(HttpSession.session_id == session_id).first()
+        if not obj:
+            return False
+        db.delete(obj)
+        db.commit()
+        return True
+
+    @staticmethod
+    def update_by_sid(db: Session, session_id: str, data: HttpSessionUpdate):
+        obj = db.query(HttpSession).filter(HttpSession.session_id == session_id).first()
+        if not obj:
+            return None
+        for k, v in data.dict().items():
+            if v is not None:
+                setattr(obj, k, v)
+        db.commit()
+        db.refresh(obj)
+        return obj

app/services/seaweedfs_service.py

@@ -0,0 +1,70 @@
+import requests
+from fastapi import UploadFile
+from app.core.logger import logger
+
+
+class SeaweedFSService:
+    MASTER_URL = "http://visafly.top:9333"  # 你的 SeaweedFS master 地址
+
+    @classmethod
+    def upload(cls, file: UploadFile):
+        """上传文件到 SeaweedFS"""
+        try:
+            # 1️⃣ 获取可上传的 volume 地址
+            assign_resp = requests.get(f"{cls.MASTER_URL}/dir/assign", timeout=5)
+            assign_data = assign_resp.json()
+            fid = assign_data["fid"]
+            public_url = assign_data["publicUrl"]
+
+            # 2️⃣ 上传文件数据
+            upload_url = f"http://{public_url}/{fid}"
+            files = {"file": (file.filename, file.file, file.content_type)}
+            upload_resp = requests.post(upload_url, files=files, timeout=10)
+
+            if upload_resp.status_code == 201:
+                return {"fid": fid, "url": upload_url}
+            else:
+                logger.error(f"上传失败: {upload_resp.text}")
+                return None
+        except Exception as e:
+            logger.exception(f"SeaweedFS 上传异常, 原因={e}")
+            return None
+
+    @classmethod
+    def get(cls, fid: str):
+        """根据 fid 读取文件"""
+        try:
+            resp = requests.get(f"{cls.MASTER_URL}/dir/lookup?volumeId={fid.split(',')[0]}", timeout=5)
+            data = resp.json()
+            if not data.get("locations"):
+                return None
+
+            public_url = data["locations"][0]["publicUrl"]
+            file_url = f"http://{public_url}/{fid}"
+
+            file_resp = requests.get(file_url, timeout=10)
+            if file_resp.status_code == 200:
+                return file_resp.content, file_resp.headers.get("Content-Type", "application/octet-stream")
+            else:
+                return None
+        except Exception as e:
+            logger.exception(f"SeaweedFS 读取异常, 原因={e}")
+            return None
+
+    @classmethod
+    def delete(cls, fid: str):
+        """删除文件"""
+        try:
+            resp = requests.get(f"{cls.MASTER_URL}/dir/lookup?volumeId={fid.split(',')[0]}", timeout=5)
+            data = resp.json()
+            if not data.get("locations"):
+                return False
+
+            public_url = data["locations"][0]["publicUrl"]
+            delete_url = f"http://{public_url}/{fid}"
+
+            del_resp = requests.delete(delete_url, timeout=10)
+            return del_resp.status_code == 202
+        except Exception as e:
+            logger.exception(f"SeaweedFS 删除异常, 原因={e}")
+            return False

app/services/short_url_service.py

@@ -0,0 +1,37 @@
+import string
+import random
+from sqlalchemy.orm import Session
+from app.models.short_url import ShortUrl
+
+
+class ShortUrlService:
+    @staticmethod
+    def generate_short_key(length: int = 8) -> str:
+        """生成随机短 Key（由字母+数字组成）"""
+        chars = string.ascii_letters + string.digits
+        return ''.join(random.choices(chars, k=length))
+
+    @staticmethod
+    def create_short_url(db: Session, long_url: str) -> ShortUrl:
+        """创建短链接"""
+        # 检查是否已经存在相同的长链接
+        existing = db.query(ShortUrl).filter(ShortUrl.long_url == long_url).first()
+        if existing:
+            return existing
+
+        # 生成唯一 short_key
+        short_key = ShortUrlService.generate_short_key()
+        while db.query(ShortUrl).filter(ShortUrl.short_key == short_key).first():
+            short_key = ShortUrlService.generate_short_key()
+
+        db_obj = ShortUrl(short_key=short_key, long_url=long_url)
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def get_long_url(db: Session, short_key: str) -> str:
+        """通过短 key 获取原始长链接"""
+        record = db.query(ShortUrl).filter(ShortUrl.short_key == short_key).first()
+        return record.long_url if record else ""

app/services/slot_service.py

@@ -0,0 +1,42 @@
+from sqlalchemy.orm import Session
+from sqlalchemy import asc, desc
+from app.models.slot import Slot
+from app.schemas.slot import SlotCreate
+from typing import List
+
+class SlotService:
+
+    @staticmethod
+    def report(db: Session, obj_in: SlotCreate) -> Slot:
+        # 可以根据 submit_city + travel_country + visa_type 查重，决定是更新还是新增
+        obj = db.query(Slot).filter(
+            Slot.submit_city == obj_in.submit_city,
+            Slot.travel_country == obj_in.travel_country,
+            Slot.visa_type == obj_in.visa_type
+        ).first()
+        if obj:
+            # 更新现有记录
+            for k, v in obj_in.dict().items():
+                setattr(obj, k, v)
+        else:
+            obj = VisaSlot(**obj_in.dict())
+            db.add(obj)
+        db.commit()
+        db.refresh(obj)
+        return obj
+
+    @staticmethod
+    def search(db: Session, submit_city: str, travel_country: str, visa_type: str, date_type: str):
+        query = db.query(Slot).filter(
+            Slot.submit_city == submit_city,
+            Slot.travel_country == travel_country,
+            Slot.visa_type == visa_type
+        )
+        if date_type == "latest":
+            obj = query.order_by(desc(Slot.update_at)).first()
+            return obj
+        elif date_type == "earliest":
+            obj = query.order_by(asc(Slot.lasted_slot_date)).first()
+            return obj
+        else:
+            return query.all()

app/services/task_service.py

@@ -0,0 +1,51 @@
+import json
+from sqlalchemy.orm import Session
+from typing import List, Optional
+from app.models.task import Task
+from app.schemas.task import TaskCreate, TaskUpdate
+
+
+class TaskService:
+    @staticmethod
+    def create(db: Session, obj_in: TaskCreate) -> Task:
+        db_obj = Task(
+            command=obj_in.command,
+            args=obj_in.args,
+            status=obj_in.status or 0,
+        )
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def get_by_id(db: Session, task_id: int) -> Optional[Task]:
+        return db.query(Task).filter(Task.id == task_id).first()
+
+    @staticmethod
+    def update(db: Session, task_id: int, obj_in: TaskUpdate) -> Optional[Task]:
+        db_obj = db.query(Task).filter(Task.id == task_id).first()
+        if not db_obj:
+            return None
+
+        if obj_in.result is not None:
+            db_obj.result = obj_in.result
+        if obj_in.status is not None:
+            db_obj.status = obj_in.status
+
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def get_pending(db: Session, command: str, page: int, size: int) -> List[Task]:
+        offset = page * size
+        return (
+            db.query(Task)
+            .filter(Task.command == command, Task.status == 0)
+            .order_by(Task.create_at.asc())
+            .offset(offset)
+            .limit(size)
+            .all()
+        )

app/services/visafly_config_service.py

@@ -0,0 +1,50 @@
+from sqlalchemy.orm import Session
+from app.models.visafly_config import VisaflyConfig
+from app.schemas.visafly_config import VisaflyConfigCreate
+from typing import List
+
+class VisaflyConfigService:
+
+    @staticmethod
+    def create(db: Session, obj_in: VisaflyConfigCreate) -> VisaflyConfig:
+        db_obj = VisaflyConfig(**obj_in.dict())
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    @staticmethod
+    def get_submission_countries(db: Session) -> List[dict]:
+        rows = db.query(
+            VisaflyConfig.submission_country,
+            VisaflyConfig.submission_country_code
+        ).distinct().all()
+        return [{"country": r[0], "country_code": r[1]} for r in rows]
+
+    @staticmethod
+    def get_cities_by_country(db: Session, country_code: str) -> List[dict]:
+        rows = db.query(
+            VisaflyConfig.submission_city,
+            VisaflyConfig.submission_city_code
+        ).filter(VisaflyConfig.submission_country_code == country_code).distinct().all()
+        return [{"city": r[0], "city_code": r[1]} for r in rows]
+
+    @staticmethod
+    def get_travel_countries_by_city(db: Session, city_code: str) -> List[dict]:
+        rows = db.query(
+            VisaflyConfig.travel_country,
+            VisaflyConfig.travel_country_code,
+            VisaflyConfig.visa_type,
+            VisaflyConfig.provider,
+            VisaflyConfig.field_requirement_type
+        ).filter(VisaflyConfig.submission_city_code == city_code).all()
+
+        return [
+            {
+                "travel_country": r[0],
+                "travel_country_code": r[1],
+                "visa_type": r[2],
+                "provider": r[3],
+                "field_requirement_type": r[4]
+            } for r in rows
+        ]