"""审计日志服务""" import os from datetime import datetime, timedelta from typing import Optional from sqlalchemy.orm import Session from sqlalchemy import and_ from app.models.audit_log import AuditLog AUDIT_LOG_DIR = os.environ.get("AUDIT_LOG_DIR", "/app/logs/audit") AUDIT_LOG_RETENTION_DAYS = int(os.environ.get("AUDIT_LOG_RETENTION_DAYS", "90")) # 路由分类规则：(method, path_pattern, exact, action_type, action_subtype, resource_type, description) # exact=True 精确匹配路径；exact=False 前缀匹配。 # 精确匹配规则排在前，前缀匹配按从长到短排列，避免短前缀误匹配。 _ROUTE_MAP = [ # ── 认证 ────────────────────────────────────────────────────── ("POST", "/api/auth/callback", True, "auth", "login", "user", "用户登录"), ("GET", "/api/auth/profile", True, "auth", "profile", "user", "查看个人信息"), # ── OLT：精确路径（不含 ID 段）──────────────────────────────── ("GET", "/api/olt/devices", True, "olt", "list", "olt", "查询 OLT 设备列表"), ("POST", "/api/olt/devices", True, "olt", "create", "olt", "新建 OLT 设备"), ("POST", "/api/olt/import", True, "olt", "import", "olt", "批量导入 OLT 设备"), ("POST", "/api/olt/quick-scan", True, "olt", "quick_scan", "olt", "触发快速扫描"), ("POST", "/api/olt/loopback-detection", True, "olt", "loopback", "olt", "触发环路检测"), # OLT 端口操作（路径含 ID，前缀匹配；/ports/toggle 比 /devices/ 更具体，先列） ("POST", "/api/olt/devices/", False, "olt", "port_toggle", "olt", "切换 OLT 端口状态"), ("GET", "/api/olt/devices/", False, "olt", "port_list", "olt", "查询 OLT 端口列表"), ("PUT", "/api/olt/devices/", False, "olt", "update", "olt", "更新 OLT 设备信息"), ("DELETE", "/api/olt/devices/", False, "olt", "delete", "olt", "删除 OLT 设备"), # 重复 MAC ("POST", "/api/olt/duplicate-macs/", False, "olt", "port_clear", "olt", "清除重复 MAC 端口占用"), ("DELETE", "/api/olt/duplicate-macs/", False, "olt", "mac_delete", "olt", "删除重复 MAC 记录"), # 新发现设备 ("PUT", "/api/olt/new-devices/", False, "olt", "device_fill", "olt", "补全新发现设备信息"), ("DELETE", "/api/olt/new-devices/", False, "olt", "device_ignore", "olt", "忽略新发现设备"), # ── ONU 设备 ────────────────────────────────────────────────── ("POST", "/api/check/status", True, "system", "scan_trigger", "device", "手动触发全量状态扫描"), ("POST", "/api/import/upload", True, "device", "import", "device", "批量导入 ONU 设备"), ("DELETE", "/api/devices/status/all", True, "device", "status_clear", "device", "清除所有设备状态"), ("POST", "/api/devices/", False, "device", "refresh", "device", "刷新单台设备在线状态"), ("PUT", "/api/devices/", False, "device", "update", "device", "更新 ONU 设备信息"), ("DELETE", "/api/devices/", False, "device", "delete", "device", "删除 ONU 设备"), # 设备更换（路径含 /replace，需在 refresh 前匹配，通过 _classify 特殊处理） ("POST", "/replace", False, "device", "replace", "device", "更换设备 MAC 地址"), # ── 用户管理 ────────────────────────────────────────────────── ("POST", "/api/users", True, "user", "create", "user", "创建用户"), ("PUT", "/api/users/", False, "user", "update", "user", "更新用户信息"), ("DELETE", "/api/users/", False, "user", "delete", "user", "删除用户"), # ── 角色权限 ────────────────────────────────────────────────── ("PUT", "/api/roles/", False, "user", "role_update", "role", "更新角色权限配置"), # ── 系统设置 ────────────────────────────────────────────────── ("PUT", "/api/settings/check_interval", True, "system", "config_update", "system", "更新定时扫描间隔"), ("PUT", "/api/settings/", False, "system", "config_update", "system", "更新系统配置"), # ── 库存管理（精确路径优先）─────────────────────────────────── ("POST", "/api/inventory/transactions/purchase", True, "inventory", "purchase", "inventory", "物料采购入库"), ("POST", "/api/inventory/transactions/allocate", True, "inventory", "allocate", "inventory", "物料分配出库"), ("POST", "/api/inventory/transactions/return", True, "inventory", "return", "inventory", "物料退库"), ("POST", "/api/inventory/categories", True, "inventory", "cat_create", "inventory", "新建库存分类"), ("POST", "/api/inventory/materials", True, "inventory", "mat_create", "inventory", "新建物料"), ("PUT", "/api/inventory/materials/", False, "inventory", "mat_update", "inventory", "更新物料信息"), ("DELETE", "/api/inventory/materials/", False, "inventory", "mat_delete", "inventory", "删除物料"), ("POST", "/api/inventory/checks", True, "inventory", "check_create", "inventory", "发起库存盘点"), ("POST", "/api/inventory/checks/", False, "inventory", "check_adjust", "inventory", "盘点差异调整"), ("PUT", "/api/inventory/checks/", False, "inventory", "check_update", "inventory", "更新盘点记录"), ] def _classify(method: str, path: str): """根据请求方法和路径推断操作分类，精确匹配优先于前缀匹配""" # 第一轮：精确匹配 for m, p, exact, atype, subtype, rtype, desc in _ROUTE_MAP: if exact and method == m and path == p: return atype, subtype, rtype, desc # 第二轮：后缀匹配（path_pattern 以 "/" 开头但不含 "/api"，视为后缀） for m, p, exact, atype, subtype, rtype, desc in _ROUTE_MAP: if not exact and not p.startswith('/api') and method == m and path.endswith(p): return atype, subtype, rtype, desc # 第三轮：前缀匹配（规则列表已按从具体到宽泛排列） for m, p, exact, atype, subtype, rtype, desc in _ROUTE_MAP: if not exact and p.startswith('/api') and method == m and path.startswith(p): return atype, subtype, rtype, desc return "system", "request", "unknown", f"{method} {path}" def _get_ip(request) -> str: forwarded = request.headers.get("x-forwarded-for") if forwarded: return forwarded.split(",")[0].strip() if request.client: return request.client.host return "" def write_audit_log( db: Session, *, user_id: str, username: str, user_role: str, method: str, path: str, ip_address: str = "", user_agent: str = "", status_code: int, request_params: Optional[dict] = None, response_data: Optional[dict] = None, error_message: Optional[str] = None, description: Optional[str] = None, resource_id: Optional[str] = None, resource_name: Optional[str] = None, ): action_type, action_subtype, resource_type, default_desc = _classify(method, path) status = "success" if status_code < 400 else ("failed" if status_code < 500 else "error") log = AuditLog( user_id=user_id, username=username, user_role=user_role, action_type=action_type, action_subtype=action_subtype, ip_address=ip_address, user_agent=user_agent[:500] if user_agent else "", request_method=method, request_path=path, status=status, status_code=status_code, resource_type=resource_type, resource_id=resource_id, resource_name=resource_name, description=description or default_desc, request_params=request_params, response_data=response_data if status_code < 400 else None, error_message=error_message, ) db.add(log) db.commit() return log.id def query_logs( db: Session, *, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, user_id: Optional[str] = None, username: Optional[str] = None, action_type: Optional[str] = None, resource_type: Optional[str] = None, status: Optional[str] = None, page: int = 1, page_size: int = 50, ): q = db.query(AuditLog) filters = [] if start_time: filters.append(AuditLog.action_time >= start_time) if end_time: filters.append(AuditLog.action_time <= end_time) if user_id: filters.append(AuditLog.user_id == user_id) if username: filters.append(AuditLog.username.ilike(f"%{username}%")) if action_type: filters.append(AuditLog.action_type == action_type) if resource_type: filters.append(AuditLog.resource_type == resource_type) if status: filters.append(AuditLog.status == status) if filters: q = q.filter(and_(*filters)) total = q.count() items = q.order_by(AuditLog.action_time.desc()).offset((page - 1) * page_size).limit(page_size).all() return total, items def cleanup_old_logs(db: Session, retention_days: int = AUDIT_LOG_RETENTION_DAYS): """清理超过保留期的日志""" cutoff = datetime.utcnow() - timedelta(days=retention_days) deleted = db.query(AuditLog).filter(AuditLog.created_at < cutoff).delete() db.commit() return deleted