python-fastapi-mm-zhcs-yj-api/ws_main.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from fastapi import FastAPI, Request #, Response,HTTPException
from fastapi.responses import PlainTextResponse, JSONResponse, RedirectResponse, FileResponse
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.staticfiles import StaticFiles
from routers.api.pattern import router as ws_router
from exceptions import *
from pymysql.err import OperationalError
import os
import platform
from config import settings
from extensions import logger
from jobs import register_jobs
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from common.websocketManager import manager
from utils.redis_util import *

app = FastAPI(docs_url=None, redoc_url=None)
# app.include_router(home.router)
app.include_router(ws_router, prefix="/ws/api/pattern")
# app.include_router(prod_api.router, prefix="")

app.add_middleware(SessionMiddleware, secret_key='MM-ZHCS-YJ-API', max_age=36000 * 24)
app.add_middleware(CORSMiddleware,allow_origins=['*'],allow_credentials=False,allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],allow_headers=['*'])

app.mount('/static', StaticFiles(directory='static'), name='static')

@app.middleware("http")
async def process_authorization(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = (time.time() - start_time) * 1000  # 毫秒
    logger.info(f"Request {request.method} {request.url} processed in {process_time:.2f}ms")
    response.headers["X-Process-Time"] = str(process_time)
    return response


@app.get("/favicon.ico", response_class=FileResponse)
async def favicon():
    return FileResponse("./static/favicon.ico")


# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
#     return JSONResponse(status_code=200, content={ "errcode": 100060, "errmsg": "参数错误" }, )

@app.exception_handler(TokenException)
async def exception_handler(request: Request, exc: TokenException):
    return JSONResponse(
        status_code=200,
        content={"code": 401, "msg":"Token过期"}
        # content={"errcode": "4001", "errmsg":"验证不通过"}
    )
@app.exception_handler(RoleException)
async def role_exception_handler(request: Request, exc: RoleException):
    # 这样界面才有文字提示
    return JSONResponse(
        status_code=200,
        content={"code": 403, "msg":"表示权限不足"}
    )
    # return JSONResponse(
    #     status_code=403,  # 使用 403 Forbidden 表示权限不足
    #     content={"errcode": exc.errcode, "errmsg": exc.errmsg}
    # )
@app.exception_handler(OperationalError)
async def exception_handler(request: Request, exc: OperationalError):
    return JSONResponse(
        status_code=412,
        content={"code": "412", "msg":f"接口对应sql异常，请检查{exc}"}
    )


# @app.exception_handler(AttributeError)
# async def exception_handler(request: Request, exc: TokenException):
#     return JSONResponse(
#         status_code=200,
#         content={"errcode": "500", "errmsg":"Internal Server Error"}
#     )


@app.exception_handler(AuthException)
async def exception_handler(request: Request, exc: TokenException):
    return RedirectResponse("/")
    
@app.exception_handler(AlertException)
async def exception_handler(request: Request, exc: AlertException):
    return JSONResponse(
        status_code=200,
        content={"ret": 1, "msg": exc.msg}
    )


    
if __name__ == '__main__':
    import uvicorn
    sys = platform.system()
    if sys == "Windows":
        os.system("title 茂名应急数据中台 DEV")
    
    uvicorn.run(app='main:app', host='0.0.0.0',port=9987, reload=True)
    # uvicorn.run(app='main:app', host='0.0.0.0', port=9988)