feat: 增加fluent日志

config/exceptions.py

@@ -0,0 +1,56 @@
+import time
+
+from fastapi import Request, HTTPException, status
+from fastapi.responses import JSONResponse
+from config.logging import logger
+
+
+# 自定义异常类
+class AppException(Exception):
+    def __init__(self, message: str, details=None):
+        self.message = message
+        self.details = details
+
+
+# 全局异常处理中间件
+async def global_exception_handler(request: Request, call_next):
+    try:
+        start_time = time.time()
+
+        # 请求摘要
+        logger.info(
+            f"<AuditSummary> {request.method} {request.url.path}"
+        )
+
+        response = await call_next(request)
+
+        # 耗时
+        time_taken = int((time.time() - start_time) * 1000)
+        logger.info(
+            f"<AuditSummary> Request URL {request.url.path} | Time Taken {time_taken} ms"
+        )
+
+        if response.status_code >= 400:
+            logger.warning(f"<AuditError> status={response.status_code}")
+
+        return response
+
+    except AppException as e:
+        # 记录业务异常
+        logger.error(f"<AuditError> {e.message} | details={e.details}")
+        return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST,
+                            content={"message": e.message, "details": e.details})
+
+    except Exception as e:
+        # 记录未知异常（带堆栈信息）
+        logger.critical(f"<AuditError> {str(e)}", exc_info=True)
+        return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                            content={"code": 500, "message": "服务器内部错误", "details": str(e)})
+
+
+def get_credentials_exception() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )

config/logging.py

@@ -0,0 +1,84 @@
+import atexit
+import os
+import sys
+
+from dotenv import load_dotenv
+from fluent import sender
+from loguru import logger
+
+load_dotenv()
+
+# 日志格式
+STDOUT_FORMAT = (
+    "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
+    "<level>{level: <8}</level> | "
+    "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
+    "<level>{message}</level>"
+)
+
+FILE_FORMAT = (
+    "{time:YYYY-MM-DD HH:mm:ss.SSS} | "
+    "{level: <8} | "
+    "{name}:{function}:{line} - {message}"
+)
+
+fluent_sender = sender.FluentSender(
+    tag=os.getenv("FLUENTD_TOPIC"),
+    host=os.getenv("FLUENTD_HOST"),
+    port=int(os.getenv("FLUENTD_PORT", 24224)),
+    buffer_max_size=8 * 1024 * 1024,
+    timeout=3.0,
+    retry_timeout=60
+)
+
+
+def log_to_fluent(message):
+    try:
+        record = message.record
+
+        # 构建结构化日志数据
+        log_data = {
+            'topic': os.getenv("FLUENTD_TOPIC"),
+            'timestamp': record['time'].timestamp(),
+            'level': record['level'].name.lower(),
+            'message': record['message'],
+            'source': f"{record['file'].path}:{record['line']}",
+            'module': record['module'],
+            'function': record['function'],
+            'process_id': record['process'].id,
+            'thread_id': record['thread'].id,
+            **record['extra']
+        }
+
+        if not fluent_sender.emit(os.getenv("FLUENTD_TOPIC"), log_data):
+            print(f"Fluentd 发送失败: {fluent_sender.last_error}")
+
+    except Exception as e:
+        print(f"日志处理异常: {str(e)}")
+
+
+# 移除默认处理器
+logger.remove()
+
+# 添加控制台处理器
+logger.add(
+    sink=sys.stdout,
+    level="INFO",
+    format=STDOUT_FORMAT,
+    colorize=True,
+    backtrace=True,  # 显示完整异常堆栈
+    diagnose=True,  # 显示详细异常信息
+)
+
+logger.add(
+    log_to_fluent,
+    level="INFO",  # 处理 INFO 及以上级别
+    format="{message}",  # 原始消息（实际使用结构化数据）
+    backtrace=True,  # 启用堆栈回溯
+    diagnose=True  # 显示诊断信息
+)
+
+atexit.register(fluent_sender.close)
+
+# 导出配置好的logger
+__all__ = ["logger"]

main.py

@@ -1,6 +1,7 @@
 from fastapi import FastAPI
 from starlette.middleware.cors import CORSMiddleware
 
+from config.exceptions import global_exception_handler
 from routers import routers
 
 app = FastAPI(title="Family Service")
@@ -12,5 +13,7 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+app.middleware("http")(global_exception_handler)
+
 for router in routers:
     app.include_router(router)

requirements.txt

@@ -9,4 +9,6 @@ uvicorn~=0.23.0
 pymysql~=1.2.0
 boto3~=1.40.59
 botocore~=1.40.59
-python-multipart~=0.0.20
+python-multipart~=0.0.20
+fluent-logger~=0.10.0
+loguru~=0.7.3

service/comment_service.py

@@ -8,6 +8,7 @@ from service.record_service import get_record_by_id
 
 def get_comment_by_id(db: Session, comment_id: int) -> Comments | None:
     result = db.execute(select(Comments).where(Comments.id == comment_id))
+
     return result.scalar_one_or_none()
 
 
@@ -33,6 +34,7 @@ def update_comment(db: Session, comment_id: int, comment_dto: CommentUpdate) ->
     comment.content = comment_dto.content
     db.commit()
     db.refresh(comment)
+
     return True
 
 

service/user_service.py

@@ -19,10 +19,10 @@ def create_user(db: Session, user: UserCreate) -> Users:
     return user
 
 
-def update_user(db: Session, user_id: int, obj_in: UserUpdate) -> Users:
+def update_user(db: Session, user_id: int, obj_in: UserUpdate) -> Users | None:
     user = get_user_by_id(db, user_id)
     if not user:
-        return False
+        return None
 
     for field, value in obj_in.model_dump(exclude_unset=True).items():
         setattr(user, field, value)