初始化
This commit is contained in:
18
core/__init__.py
Normal file
18
core/__init__.py
Normal file
@@ -0,0 +1,18 @@
|
||||
"""
|
||||
核心模块
|
||||
- 配置管理
|
||||
- 日志记录
|
||||
- 错误处理
|
||||
"""
|
||||
from .config import config, Config
|
||||
from .logger import setup_logging, get_all_log_files, read_log_file
|
||||
from .error_handler import ErrorHandler
|
||||
|
||||
__all__ = [
|
||||
'config',
|
||||
'Config',
|
||||
'setup_logging',
|
||||
'get_all_log_files',
|
||||
'read_log_file',
|
||||
'ErrorHandler'
|
||||
]
|
||||
101
core/config.py
Normal file
101
core/config.py
Normal file
@@ -0,0 +1,101 @@
|
||||
"""
|
||||
全局配置管理模块
|
||||
- 保存/读取用户配置
|
||||
- 配置全局文件保存位置
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
class Config:
|
||||
"""全局配置管理器"""
|
||||
|
||||
_instance = None
|
||||
_config_file = None
|
||||
_config_data = {}
|
||||
|
||||
# 默认配置
|
||||
DEFAULT_CONFIG = {
|
||||
"output_directory": "", # 空字符串表示每次询问
|
||||
"auto_save_to_default": False, # 是否自动保存到默认目录
|
||||
"image_quality": 85, # 默认图片质量
|
||||
"enable_lossless_compression": True, # 启用无损压缩
|
||||
"show_preview": True, # 显示预览
|
||||
"animation_enabled": True, # 启用动画
|
||||
"animation_duration": 300, # 动画时长(ms)
|
||||
}
|
||||
|
||||
def __new__(cls):
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
cls._instance._initialized = False
|
||||
return cls._instance
|
||||
|
||||
def __init__(self):
|
||||
if self._initialized:
|
||||
return
|
||||
|
||||
# 配置文件路径
|
||||
app_root = Path(__file__).parent.parent
|
||||
config_dir = app_root / "config"
|
||||
config_dir.mkdir(exist_ok=True)
|
||||
self._config_file = config_dir / "settings.json"
|
||||
|
||||
# 加载配置
|
||||
self._load_config()
|
||||
self._initialized = True
|
||||
|
||||
def _load_config(self):
|
||||
"""加载配置文件"""
|
||||
if self._config_file.exists():
|
||||
try:
|
||||
with open(self._config_file, "r", encoding="utf-8") as f:
|
||||
self._config_data = json.load(f)
|
||||
logging.info(f"已加载配置文件: {self._config_file}")
|
||||
except Exception as e:
|
||||
logging.error(f"加载配置文件失败: {e}")
|
||||
self._config_data = {}
|
||||
else:
|
||||
self._config_data = {}
|
||||
|
||||
# 合并默认配置
|
||||
for key, value in self.DEFAULT_CONFIG.items():
|
||||
if key not in self._config_data:
|
||||
self._config_data[key] = value
|
||||
|
||||
def _save_config(self):
|
||||
"""保存配置文件"""
|
||||
try:
|
||||
with open(self._config_file, "w", encoding="utf-8") as f:
|
||||
json.dump(self._config_data, f, ensure_ascii=False, indent=2)
|
||||
logging.info("配置已保存")
|
||||
except Exception as e:
|
||||
logging.error(f"保存配置失败: {e}")
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
"""获取配置值"""
|
||||
return self._config_data.get(key, default)
|
||||
|
||||
def set(self, key: str, value: Any):
|
||||
"""设置配置值"""
|
||||
self._config_data[key] = value
|
||||
self._save_config()
|
||||
|
||||
def get_output_directory(self) -> str:
|
||||
"""获取输出目录"""
|
||||
path = self.get("output_directory", "")
|
||||
if path and os.path.isdir(path):
|
||||
return path
|
||||
return ""
|
||||
|
||||
def set_output_directory(self, path: str):
|
||||
"""设置输出目录"""
|
||||
self.set("output_directory", path)
|
||||
|
||||
|
||||
# 全局配置实例
|
||||
config = Config()
|
||||
|
||||
162
core/error_handler.py
Normal file
162
core/error_handler.py
Normal file
@@ -0,0 +1,162 @@
|
||||
"""
|
||||
全局错误处理模块
|
||||
- 重写 sys.excepthook 捕获主线程未处理异常
|
||||
- 使用 threading.excepthook 捕获子线程异常
|
||||
- 所有异常统一进入 ErrorHandler 处理
|
||||
- 弹框提示用户,不闪退
|
||||
"""
|
||||
import sys
|
||||
import logging
|
||||
import threading
|
||||
import traceback
|
||||
from PySide6.QtWidgets import QMessageBox, QTextEdit, QPushButton, QVBoxLayout, QDialog
|
||||
from PySide6.QtCore import QObject, Signal
|
||||
|
||||
|
||||
class ErrorSignal(QObject):
|
||||
"""用于跨线程传递错误信号"""
|
||||
error_occurred = Signal(str, str) # (error_message, error_detail)
|
||||
|
||||
|
||||
class ErrorDialog(QDialog):
|
||||
"""自定义错误对话框,支持显示详细堆栈"""
|
||||
|
||||
def __init__(self, title: str, message: str, detail: str, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle(title)
|
||||
self.setMinimumSize(500, 200)
|
||||
self.setup_ui(message, detail)
|
||||
|
||||
def setup_ui(self, message: str, detail: str):
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setSpacing(16)
|
||||
|
||||
# 错误消息
|
||||
from PySide6.QtWidgets import QLabel
|
||||
msg_label = QLabel(message)
|
||||
msg_label.setWordWrap(True)
|
||||
msg_label.setStyleSheet("font-size: 14px; color: #f87171;")
|
||||
layout.addWidget(msg_label)
|
||||
|
||||
# 详细信息(可展开)
|
||||
self.detail_edit = QTextEdit()
|
||||
self.detail_edit.setPlainText(detail)
|
||||
self.detail_edit.setReadOnly(True)
|
||||
self.detail_edit.setVisible(False)
|
||||
self.detail_edit.setMinimumHeight(200)
|
||||
layout.addWidget(self.detail_edit)
|
||||
|
||||
# 按钮区域
|
||||
from PySide6.QtWidgets import QHBoxLayout
|
||||
btn_layout = QHBoxLayout()
|
||||
|
||||
self.detail_btn = QPushButton("显示详情")
|
||||
self.detail_btn.setObjectName("secondary_btn")
|
||||
self.detail_btn.clicked.connect(self.toggle_detail)
|
||||
btn_layout.addWidget(self.detail_btn)
|
||||
|
||||
btn_layout.addStretch()
|
||||
|
||||
ok_btn = QPushButton("确定")
|
||||
ok_btn.setObjectName("primary_btn")
|
||||
ok_btn.clicked.connect(self.accept)
|
||||
btn_layout.addWidget(ok_btn)
|
||||
|
||||
layout.addLayout(btn_layout)
|
||||
|
||||
def toggle_detail(self):
|
||||
"""切换详情显示"""
|
||||
visible = not self.detail_edit.isVisible()
|
||||
self.detail_edit.setVisible(visible)
|
||||
self.detail_btn.setText("隐藏详情" if visible else "显示详情")
|
||||
|
||||
# 调整窗口大小
|
||||
if visible:
|
||||
self.resize(600, 450)
|
||||
else:
|
||||
self.resize(500, 200)
|
||||
|
||||
|
||||
class ErrorHandler:
|
||||
"""全局错误处理器"""
|
||||
|
||||
_instance = None
|
||||
|
||||
def __new__(cls, *args, **kwargs):
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
return cls._instance
|
||||
|
||||
def __init__(self, app=None):
|
||||
if hasattr(self, '_initialized') and self._initialized:
|
||||
return
|
||||
|
||||
self.app = app
|
||||
self.error_signal = ErrorSignal()
|
||||
self.error_signal.error_occurred.connect(self._show_error_dialog)
|
||||
|
||||
# 设置全局异常处理
|
||||
sys.excepthook = self.handle_exception
|
||||
threading.excepthook = self.handle_thread_exception
|
||||
|
||||
self._initialized = True
|
||||
logging.info("全局错误处理器已初始化")
|
||||
|
||||
def handle_exception(self, exc_type, exc_value, exc_tb):
|
||||
"""处理主线程未捕获的异常"""
|
||||
# 忽略键盘中断
|
||||
if issubclass(exc_type, KeyboardInterrupt):
|
||||
sys.__excepthook__(exc_type, exc_value, exc_tb)
|
||||
return
|
||||
|
||||
# 格式化错误信息
|
||||
error_msg = str(exc_value)
|
||||
error_detail = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))
|
||||
|
||||
# 记录日志
|
||||
logging.error(f"未捕获的异常:\n{error_detail}")
|
||||
|
||||
# 显示错误对话框
|
||||
self._show_error_dialog(error_msg, error_detail)
|
||||
|
||||
def handle_thread_exception(self, args):
|
||||
"""处理子线程未捕获的异常"""
|
||||
exc_type = args.exc_type
|
||||
exc_value = args.exc_value
|
||||
exc_tb = args.exc_traceback
|
||||
|
||||
# 格式化错误信息
|
||||
error_msg = str(exc_value) if exc_value else str(exc_type)
|
||||
error_detail = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))
|
||||
|
||||
# 记录日志
|
||||
logging.error(f"线程异常 ({args.thread.name}):\n{error_detail}")
|
||||
|
||||
# 通过信号在主线程显示对话框
|
||||
self.error_signal.error_occurred.emit(error_msg, error_detail)
|
||||
|
||||
def _show_error_dialog(self, error_msg: str, error_detail: str):
|
||||
"""显示错误对话框"""
|
||||
try:
|
||||
dialog = ErrorDialog(
|
||||
title="发生错误",
|
||||
message=f"程序遇到了一个问题:\n\n{error_msg}\n\n详情已记录到日志文件。",
|
||||
detail=error_detail
|
||||
)
|
||||
dialog.exec()
|
||||
except Exception as e:
|
||||
# 如果对话框也出错,至少打印到控制台
|
||||
print(f"显示错误对话框失败: {e}")
|
||||
print(f"原始错误: {error_msg}")
|
||||
print(error_detail)
|
||||
|
||||
@staticmethod
|
||||
def safe_execute(func, *args, error_msg: str = "操作失败", **kwargs):
|
||||
"""安全执行函数,捕获异常并显示友好提示"""
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except Exception as e:
|
||||
logging.error(f"{error_msg}: {e}\n{traceback.format_exc()}")
|
||||
QMessageBox.warning(None, "警告", f"{error_msg}\n\n{str(e)}")
|
||||
return None
|
||||
|
||||
91
core/logger.py
Normal file
91
core/logger.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""
|
||||
日志配置模块
|
||||
- 日志文件: logs/app_YYYYMMDD.log
|
||||
- 日志格式: [时间] [级别] [模块] 消息
|
||||
- 自动清理30天前的旧日志
|
||||
"""
|
||||
import os
|
||||
import logging
|
||||
import datetime
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def get_logs_dir() -> Path:
|
||||
"""获取日志目录路径"""
|
||||
# 获取应用根目录
|
||||
app_root = Path(__file__).parent.parent
|
||||
logs_dir = app_root / "logs"
|
||||
logs_dir.mkdir(exist_ok=True)
|
||||
return logs_dir
|
||||
|
||||
|
||||
def get_log_file_path() -> Path:
|
||||
"""获取当天的日志文件路径"""
|
||||
today = datetime.date.today().strftime("%Y%m%d")
|
||||
return get_logs_dir() / f"app_{today}.log"
|
||||
|
||||
|
||||
def setup_logging():
|
||||
"""配置日志系统"""
|
||||
log_file = get_log_file_path()
|
||||
|
||||
# 日志格式
|
||||
log_format = "[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s"
|
||||
date_format = "%Y-%m-%d %H:%M:%S"
|
||||
|
||||
# 配置根日志器
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format=log_format,
|
||||
datefmt=date_format,
|
||||
handlers=[
|
||||
# 文件处理器
|
||||
logging.FileHandler(log_file, encoding="utf-8"),
|
||||
# 控制台处理器
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
|
||||
# 记录启动日志
|
||||
logging.info("=" * 50)
|
||||
logging.info("奶酪云工具箱 启动")
|
||||
logging.info("=" * 50)
|
||||
|
||||
# 清理旧日志
|
||||
cleanup_old_logs()
|
||||
|
||||
|
||||
def cleanup_old_logs(days: int = 30):
|
||||
"""清理指定天数前的旧日志"""
|
||||
logs_dir = get_logs_dir()
|
||||
cutoff_date = datetime.date.today() - datetime.timedelta(days=days)
|
||||
|
||||
for log_file in logs_dir.glob("app_*.log"):
|
||||
try:
|
||||
# 从文件名提取日期
|
||||
date_str = log_file.stem.replace("app_", "")
|
||||
file_date = datetime.datetime.strptime(date_str, "%Y%m%d").date()
|
||||
|
||||
if file_date < cutoff_date:
|
||||
log_file.unlink()
|
||||
logging.info(f"已清理旧日志文件: {log_file.name}")
|
||||
except (ValueError, OSError) as e:
|
||||
logging.warning(f"清理日志文件失败 {log_file}: {e}")
|
||||
|
||||
|
||||
def get_all_log_files() -> list:
|
||||
"""获取所有日志文件列表(按时间倒序)"""
|
||||
logs_dir = get_logs_dir()
|
||||
log_files = list(logs_dir.glob("app_*.log"))
|
||||
log_files.sort(key=lambda x: x.name, reverse=True)
|
||||
return log_files
|
||||
|
||||
|
||||
def read_log_file(file_path: Path) -> str:
|
||||
"""读取日志文件内容"""
|
||||
try:
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
except Exception as e:
|
||||
return f"读取日志文件失败: {e}"
|
||||
|
||||
Reference in New Issue
Block a user