250617-1代码。

ws_main.py

@@ -0,0 +1,105 @@
+#!/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, debug=True)
+    # uvicorn.run(app='main:app', host='0.0.0.0', port=9988)