update app/grabber/parsers.py

app/grabber/parsers.py

@@ -0,0 +1,85 @@
+import base64
+import json
+from typing import Optional
+from app.models.proxy import Proxy
+
+
+class VMessParser:
+    @staticmethod
+    def parse(url: str) -> Proxy:
+        if not url.startswith("vmess://"):
+            raise ValueError("Invalid VMess URL")
+
+        try:
+            encoded = url.replace("vmess://", "")
+            missing_padding = len(encoded) % 4
+            if missing_padding:
+                encoded += "=" * (4 - missing_padding)
+
+            decoded = base64.b64decode(encoded).decode("utf-8")
+            config = json.loads(decoded)
+
+            return Proxy(
+                ip=config.get("add", ""),
+                port=int(config.get("port", 0)),
+                protocol="vmess",
+                source="subscription",
+            )
+        except Exception as e:
+            raise ValueError(f"Failed to parse VMess URL: {e}")
+
+
+class VLESSParser:
+    @staticmethod
+    def parse(url: str) -> Proxy:
+        if not url.startswith("vless://"):
+            raise ValueError("Invalid VLESS URL")
+
+        try:
+            url_parts = url.replace("vless://", "")
+            uuid_part, server_part = url_parts.split("@", 1)
+            server_port = server_part.split("?")[0]
+            server, port = server_port.rsplit(":", 1)
+
+            return Proxy(
+                ip=server, port=int(port), protocol="vless", source="subscription"
+            )
+        except Exception as e:
+            raise ValueError(f"Failed to parse VLESS URL: {e}")
+
+
+class TrojanParser:
+    @staticmethod
+    def parse(url: str) -> Proxy:
+        if not url.startswith("trojan://"):
+            raise ValueError("Invalid Trojan URL")
+
+        try:
+            url_parts = url.replace("trojan://", "")
+            password_part, server_part = url_parts.split("@", 1)
+            server_port = server_part.split("?")[0]
+            server, port = server_port.rsplit(":", 1)
+
+            return Proxy(
+                ip=server, port=int(port), protocol="trojan", source="subscription"
+            )
+        except Exception as e:
+            raise ValueError(f"Failed to parse Trojan URL: {e}")
+
+
+class SSParser:
+    @staticmethod
+    def parse(url: str) -> Proxy:
+        if not url.startswith("ss://"):
+            raise ValueError("Invalid Shadowsocks URL")
+
+        try:
+            url_parts = url.replace("ss://", "")
+            config_part, server_part = url_parts.split("@", 1)
+            server, port = server_part.split(":", 1)
+
+            return Proxy(
+                ip=server, port=int(port), protocol="shadowsocks", source="subscription"
+            )
+        except Exception as e:
+            raise ValueError(f"Failed to parse Shadowsocks URL: {e}")