日志与审计系统文档

概述

邮轮穿舱件管理系统后台采用分层架构设计，实现了完整的日志记录、访问审计和错误追踪功能。系统基于FastAPI框架构建，使用Loguru进行日志管理，Tortoise-ORM进行数据持久化，提供了全面的操作审计和错误监控能力。

系统架构

整体架构图

flowchart TD
    subgraph 前端层
        A[Web客户端]
        B[移动端应用]
    end
    
    subgraph API网关层
        C[FastAPI应用]
    end
    
    subgraph 业务逻辑层
        D[日志路由器]
        E[访问日志服务]
        F[异常处理中间件]
    end
    
    subgraph 数据访问层
        G[访问日志模型]
        H[数据库]
    end
    
    subgraph 基础设施层
        I[日志配置]
        J[文件系统]
    end
    
    A --> C
    B --> C
    C --> D
    D --> E
    E --> G
    G --> H
    C --> F
    F --> I
    I --> J

核心组件关系

classDiagram
    class AccessLog {
        +IntField log_id
        +CharField action_type
        +CharField action_title
        +CharField action_description
        +CharField related_users
        +CharField related_tickets
        +CharField related_workpieces
        +IntField created_by
        +IntField updated_by
        +DatetimeField created_at
        +DatetimeField updated_at
    }
    
    class AccessLogService {
        +create_access_log()
        +get_access_log_by_id()
        +get_all_access_logs()
        +get_logs_by_action_type()
        +get_logs_by_user()
        +update_access_log()
        +delete_access_log()
        +search_logs_by_keyword()
    }
    
    class LogRouter {
        +get_access_log_detail()
        +get_all_access_logs()
        +get_logs_by_type()
        +get_logs_by_user()
        +search_logs_by_keyword()
        +delete_access_log()
    }
    
    class LoggerConfig {
        +setup_logger()
        +LOG_FORMAT
        +LOG_RETENTION_DAYS
    }
    
    class GlobalExceptionMiddleware {
        +dispatch()
        +_handle_exception()
    }
    
    AccessLogService --> AccessLog
    LogRouter --> AccessLogService
    GlobalExceptionMiddleware --> LoggerConfig

日志系统实现

日志配置模块

系统使用Loguru库进行日志管理，提供了灵活的配置选项和强大的日志功能。

核心配置参数：

日志级别：DEBUG/INFO模式可配置
日志格式：包含时间戳、级别、模块名、函数名和行号
日志轮转：每天0点自动创建新文件
日志保留：默认保留7天日志文件
日志压缩：自动压缩为zip格式

配置文件： app/core/loggers.py

# 日志配置示例
LOG_FORMAT = (
    "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
    "<level>{level: <8}</level> | "
    "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
    "<level>{message}</level>"
)

def setup_logger(debug: bool = False) -> None:
    """初始化日志配置"""
    _logger.remove()
    
    # 控制台日志配置
    _logger.add(
        sys.stdout,
        level="DEBUG" if debug else "INFO",
        format=LOG_FORMAT,
        enqueue=True,  # 多线程安全
        backtrace=True,  # 异常堆栈
        diagnose=True,  # 变量诊断
    )
    
    # 文件日志配置
    _logger.add(
        str(LOG_DIR / "fastapi_{time:YYYY-MM-DD}.log"),
        rotation="00:00",  # 每天轮转
        retention=f"{LOG_RETENTION_DAYS} days",
        compression="zip",
        level="DEBUG" if debug else "INFO",
        format=LOG_FORMAT,
        encoding="utf-8",
        enqueue=True,
    )

日志数据流

sequenceDiagram
    participant Client as 客户端
    participant Router as 路由层
    participant Service as 服务层
    participant Model as 数据模型
    participant Logger as 日志系统
    participant File as 文件系统
    
    Client->>Router: HTTP请求
    Router->>Service: 调用业务逻辑
    Service->>Model: 数据操作
    Model->>Service: 返回结果
    Service->>Logger: 记录操作日志
    Logger->>File: 写入日志文件
    Service->>Router: 返回业务结果
    Router->>Client: HTTP响应

访问审计系统

数据模型设计

访问审计系统采用关系型数据模型，记录用户操作的关键信息。

访问日志实体关系图：

erDiagram
    ACCESS_LOG ||--o{ USER : created_by
    ACCESS_LOG ||--o{ TICKET : related_tickets
    ACCESS_LOG ||--o{ WORKPIECE : related_workpieces
    ACCESS_LOG {
        int log_id PK
        string action_type
        string action_title
        string action_description
        string related_users
        string related_tickets
        string related_workpieces
        int created_by FK
        int updated_by FK
        datetime created_at
        datetime updated_at
    }

数据模型定义： app/models/access_log.py

class AccessLog(Model):
    """访问日志数据模型"""
    log_id = fields.IntField(description="日志ID", primary_key=True)
    action_type = fields.CharField(max_length=255, description="操作类型")
    action_title = fields.CharField(max_length=255, description="操作标题")
    action_description = fields.CharField(max_length=255, description="操作描述")
    related_users = fields.CharField(max_length=255, null=True, description="相关用户")
    related_tickets = fields.CharField(max_length=255, null=True, description="相关工单")
    related_workpieces = fields.CharField(max_length=255, null=True, description="相关工件")
    created_by = fields.IntField(description="创建人", null=True, default=-1)
    updated_by = fields.IntField(description="更新人", null=True, default=-1)
    created_at = fields.DatetimeField(description="创建时间", auto_now_add=True)
    updated_at = fields.DatetimeField(description="更新时间", auto_now=True)

服务层实现

访问日志服务提供了完整的CRUD操作和查询功能。

服务层功能： app/service/access_log_service.py

async def service_create_access_log(
    action_type: str,
    action_title: str,
    action_description: str = "",
    related_users: str = "",
    related_tickets: str = "",
    related_workpieces: str = "",
    created_by: int = -1
) -> AccessLog:
    """创建访问日志记录"""
    try:
        access_log = await AccessLog.create(
            action_type=action_type,
            action_title=action_title,
            action_description=action_description,
            related_users=related_users,
            related_tickets=related_tickets,
            related_workpieces=related_workpieces,
            created_by=created_by
        )
        logger.info(f"Created access log with ID: {access_log.log_id}")
        return access_log
    except IntegrityError as e:
        logger.error(f"Error creating access log: {e}")
        raise HTTPException(status_code=400, detail="创建访问日志失败")

服务方法列表：

service_create_access_log() - 创建访问日志
service_get_access_log_by_id() - 根据ID查询日志
service_get_all_access_logs() - 获取所有日志（分页）
service_get_access_logs_by_action_type() - 按操作类型过滤
service_get_access_logs_by_user() - 按用户过滤
service_update_access_log() - 更新日志信息
service_delete_access_log() - 删除日志记录
service_search_access_logs_by_keyword() - 关键词搜索

API接口设计

系统提供了RESTful风格的API接口进行访问日志管理。

路由配置： app/routers/log_router.py

log_router = APIRouter(
    prefix="/log",
    tags=["审计日志"],
)

@log_router.get("/detail/{log_id}")
async def get_access_log_detail(log_id: int):
    """获取访问日志详情"""
    access_log = await service_get_access_log_by_id(log_id)
    if not access_log:
        raise HTTPException(status_code=404, detail="访问日志不存在")
    return access_log

API端点：

GET /log/detail/{log_id} - 获取日志详情
GET /log/all - 获取所有日志（分页）
GET /log/type/{action_type} - 按类型查询日志
GET /log/user/{user_id} - 按用户查询日志
GET /log/search - 关键词搜索日志
DELETE /log/{log_id} - 删除日志记录

错误追踪系统

全局异常处理

系统实现了全局异常处理中间件，确保所有未捕获的异常都能被正确处理和记录。

异常处理流程： app/core/middleware.py

stateDiagram-v2
    [*] --> 请求接收
    请求接收 --> 异常处理: 调用业务逻辑
    异常处理 --> 值错误: ValueError
    异常处理 --> 权限错误: PermissionError
    异常处理 --> 文件未找到: FileNotFoundError
    异常处理 --> 未知异常: 其他异常类型
    
    值错误 --> 记录日志: 记录错误信息
    权限错误 --> 记录日志: 记录错误信息
    文件未找到 --> 记录日志: 记录错误信息
    未知异常 --> 记录日志: 记录错误信息
    
    记录日志 --> 返回响应: 生成错误响应
    返回响应 --> [*]

异常处理实现：

class GlobalExceptionMiddleware(BaseHTTPMiddleware):
    """全局异常处理中间件"""
    
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        try:
            response = await call_next(request)
            return response
        except Exception as exc:
            return await self._handle_exception(request, exc)
    
    async def _handle_exception(self, request: Request, exc: Exception) -> JSONResponse:
        # 记录详细错误日志
        error_msg = f"Unhandled exception in {request.method} {request.url}: {str(exc)}"
        logger.error(error_msg)
        logger.error(f"Exception traceback: {traceback.format_exc()}")
        
        # 根据异常类型返回相应错误响应
        if isinstance(exc, ValueError):
            return JSONResponse(status_code=400, content={"error": "Bad Request"})
        elif isinstance(exc, PermissionError):
            return JSONResponse(status_code=403, content={"error": "Forbidden"})
        elif isinstance(exc, FileNotFoundError):
            return JSONResponse(status_code=404, content={"error": "Not Found"})
        else:
            return JSONResponse(status_code=500, content={"error": "Internal Server Error"})

错误处理策略

系统采用分层次的错误处理策略：

业务层错误：在服务层进行捕获和处理
数据层错误：通过ORM异常进行处理
系统层错误：通过全局异常中间件处理

系统依赖关系分析

模块依赖图

flowchart TD
    subgraph 核心模块
        A[loggers.py]
        B[access_log.py]
    end
    
    subgraph 服务模块
        C[access_log_service.py]
    end
    
    subgraph 路由模块
        D[log_router.py]
    end
    
    subgraph 中间件模块
        E[middleware.py]
    end
    
    subgraph 外部依赖
        F[Loguru]
        G[Tortoise-ORM]
        H[FastAPI]
    end
    
    A --> F
    B --> G
    C --> A
    C --> B
    D --> C
    E --> A
    D --> H
    E --> H

依赖关系说明

直接依赖：

access_log_service.py 依赖 loggers.py 和 access_log.py
log_router.py 依赖 access_log_service.py
middleware.py 依赖 loggers.py

外部依赖：

Loguru：日志记录功能
Tortoise-ORM：数据持久化
FastAPI：Web框架和路由功能

性能分析与优化建议

性能考虑

日志写入性能：使用异步写入和队列机制确保高性能
数据库查询性能：通过分页和索引优化查询效率
内存使用：合理的日志保留策略减少存储压力

优化建议

日志级别动态调整：生产环境建议关闭DEBUG级别日志
数据库索引优化：为常用查询字段添加索引
日志文件压缩：启用压缩减少存储空间占用
监控告警：集成监控系统实现实时告警

索引

邮轮穿舱件管理系统的日志与审计模块实现了完整的操作追踪、错误监控和安全审计功能。系统采用分层架构设计，各模块职责清晰，依赖关系合理。通过Loguru和Tortoise-ORM的组合，提供了高性能的日志记录和数据持久化能力。全局异常处理机制确保了系统的稳定性和可维护性。

关键特性：

完整的操作审计追踪
多层次的错误处理机制
灵活的日志配置选项
RESTful风格的API接口
高性能的日志记录系统

该系统为邮轮穿舱件管理提供了可靠的安全审计和运维监控基础，满足了企业级应用对日志和审计的高标准要求。

概述​

系统架构​

整体架构图​

核心组件关系​

日志系统实现​

日志配置模块​

日志数据流​

访问审计系统​

数据模型设计​

服务层实现​

API接口设计​

错误追踪系统​

全局异常处理​

错误处理策略​

系统依赖关系分析​

模块依赖图​

依赖关系说明​

性能分析与优化建议​

性能考虑​

优化建议​

索引​

概述