import requests
|
import random
|
import asyncio
|
import aiohttp
|
from bs4 import BeautifulSoup
|
from auto_media_publisher.config.conf_base import PUBLIC_IP_CITY_URL,DEFAULT_USER_AGENT
|
|
|
def check_proxy_valid(proxy=None):
|
if proxy:
|
ip, city = get_public_ip_info(proxy)
|
else:
|
ip, city = get_public_ip_info()
|
return ip is not None,ip,city
|
|
def get_public_ip_info(proxy=None):
|
if isinstance(proxy, str) and proxy.strip().lower().startswith(('http://', 'https://')):
|
proxies = {"http": proxy, "https": proxy}
|
else:
|
proxies = None
|
print(f"checking proxy... {proxy}")
|
try:
|
url = random.choice(PUBLIC_IP_CITY_URL)
|
response = requests.get(url, proxies=proxies, timeout=5)
|
response.raise_for_status()
|
data = response.json()
|
|
# 不同服务字段不同
|
ip=None
|
city=None
|
if "query" in data: # ip-api 风格
|
ip = data["query"]
|
elif "ip" in data: # ip-api 风格
|
ip = data["ip"]
|
if "city" in data:
|
city = data["city"]
|
|
return ip, city
|
|
except Exception as e:
|
print("[get_public_ip_info] 请求失败:", e)
|
return None,None
|
|
async def get_free_valid_proxies(protocol='http',count = 50):
|
tasks = [
|
_get_free_valid_proxies_scdn(protocol,count),
|
_get_free_valid_proxies_proxyhub(),
|
_get_free_valid_proxies_proxy5(),
|
_get_free_valid_proxies_proxy_list(),
|
_get_free_valid_proxies_proxy_kuaidaili_intr(),
|
_get_free_valid_proxies_proxy_kuaidaili_inha(),
|
_get_free_valid_proxies_proxy_kuaidaili_dps(),
|
_get_free_valid_proxies_proxy_kuaidaili_89ip()
|
]
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
all_proxies = []
|
for result in results:
|
if isinstance(result, Exception):
|
print("一个源出错:", result)
|
continue
|
if result is not None:
|
all_proxies.extend(result)
|
|
return all_proxies
|
|
|
async def _test_proxy(session, proxy_raw):
|
proxy = f"http://{proxy_raw}"
|
|
test_url = random.choice(PUBLIC_IP_CITY_URL)
|
|
try:
|
async with session.get(test_url, proxy=proxy, timeout=5) as response:
|
if response.status != 200:
|
return None
|
data = await response.json()
|
city = data.get("city")
|
return proxy_raw
|
except Exception as e:
|
print(f"❌ {proxy_raw} 无效: {e}")
|
return None
|
|
async def _get_free_valid_proxies_scdn(protocol='http',count = 50):
|
try:
|
url = 'https://proxy.scdn.io/api/get_proxy.php'
|
params = {
|
'protocol': protocol,
|
'count': count
|
}
|
response = requests.get(url, params=params)
|
response.raise_for_status()
|
data = response.json()
|
valid_proxies = []
|
if data and data['code']==200:
|
proxies_raw = data['data']['proxies']
|
async with aiohttp.ClientSession() as session:
|
tasks = [_test_proxy(session, p) for p in proxies_raw]
|
results = await asyncio.gather(*tasks)
|
# 筛选有效结果
|
valid_proxies = [r for r in results if r is not None]
|
return valid_proxies
|
return ["14.29.47.90:3128"]
|
except:
|
return ["14.29.47.90:3128"]
|
|
async def _get_free_valid_proxies_proxyhub():
|
return await _get_free_valid_proxies_("https://proxyhub.me/zh/cn-http-proxy-list.html")
|
|
async def _get_free_valid_proxies_proxy5():
|
return await _get_free_valid_proxies_("https://proxy5.net/cn/free-proxy/china")
|
|
async def _get_free_valid_proxies_proxy_list():
|
return await _get_free_valid_proxies_("https://free-proxy-list.net/zh-cn/")
|
|
async def _get_free_valid_proxies_proxy_kuaidaili_intr():
|
return await _get_free_valid_proxies_("https://www.kuaidaili.com/free/intr/")
|
|
async def _get_free_valid_proxies_proxy_kuaidaili_inha():
|
return await _get_free_valid_proxies_("https://www.kuaidaili.com/free/inha/")
|
|
async def _get_free_valid_proxies_proxy_kuaidaili_dps():
|
return await _get_free_valid_proxies_("https://www.kuaidaili.com/free/dps/")
|
|
async def _get_free_valid_proxies_proxy_kuaidaili_89ip():
|
return await _get_free_valid_proxies_("https://www.89ip.cn/")
|
|
|
async def _get_free_valid_proxies_(url):
|
try:
|
headers = {"User-Agent": DEFAULT_USER_AGENT}
|
response = requests.get(url, headers=headers)
|
response.raise_for_status()
|
soup = BeautifulSoup(response.text, "html.parser")
|
proxy_list = []
|
tbody = soup.find("tbody")
|
if tbody:
|
rows = tbody.find_all("tr")
|
for tr in rows:
|
tds = tr.find_all("td")
|
if len(tds) >= 2:
|
ip = tds[0].get_text(strip=True)
|
port = tds[1].get_text(strip=True)
|
proxy = f"{ip}:{port}"
|
proxy_list.append(proxy)
|
if len(proxy_list) > 0:
|
async with aiohttp.ClientSession() as session:
|
tasks = [_test_proxy(session, p) for p in proxy_list]
|
results = await asyncio.gather(*tasks)
|
valid_proxies = [r for r in results if r is not None]
|
return valid_proxies
|
return None
|
except:
|
return None
|
