feat: add support for socks and http/https proxy protocols

app/checker.py

@@ -265,6 +265,10 @@ def _link_to_xray_outbound(link: str) -> dict | None:
                 return ProxyChecker._ss_to_xray(link)
             elif link.startswith("hysteria2://") or link.startswith("hy2://"):
                 return ProxyChecker._hysteria2_to_xray(link)
+            elif link.startswith(("socks5://", "socks4://", "socks4a://")):
+                return ProxyChecker._socks_to_xray(link)
+            elif link.startswith(("http://", "https://")) and "#" in link:
+                return ProxyChecker._http_to_xray(link)
         except Exception:
             pass
         return None
@@ -629,6 +633,52 @@ def _hysteria2_to_xray(link: str) -> dict | None:
         except Exception:
             return None
 
+    @staticmethod
+    def _socks_to_xray(link: str) -> dict | None:
+        """socks5:// / socks4:// / socks4a:// 分享链接转内核 outbound"""
+        try:
+            parsed = urlparse(link)
+            address = parsed.hostname or ""
+            port = parsed.port or 0
+            if not address or not port:
+                return None
+
+            servers = [{"address": address, "port": port}]
+            user = unquote(parsed.username) if parsed.username else ""
+            password = unquote(parsed.password) if parsed.password else ""
+            if user:
+                servers[0]["users"] = [{"user": user, "pass": password}]
+
+            return {
+                "protocol": "socks",
+                "settings": {"servers": servers},
+            }
+        except Exception:
+            return None
+
+    @staticmethod
+    def _http_to_xray(link: str) -> dict | None:
+        """http:// / https:// 代理链接转内核 outbound"""
+        try:
+            parsed = urlparse(link)
+            address = parsed.hostname or ""
+            port = parsed.port or 0
+            if not address or not port:
+                return None
+
+            servers = [{"address": address, "port": port}]
+            user = unquote(parsed.username) if parsed.username else ""
+            password = unquote(parsed.password) if parsed.password else ""
+            if user:
+                servers[0]["users"] = [{"user": user, "pass": password}]
+
+            return {
+                "protocol": "http",
+                "settings": {"servers": servers},
+            }
+        except Exception:
+            return None
+
     # ---- TCP/TLS 直连检测（回退方案） ----
 
     async def _check_tcp_tls(self, info: ProxyConnInfo) -> float | None:
@@ -734,6 +784,10 @@ def _parse_link(self, link: str) -> ProxyConnInfo | None:
                 return self._parse_ss(link)
             elif link.startswith("hysteria2://") or link.startswith("hy2://"):
                 return self._parse_hysteria2(link)
+            elif link.startswith(("socks5://", "socks4://", "socks4a://")):
+                return self._parse_socks(link)
+            elif link.startswith(("http://", "https://")) and "#" in link:
+                return self._parse_http_proxy(link)
         except Exception:
             pass
         return None
@@ -847,6 +901,36 @@ def _parse_hysteria2(self, link: str) -> ProxyConnInfo | None:
         except Exception:
             return None
 
+    def _parse_socks(self, link: str) -> ProxyConnInfo | None:
+        try:
+            parsed = urlparse(link)
+            address = parsed.hostname or ""
+            port = parsed.port or 0
+            if not address or not port:
+                return None
+            protocol = "socks5" if link.lower().startswith("socks5://") else "socks4"
+            return ProxyConnInfo(
+                host=address, port=int(port),
+                use_tls=False, protocol=protocol,
+            )
+        except Exception:
+            return None
+
+    def _parse_http_proxy(self, link: str) -> ProxyConnInfo | None:
+        try:
+            parsed = urlparse(link)
+            address = parsed.hostname or ""
+            port = parsed.port or 0
+            if not address or not port:
+                return None
+            protocol = "https" if link.lower().startswith("https://") else "http"
+            return ProxyConnInfo(
+                host=address, port=int(port),
+                use_tls=(protocol == "https"), protocol=protocol,
+            )
+        except Exception:
+            return None
+
 
 class ConnectivityMonitor:
     """连通性监控器 - 定期检测本地转发端口是否可达"""

app/generator.py

@@ -85,11 +85,52 @@ def _link_to_clash_proxy(link: str, fallback_name: str) -> dict | None:
             return _ss_link_to_clash(link, fallback_name)
         elif link.startswith("hysteria2://") or link.startswith("hy2://"):
             return _hysteria2_link_to_clash(link, fallback_name)
+        elif link.startswith(("socks5://", "socks4://", "socks4a://")):
+            return _socks_link_to_clash(link, fallback_name)
+        elif link.startswith(("http://", "https://")) and "#" in link:
+            return _http_link_to_clash(link, fallback_name)
     except Exception as e:
         logger.debug("转换 Clash 格式失败: %s - %s", link[:50], e)
     return None
 
 
+def _socks_link_to_clash(link: str, fallback_name: str) -> dict:
+    """socks5:// / socks4:// / socks4a:// 转 Clash proxy"""
+    parsed = urlparse(link)
+    name = unquote(parsed.fragment) if parsed.fragment else fallback_name
+    proxy = {
+        "name": name or fallback_name,
+        "type": "socks5",
+        "server": parsed.hostname or "",
+        "port": parsed.port or 1080,
+        "udp": True,
+    }
+    if parsed.username:
+        proxy["username"] = unquote(parsed.username)
+    if parsed.password:
+        proxy["password"] = unquote(parsed.password)
+    return proxy
+
+
+def _http_link_to_clash(link: str, fallback_name: str) -> dict:
+    """http:// / https:// 代理链接转 Clash proxy"""
+    parsed = urlparse(link)
+    name = unquote(parsed.fragment) if parsed.fragment else fallback_name
+    proxy = {
+        "name": name or fallback_name,
+        "type": "http",
+        "server": parsed.hostname or "",
+        "port": parsed.port or 8080,
+    }
+    if parsed.username:
+        proxy["username"] = unquote(parsed.username)
+    if parsed.password:
+        proxy["password"] = unquote(parsed.password)
+    if link.lower().startswith("https://"):
+        proxy["tls"] = True
+    return proxy
+
+
 def _vmess_link_to_clash(link: str, fallback_name: str) -> dict:
     """vmess:// 转 Clash proxy"""
     config_b64 = link[8:]

app/parser.py

@@ -2,7 +2,7 @@
 """订阅解析模块 - 拉取和解析节点订阅链接
 
 核心解析逻辑移植自 Proxy_List/get_connected_proxies/get_connected_proxies.py
-支持 vmess / vless / trojan / ss / hysteria2 五种协议
+支持 vmess / vless / trojan / ss / hysteria2 / socks5 / socks4 / http / https 协议
 """
 
 import base64
@@ -29,18 +29,23 @@ async def fetch_subscription(url: str, timeout: float = 15.0) -> str:
 def parse_subscription(content: str) -> list[ProxyInfo]:
     """解析订阅内容，返回 ProxyInfo 列表
 
-    支持 vmess / vless / trojan / ss / hysteria2 协议
+    支持 vmess / vless / trojan / ss / hysteria2 / socks5 / socks4 / http / https 协议
     自动检测 base64 编码的订阅内容并解码
     """
     share_links: list[ProxyInfo] = []
     lines = content.strip().split("\n")
 
+    # 支持的协议前缀
+    supported_prefixes = (
+        "vmess://", "vless://", "trojan://", "ss://",
+        "hysteria2://", "hy2://",
+        "socks5://", "socks4://", "socks4a://",
+        "http://", "https://",
+    )
+
     # 检测是否为 base64 编码的订阅内容
     has_protocol_prefix = any(
-        line.strip().startswith((
-            "vmess://", "vless://", "trojan://", "ss://",
-            "hysteria2://", "hy2://",
-        ))
+        line.strip().startswith(supported_prefixes)
         for line in lines if line.strip()
     )
 
@@ -67,6 +72,11 @@ def parse_subscription(content: str) -> list[ProxyInfo]:
             info = _parse_ss(line)
         elif line.startswith("hysteria2://") or line.startswith("hy2://"):
             info = _parse_hysteria2(line)
+        elif line.startswith(("socks5://", "socks4://", "socks4a://")):
+            info = _parse_socks(line)
+        elif line.startswith(("http://", "https://")) and "#" in line:
+            # 仅当 http/https 链接带 #fragment 时视为代理节点（避免误判普通 URL）
+            info = _parse_http_proxy(line)
         else:
             info = None
 
@@ -214,6 +224,56 @@ def _parse_hysteria2(line: str) -> ProxyInfo | None:
         return None
 
 
+def _parse_socks(line: str) -> ProxyInfo | None:
+    """解析 socks5:// / socks4:// / socks4a:// 链接
+
+    格式: socks5://[user:pass@]host:port[#name]
+    """
+    try:
+        parsed = urlparse(line)
+        address = parsed.hostname or ""
+        port = str(parsed.port) if parsed.port else ""
+        if not address or not port:
+            return None
+        name = unquote(parsed.fragment) if parsed.fragment else f"{address}:{port}"
+        # 统一协议名
+        protocol = "socks5" if line.lower().startswith("socks5://") else "socks4"
+        return ProxyInfo(
+            protocol=protocol,
+            name=name,
+            address=address,
+            port=port,
+            link=line,
+        )
+    except Exception:
+        return None
+
+
+def _parse_http_proxy(line: str) -> ProxyInfo | None:
+    """解析 http:// / https:// 代理链接
+
+    格式: http://[user:pass@]host:port[#name]
+    仅当带 #fragment 时视为代理节点（避免误判普通 URL）
+    """
+    try:
+        parsed = urlparse(line)
+        address = parsed.hostname or ""
+        port = str(parsed.port) if parsed.port else ""
+        if not address or not port:
+            return None
+        name = unquote(parsed.fragment) if parsed.fragment else f"{address}:{port}"
+        protocol = "https" if line.lower().startswith("https://") else "http"
+        return ProxyInfo(
+            protocol=protocol,
+            name=name,
+            address=address,
+            port=port,
+            link=line,
+        )
+    except Exception:
+        return None
+
+
 # ---- 服务实例 API 获取 ----
 
 async def instance_login(base_url: str, username: str, password: str,