211 lines
6.8 KiB
Python
211 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
系統日誌資料模型
|
|
|
|
Author: PANJIT IT Team
|
|
Created: 2024-01-28
|
|
Modified: 2024-01-28
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy.sql import func
|
|
from app import db
|
|
|
|
|
|
class SystemLog(db.Model):
|
|
"""系統日誌表 (dt_system_logs)"""
|
|
__tablename__ = 'dt_system_logs'
|
|
|
|
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
|
|
level = db.Column(
|
|
db.Enum('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL', name='log_level'),
|
|
nullable=False,
|
|
comment='日誌等級'
|
|
)
|
|
module = db.Column(db.String(100), nullable=False, comment='模組名稱')
|
|
user_id = db.Column(db.Integer, db.ForeignKey('dt_users.id'), comment='使用者ID')
|
|
job_id = db.Column(db.Integer, db.ForeignKey('dt_translation_jobs.id'), comment='任務ID')
|
|
message = db.Column(db.Text, nullable=False, comment='日誌訊息')
|
|
extra_data = db.Column(db.JSON, comment='額外資料')
|
|
created_at = db.Column(db.DateTime, default=func.now(), comment='建立時間')
|
|
|
|
def __repr__(self):
|
|
return f'<SystemLog {self.level} {self.module}>'
|
|
|
|
def to_dict(self):
|
|
"""轉換為字典格式"""
|
|
return {
|
|
'id': self.id,
|
|
'level': self.level,
|
|
'module': self.module,
|
|
'user_id': self.user_id,
|
|
'job_id': self.job_id,
|
|
'message': self.message,
|
|
'extra_data': self.extra_data,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None
|
|
}
|
|
|
|
@classmethod
|
|
def log(cls, level, module, message, user_id=None, job_id=None, extra_data=None):
|
|
"""記錄日誌"""
|
|
log_entry = cls(
|
|
level=level.upper(),
|
|
module=module,
|
|
message=message,
|
|
user_id=user_id,
|
|
job_id=job_id,
|
|
extra_data=extra_data
|
|
)
|
|
|
|
db.session.add(log_entry)
|
|
db.session.commit()
|
|
return log_entry
|
|
|
|
@classmethod
|
|
def debug(cls, module, message, user_id=None, job_id=None, extra_data=None):
|
|
"""記錄除錯日誌"""
|
|
return cls.log('DEBUG', module, message, user_id, job_id, extra_data)
|
|
|
|
@classmethod
|
|
def info(cls, module, message, user_id=None, job_id=None, extra_data=None):
|
|
"""記錄資訊日誌"""
|
|
return cls.log('INFO', module, message, user_id, job_id, extra_data)
|
|
|
|
@classmethod
|
|
def warning(cls, module, message, user_id=None, job_id=None, extra_data=None):
|
|
"""記錄警告日誌"""
|
|
return cls.log('WARNING', module, message, user_id, job_id, extra_data)
|
|
|
|
@classmethod
|
|
def error(cls, module, message, user_id=None, job_id=None, extra_data=None):
|
|
"""記錄錯誤日誌"""
|
|
return cls.log('ERROR', module, message, user_id, job_id, extra_data)
|
|
|
|
@classmethod
|
|
def critical(cls, module, message, user_id=None, job_id=None, extra_data=None):
|
|
"""記錄嚴重錯誤日誌"""
|
|
return cls.log('CRITICAL', module, message, user_id, job_id, extra_data)
|
|
|
|
@classmethod
|
|
def get_logs(cls, level=None, module=None, user_id=None, start_date=None, end_date=None, limit=100, offset=0):
|
|
"""查詢日誌"""
|
|
query = cls.query
|
|
|
|
if level:
|
|
query = query.filter_by(level=level.upper())
|
|
|
|
if module:
|
|
query = query.filter(cls.module.like(f'%{module}%'))
|
|
|
|
if user_id:
|
|
query = query.filter_by(user_id=user_id)
|
|
|
|
if start_date:
|
|
query = query.filter(cls.created_at >= start_date)
|
|
|
|
if end_date:
|
|
query = query.filter(cls.created_at <= end_date)
|
|
|
|
# 按時間倒序排列
|
|
query = query.order_by(cls.created_at.desc())
|
|
|
|
if limit:
|
|
query = query.limit(limit)
|
|
if offset:
|
|
query = query.offset(offset)
|
|
|
|
return query.all()
|
|
|
|
@classmethod
|
|
def get_log_statistics(cls, days=7):
|
|
"""取得日誌統計資料"""
|
|
end_date = datetime.utcnow()
|
|
start_date = end_date - timedelta(days=days)
|
|
|
|
# 按等級統計
|
|
level_stats = db.session.query(
|
|
cls.level,
|
|
func.count(cls.id).label('count')
|
|
).filter(
|
|
cls.created_at >= start_date
|
|
).group_by(cls.level).all()
|
|
|
|
# 按模組統計
|
|
module_stats = db.session.query(
|
|
cls.module,
|
|
func.count(cls.id).label('count')
|
|
).filter(
|
|
cls.created_at >= start_date
|
|
).group_by(cls.module).order_by(
|
|
func.count(cls.id).desc()
|
|
).limit(10).all()
|
|
|
|
# 每日統計
|
|
daily_stats = db.session.query(
|
|
func.date(cls.created_at).label('date'),
|
|
cls.level,
|
|
func.count(cls.id).label('count')
|
|
).filter(
|
|
cls.created_at >= start_date
|
|
).group_by(
|
|
func.date(cls.created_at), cls.level
|
|
).order_by(
|
|
func.date(cls.created_at)
|
|
).all()
|
|
|
|
return {
|
|
'level_stats': [
|
|
{'level': stat.level, 'count': stat.count}
|
|
for stat in level_stats
|
|
],
|
|
'module_stats': [
|
|
{'module': stat.module, 'count': stat.count}
|
|
for stat in module_stats
|
|
],
|
|
'daily_stats': [
|
|
{
|
|
'date': stat.date.isoformat(),
|
|
'level': stat.level,
|
|
'count': stat.count
|
|
}
|
|
for stat in daily_stats
|
|
]
|
|
}
|
|
|
|
@classmethod
|
|
def cleanup_old_logs(cls, days_to_keep=30):
|
|
"""清理舊日誌"""
|
|
cutoff_date = datetime.utcnow() - timedelta(days=days_to_keep)
|
|
|
|
deleted_count = cls.query.filter(
|
|
cls.created_at < cutoff_date
|
|
).delete(synchronize_session=False)
|
|
|
|
db.session.commit()
|
|
return deleted_count
|
|
|
|
@classmethod
|
|
def get_error_summary(cls, days=1):
|
|
"""取得錯誤摘要"""
|
|
start_date = datetime.utcnow() - timedelta(days=days)
|
|
|
|
error_logs = cls.query.filter(
|
|
cls.level.in_(['ERROR', 'CRITICAL']),
|
|
cls.created_at >= start_date
|
|
).order_by(cls.created_at.desc()).limit(50).all()
|
|
|
|
# 按模組分組錯誤
|
|
error_by_module = {}
|
|
for log in error_logs:
|
|
module = log.module
|
|
if module not in error_by_module:
|
|
error_by_module[module] = []
|
|
error_by_module[module].append(log.to_dict())
|
|
|
|
return {
|
|
'total_errors': len(error_logs),
|
|
'error_by_module': error_by_module,
|
|
'recent_errors': [log.to_dict() for log in error_logs[:10]]
|
|
} |