| from flask import Flask, request, send_from_directory, jsonify |
| from flask_cors import CORS |
| import json |
| import re |
| import os |
| import shutil |
| from datetime import datetime |
| import glob |
| |
| app = Flask(__name__, static_folder='public') |
| CORS(app, resources={ |
| r"/*": { |
| "origins": "*", |
| "methods": ["GET", "POST", "OPTIONS"], |
| "allow_headers": ["Content-Type", "Authorization", "Access-Control-Allow-Origin"], |
| "expose_headers": ["Content-Type", "Authorization"], |
| "supports_credentials": True |
| } |
| }) |
| |
| def read_config(filename): |
| try: |
| # 使用 os.path.join 来正确处理路径 |
| filepath = filename # 现在 filename 已经包含完整路径 |
| print(f"Reading config from: {filepath}") # 添加日志 |
| |
| with open(filepath, 'r', encoding='utf-8') as f: |
| content = f.read() |
| |
| # 移除注释 |
| json_content = re.sub(r'/\*[\s\S]*?\*/', '', content) # 移除多行注释 |
| json_content = '\n'.join( |
| line for line in json_content.split('\n') |
| if not line.strip().startswith('//') # 移除单行注释 |
| ) |
| json_content = re.sub(r',(\s*[}\]])', r'\1', json_content) # 移除末尾逗号 |
| |
| try: |
| config = json.loads(json_content) |
| print(f"Successfully parsed config: {config}") # 添加日志 |
| except json.JSONDecodeError as e: |
| print(f"JSON parse error: {str(e)}") # 添加日志 |
| print(f"Content being parsed: {json_content}") |
| raise |
| |
| # 确保必要字段存在 |
| config.setdefault('rncn-config-version', 1) |
| config.setdefault('geodata-mode', False) |
| config.setdefault('geo-auto-update', True) |
| config.setdefault('geo-update-interval', 24) |
| config.setdefault('geox-url', {'mmdb': ''}) |
| config.setdefault('hosts', {}) |
| config.setdefault('dns', {}) |
| |
| return config |
| except Exception as e: |
| print(f"解析 {filename} 失败:", str(e)) |
| print("文件内容:", content) |
| raise Exception(f"解析 {filename} 失败: {str(e)}") |
| |
| def write_config(filename, data): |
| try: |
| filepath = filename |
| print(f"Writing config to: {filepath}") |
| |
| # 格式化 hosts 对象 |
| hosts_entries = [] |
| for domain, ip in sorted(data['hosts'].items()): |
| hosts_entries.append(f'\t\t"{domain}"\t\t:\t"{ip}"') |
| hosts_content = ',\n'.join(hosts_entries) |
| |
| # 根据文件名判断是国内还是海外配置 |
| is_cn_config = 'cn-config' in filepath |
| |
| # 构建新的文件内容 |
| config = { |
| "rncn-config-version": data["rncn-config-version"], |
| "geodata-mode": data.get("geodata-mode", False), |
| "geo-auto-update": data.get("geo-auto-update", True), |
| "geo-update-interval": data.get("geo-update-interval", 24), |
| "geox-url": { |
| "mmdb": data.get("geox-url", {}).get("mmdb", "") |
| }, |
| "hosts": data["hosts"], |
| "dns": { |
| "use-hosts": True, |
| "use-system-hosts": False, |
| "respect-rules": False, |
| "enhanced-mode": "normal", |
| "ipv6": True |
| } |
| } |
| |
| # 根据配置类型设置不同的 DNS 配置 |
| if is_cn_config: |
| config["dns"].update({ |
| "default-nameserver": ["223.5.5.5"], |
| "nameserver": [ |
| "https://223.5.5.5/dns-query", |
| "https://223.6.6.6/dns-query" |
| ], |
| "fallback": [ |
| "https://1.0.0.1/dns-query" |
| ], |
| "fallback-filter": { |
| "geoip": True, |
| "geoip-code": "CN" |
| } |
| }) |
| else: |
| config["dns"].update({ |
| "default-nameserver": ["1.1.1.1", "1.0.0.1"], |
| "nameserver": [ |
| "https://1.1.1.1/dns-query", |
| "https://1.0.0.1/dns-query" |
| ] |
| }) |
| |
| # 使用 json.dumps 生成基本 JSON 字符串 |
| json_str = json.dumps(config, indent='\t') |
| |
| # 替换 hosts 部分的格式 |
| json_str = re.sub( |
| r'"hosts":\s*{[^}]*}', |
| f'"hosts": {{\n{hosts_content}\n\t}}', |
| json_str |
| ) |
| |
| # 写入文件 |
| with open(filepath, 'w', encoding='utf-8') as f: |
| f.write(json_str) |
| |
| except Exception as e: |
| print(f"写入 {filename} 失败:", str(e)) |
| print("尝试写入的内容:", json_str) |
| raise Exception(f"写入 {filename} 失败: {str(e)}") |
| |
| def load_app_config(): |
| try: |
| with open('config.json', 'r', encoding='utf-8') as f: |
| return json.load(f) |
| except Exception as e: |
| print(f"加载配置文件失败: {str(e)}") |
| return { |
| "title": "DNS 配置管理", |
| "files": { |
| "oversea": "gen/rncn-dns-oversea-config.js", |
| "cn": "gen/rncn-dns-cn-config.js", |
| "history": { |
| "dir": "gen/history", |
| "maxVersions": 10 |
| } |
| }, |
| "server": { |
| "host": "0.0.0.0", |
| "port": 3001 |
| } |
| } |
| |
| # 加载配置 |
| app_config = load_app_config() |
| |
| @app.route('/') |
| def index(): |
| return send_from_directory('public', 'index.html') |
| |
| @app.route('/api/config', methods=['GET']) |
| def get_config(): |
| try: |
| # 读取配置文件 |
| oversea_config = read_config(app_config['files']['oversea']) |
| cn_config = read_config(app_config['files']['cn']) |
| |
| # 提供默认的 hints |
| default_hints = { |
| "domain": "[默认提示] 域名格式如 \"+.example.com\" 或 \"specific.example.com\"", |
| "ip": { |
| "cn": "[默认提示] 国内解锁集群: 10.100.253.2", |
| "overseas": "[默认提示] 海外解锁集群: 10.100.253.1" |
| } |
| } |
| |
| # 检查是否使用了默认值 |
| using_default_hints = 'hints' not in app_config |
| if using_default_hints: |
| print("警告: 未找到配置文件中的 hints 配置，使用默认值") |
| |
| # 构建域名映射 |
| hosts = { |
| domain: { |
| 'cnIp': cn_config['hosts'].get(domain, ''), |
| 'overseasIp': oversea_config['hosts'].get(domain, '') |
| } |
| for domain in set(list(cn_config['hosts'].keys()) + list(oversea_config['hosts'].keys())) |
| } |
| |
| return jsonify({ |
| 'title': app_config.get('title', 'DNS 配置管理'), |
| 'hosts': hosts, # 使用合并后的域名映射 |
| 'hints': app_config.get('hints', default_hints), |
| 'cnVersion': cn_config['rncn-config-version'], |
| 'overseaVersion': oversea_config['rncn-config-version'], |
| 'usingDefaultHints': using_default_hints |
| }) |
| except Exception as e: |
| print(f"Error in get_config: {str(e)}") |
| return jsonify({'error': str(e)}), 500 |
| |
| @app.route('/api/config', methods=['POST']) |
| def update_config(): |
| try: |
| data = request.json |
| cn_hosts = data['hosts']['cn'] |
| overseas_hosts = data['hosts']['overseas'] |
| |
| # 读取现有配置 |
| oversea_config = read_config(app_config['files']['oversea']) |
| cn_config = read_config(app_config['files']['cn']) |
| |
| # 检查是否有实际更改 |
| cn_changed = cn_hosts != cn_config['hosts'] |
| oversea_changed = overseas_hosts != oversea_config['hosts'] |
| |
| if not cn_changed and not oversea_changed: |
| return jsonify({ |
| 'success': True, |
| 'message': '配置未发生变化' |
| }) |
| |
| # 只更新发生变化的配置 |
| if cn_changed: |
| cn_config['rncn-config-version'] += 1 |
| cn_config['hosts'] = cn_hosts |
| save_history('cn', cn_config) |
| write_config(app_config['files']['cn'], cn_config) |
| |
| if oversea_changed: |
| oversea_config['rncn-config-version'] += 1 |
| oversea_config['hosts'] = overseas_hosts |
| save_history('oversea', oversea_config) |
| write_config(app_config['files']['oversea'], oversea_config) |
| |
| return jsonify({ |
| 'success': True, |
| 'message': '配置已更新', |
| 'updated': { |
| 'cn': cn_changed, |
| 'oversea': oversea_changed |
| } |
| }) |
| except Exception as e: |
| print(f"Error in update_config: {str(e)}") |
| return jsonify({'error': str(e)}), 500 |
| |
| def save_history(config_type, config_data): |
| history_dir = os.path.join(app_config['files']['history']['dir'], config_type) |
| os.makedirs(history_dir, exist_ok=True) |
| |
| # 只使用版本号命名文件 |
| version = config_data['rncn-config-version'] |
| history_file = f"{config_type}_v{version}.js" |
| |
| # 保存历史文件 |
| filepath = os.path.join(history_dir, history_file) |
| with open(filepath, 'w', encoding='utf-8') as f: |
| json_str = json.dumps(config_data, indent='\t') |
| f.write(json_str) |
| |
| # 清理旧版本 |
| max_versions = app_config['files']['history'].get('maxVersions', 10) |
| files = glob.glob(os.path.join(history_dir, f"{config_type}_v*.js")) |
| files.sort(key=lambda x: os.path.getmtime(x), reverse=True) # 按修改时间排序 |
| for old_file in files[max_versions:]: |
| os.remove(old_file) |
| |
| @app.route('/api/history', methods=['GET']) |
| def get_history(): |
| try: |
| history = {'cn': [], 'oversea': []} |
| history_dir = app_config['files']['history']['dir'] |
| |
| for config_type in ['cn', 'oversea']: |
| type_dir = os.path.join(history_dir, config_type) |
| if os.path.exists(type_dir): |
| files = glob.glob(os.path.join(type_dir, f"{config_type}_v*.js")) |
| # 按修改时间排序 |
| files.sort(key=lambda x: os.path.getmtime(x), reverse=True) |
| |
| for file in files: |
| with open(file, 'r', encoding='utf-8') as f: |
| config = json.load(f) |
| filename = os.path.basename(file) |
| mtime = os.path.getmtime(file) |
| history[config_type].append({ |
| 'version': config['rncn-config-version'], |
| 'timestamp': int(mtime), # 使用文件修改时间 |
| 'filename': filename, |
| 'config': config |
| }) |
| |
| return jsonify(history) |
| except Exception as e: |
| print(f"Error in get_history: {str(e)}") |
| return jsonify({'error': str(e)}), 500 |
| |
| @app.route('/api/history/restore', methods=['POST']) |
| def restore_history(): |
| try: |
| data = request.json |
| config_type = data['type'] # 'cn' 或 'oversea' |
| version = data['version'] |
| |
| # 读取历史文件 |
| history_dir = os.path.join(app_config['files']['history']['dir'], config_type) |
| files = glob.glob(os.path.join(history_dir, f"{config_type}_v{version}.js")) |
| if not files: |
| raise ValueError(f"找不到版本 {version} 的历史记录") |
| |
| # 读取历史配置 |
| with open(files[0], 'r', encoding='utf-8') as f: |
| history_config = json.load(f) |
| |
| # 读取当前配置 |
| current_file = app_config['files'][config_type] |
| current_config = read_config(current_file) |
| |
| # 保存当前配置到历史记录 |
| save_history(config_type, current_config) |
| |
| # 更新历史配置的版本号为当前版本号+1 |
| history_config['rncn-config-version'] = current_config['rncn-config-version'] + 1 |
| |
| # 恢复历史配置 |
| write_config(current_file, history_config) |
| |
| # 保存恢复后的配置到历史记录 |
| save_history(config_type, history_config) |
| |
| return jsonify({ |
| 'success': True, |
| 'newVersion': history_config['rncn-config-version'] |
| }) |
| except Exception as e: |
| print(f"Error in restore_history: {str(e)}") |
| return jsonify({'error': str(e)}), 500 |
| |
| if __name__ == '__main__': |
| app.run( |
| host=app_config.get('server', {}).get('host', '0.0.0.0'), |
| port=app_config.get('server', {}).get('port', 3001), |
| debug=True |
| ) |