125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
from fastapi import FastAPI, HTTPException
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
import os
|
||
import logging
|
||
from dotenv import load_dotenv
|
||
|
||
from app.routers import qcloud_router, dress_router, tryon_router
|
||
from app.utils.config import get_settings
|
||
from app.utils.response import APIResponse
|
||
from app.database import Base, engine
|
||
from app.middleware import ResponseWrapperMiddleware
|
||
from app.exceptions import setup_exception_handlers
|
||
|
||
# 创建数据库表
|
||
Base.metadata.create_all(bind=engine)
|
||
|
||
# 加载环境变量
|
||
load_dotenv(dotenv_path=".env")
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 全局配置：是否使用标准响应格式，默认禁用直接返回数据
|
||
os.environ["USE_STANDARD_RESPONSE"] = os.environ.get("USE_STANDARD_RESPONSE", "0")
|
||
logger.info(f"标准响应格式: {'已启用' if os.environ['USE_STANDARD_RESPONSE'] == '1' else '已禁用'}")
|
||
|
||
app = FastAPI(
|
||
title="AI-Dressing API",
|
||
description="基于 DashScope 的 AI 服务 API",
|
||
version="0.1.0",
|
||
)
|
||
|
||
# 添加 CORS 中间件
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"], # 在生产环境中，应该指定确切的域名
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# 添加响应包装中间件
|
||
app.add_middleware(ResponseWrapperMiddleware)
|
||
|
||
# 设置异常处理
|
||
setup_exception_handlers(app)
|
||
|
||
# 注册路由
|
||
app.include_router(qcloud_router.router, prefix="/api/qcloud", tags=["腾讯云"])
|
||
app.include_router(dress_router.router, prefix="/api/dresses", tags=["服装"])
|
||
app.include_router(tryon_router.router, prefix="/api/tryons", tags=["试穿"])
|
||
|
||
@app.get("/", tags=["健康检查"])
|
||
async def root():
|
||
"""API 根端点"""
|
||
return {"message": "服务运行中"}
|
||
|
||
@app.get("/health", tags=["健康检查"])
|
||
async def health_check():
|
||
"""健康检查端点，用于Docker容器健康监控"""
|
||
health_data = {"checks": {}}
|
||
health_status = "healthy"
|
||
health_message = "服务运行正常"
|
||
|
||
# 检查数据库连接
|
||
try:
|
||
from app.database import SessionLocal
|
||
db = SessionLocal()
|
||
db.execute("SELECT 1")
|
||
db.close()
|
||
health_data["checks"]["database"] = {"status": "healthy", "message": "数据库连接正常"}
|
||
except Exception as e:
|
||
health_status = "unhealthy"
|
||
health_message = "服务异常"
|
||
health_data["checks"]["database"] = {"status": "unhealthy", "message": f"数据库连接失败: {str(e)}"}
|
||
|
||
# 检查API密钥
|
||
if os.getenv("DASHSCOPE_API_KEY"):
|
||
health_data["checks"]["dashscope"] = {"status": "configured", "message": "DashScope API密钥已配置"}
|
||
else:
|
||
health_data["checks"]["dashscope"] = {"status": "missing", "message": "DashScope API密钥未配置"}
|
||
|
||
if os.getenv("QCLOUD_SECRET_ID") and os.getenv("QCLOUD_SECRET_KEY"):
|
||
health_data["checks"]["qcloud"] = {"status": "configured", "message": "腾讯云凭证已配置"}
|
||
else:
|
||
health_data["checks"]["qcloud"] = {"status": "missing", "message": "腾讯云凭证未配置"}
|
||
|
||
if health_status == "healthy":
|
||
return {
|
||
"status": health_status,
|
||
"message": health_message,
|
||
"data": health_data
|
||
}
|
||
else:
|
||
return {
|
||
"status": health_status,
|
||
"message": health_message,
|
||
"data": health_data
|
||
}
|
||
|
||
@app.get("/info", tags=["服务信息"])
|
||
async def get_info():
|
||
"""获取服务基本信息"""
|
||
settings = get_settings()
|
||
info_data = {
|
||
"app_name": "AI-Dressing API",
|
||
"version": "0.1.0",
|
||
"debug_mode": settings.debug,
|
||
"standard_response": os.environ.get("USE_STANDARD_RESPONSE") == "1"
|
||
}
|
||
return info_data
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
settings = get_settings()
|
||
uvicorn.run(
|
||
"app.main:app",
|
||
host=settings.host,
|
||
port=settings.port,
|
||
reload=settings.debug,
|
||
) |