66 lines
2.4 KiB
Python
Executable File
66 lines
2.4 KiB
Python
Executable File
import logging
|
|
from datetime import datetime
|
|
from database import get_db_cursor
|
|
from libs.exception import OperationLogError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class OperationLogger:
|
|
def log_operation(self, user_id: int, operation_type: str, endpoint: str,
|
|
parameters: str = None, status: str = "SUCCESS"):
|
|
"""记录API操作日志"""
|
|
try:
|
|
with get_db_cursor(db_type='sqlite') as cursor:
|
|
cursor.execute(
|
|
"""INSERT INTO api_operations
|
|
(user_id, operation_type, endpoint, parameters, status)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(user_id, operation_type, endpoint, parameters, status)
|
|
)
|
|
return cursor.lastrowid
|
|
except Exception as e:
|
|
logger.error(f"记录操作日志失败: {e}")
|
|
raise OperationLogError("操作日志记录失败")
|
|
|
|
def get_operations(self, user_id: int = None,
|
|
start_time: datetime = None,
|
|
end_time: datetime = None,
|
|
limit: int = 100):
|
|
"""查询操作日志"""
|
|
query = "SELECT * FROM api_operations WHERE 1=1"
|
|
params = []
|
|
|
|
if user_id:
|
|
query += " AND user_id = ?"
|
|
params.append(user_id)
|
|
if start_time:
|
|
query += " AND created_at >= ?"
|
|
params.append(start_time)
|
|
if end_time:
|
|
query += " AND created_at <= ?"
|
|
params.append(end_time)
|
|
|
|
query += " ORDER BY created_at DESC LIMIT ?"
|
|
params.append(limit)
|
|
|
|
try:
|
|
with get_db_cursor(db_type='sqlite') as cursor:
|
|
cursor.execute(query, params)
|
|
return cursor.fetchall()
|
|
except Exception as e:
|
|
logger.error(f"查询操作日志失败: {e}")
|
|
raise OperationLogError("操作日志查询失败")
|
|
|
|
def clean_old_logs(self, days: int = 30):
|
|
"""清理过期日志"""
|
|
try:
|
|
with get_db_cursor(db_type='sqlite') as cursor:
|
|
cursor.execute(
|
|
"DELETE FROM api_operations WHERE created_at < datetime('now', ?)",
|
|
(f"-{days} days",)
|
|
)
|
|
return cursor.rowcount
|
|
except Exception as e:
|
|
logger.error(f"清理操作日志失败: {e}")
|
|
raise OperationLogError("操作日志清理失败")
|