""" 配置管理 — 全局配置读写 """ import json import os CONFIG_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'config.json') DEFAULT_CONFIG = { "dns_servers": ["8.8.8.8", "114.114.114.114"], "thread_count": 10, "send_interval": 0, "send_mode": "smtp", "send_timeout": 60, "helo_domain": "", "reply_to": "", "web_port": 49800, "ai_server": "http://127.0.0.1:8000", "ai_enabled": True, "api_url": "", "api_key": "", "api_batch": 100, "api_interval": 60, "api_callback": "", "smtp_presets": [ {"type": "Gmail", "smtp": "smtp.gmail.com", "port": 465, "ssl": "ssl", "is_full_user_name": True}, {"type": "Outlook", "smtp": "smtp-mail.outlook.com", "port": 587, "ssl": "tls", "is_full_user_name": True}, {"type": "QQ", "smtp": "smtp.qq.com", "port": 465, "ssl": "ssl", "is_full_user_name": True}, {"type": "163", "smtp": "smtp.163.com", "port": 465, "ssl": "ssl", "is_full_user_name": True}, {"type": "126", "smtp": "smtp.126.com", "port": 465, "ssl": "ssl", "is_full_user_name": True}, {"type": "Yahoo", "smtp": "smtp.mail.yahoo.com", "port": 465, "ssl": "ssl", "is_full_user_name": True}, ], "magic_config": { "enabled": True, "level": "mid", "html_entity": True, "hex_encode": False, "wide_char": False, "full_half": False, "zero_char": True, "magic_code": True, "html_tag": False, "random_cn": False, "random_jp": False, "random_en": False, "random_emoji": False, "random_css": False, "random_html": False, }, "smart_rules": { "enabled": True, "auto_change_ip": True, "auto_change_box": True, "auto_reduce_speed": True, "auto_change_content": True, "auto_fix_dns": True, "auto_disable": True, "auto_adsl": False, "series_fail_threshold": 5, }, "adsl_config": { "enabled": False, "connection_name": "宽带连接", "username": "", "password": "", "auto_dial": False, "fail_threshold": 5, "wait_seconds": 10, }, "proxy_config": { "enabled": False, "auto_switch": False, "per_ip_count": 50, "expire_seconds": 300, }, "track_config": { "enabled": False, "server_url": "", "method": "pixel", "track_clicks": False, }, "bounce_config": { "enabled": False, "bounce_email": "", "auto_clean": False, "callback_api": False, }, "ipv6_enabled": False, "ssh_retry": True, "ssh_retry_count": 3, "ssh_retry_interval": 2, "mail_cache": True, "random_delay_deploy": True, "deploy_concurrency": 40, "ssh_timeout": 120, "command_timeout": 3000, } class Config: _instance = None _data = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): if self._data is None: self.load() def load(self): if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, 'r', encoding='utf-8') as f: saved = json.load(f) self._data = self._deep_merge(DEFAULT_CONFIG.copy(), saved) except Exception: self._data = DEFAULT_CONFIG.copy() else: self._data = DEFAULT_CONFIG.copy() self.save() @staticmethod def _deep_merge(base: dict, override: dict) -> dict: """深度合并两个字典""" for key, value in override.items(): if key in base and isinstance(base[key], dict) and isinstance(value, dict): base[key] = Config._deep_merge(base[key], value) else: base[key] = value return base def save(self): try: with open(CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(self._data, f, ensure_ascii=False, indent=2) except Exception: pass def get(self, key, default=None): keys = key.split('.') val = self._data for k in keys: if isinstance(val, dict) and k in val: val = val[k] else: return default return val def set(self, key, value): keys = key.split('.') val = self._data for k in keys[:-1]: if k not in val: val[k] = {} val = val[k] val[keys[-1]] = value self.save() @property def data(self): return self._data # ===== 数据持久化: 服务器/发件箱/收件人/变量 ===== DATA_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'data.json') def save_data(self, servers=None, boxes=None, emails=None, variables=None, dns_accounts=None, proxies=None, word_replacements=None): """持久化业务数据到 data.json""" import dataclasses data = {} if servers is not None: data['servers'] = [dataclasses.asdict({ k: v for k, v in s.__dict__.items()}) for s in servers] if boxes is not None: data['boxes'] = [] for b in boxes: d = {k: v for k, v in b.__dict__.items() if k != 'locker'} data['boxes'].append(d) if emails is not None: data['emails'] = [dataclasses.asdict(e) for e in emails] if variables is not None: data['variables'] = [] for v in variables: d = {k: v for k, v in v.__dict__.items() if k != 'locker'} data['variables'].append(d) if dns_accounts is not None: data['dns_accounts'] = [dataclasses.asdict(a) for a in dns_accounts] if proxies is not None: data['proxies'] = [dataclasses.asdict(p) for p in proxies] if word_replacements is not None: data['word_replacements'] = list(word_replacements) try: with open(self.DATA_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) except Exception: pass def load_data(self): """从 data.json 加载业务数据，返回 dict""" if not os.path.exists(self.DATA_FILE): return {} try: with open(self.DATA_FILE, 'r', encoding='utf-8') as f: return json.load(f) except Exception: return {} @staticmethod def dict_to_server(d): from core.models import ServerConfig s = ServerConfig() for k, v in d.items(): if hasattr(s, k): setattr(s, k, v) return s @staticmethod def dict_to_box(d): from core.models import MailBox b = MailBox() for k, v in d.items(): if hasattr(b, k) and k != 'locker': setattr(b, k, v) return b @staticmethod def dict_to_email(d): from core.models import Email e = Email() for k, v in d.items(): if hasattr(e, k): setattr(e, k, v) return e @staticmethod def dict_to_variable(d): from core.models import Variable v = Variable() for k, val in d.items(): if hasattr(v, k) and k != 'locker': setattr(v, k, val) return v