# 操作日志系统(审计追踪模式) > **适用场景**: 任何需要审计追踪的 CRUD 管理系统 > **创建日期**: 2026-06-29 --- ## 四层架构 | 层次 | 文件 | 职责 | |:----|:-----|:-----| | 模型 | `models/operation_log.py` | 定义数据库表结构 | | 服务 | `services/operation_log.py` | 工具函数 `log_operation()` | | API | `routers/logs.py` | `GET /api/logs` 查询接口 | | 埋点 | 各 CRUD 路由中 | 在 create/update/delete 后调用 | --- ## 模型定义 ```python from sqlalchemy import Column, Integer, String, DateTime, Text from sqlalchemy.sql import func from app.database import Base class OperationLog(Base): __tablename__ = "operation_logs" id = Column(Integer, primary_key=True, index=True) user = Column(String(64), default="admin", comment="操作人") action = Column(String(32), nullable=False, comment="操作类型: create/update/delete") entity_type = Column(String(32), nullable=False, comment="实体类型") entity_id = Column(Integer, nullable=True, comment="实体ID") entity_name = Column(String(128), nullable=True, comment="实体名称(冗余,方便展示)") detail = Column(Text, nullable=True, comment="操作详情") created_at = Column(DateTime(timezone=True), server_default=func.now(), comment="操作时间") ``` ## 服务层 ```python from sqlalchemy.orm import Session from app.models.operation_log import OperationLog def log_operation( db: Session, action: str, entity_type: str, entity_id: int = None, entity_name: str = None, detail: str = None, user: str = "admin", ): log = OperationLog( user=user, action=action, entity_type=entity_type, entity_id=entity_id, entity_name=entity_name, detail=detail, ) db.add(log) db.commit() ``` ## API 路由 ```python from fastapi import APIRouter, Depends, Query from sqlalchemy.orm import Session from app.database import get_db from app.models.operation_log import OperationLog from typing import Optional router = APIRouter(prefix="/api/logs", tags=["操作日志"]) @router.get("") def list_logs( entity_type: Optional[str] = Query(None), action: Optional[str] = Query(None), limit: int = Query(100, ge=1, le=500), offset: int = Query(0, ge=0), db: Session = Depends(get_db), ): q = db.query(OperationLog).order_by(OperationLog.created_at.desc()) if entity_type: q = q.filter(OperationLog.entity_type == entity_type) if action: q = q.filter(OperationLog.action == action) total = q.count() logs = q.offset(offset).limit(limit).all() return { "total": total, "logs": [ { "id": log.id, "user": log.user, "action": log.action, "entity_type": log.entity_type, "entity_id": log.entity_id, "entity_name": log.entity_name, "detail": log.detail, "created_at": log.created_at.isoformat() if log.created_at else None, } for log in logs ], } ``` ## 埋点注入模式 ### 创建后 ```python log_operation(db, "create", "project", p.id, p.project_name, f"创建项目: {p.project_code}") ``` ### 更新后 ```python log_operation(db, "update", "project", p.id, p.project_name, f"更新项目: {p.project_code}") ``` ### 删除前(需要 entity_name 做记录) ```python log_operation(db, "delete", "project", project_id, p.project_name if p else None, f"删除项目: ID={project_id}") ``` ### 批量导入 ```python log_operation(db, "create", "import", None, filename, f"Excel导入: {stats}") ``` ## 注册路由 在 `main.py` 中: ```python from app.routers import auth, groups, personnel, projects, settings, dashboard, reports, search, finance, logs # 受保护路由 app.include_router(logs.router, dependencies=[Depends(verify_token)]) ``` ## 前端页面 ```vue ``` ## 路由注册 + 侧边栏 ```javascript // router/index.js { path: 'logs', name: 'Logs', component: () => import('../views/logs/Index.vue'), } // Layout.vue 侧边栏 操作日志 ``` ## 适用实体类型 | entity_type | 含义 | 埋点位置 | |:------------|:-----|:---------| | project | 项目 | projects.py CRUD | | task | WBS 任务 | projects.py task CRUD | | milestone | 里程碑 | projects.py milestone CRUD | | income | 月度收益 | projects.py income CRUD | | personnel | 人员 | personnel.py CRUD | | group | 组别 | groups.py CRUD | | setting | 系统设置 | settings.py CRUD | | import | 数据导入 | import_data.py | ## 注意事项 1. **删除操作需在 delete 前记录**:`db.delete()` 后对象属性仍可访问,但 `db.commit()` 后不可用 2. **SQLAlchemy Column 类型警告**:Pyright 会报 `Column[int]` 不能赋值给 `int` 的参数类型错误,这是已知假阳性,不影响运行 3. **不要过度埋点**:GET 查询操作不需要记录,只记录 create/update/delete 4. **entity_name 冗余存储**:即使关联实体被删除,日志中仍保留名称用于展示