-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathparser.py
More file actions
116 lines (96 loc) · 3.44 KB
/
Copy pathparser.py
File metadata and controls
116 lines (96 loc) · 3.44 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""Link parser for MTProto and WEB proxy URLs."""
import re
from urllib.parse import parse_qs, urlparse
from models import ProxyBase, ProxyType
class ProxyLinkParser:
"""Parser for MTProto proxy links."""
TG_PATTERN = re.compile(
r"tg://(?:proxy|webproxy)\?[^\s<>\"'\]\)]+",
re.IGNORECASE,
)
HTTPS_PATTERN = re.compile(
r"https?://t\.me/(?:proxy|webproxy)\?[^\s<>\"'\]\)]+",
re.IGNORECASE,
)
@classmethod
def clean_link(cls, url: str) -> str:
"""
Clean a proxy URL by removing trailing punctuation and artifacts
that might be captured from messy text.
"""
url = url.strip()
while url and url[-1] in ".,;:!?)]'\"":
url = url[:-1]
return url
@classmethod
def parse_single(cls, url: str) -> ProxyBase | None:
"""
Parse a single proxy URL.
Returns ProxyBase or None if invalid.
"""
url = cls.clean_link(url)
if url.lower().startswith("tg://"):
url = "https://t.me/" + url.split("://", 1)[1]
try:
parsed = urlparse(url)
params = parse_qs(parsed.query)
server = params.get("server", [None])[0]
port_str = params.get("port", [None])[0]
secret = params.get("secret", [None])[0]
proxy_type = (
ProxyType.WEB
if parsed.path.lower() == "/webproxy"
else ProxyType.MT_PROTO
)
if not all([server, secret]):
return None
if proxy_type == ProxyType.WEB:
port = 443
elif not port_str:
return None
else:
port = int(port_str)
return ProxyBase(
server=server, port=port, secret=secret, proxy_type=proxy_type
)
except (ValueError, TypeError):
return None
@classmethod
def parse_text(cls, text: str) -> tuple[list[ProxyBase], list[str]]:
"""
Extract all proxy links from text.
Returns (list of valid proxies, list of error messages).
"""
proxies: list[ProxyBase] = []
errors: list[str] = []
seen: set[tuple[str, int, str, ProxyType]] = set()
tg_links = cls.TG_PATTERN.findall(text)
https_links = cls.HTTPS_PATTERN.findall(text)
all_links = tg_links + https_links
for link in all_links:
proxy = cls.parse_single(link)
if proxy:
key = (proxy.server, proxy.port, proxy.secret, proxy.proxy_type)
if key not in seen:
seen.add(key)
proxies.append(proxy)
else:
errors.append("Invalid proxy link")
return proxies, errors
@classmethod
def generate_link(
cls,
server: str,
port: int,
secret: str,
format: str = "tg",
proxy_type: ProxyType = ProxyType.MT_PROTO,
) -> str:
"""Generate proxy link in specified format."""
if proxy_type == ProxyType.WEB:
if format == "https":
return f"https://t.me/webproxy?server={server}&secret={secret}"
return f"tg://webproxy?server={server}&secret={secret}"
if format == "https":
return f"https://t.me/proxy?server={server}&port={port}&secret={secret}"
return f"tg://proxy?server={server}&port={port}&secret={secret}"