danial/kami_itunes_third_api
1
0
Fork 0
mirror of https://git.oceanpay.cc/danial/kami_itunes_third_api.git synced 2025-12-18 22:20:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(proxy): 增强代理服务和错误处理机制

Browse Source 
- 在 ProxyService 类中添加了代理失败标记和重试机制
- 优化了代理测试逻辑，增加了对多个测试 URL 的支持
- 在 AppleClient 类中实现了 SSL 证书验证禁用和重试策略
- 更新了请求头设置，提升了请求的兼容性和稳定性
- 增加了对登录重试次数的限制和错误处理
This commit is contained in:
danial
2025-05-01 19:09:08 +08:00
parent f813d4208a
commit 108aeb8a8a
3 changed files with 183 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
config/config.yml
View File

@@ -28,3 +28,8 @@ proxies:
- http://fga3568:fga3568@219.152.50.19:6588
- http://fga3568:fga3568@125.74.88.251:6588
- http://fga3568:fga3568@36.111.202.63:6588
- http://fga3568:fga3568@219.152.50.20:6588
- http://fga3568:fga3568@219.152.50.21:6588
- http://fga3568:fga3568@219.152.50.22:6588
- http://fga3568:fga3568@219.152.50.23:6588
- http://fga3568:fga3568@219.152.50.24:6588

116
src/integrations/itunes/api.py
View File

@@ -1,9 +1,11 @@
import pickle
import re
import traceback
import urllib3
import requests
from loguru import logger
from urllib3.util import Retry
from requests.adapters import HTTPAdapter
from src.integrations.itunes.models.login import (
ItunesLoginResponse,
@@ -19,15 +21,34 @@ from src.integrations.june.models.login import LoginSignatureModel, ItunesLoginM
from src.integrations.june.models.redeem import AuthenticateModel, ItunesRedeemModel
from src.service.proxy import ProxyService
# 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class AppleClient:
def __init__(self):
self.__session = requests.Session()
self.__session.adapters.update(
{
"DEFAULT_RETRIES": 5,
}
self.__session.verify = False # 禁用 SSL 证书验证
# 设置重试策略
retry_strategy = Retry(
total=3, # 总重试次数
backoff_factor=1, # 重试间隔，增加到1秒
status_forcelist=[429, 500, 502, 503, 504], # 需要重试的HTTP状态码
allowed_methods=["GET", "POST"], # 允许重试的 HTTP 方法
raise_on_status=False, # 不抛出 HTTP 错误状态的异常
raise_on_redirect=False, # 不抛出重定向异常
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.__session.mount("http://", adapter)
self.__session.mount("https://", adapter)
# 设置默认请求头
self.__session.headers.update({
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive'
})
def query_sign_sap_setup(self, signature: LoginSignatureModel, retries=3) -> str:
if retries <= 0:
@@ -68,6 +89,18 @@ class AppleClient:
def login(
self, sign_map: AuthenticateModel, account_info: ItunesAccountInfo, server_id: str = "", retries: int = 5
) -> ItunesLoginResponse:
if retries <= 0:
logger.error("登录重试次数已用完")
return ItunesLoginResponse(
serverId="",
response=ItunesFailLoginPlistData(
status=30,
failureType="MAX_RETRIES_EXCEEDED",
customerMessage="登录重试次数已用完"
),
originLog="登录重试次数已用完"
)
headers = {
"X-Apple-ActionSignature": sign_map.signature,
"X-Apple-Store-Front": "143465-19,17",
@@ -83,26 +116,24 @@ class AppleClient:
"Content-Type": "application/x-apple-plist; Charset=UTF-8",
"Referer": "https://buy.itunes.apple.com/WebObjects/MZFinance.woa/wa/authenticate",
}
# # 生成cookie
cookies = {
"mzf_in": "07281",
"s_vi": "",
"itsMetricsR": "Genre-CN-Mobile Software Applications-29099@@Mobile Software Applications-main@@@@",
"s_vnum_n2_us": "0|1",
}
params = {}
if server_id != "":
url = (
"https://p"
+ server_id
+ "-buy.itunes.apple.com/WebObjects/MZFinance.woa/wa/authenticate"
)
url = f"https://p{server_id}-buy.itunes.apple.com/WebObjects/MZFinance.woa/wa/authenticate"
params = {"Pod": server_id, "PRH": server_id}
else:
url = (
"https://buy.itunes.apple.com/WebObjects/MZFinance.woa/wa/authenticate"
)
url = "https://buy.itunes.apple.com/WebObjects/MZFinance.woa/wa/authenticate"
self.__session.headers["X-Apple-Store-Front"] = "143465-19,12"
try:
response = self.__session.post(
url,
headers=headers,
@@ -113,38 +144,65 @@ class AppleClient:
timeout=30,
proxies=ProxyService().get_wrap_proxy(account_info.account_name),
)
except (requests.exceptions.SSLError, requests.exceptions.ProxyError) as e:
logger.error(f"SSL/代理错误: {str(e)}")
# 代理错误时，清除当前代理并重试
ProxyService().set_expire_strategy()
return self.login(sign_map, account_info, server_id, retries - 1)
except requests.exceptions.RequestException as e:
logger.error(f"请求错误: {str(e)}")
return self.login(sign_map, account_info, server_id, retries - 1)
if response.is_redirect:
redirect_url = response.headers["Location"]
groups = re.search(r"https://p(\d+)-buy.itunes.apple.com", redirect_url)
if groups:
server_id = groups.group(1)
return self.login(sign_map, account_info, server_id, retries - 1)
else:
logger.error(f"无法解析重定向URL: {redirect_url}")
return ItunesLoginResponse(
serverId="",
response=ItunesFailLoginPlistData(
status=30,
failureType="INVALID_REDIRECT",
customerMessage="无法解析重定向URL"
),
originLog=response.text
)
try:
response_dict_data = parse_xml(response.text)
except Exception as e:
logger.error(f"解析响应XML失败: {str(e)}")
return ItunesLoginResponse(
serverId=server_id,
response=ItunesFailLoginPlistData(
status=30,
failureType="PARSE_ERROR",
customerMessage="解析响应失败"
),
originLog=response.text
)
if "failureType" in response_dict_data:
status = 31
# 账户被禁用
if (
response_dict_data.get("metrics", {}).get("dialogId")
== "MZFinance.AccountDisabled"
):
if response_dict_data.get("metrics", {}).get("dialogId") == "MZFinance.AccountDisabled":
status = 14
# 账户被锁定
if (
response_dict_data.get("metrics", {}).get("dialogId")
== "MZFinance.DisabledAndFraudLocked"
):
if response_dict_data.get("metrics", {}).get("dialogId") == "MZFinance.DisabledAndFraudLocked":
status = 14
# 密码错误
if response_dict_data.get("failureType") == "-5000":
status = 13
if status == 31:
logger.warning("登录状态未知：", response_dict_data)
response_model = ItunesFailLoginPlistData(
**{"status": status, **response_dict_data}
)
response_model = ItunesFailLoginPlistData(**{"status": status, **response_dict_data})
else:
response_model = ItunesSuccessLoginPlistData(**response_dict_data)
return ItunesLoginResponse(
serverId=server_id, response=response_model, originLog=response.text
serverId=server_id,
response=response_model,
originLog=response.text
)
def redeem(

101
src/service/proxy.py
View File

@@ -1,19 +1,27 @@
import copy
import time
import requests
from loguru import logger
from src.initialization import setting
import random
import requests
import urllib3
from loguru import logger
from src.initialization import setting
# 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 账户列表
account_list = {}
class ProxyService:
def __init__(self) -> None:
self.proxy_list = setting.proxies.address
self.proxy_timeout = 5 # 代理超时时间
self.test_urls = [
"https://www.baidu.com",
"https://www.apple.com",
"https://buy.itunes.apple.com"
]
self.failed_proxies = {} # 记录失败的代理
def get_wrap_proxy(self, account_name: str):
proxy = self.get_proxy(account_name)
@@ -36,31 +44,78 @@ class ProxyService:
def set_proxy(self, account_name: str):
proxy = ""
if self.proxy_list:
for _ in range(len(self.proxy_list)):
proxy = random.choice(self.proxy_list)
if self.test_baidu(proxy):
# 随机打乱代理列表顺序
available_proxies = [p for p in self.proxy_list if not self.is_proxy_failed(p)]
if not available_proxies:
# 如果所有代理都失败，重置失败记录
self.failed_proxies.clear()
available_proxies = self.proxy_list
random.shuffle(available_proxies)
for proxy in available_proxies:
if self.test_proxy(proxy):
account_list[account_name] = {
"address": proxy,
"timeout": time.time() + 60 * 60 * 24,
"timeout": time.time() + 60 * 30, # 30分钟过期
}
return proxy
return proxy
# 设置触发账号过期策略
def set_expire_strategy(self):
current_time = time.time()
for key, value in copy.deepcopy(account_list).items():
if value.get("timeout") > time.time():
if value.get("timeout", 0) < current_time:
account_list.pop(key)
# 测试下百度是否可以联通
def test_baidu(self, proxy: str):
try:
return (
requests.get(
"https://www.baidu.com", proxies=self.warp_proxy(proxy), timeout=5
).status_code == 200
)
except Exception as e:
logger.info(f"[+] 获取签到证书耗时: 测试proxy失败，当前proxy：", self.warp_proxy(proxy))
def is_proxy_failed(self, proxy: str) -> bool:
"""检查代理是否在失败列表中且未过期"""
if proxy in self.failed_proxies:
if time.time() - self.failed_proxies[proxy] < 300: # 5分钟内不重用失败的代理
return True
else:
del self.failed_proxies[proxy]
return False
def mark_proxy_failed(self, proxy: str):
"""标记代理为失败"""
self.failed_proxies[proxy] = time.time()
def test_proxy(self, proxy: str) -> bool:
"""测试代理是否可用
Args:
proxy: 代理地址
Returns:
bool: 代理是否可用
"""
proxies = self.warp_proxy(proxy)
if not proxies:
return False
for url in self.test_urls:
try:
response = requests.get(
url,
proxies=proxies,
timeout=self.proxy_timeout,
verify=False, # 禁用 SSL 证书验证
headers={
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive'
}
)
if response.status_code == 200:
return True
except (requests.exceptions.SSLError,
requests.exceptions.ProxyError,
requests.exceptions.ConnectionError,
requests.exceptions.Timeout) as e:
logger.debug(f"代理测试失败: {proxy}, URL: {url}, 错误: {str(e)}")
continue
# 如果所有URL都测试失败，标记代理为失败
self.mark_proxy_failed(proxy)
return False