Compare commits
2 Commits
e813199501
...
104eb8d949
Author | SHA1 | Date | |
---|---|---|---|
104eb8d949 | |||
c7a9931561 |
@ -6,6 +6,22 @@ import requests
|
|||||||
|
|
||||||
|
|
||||||
def fetch_list(url):
|
def fetch_list(url):
|
||||||
|
"""Fetches proxy list from the given URL.
|
||||||
|
|
||||||
|
The HTTP response text will be searched for `ip:port` and `username:password@ip:port`
|
||||||
|
patterns to accommodate different source list formats.
|
||||||
|
|
||||||
|
If an error occurs while fetching the list, it will be logged with WARNING
|
||||||
|
and an empty list will be returned.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
url (str): The URL to fetch proxy list from.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
list: A list of proxy server addresses fetched from the URL.
|
||||||
|
If an error occurs while fetching the list, it will be logged
|
||||||
|
with WARNING and an empty list will be returned.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
response = requests.get(url)
|
response = requests.get(url)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
@ -14,9 +30,9 @@ def fetch_list(url):
|
|||||||
return []
|
return []
|
||||||
|
|
||||||
proxy_regex = r"(?:\b(?:[\S]+:)?(?:[\S]+)?@\b)?(?:\d{1,3}\.){3}\d{1,3}:\d+"
|
proxy_regex = r"(?:\b(?:[\S]+:)?(?:[\S]+)?@\b)?(?:\d{1,3}\.){3}\d{1,3}:\d+"
|
||||||
proxies = re.findall(proxy_regex, response.text)
|
proxies = set(re.findall(proxy_regex, response.text))
|
||||||
logging.info(f'Fetched {len(proxies)} proxies from {url}')
|
logging.info(f'Fetched {len(proxies)} proxies from {url}')
|
||||||
return proxies
|
return list(proxies)
|
||||||
|
|
||||||
|
|
||||||
def fetch_all(urls, max_workers=8):
|
def fetch_all(urls, max_workers=8):
|
||||||
|
@ -17,7 +17,8 @@ def start_web_server():
|
|||||||
def test_fetch_list():
|
def test_fetch_list():
|
||||||
expected = ['127.0.0.1:9000', '127.0.0.1:9001', 'username:pa$$@word@127.0.0.1:9002']
|
expected = ['127.0.0.1:9000', '127.0.0.1:9001', 'username:pa$$@word@127.0.0.1:9002']
|
||||||
result = fetch_list('http://localhost:8888/proxies1.txt')
|
result = fetch_list('http://localhost:8888/proxies1.txt')
|
||||||
assert result == expected
|
for proxy in expected:
|
||||||
|
assert proxy in result
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_list_fail():
|
def test_fetch_list_fail():
|
||||||
@ -29,7 +30,8 @@ def test_fetch_list_fail():
|
|||||||
def test_fetch_list_only_valid():
|
def test_fetch_list_only_valid():
|
||||||
expected = ['127.0.0.1:9000', '127.0.0.1:9001', 'username:pa$$@word@127.0.0.1:9002']
|
expected = ['127.0.0.1:9000', '127.0.0.1:9001', 'username:pa$$@word@127.0.0.1:9002']
|
||||||
result = fetch_list('http://localhost:8888/proxies1.txt')
|
result = fetch_list('http://localhost:8888/proxies1.txt')
|
||||||
assert result == expected
|
for proxy in expected:
|
||||||
|
assert proxy in result
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_all():
|
def test_fetch_all():
|
||||||
|
Loading…
Reference in New Issue
Block a user