-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* 增加私人代理池场景中,带身份认证的代理的支持 * 修复忘记修改部分 --------- Co-authored-by: dashuai xu <[email protected]>
- Loading branch information
Showing
1 changed file
with
35 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -5,7 +5,10 @@ def is_valid_proxy(data): | |
""" | ||
check this string is within proxy format | ||
""" | ||
if data.__contains__(':'): | ||
if is_auth_proxy(data): | ||
host, port = extract_auth_proxy(data) | ||
return is_ip_valid(host) and is_port_valid(port) | ||
elif data.__contains__(':'): | ||
ip = data.split(':')[0] | ||
port = data.split(':')[1] | ||
return is_ip_valid(ip) and is_port_valid(port) | ||
|
@@ -17,6 +20,8 @@ def is_ip_valid(ip): | |
""" | ||
check this string is within ip format | ||
""" | ||
if is_auth_proxy(ip): | ||
ip = ip.split('@')[1] | ||
a = ip.split('.') | ||
if len(a) != 4: | ||
return False | ||
|
@@ -48,9 +53,36 @@ def convert_proxy_or_proxies(data): | |
# skip invalid item | ||
item = item.strip() | ||
if not is_valid_proxy(item): continue | ||
host, port = item.split(':') | ||
if is_auth_proxy(item): | ||
host, port = extract_auth_proxy(item) | ||
else: | ||
host, port = item.split(':') | ||
result.append(Proxy(host=host, port=int(port))) | ||
return result | ||
if isinstance(data, str) and is_valid_proxy(data): | ||
host, port = data.split(':') | ||
if is_auth_proxy(data): | ||
host, port = extract_auth_proxy(data) | ||
else: | ||
host, port = data.split(':') | ||
return Proxy(host=host, port=int(port)) | ||
|
||
|
||
def is_auth_proxy(data: str) -> bool: | ||
return '@' in data | ||
|
||
|
||
def extract_auth_proxy(data: str) -> (str, str): | ||
""" | ||
extract host and port from a proxy with authentication | ||
""" | ||
auth = data.split('@')[0] | ||
ip_port = data.split('@')[1] | ||
ip = ip_port.split(':')[0] | ||
port = ip_port.split(':')[1] | ||
host = auth + '@' + ip | ||
return host, port | ||
|
||
|
||
if __name__ == '__main__': | ||
proxy = 'test1234:[email protected]:32425' | ||
print(extract_auth_proxy(proxy)) |