Skip to content

Commit f1e7124

Browse files
NanmiCoderNanmiCoder
committed
fix: proxy extract error
1 parent 6eef02d commit f1e7124

File tree

4 files changed

+268
-16
lines changed

4 files changed

+268
-16
lines changed

‎proxy/base_proxy.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ async def get_proxy(self, num: int) -> List[IpInfoModel]:
5353

5454
class IpCache:
5555
def __init__(self):
56-
self.cache_client: AbstractCache = CacheFactory.create_cache(cache_type=config.CACHE_TYPE_MEMORY)
56+
self.cache_client: AbstractCache = CacheFactory.create_cache(cache_type=config.CACHE_TYPE_REDIS)
5757

5858
def set_ip(self, ip_key: str, ip_value_info: str, ex: int):
5959
"""

‎proxy/providers/kuaidl_proxy.py‎

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -33,11 +33,14 @@
3333
from proxy.types import ProviderNameEnum
3434
from tools import utils
3535

36+
# 快代理的IP代理过期时间向前推移5秒，避免临界时间使用失败
37+
DELTA_EXPIRED_SECOND = 5
38+
3639

3740
class KuaidailiProxyModel(BaseModel):
3841
ip: str = Field("ip")
3942
port: int = Field("端口")
40-
expire_ts: int = Field("过期时间")
43+
expire_ts: int = Field("过期时间，单位秒，多少秒后过期")
4144

4245

4346
def parse_kuaidaili_proxy(proxy_info: str) -> KuaidailiProxyModel:
@@ -114,7 +117,7 @@ async def get_proxy(self, num: int) -> List[IpInfoModel]:
114117
response = await client.get(self.api_base + uri, params=self.params)
115118

116119
if response.status_code != 200:
117-
utils.logger.error(f"[KuaiDaiLiProxy.get_proxies] statuc code not 200 and response.txt:{response.text}")
120+
utils.logger.error(f"[KuaiDaiLiProxy.get_proxies] statuc code not 200 and response.txt:{response.text}, status code: {response.status_code}")
118121
raise Exception("get ip error from proxy provider and status code not 200 ...")
119122

120123
ip_response: Dict = response.json()
@@ -125,16 +128,19 @@ async def get_proxy(self, num: int) -> List[IpInfoModel]:
125128
proxy_list: List[str] = ip_response.get("data", {}).get("proxy_list")
126129
for proxy in proxy_list:
127130
proxy_model = parse_kuaidaili_proxy(proxy)
131+
# expire_ts是相对时间（秒数），需要转换为绝对时间戳
132+
# 提前DELTA_EXPIRED_SECOND秒认为过期，避免临界时间使用失败
128133
ip_info_model = IpInfoModel(
129134
ip=proxy_model.ip,
130135
port=proxy_model.port,
131136
user=self.kdl_user_name,
132137
password=self.kdl_user_pwd,
133-
expired_time_ts=proxy_model.expire_ts,
138+
expired_time_ts=proxy_model.expire_ts + utils.get_unix_timestamp() - DELTA_EXPIRED_SECOND,
134139

135140
)
136141
ip_key = f"{self.proxy_brand_name}_{ip_info_model.ip}_{ip_info_model.port}"
137-
self.ip_cache.set_ip(ip_key, ip_info_model.model_dump_json(), ex=ip_info_model.expired_time_ts)
142+
# 缓存过期时间使用相对时间（秒数），也需要减去缓冲时间
143+
self.ip_cache.set_ip(ip_key, ip_info_model.model_dump_json(), ex=proxy_model.expire_ts - DELTA_EXPIRED_SECOND)
138144
ip_infos.append(ip_info_model)
139145

140146
return ip_cache_list + ip_infos
@@ -143,12 +149,22 @@ async def get_proxy(self, num: int) -> List[IpInfoModel]:
143149
def new_kuai_daili_proxy() -> KuaiDaiLiProxy:
144150
"""
145151
构造快代理HTTP实例
152+
支持两种环境变量命名格式：
153+
1. 大写格式：KDL_SECERT_ID, KDL_SIGNATURE, KDL_USER_NAME, KDL_USER_PWD
154+
2. 小写格式：kdl_secret_id, kdl_signature, kdl_user_name, kdl_user_pwd
155+
优先使用大写格式，如果不存在则使用小写格式
146156
Returns:
147157
148158
"""
159+
# 支持大小写两种环境变量格式，优先使用大写
160+
kdl_secret_id = os.getenv("KDL_SECERT_ID") or os.getenv("kdl_secret_id", "你的快代理secert_id")
161+
kdl_signature = os.getenv("KDL_SIGNATURE") or os.getenv("kdl_signature", "你的快代理签名")
162+
kdl_user_name = os.getenv("KDL_USER_NAME") or os.getenv("kdl_user_name", "你的快代理用户名")
163+
kdl_user_pwd = os.getenv("KDL_USER_PWD") or os.getenv("kdl_user_pwd", "你的快代理密码")
164+
149165
return KuaiDaiLiProxy(
150-
kdl_secret_id=os.getenv("kdl_secret_id", "你的快代理secert_id"),
151-
kdl_signature=os.getenv("kdl_signature", "你的快代理签名"),
152-
kdl_user_name=os.getenv("kdl_user_name", "你的快代理用户名"),
153-
kdl_user_pwd=os.getenv("kdl_user_pwd", "你的快代理密码"),
166+
kdl_secret_id=kdl_secret_id,
167+
kdl_signature=kdl_signature,
168+
kdl_user_name=kdl_user_name,
169+
kdl_user_pwd=kdl_user_pwd,
154170
)

‎proxy/providers/wandou_http_proxy.py‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -109,11 +109,14 @@ async def get_proxy(self, num: int) -> List[IpInfoModel]:
109109
def new_wandou_http_proxy() -> WanDouHttpProxy:
110110
"""
111111
构造豌豆HTTP实例
112+
支持两种环境变量命名格式：
113+
1. 大写格式：WANDOU_APP_KEY
114+
2. 小写格式：wandou_app_key
115+
优先使用大写格式，如果不存在则使用小写格式
112116
Returns:
113117
114118
"""
115-
return WanDouHttpProxy(
116-
app_key=os.getenv(
117-
"wandou_app_key", "你的豌豆HTTP app_key"
118-
), # 通过环境变量的方式获取豌豆HTTP app_key
119-
)
119+
# 支持大小写两种环境变量格式，优先使用大写
120+
app_key = os.getenv("WANDOU_APP_KEY") or os.getenv("wandou_app_key", "你的豌豆HTTP app_key")
121+
122+
return WanDouHttpProxy(app_key=app_key)

‎test/test_proxy_ip_pool.py‎

Lines changed: 235 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,250 @@
2222
# @Author : [email protected]
2323
# @Time : 2023/12/2 14:42
2424
# @Desc :
25+
import time
2526
from unittest import IsolatedAsyncioTestCase
27+
from unittest.mock import AsyncMock, MagicMock
2628

27-
from proxy.proxy_ip_pool import create_ip_pool
29+
from proxy.proxy_ip_pool import create_ip_pool, ProxyIpPool
2830
from proxy.types import IpInfoModel
2931

3032

3133
class TestIpPool(IsolatedAsyncioTestCase):
3234
async def test_ip_pool(self):
3335
pool = await create_ip_pool(ip_pool_count=1, enable_validate_ip=True)
3436
print("\n")
35-
for i in range(3):
37+
for _ in range(3):
3638
ip_proxy_info: IpInfoModel = await pool.get_proxy()
3739
print(ip_proxy_info)
3840
self.assertIsNotNone(ip_proxy_info.ip, msg="验证 ip 是否获取成功")
41+
42+
async def test_ip_expiration(self):
43+
"""测试IP代理过期检测功能"""
44+
print("\n=== 开始测试IP代理过期检测 ===")
45+
46+
# 1. 创建IP池并获取一个代理
47+
pool = await create_ip_pool(ip_pool_count=2, enable_validate_ip=True)
48+
ip_proxy_info: IpInfoModel = await pool.get_proxy()
49+
print(f"获取到的代理: {ip_proxy_info.ip}:{ip_proxy_info.port}")
50+
51+
# 2. 测试未过期的情况
52+
if ip_proxy_info.expired_time_ts:
53+
print(f"代理过期时间戳: {ip_proxy_info.expired_time_ts}")
54+
print(f"当前时间戳: {int(time.time())}")
55+
print(f"剩余有效时间: {ip_proxy_info.expired_time_ts - int(time.time())} 秒")
56+
57+
is_expired = ip_proxy_info.is_expired(buffer_seconds=30)
58+
print(f"代理是否过期(缓冲30秒): {is_expired}")
59+
self.assertFalse(is_expired, msg="新获取的IP应该未过期")
60+
else:
61+
print("当前代理未设置过期时间，跳过过期检测")
62+
63+
# 3. 测试即将过期的情况（设置为5分钟后过期）
64+
current_ts = int(time.time())
65+
five_minutes_later = current_ts + 300 # 5分钟 = 300秒
66+
ip_proxy_info.expired_time_ts = five_minutes_later
67+
print(f"\n设置代理过期时间为5分钟后: {five_minutes_later}")
68+
69+
# 不应该过期（缓冲30秒）
70+
is_expired_30s = ip_proxy_info.is_expired(buffer_seconds=30)
71+
print(f"代理是否过期(缓冲30秒): {is_expired_30s}")
72+
self.assertFalse(is_expired_30s, msg="5分钟后过期的IP，缓冲30秒不应该过期")
73+
74+
# 4. 测试已过期的情况（设置为已经过期）
75+
expired_ts = current_ts - 60 # 1分钟前已过期
76+
ip_proxy_info.expired_time_ts = expired_ts
77+
print(f"\n设置代理过期时间为1分钟前: {expired_ts}")
78+
79+
is_expired = ip_proxy_info.is_expired(buffer_seconds=30)
80+
print(f"代理是否过期(缓冲30秒): {is_expired}")
81+
self.assertTrue(is_expired, msg="已过期的IP应该被检测为过期")
82+
83+
# 5. 测试临界过期情况（29秒后过期，缓冲30秒应该认为已过期）
84+
almost_expired_ts = current_ts + 29
85+
ip_proxy_info.expired_time_ts = almost_expired_ts
86+
print(f"\n设置代理过期时间为29秒后: {almost_expired_ts}")
87+
88+
is_expired_critical = ip_proxy_info.is_expired(buffer_seconds=30)
89+
print(f"代理是否过期(缓冲30秒): {is_expired_critical}")
90+
self.assertTrue(is_expired_critical, msg="29秒后过期的IP，缓冲30秒应该被认为已过期")
91+
92+
print("\n=== IP代理过期检测测试完成 ===")
93+
94+
async def test_proxy_pool_auto_refresh(self):
95+
"""测试代理池自动刷新过期代理的功能"""
96+
print("\n=== 开始测试代理池自动刷新功能 ===")
97+
98+
# 1. 创建IP池
99+
pool = await create_ip_pool(ip_pool_count=2, enable_validate_ip=True)
100+
101+
# 2. 获取一个代理
102+
first_proxy = await pool.get_proxy()
103+
print(f"首次获取代理: {first_proxy.ip}:{first_proxy.port}")
104+
105+
# 验证当前代理未过期
106+
is_expired = pool.is_current_proxy_expired(buffer_seconds=30)
107+
print(f"当前代理是否过期: {is_expired}")
108+
109+
if first_proxy.expired_time_ts:
110+
print(f"当前代理过期时间戳: {first_proxy.expired_time_ts}")
111+
112+
# 3. 手动设置当前代理为已过期
113+
current_ts = int(time.time())
114+
pool.current_proxy.expired_time_ts = current_ts - 60
115+
print(f"\n手动设置代理为已过期（1分钟前）")
116+
117+
# 4. 检测是否过期
118+
is_expired_after = pool.is_current_proxy_expired(buffer_seconds=30)
119+
print(f"设置后代理是否过期: {is_expired_after}")
120+
self.assertTrue(is_expired_after, msg="手动设置过期后应该被检测为过期")
121+
122+
# 5. 使用 get_or_refresh_proxy 自动刷新
123+
print("\n调用 get_or_refresh_proxy 自动刷新过期代理...")
124+
refreshed_proxy = await pool.get_or_refresh_proxy(buffer_seconds=30)
125+
print(f"刷新后的代理: {refreshed_proxy.ip}:{refreshed_proxy.port}")
126+
127+
# 6. 验证新代理未过期
128+
is_new_expired = pool.is_current_proxy_expired(buffer_seconds=30)
129+
print(f"新代理是否过期: {is_new_expired}")
130+
self.assertFalse(is_new_expired, msg="刷新后的新代理应该未过期")
131+
132+
print("\n=== 代理池自动刷新测试完成 ===")
133+
else:
134+
print("当前代理未设置过期时间，跳过自动刷新测试")
135+
136+
async def test_ip_expiration_standalone(self):
137+
"""独立测试IP过期检测功能（不依赖真实代理提供商）"""
138+
print("\n=== 开始独立测试IP代理过期检测功能 ===")
139+
140+
current_ts = int(time.time())
141+
142+
# 1. 测试未设置过期时间的IP（永不过期）
143+
ip_no_expire = IpInfoModel(
144+
ip="192.168.1.1",
145+
port=8080,
146+
user="test_user",
147+
password="test_pwd",
148+
expired_time_ts=None
149+
)
150+
print(f"\n测试1: IP未设置过期时间")
151+
is_expired = ip_no_expire.is_expired(buffer_seconds=30)
152+
print(f" 代理: {ip_no_expire.ip}:{ip_no_expire.port}")
153+
print(f" 过期时间: {ip_no_expire.expired_time_ts}")
154+
print(f" 是否过期: {is_expired}")
155+
self.assertFalse(is_expired, msg="未设置过期时间的IP应该永不过期")
156+
157+
# 2. 测试5分钟后过期的IP（应该未过期）
158+
five_minutes_later = current_ts + 300
159+
ip_valid = IpInfoModel(
160+
ip="192.168.1.2",
161+
port=8080,
162+
user="test_user",
163+
password="test_pwd",
164+
expired_time_ts=five_minutes_later
165+
)
166+
print(f"\n测试2: IP将在5分钟后过期")
167+
is_expired = ip_valid.is_expired(buffer_seconds=30)
168+
print(f" 代理: {ip_valid.ip}:{ip_valid.port}")
169+
print(f" 当前时间戳: {current_ts}")
170+
print(f" 过期时间戳: {ip_valid.expired_time_ts}")
171+
print(f" 剩余时间: {ip_valid.expired_time_ts - current_ts} 秒")
172+
print(f" 是否过期(缓冲30秒): {is_expired}")
173+
self.assertFalse(is_expired, msg="5分钟后过期的IP，缓冲30秒不应该过期")
174+
175+
# 3. 测试已过期的IP
176+
already_expired = current_ts - 60
177+
ip_expired = IpInfoModel(
178+
ip="192.168.1.3",
179+
port=8080,
180+
user="test_user",
181+
password="test_pwd",
182+
expired_time_ts=already_expired
183+
)
184+
print(f"\n测试3: IP已经过期（1分钟前）")
185+
is_expired = ip_expired.is_expired(buffer_seconds=30)
186+
print(f" 代理: {ip_expired.ip}:{ip_expired.port}")
187+
print(f" 当前时间戳: {current_ts}")
188+
print(f" 过期时间戳: {ip_expired.expired_time_ts}")
189+
print(f" 已过期: {current_ts - ip_expired.expired_time_ts} 秒")
190+
print(f" 是否过期(缓冲30秒): {is_expired}")
191+
self.assertTrue(is_expired, msg="已过期的IP应该被检测为过期")
192+
193+
# 4. 测试临界过期（29秒后过期，缓冲30秒应该认为已过期）
194+
almost_expired = current_ts + 29
195+
ip_critical = IpInfoModel(
196+
ip="192.168.1.4",
197+
port=8080,
198+
user="test_user",
199+
password="test_pwd",
200+
expired_time_ts=almost_expired
201+
)
202+
print(f"\n测试4: IP即将过期（29秒后）")
203+
is_expired = ip_critical.is_expired(buffer_seconds=30)
204+
print(f" 代理: {ip_critical.ip}:{ip_critical.port}")
205+
print(f" 当前时间戳: {current_ts}")
206+
print(f" 过期时间戳: {ip_critical.expired_time_ts}")
207+
print(f" 剩余时间: {ip_critical.expired_time_ts - current_ts} 秒")
208+
print(f" 是否过期(缓冲30秒): {is_expired}")
209+
self.assertTrue(is_expired, msg="29秒后过期的IP，缓冲30秒应该被认为已过期")
210+
211+
# 5. 测试31秒后过期（缓冲30秒应该未过期）
212+
just_safe = current_ts + 31
213+
ip_just_safe = IpInfoModel(
214+
ip="192.168.1.5",
215+
port=8080,
216+
user="test_user",
217+
password="test_pwd",
218+
expired_time_ts=just_safe
219+
)
220+
print(f"\n测试5: IP在安全范围内（31秒后过期）")
221+
is_expired = ip_just_safe.is_expired(buffer_seconds=30)
222+
print(f" 代理: {ip_just_safe.ip}:{ip_just_safe.port}")
223+
print(f" 当前时间戳: {current_ts}")
224+
print(f" 过期时间戳: {ip_just_safe.expired_time_ts}")
225+
print(f" 剩余时间: {ip_just_safe.expired_time_ts - current_ts} 秒")
226+
print(f" 是否过期(缓冲30秒): {is_expired}")
227+
self.assertFalse(is_expired, msg="31秒后过期的IP，缓冲30秒应该未过期")
228+
229+
# 6. 测试ProxyIpPool的过期检测
230+
print(f"\n测试6: ProxyIpPool的过期检测功能")
231+
mock_provider = MagicMock()
232+
mock_provider.get_proxy = AsyncMock(return_value=[])
233+
234+
pool = ProxyIpPool(
235+
ip_pool_count=1,
236+
enable_validate_ip=False,
237+
ip_provider=mock_provider
238+
)
239+
240+
# 6.1 测试无当前代理时
241+
is_expired = pool.is_current_proxy_expired(buffer_seconds=30)
242+
print(f" 无当前代理时是否过期: {is_expired}")
243+
self.assertTrue(is_expired, msg="无当前代理时应该返回True")
244+
245+
# 6.2 设置一个有效的代理
246+
valid_proxy = IpInfoModel(
247+
ip="192.168.1.6",
248+
port=8080,
249+
user="test_user",
250+
password="test_pwd",
251+
expired_time_ts=current_ts + 300 # 5分钟后过期
252+
)
253+
pool.current_proxy = valid_proxy
254+
is_expired = pool.is_current_proxy_expired(buffer_seconds=30)
255+
print(f" 设置有效代理后是否过期: {is_expired}")
256+
self.assertFalse(is_expired, msg="有效的代理应该返回False")
257+
258+
# 6.3 设置一个已过期的代理
259+
expired_proxy = IpInfoModel(
260+
ip="192.168.1.7",
261+
port=8080,
262+
user="test_user",
263+
password="test_pwd",
264+
expired_time_ts=current_ts - 60 # 1分钟前已过期
265+
)
266+
pool.current_proxy = expired_proxy
267+
is_expired = pool.is_current_proxy_expired(buffer_seconds=30)
268+
print(f" 设置已过期代理后是否过期: {is_expired}")
269+
self.assertTrue(is_expired, msg="已过期的代理应该返回True")
270+
271+
print("\n=== 独立IP代理过期检测测试完成 ===\n")

0 commit comments

Comments
 (0)