personal-consumption-service/run.py

# @description: 
# @author: licanglong
# @date: 2025/11/20 11:41
import logging
import os

import uvicorn
from fastapi import FastAPI
from pydantic import ValidationError
from starlette.requests import Request
from starlette.responses import JSONResponse

# 设置环境变量
os.environ["APP_PATH"] = os.getenv('APP_PATH') or os.path.abspath(os.path.dirname(__file__))  # noqa

from app.App import App
from app.core import BizException, CTX

CTX.DEFAULT_LOG_FILE = "logs/app.log"  # noqa


class AppImpl(App):

    def create_app(self):
        from app.models.Result import SysResult
        from app.routes.base import base_router
        from app.routes.risk import risk_router
        from app.routes.vector_store import vector_store_router
        from app.routes.invoice import invoice_router

        fast_app = FastAPI()

        @fast_app.exception_handler(404)
        async def handle_404_error(request: Request, exc: Exception):
            # 打印请求路径
            logging.error(f"error request url [404]: {request.url}")
            logging.exception(exc)
            result = SysResult.fail(code=404, msg=str(exc))
            return JSONResponse(
                status_code=404,
                content=result.model_dump()
            )

        @fast_app.exception_handler(BizException)
        async def biz_error_handle(request: Request, exc: BizException):
            logging.exception(exc.message)
            result = SysResult.fail(msg=exc.message or "服务异常", code=exc.code)
            return JSONResponse(
                status_code=500,
                content=result.model_dump()
            )

        @fast_app.exception_handler(ValidationError)
        async def biz_error_handle(request: Request, exc: Exception):
            logging.exception("数据类型不匹配")
            return JSONResponse(
                status_code=500,
                content=SysResult.fail(msg="数据类型不匹配", code=5001).model_dump()
            )

        @fast_app.exception_handler(Exception)
        async def exception_handle(request: Request, exc: Exception):
            logging.exception("服务异常")
            return JSONResponse(
                status_code=500,
                content=SysResult.fail(msg="服务异常").model_dump()
            )

        # 注册路由
        fast_app.include_router(base_router)
        fast_app.include_router(invoice_router)
        fast_app.include_router(risk_router)
        fast_app.include_router(vector_store_router)

        return fast_app

    def run(self, *args, **kwargs):
        app = self.create_app()
        port = CTX.ENV.getprop("server.port", 7699)
        try:
            uvicorn.run(
                app=app,
                host="0.0.0.0",
                port=port,
            )
        except KeyboardInterrupt:
            logging.warning("程序终止!!")


if __name__ == '__main__':
    AppImpl().start()