263 lines
9.8 KiB
Python
263 lines
9.8 KiB
Python
'''工具函数集合'''
|
||
import os
|
||
import json
|
||
import shutil
|
||
import datetime
|
||
import re
|
||
import ast
|
||
from pathlib import Path
|
||
from typing import Dict, Any, List, Tuple, Optional
|
||
|
||
def merge_rules(existing_rules: Dict, default_rules: Dict) -> Dict:
|
||
"""智能合并规则：保留现有规则，只添加默认规则中的新字段"""
|
||
merged_rules = existing_rules.copy()
|
||
|
||
# 确保所有必需的字段都存在
|
||
for key in default_rules:
|
||
if key not in merged_rules:
|
||
merged_rules[key] = default_rules[key]
|
||
|
||
# 确保"未分类"分组始终存在
|
||
if "categories" not in merged_rules:
|
||
merged_rules["categories"] = {}
|
||
if "未分类" not in merged_rules["categories"]:
|
||
merged_rules["categories"]["未分类"] = []
|
||
|
||
# 确保其他必要字段存在
|
||
for field in ["display_names", "tooltips", "field_types", "field_decimals",
|
||
"hidden", "validations", "field_order", "last_saved_values", "encrypted_fields"]:
|
||
if field not in merged_rules:
|
||
if field == "hidden" or field == "encrypted_fields":
|
||
merged_rules[field] = []
|
||
else:
|
||
merged_rules[field] = {}
|
||
|
||
return merged_rules
|
||
|
||
def backup_existing_rules(rules_file: str, program_dir: str) -> str:
|
||
"""备份现有规则文件"""
|
||
if os.path.exists(rules_file):
|
||
try:
|
||
# 创建备份目录
|
||
backup_dir = os.path.join(program_dir, "backups")
|
||
os.makedirs(backup_dir, exist_ok=True)
|
||
|
||
# 生成带时间戳的备份文件名
|
||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
backup_file = os.path.join(backup_dir, f"config_editor_rules_{timestamp}.json")
|
||
|
||
# 备份文件
|
||
shutil.copy2(rules_file, backup_file)
|
||
print(f"已备份规则文件到: {backup_file}")
|
||
|
||
# 清理旧的备份文件
|
||
cleanup_old_backups(backup_dir)
|
||
|
||
return backup_file
|
||
except Exception as e:
|
||
print(f"备份规则文件失败: {e}")
|
||
return ""
|
||
|
||
def cleanup_old_backups(backup_dir: str, keep_count: int = 5):
|
||
"""清理旧的备份文件"""
|
||
try:
|
||
# 获取所有备份文件
|
||
backup_files = []
|
||
for filename in os.listdir(backup_dir):
|
||
if filename.startswith("config_editor_rules_") and filename.endswith(".json"):
|
||
filepath = os.path.join(backup_dir, filename)
|
||
backup_files.append((filepath, os.path.getmtime(filepath)))
|
||
|
||
# 按修改时间排序（从旧到新）
|
||
backup_files.sort(key=lambda x: x[1])
|
||
|
||
# 删除旧的备份文件，保留最近几个
|
||
if len(backup_files) > keep_count:
|
||
files_to_delete = backup_files[:-keep_count]
|
||
for filepath, _ in files_to_delete:
|
||
os.remove(filepath)
|
||
print(f"已删除旧备份文件: {filepath}")
|
||
|
||
except Exception as e:
|
||
print(f"清理旧备份文件失败: {e}")
|
||
|
||
def get_validation_tooltip(validation: Dict) -> str:
|
||
"""生成校验规则的提示信息"""
|
||
if not validation:
|
||
return ""
|
||
|
||
rules = []
|
||
if validation.get("required"):
|
||
rules.append("• 必填项")
|
||
if "min" in validation:
|
||
rules.append(f"• 最小值: {validation['min']}")
|
||
if "max" in validation:
|
||
rules.append(f"• 最大值: {validation['max']}")
|
||
if "regex" in validation:
|
||
rules.append(f"• 正则表达式: {validation['regex']}")
|
||
|
||
return "\n".join(rules)
|
||
|
||
def format_dict_value(value: Dict, original_lines: List[str] = None) -> str:
|
||
"""专门格式化字典值，保持多行格式"""
|
||
if not isinstance(value, dict):
|
||
return json.dumps(value, ensure_ascii=False)
|
||
|
||
# 如果有原始行信息，尝试保持原始格式
|
||
if original_lines and len(original_lines) > 1:
|
||
# 检查原始格式是否是漂亮的多行格式
|
||
first_line = original_lines[0].strip()
|
||
if first_line.endswith('{') and len(original_lines) > 2:
|
||
# 原始是多行格式，我们也使用多行格式
|
||
result = "{\n"
|
||
for i, (key, val) in enumerate(value.items()):
|
||
# 使用4个空格缩进
|
||
result += f' "{key}": {json.dumps(val, ensure_ascii=False)}'
|
||
if i < len(value) - 1:
|
||
result += ",\n"
|
||
else:
|
||
result += "\n"
|
||
result += "}"
|
||
return result
|
||
|
||
# 如果没有原始格式信息或不是多行格式，使用漂亮的JSON格式
|
||
return json.dumps(value, indent=4, ensure_ascii=False)
|
||
|
||
def format_value(value: Any, config_field=None) -> str:
|
||
"""格式化值，保持与Python兼容的格式"""
|
||
if value is None:
|
||
return "None"
|
||
elif isinstance(value, bool):
|
||
return "True" if value else "False"
|
||
elif isinstance(value, (int, float)):
|
||
# 根据小数位数格式化浮点数
|
||
if isinstance(value, float) and config_field and config_field.decimals is not None:
|
||
format_str = f"{{:.{config_field.decimals}f}}"
|
||
return format_str.format(value)
|
||
else:
|
||
return str(value)
|
||
elif isinstance(value, str):
|
||
# 检查字符串是否已经用引号包围
|
||
if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):
|
||
return value
|
||
# 检查字符串是否包含特殊字符
|
||
elif any(char in value for char in [' ', '\t', '\n', '#', '=']):
|
||
return f'"{value}"'
|
||
else:
|
||
return f'"{value}"'
|
||
elif isinstance(value, dict):
|
||
# 对于字典，使用专门的格式化函数
|
||
original_lines = config_field.original_lines if config_field else None
|
||
return format_dict_value(value, original_lines)
|
||
elif isinstance(value, list):
|
||
return json.dumps(value, ensure_ascii=False, indent=4)
|
||
else:
|
||
return repr(value)
|
||
|
||
def get_default_rules() -> Dict:
|
||
"""获取默认规则"""
|
||
return {
|
||
"categories": {"未分类": []},
|
||
"display_names": {},
|
||
"tooltips": {},
|
||
"field_types": {},
|
||
"field_decimals": {},
|
||
"hidden": [],
|
||
"encrypted_fields": [], #加密字段列表
|
||
"validations": {},
|
||
"field_order": {},
|
||
"last_saved_values": {}
|
||
}
|
||
|
||
def validate_config_data(config_data: Dict[str, Any], all_config_fields: Dict[str, Any]) -> List[str]:
|
||
"""校验配置数据"""
|
||
errors = []
|
||
|
||
for field_name, value in config_data.items():
|
||
config_field = all_config_fields.get(field_name)
|
||
if not config_field:
|
||
continue
|
||
|
||
validation = config_field.validation
|
||
if not validation:
|
||
continue
|
||
|
||
# 必填校验
|
||
if validation.get("required") and (value is None or value == ""):
|
||
errors.append(f"配置项 '{config_field.display_name}' 是必填项")
|
||
continue
|
||
|
||
# 数字范围校验
|
||
if isinstance(value, (int, float)):
|
||
if "min" in validation:
|
||
try:
|
||
min_val = float(validation["min"])
|
||
if value < min_val:
|
||
errors.append(f"配置项 '{config_field.display_name}' 的值不能小于 {min_val}")
|
||
except ValueError:
|
||
pass
|
||
|
||
if "max" in validation:
|
||
try:
|
||
max_val = float(validation["max"])
|
||
if value > max_val:
|
||
errors.append(f"配置项 '{config_field.display_name}' 的值不能大于 {max_val}")
|
||
except ValueError:
|
||
pass
|
||
|
||
# 字符串正则校验
|
||
elif isinstance(value, str) and "regex" in validation:
|
||
try:
|
||
if not re.match(validation["regex"], value):
|
||
errors.append(f"配置项 '{config_field.display_name}' 的值不符合格式要求")
|
||
except re.error:
|
||
errors.append(f"配置项 '{config_field.display_name}' 的正则表达式格式错误")
|
||
|
||
return errors
|
||
|
||
def get_field_type_from_value(value: Any) -> str:
|
||
"""根据值推断字段类型"""
|
||
if isinstance(value, bool):
|
||
return "bool"
|
||
elif isinstance(value, int):
|
||
return "int"
|
||
elif isinstance(value, float):
|
||
return "float"
|
||
elif isinstance(value, (list, dict)):
|
||
return "json"
|
||
else:
|
||
return "str"
|
||
|
||
def ensure_rule_structure(rules: Dict) -> Dict:
|
||
"""确保规则结构完整"""
|
||
# 确保规则中有"未分类"分组
|
||
if "categories" not in rules:
|
||
rules["categories"] = {}
|
||
if "未分类" not in rules["categories"]:
|
||
rules["categories"]["未分类"] = []
|
||
|
||
# 确保规则中有field_decimals字段
|
||
if "field_decimals" not in rules:
|
||
rules["field_decimals"] = {}
|
||
|
||
# 确保规则中有field_order字段
|
||
if "field_order" not in rules:
|
||
rules["field_order"] = {}
|
||
|
||
# 确保规则中有last_saved_values字段
|
||
if "last_saved_values" not in rules:
|
||
rules["last_saved_values"] = {}
|
||
|
||
# 确保规则中有encrypted_fields字段
|
||
if "encrypted_fields" not in rules:
|
||
rules["encrypted_fields"] = []
|
||
|
||
# 确保其他必要字段存在
|
||
for field in ["display_names", "tooltips", "field_types", "validations"]:
|
||
if field not in rules:
|
||
rules[field] = {}
|
||
|
||
if "hidden" not in rules:
|
||
rules["hidden"] = []
|
||
|
||
return rules |