feaet:update

.env

@@ -0,0 +1,3 @@
+DATABASE_URL=postgresql://user:password@127.0.0.1:5432/dbname
+REDIS_URL=redis://:STEs2x6ML0U1HlpE9SojM6YU7QPhqzY8@92.119.127.236:6379/0
+API_TOKEN=7x9EjFpmv7GjZc6AfVeqxuUBANpqkpkHAtxJM7CAW5oZhs0nEyCJBy39N4XXs5hgfYWXw3jFrcgXqQ42HAx9Qvwtk9vC2GvKBbWz

.gitignore

@@ -0,0 +1,2 @@
+*.pyc
+venv

app/api/router.py

@@ -0,0 +1,30 @@
+from typing import List
+from fastapi import APIRouter, Query, Depends
+from app.core.redis import get_redis_client
+from redis.asyncio import Redis
+from app.schemas.user import UserOut
+from app.schemas.troov import TroovRate
+from app.services.troov_service import get_rate_by_date
+
+# 公共路由
+public_router = APIRouter(prefix="/api", tags=["troov"])
+
+@public_router.get("/ping")
+def ping():
+    return {"message": "pong"}
+
+# 受保护路由
+protected_router = APIRouter(prefix="/api", tags=["troov"])
+
+@protected_router.get("/users", response_model=List[UserOut])
+def get_users():
+    return [
+        {"id": 1, "name": "Alice"},
+        {"id": 2, "name": "Bob"}
+    ]
+
+@protected_router.get("/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)

app/core/auth.py

@@ -0,0 +1,18 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from app.core.config import settings
+
+security = HTTPBearer()
+
+API_TOKEN = settings.api_token
+
+def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    """
+    全局鉴权依赖
+    """
+    if credentials.credentials != API_TOKEN:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or missing token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )

app/core/config.py

@@ -0,0 +1,13 @@
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    app_name: str = "MyApp"
+    debug: bool = True
+    database_url: str
+    redis_url: str
+    api_token: str
+    
+    class Config:
+        env_file = ".env"
+
+settings = Settings()

app/core/database.py

@@ -0,0 +1,15 @@
+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)
+Base = declarative_base()
+
+# 依赖注入
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/core/redis.py

@@ -0,0 +1,14 @@
+from typing import Optional
+from redis import Redis
+from app.core.config import settings
+
+_redis_client: Optional[Redis] = None
+
+def get_redis_client() -> Redis:
+    """
+    同步依赖（FastAPI 可以直接注入）
+    """
+    global _redis_client
+    if _redis_client is None:
+        _redis_client = Redis.from_url(settings.redis_url, decode_responses=True)
+    return _redis_client

app/main.py

@@ -0,0 +1,61 @@
+from fastapi import FastAPI, Depends
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.openapi.utils import get_openapi
+from fastapi.security import HTTPBearer
+
+from app.api import router
+from app.core.auth import verify_token
+from app.core.config import settings
+
+app = FastAPI(title=settings.app_name)
+
+# -----------------------
+# CORS（可选）
+# -----------------------
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# -----------------------
+# 路由注册
+# -----------------------
+# 公共路由，不鉴权
+app.include_router(router.public_router, prefix="/api")
+
+# 需要鉴权的路由
+app.include_router(
+    router.protected_router,
+    prefix="/api",
+    dependencies=[Depends(verify_token)]
+)
+
+# -----------------------
+# Swagger 支持 Bearer Token
+# -----------------------
+def custom_openapi():
+    if app.openapi_schema:
+        return app.openapi_schema
+    openapi_schema = get_openapi(
+        title=app.title,
+        version="1.0.0",
+        description="API documentation",
+        routes=app.routes,
+    )
+    # 添加全局 Bearer
+    openapi_schema["components"]["securitySchemes"] = {
+        "BearerAuth": {
+            "type": "http",
+            "scheme": "bearer",
+            "bearerFormat": "JWT"
+        }
+    }
+    for path in openapi_schema["paths"].values():
+        for method in path.values():
+            method["security"] = [{"BearerAuth": []}]
+    app.openapi_schema = openapi_schema
+    return app.openapi_schema
+
+app.openapi = custom_openapi

app/schemas/troov.py

@@ -0,0 +1,7 @@
+# app/schemas/troov.py
+from pydantic import BaseModel
+
+class TroovRate(BaseModel):
+    time: str
+    rate: str
+    capacity: int

app/schemas/user.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+class UserOut(BaseModel):
+    id: int
+    name: str
+
+class TroovRate(BaseModel):
+    time: str
+    rate: str
+    capacity: int

app/services/troov_service.py

@@ -0,0 +1,67 @@
+import json
+import time
+import requests
+from typing import List
+from fastapi import Depends
+from app.schemas.troov import TroovRate
+
+def pop_redis_value_session(redis_client):
+    lua_script = '''
+local keys = redis.call('keys', 'session:*')
+local max_ttl = -1
+local max_key = nil
+
+for _, key in ipairs(keys) do
+    local ttl = redis.call('ttl', key)
+    if ttl > max_ttl then
+        max_ttl = ttl
+        max_key = key
+    end
+end
+
+if max_key then
+    local value = redis.call('get', max_key)
+    redis.call('del', max_key)
+    return {max_key, value, max_ttl}
+else
+    return nil
+end
+'''
+    result = redis_client.eval(lua_script, 0)
+    return result
+
+def fetch_rate(session_dic, date):
+    url = f"https://api.consulat.gouv.fr/api/team/621540d353069dec25bd0045/reservations/availability?name=Visas&date={date}&places=-5&matching=&maxCapacity=-5&sessionId={session_dic['session_id']}"
+    headers = {
+        'accept': 'application/json, text/plain, */*',
+        'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8',
+        'origin': 'https://consulat.gouv.fr',
+        'referer': 'https://consulat.gouv.fr/en/ambassade-de-france-en-irlande/appointment?name=Visas',
+        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36',
+        'x-gouv-app-id': session_dic['x_gouv_app_id'],
+        'x-gouv-web': 'fr.gouv.consulat'
+    }
+    try:
+        response = requests.get(url, headers=headers)
+        return response.text
+    except Exception as e:
+        return f'Exception info={e}'
+
+def get_rate_by_date(redis_client, date: str) -> List[TroovRate]:
+    """
+    核心业务逻辑：根据日期返回 Troov 预约信息
+    """
+    result = None
+    while True:
+        result = pop_redis_value_session(redis_client)
+        if not result:
+            time.sleep(1)
+            continue
+        break
+    session_data = result[1]
+    session_dic = json.loads(session_data)
+    
+    res = fetch_rate(session_dic, date)
+    return json.loads(res)
+
+        

requirements.txt

@@ -0,0 +1,12 @@
+fastapi
+uvicorn[standard]
+sqlalchemy
+pydantic
+pydantic-settings
+psycopg2-binary  # PostgreSQL
+mysqlclient       # MySQL
+redis
+httpx
+alembic
+APScheduler
+python-dotenv

starter.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import os
+import sys
+import subprocess
+
+def main():
+    """
+    启动 FastAPI 应用，根据环境选择不同参数：
+    - DEV: 热重载，单进程
+    - PROD: 多进程，高性能 uvloop + httptools
+    """
+    env = os.getenv("ENV", "DEV").upper()  # 默认开发环境
+    host = "0.0.0.0"
+    port = "8000"
+    app_module = "app.main:app"
+
+    base_cmd = ["uvicorn", app_module, "--host", host, "--port", port]
+
+    if env == "DEV":
+        print("启动开发环境（热重载）...")
+        base_cmd += ["--reload", "--workers", "1"]
+    elif env == "PROD":
+        print("启动生产环境（多进程 + 高性能）...")
+        base_cmd += ["--workers", str(os.cpu_count()), "--loop", "uvloop", "--http", "httptools"]
+    else:
+        print(f"未知环境 {env}，使用默认开发配置")
+        base_cmd += ["--reload", "--workers", "1"]
+
+    print("执行命令:", " ".join(base_cmd))
+    print(f"Swagger UI: http://{host}:{port}/docs")
+    print(f"ReDoc UI:   http://{host}:{port}/redoc")
+    subprocess.run(base_cmd)
+
+if __name__ == "__main__":
+    main()