mirror of
https://github.com/linyqh/NarratoAI.git
synced 2025-12-11 02:12:50 +00:00
91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
"""Application implementation - ASGI."""
|
|
|
|
import os
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from loguru import logger
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from app.config import config
|
|
from app.models.exception import HttpException
|
|
from app.router import root_api_router
|
|
from app.utils import utils
|
|
from app.utils import ffmpeg_utils
|
|
|
|
|
|
def exception_handler(request: Request, e: HttpException):
|
|
return JSONResponse(
|
|
status_code=e.status_code,
|
|
content=utils.get_response(e.status_code, e.data, e.message),
|
|
)
|
|
|
|
|
|
def validation_exception_handler(request: Request, e: RequestValidationError):
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content=utils.get_response(
|
|
status=400, data=e.errors(), message="field required"
|
|
),
|
|
)
|
|
|
|
|
|
def get_application() -> FastAPI:
|
|
"""Initialize FastAPI application.
|
|
|
|
Returns:
|
|
FastAPI: Application object instance.
|
|
|
|
"""
|
|
instance = FastAPI(
|
|
title=config.project_name,
|
|
description=config.project_description,
|
|
version=config.project_version,
|
|
debug=False,
|
|
)
|
|
instance.include_router(root_api_router)
|
|
instance.add_exception_handler(HttpException, exception_handler)
|
|
instance.add_exception_handler(RequestValidationError, validation_exception_handler)
|
|
return instance
|
|
|
|
|
|
app = get_application()
|
|
|
|
# Configures the CORS middleware for the FastAPI app
|
|
cors_allowed_origins_str = os.getenv("CORS_ALLOWED_ORIGINS", "")
|
|
origins = cors_allowed_origins_str.split(",") if cors_allowed_origins_str else ["*"]
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
task_dir = utils.task_dir()
|
|
app.mount(
|
|
"/tasks", StaticFiles(directory=task_dir, html=True, follow_symlink=True), name=""
|
|
)
|
|
|
|
public_dir = utils.public_dir()
|
|
app.mount("/", StaticFiles(directory=public_dir, html=True), name="")
|
|
|
|
|
|
@app.on_event("shutdown")
|
|
def shutdown_event():
|
|
logger.info("shutdown event")
|
|
|
|
|
|
@app.on_event("startup")
|
|
def startup_event():
|
|
logger.info("startup event")
|
|
|
|
# 检测FFmpeg硬件加速
|
|
hwaccel_info = ffmpeg_utils.detect_hardware_acceleration()
|
|
if hwaccel_info["available"]:
|
|
logger.info(f"FFmpeg硬件加速检测结果: 可用 | 类型: {hwaccel_info['type']} | 编码器: {hwaccel_info['encoder']} | 独立显卡: {hwaccel_info['is_dedicated_gpu']} | 参数: {hwaccel_info['hwaccel_args']}")
|
|
else:
|
|
logger.warning(f"FFmpeg硬件加速不可用: {hwaccel_info['message']}, 将使用CPU软件编码")
|