feat: 增加处理历史保留配置

This commit is contained in:
2026-05-19 12:01:26 +08:00
parent f1075695a1
commit a3cc9f347c
9 changed files with 208 additions and 35 deletions
+49 -8
View File
@@ -2,10 +2,10 @@
配置管理模块
"""
import json
from pathlib import Path
from datetime import datetime
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List
BASE_DIR = Path(__file__).resolve().parent.parent
@@ -98,11 +98,44 @@ class RemoteDataConfig:
).normalized()
@dataclass
class HistoryRetentionConfig:
enabled: bool = False
keep_count: int = 20
def normalized(self) -> "HistoryRetentionConfig":
try:
keep_count = int(self.keep_count)
except (TypeError, ValueError):
keep_count = 20
return HistoryRetentionConfig(
enabled=bool(self.enabled),
keep_count=max(keep_count, 0),
)
def to_dict(self) -> Dict[str, Any]:
normalized = self.normalized()
return {
"enabled": normalized.enabled,
"keep_count": normalized.keep_count,
}
@classmethod
def from_dict(cls, data: Dict[str, Any] | None) -> "HistoryRetentionConfig":
data = data or {}
return cls(
enabled=bool(data.get("enabled", False)),
keep_count=data.get("keep_count", 20),
).normalized()
@dataclass
class AppConfig:
update: str = ""
mysql: MySQLConfig = field(default_factory=MySQLConfig)
remote_data: RemoteDataConfig = field(default_factory=RemoteDataConfig)
history_retention: HistoryRetentionConfig = field(default_factory=HistoryRetentionConfig)
sheet_filter: List[str] = field(default_factory=list)
extract_fields: List[Dict[str, Any]] = field(default_factory=list)
@@ -124,20 +157,28 @@ class AppConfig:
dbname=mysql_data.get("dbname", "CapacityReport")
)
remote_config = RemoteDataConfig.from_dict(data.get("RemoteData"))
history_retention = HistoryRetentionConfig.from_dict(data.get("HistoryRetention"))
return cls(
update=data.get("Update", ""),
mysql=mysql_config,
remote_data=remote_config,
history_retention=history_retention,
sheet_filter=data.get("SheetFilter", []),
extract_fields=data.get("ExtractField", [])
)
def save(self):
"""保存配置到 Configure.json,并自动更新 Update 时间"""
self.update = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
data = {
data = self.to_file_dict()
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def to_file_dict(self) -> Dict[str, Any]:
"""转换为配置文件结构(包含敏感字段,用于保存和下载)"""
return {
"Update": self.update,
"MySQL_DBInfo": {
"host": self.mysql.host,
@@ -147,12 +188,10 @@ class AppConfig:
"dbname": self.mysql.dbname
},
"RemoteData": self.remote_data.normalized().to_dict(include_password=True),
"HistoryRetention": self.history_retention.normalized().to_dict(),
"SheetFilter": self.sheet_filter,
"ExtractField": self.extract_fields
}
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def to_dict(self) -> Dict[str, Any]:
"""转换为字典(用于返回给前端,隐藏密码)"""
@@ -165,6 +204,7 @@ class AppConfig:
"dbname": self.mysql.dbname
},
"remote_data": self.remote_data.normalized().to_dict(),
"history_retention": self.history_retention.normalized().to_dict(),
"sheet_filter": self.sheet_filter,
"extract_fields": self.extract_fields
}
@@ -181,6 +221,7 @@ class AppConfig:
"dbname": self.mysql.dbname
},
"remote_data": self.remote_data.normalized().to_dict(include_password=True),
"history_retention": self.history_retention.normalized().to_dict(),
"sheet_filter": self.sheet_filter,
"extract_fields": self.extract_fields
}