python-fastapi-mm-zhcs-yj-api/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 import home, api ,prod_api
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

app = FastAPI()
# app.include_router(home.router)
app.include_router(api.router, prefix="/api")
app.include_router(prod_api.router, prefix="")

@app.get("/", response_class=PlainTextResponse)
async def main():
    return "应急后端接口9988"

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

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=['*'],allow_headers=['*'])

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

@app.on_event('startup')
def app_startup():
    logger.info('---------------------------------------')
    logger.info('--------------'+ settings.ENVIRONMENT +'---------------')
    logger.info('---------------------------------------')
    logger.info("server started.")

    sys = platform.system()
    if sys == "Windows":
        from warnings import filterwarnings
        from pytz_deprecation_shim import PytzUsageWarning
        filterwarnings('ignore', category=PytzUsageWarning)

    executors = {
        'default': ThreadPoolExecutor(20),
        'processpool': ProcessPoolExecutor(5)
    }
    scheduler = AsyncIOScheduler(timezone='Asia/Shanghai', executors=executors)
    register_jobs(scheduler)
    scheduler.start()

# @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={"errcode": "4001", "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__':
    from common import YzyApi
    # YzyApi.get_user_info("h4RxABdc-pfXIdhTT3UhTnKv-qTs-Yn6TfeSWE4IJd0")
    # YzyApi.send_textcard_message(["c3o4t44nb2e1oipbs80paw"], "title", "事件进度", "https://www.baidu.com/")
    # YzyApi.send_textcard_message(["ivj823s4b56za2jcu293f7"], "title", "事件进度", "https://www.baidu.com/")
    # YzyApi.send_textcard_message(["ivj823s4b56za2jcu293f7"], "title", "事件进度", "https://www.baidu.com/")
    # YzyApi.send_textcard_message(["eb4kehgy6wj4qn0jhx1dk6"], "预案响应消息", "预案名称: 茂名市自然灾害救助应急预案\n响应级别: Ⅰ级响应\n消息内容: 单位您好！《茂名市自然灾害救助应急预案》现已全面启动，特此通知您单位迅速响应，全力做好预案工作要点：负责救灾工作宣传报道协调工作。", "https://www.baidu.com/")
    # YzyApi.send_text_message(["eb4kehgy6wj4qn0jhx1dk6"], "事件进度")

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

    # uvicorn.run(app='main:app', host='0.0.0.0', port=9988)