跳到主要内容

邮轮穿舱件管理系统后台 - 统计API文档

文档概述​

本文档详细介绍了邮轮穿舱件管理系统后台的统计API接口，重点分析routers/stat_router.py和wechat/statistics.py两个核心文件。通过深入分析代码结构、依赖关系和实现逻辑，为开发者提供全面的技术参考。

参考源文件：

目录​

  1. 系统架构概述
  2. 统计API路由分析
  3. 依赖关系分析
  4. 数据流分析
  5. API接口详解
  6. 错误处理机制
  7. 性能优化建议

系统架构概述​

整体架构图​

flowchart TD
    subgraph 前端层
        A[Web客户端]
        B[微信小程序]
    end
    
    subgraph API路由层
        C[stat_router.py<br/>统计路由]
        D[statistics.py<br/>微信统计路由]
    end
    
    subgraph 服务层
        E[access_log_service.py]
        F[user_service.py]
        G[ticket_service.py]
        H[image_service.py]
    end
    
    subgraph 数据层
        I[AccessLog模型]
        J[User模型]
        K[Ticket模型]
        L[Image模型]
    end
    
    A --> C
    B --> D
    C --> E
    C --> F
    C --> G
    C --> H
    D --> G
    E --> I
    F --> J
    G --> K
    H --> L

组件职责说明​

  • API路由层：负责接收HTTP请求、参数验证和响应格式化
  • 服务层：封装业务逻辑，处理数据操作和业务规则
  • 数据层：通过ORM模型与数据库进行交互

统计API路由分析​

主统计路由 (stat_router.py)​

文件位置： app/routers/stat_router.py 引用源文件

from fastapi import APIRouter
from app.service.access_log_service import service_get_log_count
from app.service.user_service import service_get_all_users
from app.service.image_service import service_get_image_entity
from app.service.ticket_service import service_get_all_tickets

stat_router = APIRouter(
    prefix="/stat",
    tags=["统计管理"],
)

关键特性：

  • 路由前缀：/stat
  • API标签：统计管理
  • 无认证依赖（公开接口）

微信统计路由 (statistics.py)​

文件位置： app/routers/wechat/statistics.py 引用源文件

from fastapi import APIRouter, Depends, HTTPException, status
from app.core.authorize import verify_jwt_token
from app.service.ticket_service import service_get_all_ticket_entitys

statistics_router = APIRouter(
    prefix="/statistics",
    tags=["统计管理"],
    dependencies=[Depends(verify_jwt_token)],
)

关键特性：

  • 路由前缀：/statistics
  • API标签：统计管理
  • JWT令牌认证依赖

依赖关系分析​

模块依赖图​

flowchart TD
    subgraph 路由层
        A[stat_router.py]
        B[statistics.py]
    end
    
    subgraph 服务层
        C[access_log_service.py]
        D[user_service.py]
        E[ticket_service.py]
        F[image_service.py]
    end
    
    subgraph 核心组件
        G[authorize.py<br/>认证模块]
        H[loggers.py<br/>日志模块]
    end
    
    subgraph 数据模型层
        I[AccessLog]
        J[User]
        K[Ticket]
        L[Image]
    end
    
    A --> C
    A --> D
    A --> E
    A --> F
    B --> E
    B --> G
    C --> I
    C --> H
    D --> J
    D --> H
    E --> K
    E --> H
    F --> L
    F --> H

依赖关系表​

模块依赖模块依赖类型说明
stat_router.pyaccess_log_service强依赖统计计数功能
stat_router.pyuser_service强依赖用户数据统计
stat_router.pyticket_service强依赖工单数据统计
stat_router.pyimage_service强依赖图片数据统计
statistics.pyticket_service强依赖工单统计功能
statistics.pyauthorize强依赖JWT认证

数据流分析​

统计摘要接口数据流​

sequenceDiagram
    participant Client as 客户端
    participant StatRouter as stat_router.py
    participant UserService as user_service.py
    participant TicketService as ticket_service.py
    participant ImageService as image_service.py
    participant LogService as access_log_service.py
    participant DB as 数据库
    
    Client->>StatRouter: GET /stat/countInfo
    StatRouter->>UserService: service_get_all_users()
    UserService->>DB: User.all()
    DB-->>UserService: 用户数据
    UserService-->>StatRouter: 用户列表
    
    StatRouter->>TicketService: service_get_all_tickets()
    TicketService->>DB: Ticket.all()
    DB-->>TicketService: 工单数据
    TicketService-->>StatRouter: 工单列表
    
    StatRouter->>ImageService: service_get_image_entity()
    ImageService->>DB: Image.all()
    DB-->>ImageService: 图片数据
    ImageService-->>StatRouter: 图片列表
    
    StatRouter->>LogService: service_get_log_count()
    LogService->>DB: AccessLog.all().count()
    DB-->>LogService: 日志计数
    LogService-->>StatRouter: 日志数量
    
    StatRouter-->>Client: 统计摘要JSON

微信统计接口数据流​

sequenceDiagram
    participant Client as 微信客户端
    participant Auth as 认证模块
    participant StatRouter as statistics.py
    participant TicketService as ticket_service.py
    participant DB as 数据库
    
    Client->>StatRouter: GET /statistics/brief
    StatRouter->>Auth: verify_jwt_token()
    Auth-->>StatRouter: 认证结果
    
    StatRouter->>TicketService: service_get_all_ticket_entitys()
    TicketService->>DB: Ticket.all()
    DB-->>TicketService: 工单数据
    TicketService-->>StatRouter: 工单列表
    
    StatRouter-->>Client: 工单统计JSON

API接口详解​

1. 统计摘要接口​

端点： GET /stat/countInfo

功能描述： 获取系统统计摘要，包括用户数、工单数、图片数、访问日志数

实现代码： 引用源文件

@stat_router.get("/countInfo",
    name="统计摘要接口",
    summary="统计摘要接口",
    description="获取系统统计摘要，包括用户数、工单数、图片数、访问日志数"
)
async def get_log_count():
    """
    获取简要统计信息
    
    :return: 统计信息
    :rtype: dict
    """
    user_orm = await service_get_all_users()
    image_orm = await service_get_image_entity()
    ticket_orm = await service_get_all_tickets()
    return {
        "user_count": len(user_orm),
        "ticket_count": len(ticket_orm),
        "image_count": len(image_orm),
        "log_count": await service_get_log_count(),
    }

响应格式：

{
    "user_count": 150,
    "ticket_count": 45,
    "image_count": 320,
    "log_count": 1280
}

2. 微信工单统计接口​

端点： GET /statistics/brief

功能描述： 获取工单统计摘要，包括总工单数、打开工单数、关闭工单数

实现代码： 引用源文件

@statistics_router.get("/brief",
    name="统计摘要接口",
    summary="统计摘要接口",
    description="获取工单统计摘要，包括总工单数、打开工单数、关闭工单数"
)
async def api_get_statistics_brief() -> dict:
    """
    获取工单统计摘要
    :return: 工单统计摘要
    :rtype: dict
    """
    try:
        tickets = await service_get_all_ticket_entitys()
        total_tickets = len(tickets)
        open_tickets = len([ticket for ticket in tickets if ticket.ticket_status == "open"])
        closed_tickets = len([ticket for ticket in tickets if ticket.ticket_status == "closed"])
        return {
            "total_tickets": total_tickets,
            "open_tickets": open_tickets,
            "closed_tickets": closed_tickets,
        }
    except Exception as e:
        logger.error(f"获取工单统计摘要失败：{e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="获取工单统计摘要失败")

响应格式：

{
    "total_tickets": 45,
    "open_tickets": 12,
    "closed_tickets": 33
}

服务层实现分析​

访问日志服务 (access_log_service.py)​

核心功能： 引用源文件

async def service_get_log_count()->int:
    """
    获取访问日志总数
    
    :return: 访问日志总数
    :rtype: int
    """
    try:
        count = await AccessLog.all().count()
        logger.info(f"Total access logs count: {count}")
        return count
    except Exception as e:
        logger.error(f"Error retrieving access logs count: {e}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="服务器内部错误")

用户服务 (user_service.py)​

核心功能： 引用源文件

async def service_get_all_users(skip: int = 0, limit: int = 100) -> List[UserSchemaOut]:
    """
    获取所有用户
    
    :param skip: 跳过的记录数
    :type skip: int
    :param limit: 限制返回的记录数
    :type limit: int
    :return: 用户列表
    :rtype: List[UserSchemaOut]
    """
    try:
        users = await User.all().offset(skip).limit(limit)
        logger.info(f"成功获取用户列表，数量: {len(users)}")
        
        return [
            UserSchemaOut(
                id=user.id,
                username=user.username,
                email=user.email,
                sms=user.sms,
                status=user.status,
                openid=user.openid,
                is_system=user.is_system,
                created_by=user.created_by,
                updated_by=user.updated_by,
                created_at=user.created_at,
                updated_at=user.updated_at
            )
            for user in users
        ]
    except Exception as e:
        logger.error(f"获取用户列表失败: {str(e)}")
        raise

工单服务 (ticket_service.py)​

核心功能： 引用源文件

async def service_get_all_tickets() -> List[TicketOutSchema]:
    """
    获取所有工单
    """
    try:
        tickets = await Ticket.all()
        return tickets
    except DoesNotExist:
        logger.error(f"获取所有工单失败，工单不存在")
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="工单不存在")

图片服务 (image_service.py)​

核心功能： 引用源文件

async def service_get_image_entity() -> List[Image]:
    image_entity = await Image.all()
    return image_entity

错误处理机制​

异常处理架构​

flowchart TD
    A[API请求] --> B{参数验证}
    B -->|通过| C[业务逻辑处理]
    B -->|失败| D[返回400错误]
    
    C --> E{数据库操作}
    E -->|成功| F[返回成功响应]
    E -->|失败| G{异常类型判断}
    
    G -->|DoesNotExist| H[返回404错误]
    G -->|IntegrityError| I[返回400错误]
    G -->|其他异常| J[返回500错误]
    
    H --> K[记录错误日志]
    I --> K
    J --> K

错误码说明​

错误码错误类型说明处理方式
400Bad Request参数验证失败或数据完整性错误检查请求参数
404Not Found请求的资源不存在检查资源ID是否正确
500Internal Server Error服务器内部错误查看服务器日志

性能优化建议​

1. 数据库查询优化​

当前实现：

# 统计接口需要4次独立的数据库查询
user_orm = await service_get_all_users()      # 查询1
image_orm = await service_get_image_entity()  # 查询2  
ticket_orm = await service_get_all_tickets()   # 查询3
log_count = await service_get_log_count()     # 查询4

优化建议：

# 使用异步并发执行
import asyncio

async def get_count_info():
    # 并发执行所有查询
    user_task = service_get_all_users()
    image_task = service_get_image_entity()
    ticket_task = service_get_all_tickets()
    log_task = service_get_log_count()
    
    user_orm, image_orm, ticket_orm, log_count = await asyncio.gather(
        user_task, image_task, ticket_task, log_task
    )
    
    return {
        "user_count": len(user_orm),
        "ticket_count": len(ticket_orm),
        "image_count": len(image_orm),
        "log_count": log_count,
    }

2. 缓存策略​

建议实现缓存层：

from app.core.runtimecache import cache

@cache(ttl=300)  # 缓存5分钟
async def get_cached_count_info():
    return await get_count_info()

3. 分页统计优化​

对于大数据量的统计，建议实现分页统计接口：

@stat_router.get("/countInfo/paginated")
async def get_paginated_count_info(skip: int = 0, limit: int = 1000):
    """分页获取统计信息，避免大数据量查询"""
    # 实现分页逻辑
    pass

索引​

本文档全面分析了邮轮穿舱件管理系统后台的统计API接口，通过详细的代码结构分析、依赖关系映射和数据流可视化，为开发者提供了深入的技术参考。系统采用清晰的分层架构，具有良好的可维护性和扩展性。

关键要点：

  1. 系统采用FastAPI框架，提供RESTful API接口
  2. 统计功能分为公开接口和认证接口两种类型
  3. 依赖关系清晰，模块职责明确
  4. 错误处理机制完善，具备良好的容错性
  5. 性能优化空间较大，建议实现缓存和并发查询

下一步建议：

  • 实现统计数据的缓存机制
  • 添加更细粒度的统计维度（如时间范围统计）
  • 实现统计数据的实时更新通知
  • 添加统计数据的可视化展示接口