# -*- coding: utf-8 -*- """ 业务日志记录工具 用于记录所有 POST/PUT/DELETE 操作到 ai_logs 表 """ import logging import json from datetime import datetime from flask import request, g from database_config import get_db_manager from auth_utils import AuthUtils logger = logging.getLogger('article_server') class BusinessLogger: """业务日志记录器""" @staticmethod def log_operation( action, target_type=None, target_id=None, description=None, request_data=None, response_data=None, status='success', error_message=None ): """ 记录业务操作日志到 ai_logs 表 Args: action: 操作动作（如：create_enterprise, update_product, delete_image） target_type: 目标类型（如：enterprise, product, employee） target_id: 目标ID description: 操作描述 request_data: 请求数据（dict） response_data: 响应数据（dict） status: 状态（success/error/warning） error_message: 错误消息 """ try: # 获取当前用户信息 current_user = AuthUtils.get_current_user() user_id = current_user.get('user_id') if current_user else None # 获取IP地址 ip_address = request.environ.get('HTTP_X_FORWARDED_FOR', request.environ.get('REMOTE_ADDR', '未知')) # 处理多个代理的情况 if ip_address and ',' in ip_address: ip_address = ip_address.split(',')[0].strip() # 获取User-Agent user_agent = request.headers.get('User-Agent', '未知') # 转换数据为JSON格式（过滤敏感信息） request_json = None if request_data: # 过滤密码等敏感字段 filtered_request = BusinessLogger._filter_sensitive_data(request_data) request_json = json.dumps(filtered_request, ensure_ascii=False) response_json = None if response_data: # 过滤敏感响应数据 filtered_response = BusinessLogger._filter_sensitive_data(response_data) response_json = json.dumps(filtered_response, ensure_ascii=False) # 插入日志记录 db_manager = get_db_manager() sql = """ INSERT INTO ai_logs (user_id, action, target_type, target_id, description, ip_address, user_agent, request_data, response_data, status, error_message, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) """ params = ( user_id, action, target_type, target_id, description, ip_address[:45] if ip_address else None, # 限制长度 user_agent[:500] if user_agent else None, # 限制长度 request_json, response_json, status, error_message[:1000] if error_message else None # 限制长度 ) log_id = db_manager.execute_insert(sql, params) # 记录到应用日志 logger.info( f"[业务日志] 操作: {action} | " f"用户ID: {user_id} | " f"目标: {target_type}#{target_id} | " f"状态: {status} | " f"IP: {ip_address} | " f"日志ID: {log_id}" ) return log_id except Exception as e: # 日志记录失败不应影响业务流程 logger.error(f"[业务日志] 记录失败: {e}", exc_info=True) return None @staticmethod def _filter_sensitive_data(data): """过滤敏感数据""" if not isinstance(data, dict): return data sensitive_fields = [ 'password', 'old_password', 'new_password', 'confirm_password', 'token', 'secret', 'key' ] filtered = {} for key, value in data.items(): if key.lower() in sensitive_fields: filtered[key] = '***' # 隐藏敏感信息 elif isinstance(value, dict): filtered[key] = BusinessLogger._filter_sensitive_data(value) else: filtered[key] = value return filtered @staticmethod def log_create(target_type, target_id, description=None, request_data=None, response_data=None): """记录创建操作""" return BusinessLogger.log_operation( action=f'create_{target_type}', target_type=target_type, target_id=target_id, description=description or f'创建{target_type}', request_data=request_data, response_data=response_data, status='success' ) @staticmethod def log_update(target_type, target_id, description=None, request_data=None, response_data=None): """记录更新操作""" return BusinessLogger.log_operation( action=f'update_{target_type}', target_type=target_type, target_id=target_id, description=description or f'更新{target_type}', request_data=request_data, response_data=response_data, status='success' ) @staticmethod def log_delete(target_type, target_id, description=None, request_data=None): """记录删除操作""" return BusinessLogger.log_operation( action=f'delete_{target_type}', target_type=target_type, target_id=target_id, description=description or f'删除{target_type}', request_data=request_data, status='success' ) @staticmethod def log_error(action, target_type=None, target_id=None, error_message=None, request_data=None): """记录错误操作""" return BusinessLogger.log_operation( action=action, target_type=target_type, target_id=target_id, description=f'{action} 失败', request_data=request_data, status='error', error_message=error_message ) @staticmethod def log_login(user_id, username, success=True, error_message=None): """记录登录操作""" return BusinessLogger.log_operation( action='user_login', target_type='user', target_id=user_id, description=f'用户登录: {username}', status='success' if success else 'error', error_message=error_message ) @staticmethod def log_logout(user_id, username): """记录登出操作""" return BusinessLogger.log_operation( action='user_logout', target_type='user', target_id=user_id, description=f'用户登出: {username}', status='success' ) # 快捷方法 def log_create(target_type, target_id, **kwargs): """快捷记录创建操作""" return BusinessLogger.log_create(target_type, target_id, **kwargs) def log_update(target_type, target_id, **kwargs): """快捷记录更新操作""" return BusinessLogger.log_update(target_type, target_id, **kwargs) def log_delete(target_type, target_id, **kwargs): """快捷记录删除操作""" return BusinessLogger.log_delete(target_type, target_id, **kwargs) def log_error(action, **kwargs): """快捷记录错误操作""" return BusinessLogger.log_error(action, **kwargs) def log_operation(**kwargs): """快捷记录通用操作""" return BusinessLogger.log_operation(**kwargs)