Delete converter.py

converter.py

@@ -1,492 +0,0 @@
-import base64
-import json
-import re
-import yaml
-import requests
-from ruamel.yaml import YAML
-from urllib.parse import urlparse, parse_qs
-
-# 用于生成美观YAML的实例
-ruamel_yaml = YAML()
-ruamel_yaml.indent(mapping=2, sequence=4, offset=2)
-ruamel_yaml.width = 80
-ruamel_yaml.allow_unicode = True
-
-class SubscriptionConverter:
-    def __init__(self):
-        self.supported_types = ['clash', 'clashr', 'surge', 'quan', 'quanx', 'loon', 'ss', 'ssr', 'v2ray']
-
-    def convert(self, subscription_url, target_type, config_url=None):
-        """
-        转换订阅链接到目标格式
-        """
-        if target_type not in self.supported_types:
-            return {"status": "error", "message": f"不支持的目标类型: {target_type}"}
-            
-        try:
-            # 获取原始订阅内容
-            nodes = self._fetch_subscription(subscription_url)
-            if not nodes:
-                return {"status": "error", "message": "无法获取订阅内容或订阅内容为空"}
-            
-            # 获取配置文件 (如果提供)
-            config = {}
-            if config_url:
-                config = self._fetch_config(config_url)
-            
-            # 根据目标类型进行转换
-            if target_type == 'clash':
-                result = self._convert_to_clash(nodes, config)
-            elif target_type == 'v2ray':
-                result = self._convert_to_v2ray(nodes)
-            else:
-                # 其他格式转换逻辑
-                return {"status": "error", "message": f"目标类型 {target_type} 暂未实现具体转换逻辑"}
-            
-            return {"status": "success", "result": result}
-            
-        except Exception as e:
-            return {"status": "error", "message": f"转换过程出错: {str(e)}"}
-    
-    def _fetch_subscription(self, url):
-        """获取订阅内容并解析节点"""
-        try:
-            response = requests.get(url, timeout=15)
-            if response.status_code != 200:
-                return None
-                
-            content = response.text
-            # 尝试Base64解码（大多数订阅是Base64编码的）
-            try:
-                decoded = base64.b64decode(content).decode('utf-8')
-                content = decoded
-            except:
-                # 非Base64编码或已经解码，使用原始内容
-                pass
-                
-            # 解析节点信息（根据订阅内容类型）
-            if content.startswith('{'): # JSON格式
-                try:
-                    data = json.loads(content)
-                    return self._parse_json_subscription(data)
-                except:
-                    pass
-                    
-            elif 'proxies:' in content: # YAML (Clash)格式
-                try:
-                    data = yaml.safe_load(content)
-                    return self._parse_yaml_subscription(data)
-                except:
-                    pass
-                    
-            else: # 文本格式（每行一个节点链接）
-                return self._parse_text_subscription(content)
-                
-            return []
-        except Exception as e:
-            print(f"获取订阅内容失败: {str(e)}")
-            return []
-    
-    def _parse_json_subscription(self, data):
-        """解析JSON格式的订阅内容"""
-        nodes = []
-        
-        # 支持常见JSON订阅格式
-        if 'proxies' in data:
-            return data['proxies']  # Clash格式
-        elif 'servers' in data:
-            # SS格式
-            for server in data['servers']:
-                nodes.append({
-                    'type': 'ss',
-                    'name': server.get('remarks', f"Server {len(nodes) + 1}"),
-                    'server': server.get('server', ''),
-                    'port': server.get('server_port', 0),
-                    'password': server.get('password', ''),
-                    'cipher': server.get('method', 'aes-256-gcm')
-                })
-                
-        return nodes
-    
-    def _parse_yaml_subscription(self, data):
-        """解析YAML格式的订阅内容"""
-        if 'proxies' in data and isinstance(data['proxies'], list):
-            return data['proxies']
-        return []
-    
-    def _parse_text_subscription(self, content):
-        """解析文本格式的订阅内容（每行一个URI）"""
-        nodes = []
-        for line in content.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-                
-            # 尝试解析不同协议的URI
-            if line.startswith('ss://'):
-                node = self._parse_ss_uri(line)
-                if node:
-                    nodes.append(node)
-            elif line.startswith('ssr://'):
-                node = self._parse_ssr_uri(line)
-                if node:
-                    nodes.append(node)
-            elif line.startswith('vmess://'):
-                node = self._parse_vmess_uri(line)
-                if node:
-                    nodes.append(node)
-            elif line.startswith('trojan://'):
-                node = self._parse_trojan_uri(line)
-                if node:
-                    nodes.append(node)
-                    
-        return nodes
-    
-    def _parse_ss_uri(self, uri):
-        """解析 SS 协议URI"""
-        try:
-            if '#' in uri:
-                encoded_part, name = uri.split('#', 1)
-                name = name.strip()
-            else:
-                encoded_part = uri
-                name = f"SS {uri[:8]}"
-                
-            encoded_part = encoded_part.replace('ss://', '')
-            
-            # 处理不同格式的SS链接
-            if '@' in encoded_part:
-                # ss://method:password@server:port
-                auth_part, server_part = encoded_part.split('@', 1)
-                
-                # 处理method:password部分 (可能需要解码)
-                if ':' not in auth_part:
-                    try:
-                        auth_part = base64.b64decode(auth_part).decode('utf-8')
-                    except:
-                        pass
-                
-                if ':' in auth_part:
-                    method, password = auth_part.split(':', 1)
-                else:
-                    return None
-                    
-                # 处理server:port部分
-                server, port = server_part.split(':', 1)
-                port = int(port)
-            else:
-                # ss://BASE64(method:password@server:port)
-                try:
-                    decoded = base64.b64decode(encoded_part).decode('utf-8')
-                    if '@' in decoded:
-                        auth_part, server_part = decoded.split('@', 1)
-                        method, password = auth_part.split(':', 1)
-                        server, port_str = server_part.split(':', 1)
-                        port = int(port_str)
-                    else:
-                        return None
-                except:
-                    return None
-            
-            return {
-                'type': 'ss',
-                'name': name,
-                'server': server,
-                'port': port,
-                'password': password,
-                'cipher': method
-            }
-        except:
-            return None
-    
-    def _parse_ssr_uri(self, uri):
-        """解析 SSR 协议URI"""
-        try:
-            encoded_part = uri.replace('ssr://', '')
-            
-            # SSR链接格式: ssr://BASE64(server:port:protocol:method:obfs:BASE64(password)/?params)
-            try:
-                decoded = base64.b64decode(encoded_part).decode('utf-8')
-            except:
-                return None
-                
-            # 分离主要部分和参数部分
-            if '/' in decoded:
-                main_part, params_part = decoded.split('/', 1)
-                params = parse_qs(params_part.lstrip('?'))
-            else:
-                main_part = decoded
-                params = {}
-                
-            # 解析主要部分
-            parts = main_part.split(':')
-            if len(parts) < 6:
-                return None
-                
-            server, port, protocol, method, obfs = parts[:5]
-            password_base64 = parts[5]
-            try:
-                password = base64.b64decode(password_base64).decode('utf-8')
-            except:
-                password = password_base64
-                
-            # 获取参数
-            obfs_param = base64.b64decode(params.get('obfsparam', [''])[0]).decode('utf-8') if 'obfsparam' in params else ''
-            protocol_param = base64.b64decode(params.get('protoparam', [''])[0]).decode('utf-8') if 'protoparam' in params else ''
-            remarks = base64.b64decode(params.get('remarks', [''])[0]).decode('utf-8') if 'remarks' in params else f"SSR {server[:8]}"
-            
-            return {
-                'type': 'ssr',
-                'name': remarks,
-                'server': server,
-                'port': int(port),
-                'password': password,
-                'cipher': method,
-                'protocol': protocol,
-                'protocol-param': protocol_param,
-                'obfs': obfs,
-                'obfs-param': obfs_param
-            }
-        except:
-            return None
-    
-    def _parse_vmess_uri(self, uri):
-        """解析 VMess 协议URI"""
-        try:
-            encoded_part = uri.replace('vmess://', '')
-            
-            # VMess链接通常是Base64编码的JSON
-            try:
-                decoded = base64.b64decode(encoded_part).decode('utf-8')
-                config = json.loads(decoded)
-            except:
-                return None
-                
-            # 标准格式包含v,ps,add,port,id,aid,net等字段
-            return {
-                'type': 'vmess',
-                'name': config.get('ps', f"VMess Server"),
-                'server': config.get('add', ''),
-                'port': int(config.get('port', 0)),
-                'uuid': config.get('id', ''),
-                'alterId': int(config.get('aid', 0)),
-                'cipher': 'auto',
-                'network': config.get('net', 'tcp'),
-                'tls': True if config.get('tls') == 'tls' else False,
-                'ws-path': config.get('path', ''),
-                'ws-headers': {'Host': config.get('host', '')} if config.get('host') else {}
-            }
-        except:
-            return None
-    
-    def _parse_trojan_uri(self, uri):
-        """解析 Trojan 协议URI"""
-        try:
-            uri = uri.replace('trojan://', '')
-            
-            # trojan://password@server:port?allowInsecure=1&peer=example.com#name
-            match = re.match(r'^([^@]+)@([^:]+):(\d+)(.*)$', uri)
-            if not match:
-                return None
-                
-            password, server, port, params_part = match.groups()
-            
-            # 解析参数
-            name = ''
-            sni = ''
-            allow_insecure = False
-            
-            if '#' in params_part:
-                params_part, name = params_part.split('#', 1)
-                
-            if '?' in params_part:
-                params_str = params_part.lstrip('?')
-                params = parse_qs(params_str)
-                sni = params.get('peer', [''])[0] or params.get('sni', [''])[0]
-                allow_insecure = params.get('allowInsecure', ['0'])[0] == '1'
-            
-            return {
-                'type': 'trojan',
-                'name': name or f"Trojan {server[:8]}",
-                'server': server,
-                'port': int(port),
-                'password': password,
-                'sni': sni,
-                'skip-cert-verify': allow_insecure
-            }
-        except:
-            return None
-    
-    def _fetch_config(self, config_url):
-        """获取配置文件内容"""
-        try:
-            response = requests.get(config_url, timeout=15)
-            if response.status_code != 200:
-                return {}
-                
-            content = response.text
-            
-            # 检测配置文件格式并解析
-            if content.startswith('{'): # JSON
-                return json.loads(content)
-            elif '[' in content and ']' in content: # INI
-                return self._parse_ini_config(content)
-            else: # 尝试作为YAML解析
-                try:
-                    return yaml.safe_load(content) or {}
-                except:
-                    return {}
-        except:
-            return {}
-    
-    def _parse_ini_config(self, content):
-        """解析INI格式的配置文件"""
-        config = {
-            'rules': [],
-            'groups': []
-        }
-        
-        current_section = None
-        
-        for line in content.splitlines():
-            line = line.strip()
-            if not line or line.startswith(';') or line.startswith('#'):
-                continue
-                
-            # 检测节
-            if line.startswith('[') and line.endswith(']'):
-                current_section = line[1:-1].strip()
-                continue
-                
-            # 处理Rule节
-            if current_section == 'Rule':
-                config['rules'].append(line)
-            # 处理Proxy Group节
-            elif current_section and 'Group' in current_section:
-                config['groups'].append(line)
-                
-        return config
-    
-    def _convert_to_clash(self, nodes, config):
-        """转换为Clash配置格式"""
-        # 创建基本结构
-        clash_config = {
-            'port': 7890,
-            'socks-port': 7891,
-            'allow-lan': True,
-            'mode': 'Rule',
-            'log-level': 'info',
-            'external-controller': '127.0.0.1:9090',
-            'proxies': nodes,
-            'proxy-groups': [],
-            'rules': []
-        }
-        
-        # 根据配置创建代理组
-        if config.get('groups'):
-            default_groups = [
-                {
-                    'name': '🚀 节点选择',
-                    'type': 'select',
-                    'proxies': ['DIRECT'] + [node['name'] for node in nodes]
-                },
-                {
-                    'name': '🌍 国外媒体',
-                    'type': 'select',
-                    'proxies': ['🚀 节点选择'] + [node['name'] for node in nodes]
-                },
-                {
-                    'name': '📲 电报信息',
-                    'type': 'select',
-                    'proxies': ['🚀 节点选择'] + [node['name'] for node in nodes]
-                },
-                {
-                    'name': '🍎 苹果服务',
-                    'type': 'select',
-                    'proxies': ['DIRECT', '🚀 节点选择']
-                },
-                {
-                    'name': '🎯 全球直连',
-                    'type': 'select',
-                    'proxies': ['DIRECT', '🚀 节点选择']
-                },
-                {
-                    'name': '🛑 全球拦截',
-                    'type': 'select',
-                    'proxies': ['REJECT', 'DIRECT']
-                },
-                {
-                    'name': '⚓ 漏网之鱼',
-                    'type': 'select',
-                    'proxies': ['🚀 节点选择', 'DIRECT']
-                }
-            ]
-            clash_config['proxy-groups'] = default_groups
-        
-        # 添加规则
-        if config.get('rules'):
-            clash_config['rules'] = config.get('rules', [])
-        else:
-            # 默认规则
-            clash_config['rules'] = [
-                'DOMAIN-SUFFIX,google.com,🚀 节点选择',
-                'DOMAIN-KEYWORD,google,🚀 节点选择',
-                'DOMAIN-SUFFIX,ad.com,🛑 全球拦截',
-                'DOMAIN-SUFFIX,apple.com,🍎 苹果服务',
-                'DOMAIN-SUFFIX,telegram.org,📲 电报信息',
-                'DOMAIN-SUFFIX,youtube.com,🌍 国外媒体',
-                'DOMAIN-SUFFIX,netflix.com,🌍 国外媒体',
-                'GEOIP,CN,🎯 全球直连',
-                'MATCH,⚓ 漏网之鱼'
-            ]
-        
-        # 转换成YAML格式文本
-        yaml_str = yaml.dump(clash_config, allow_unicode=True, sort_keys=False)
-        return yaml_str
-    
-    def _convert_to_v2ray(self, nodes):
-        """转换为V2Ray格式"""
-        # 实际情况下，可能需要更复杂的处理逻辑
-        v2ray_config = {
-            "outbounds": []
-        }
-        
-        for node in nodes:
-            if node['type'] == 'vmess':
-                outbound = {
-                    "protocol": "vmess",
-                    "settings": {
-                        "vnext": [
-                            {
-                                "address": node['server'],
-                                "port": node['port'],
-                                "users": [
-                                    {
-                                        "id": node['uuid'],
-                                        "alterId": node.get('alterId', 0),
-                                        "security": node.get('cipher', 'auto')
-                                    }
-                                ]
-                            }
-                        ]
-                    },
-                    "tag": node['name']
-                }
-                
-                # 添加额外设置
-                if node.get('network') == 'ws':
-                    outbound["streamSettings"] = {
-                        "network": "ws",
-                        "security": "tls" if node.get('tls') else "none",
-                        "wsSettings": {
-                            "path": node.get('ws-path', ''),
-                            "headers": node.get('ws-headers', {})
-                        }
-                    }
-                
-                v2ray_config["outbounds"].append(outbound)
-        
-        return json.dumps(v2ray_config, ensure_ascii=False, indent=2)
-
-# 实例化供使用
-converter = SubscriptionConverter()