timxx/Yangmao_Script
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新 Points_Based/Proxy_iPzan.py

Browse Source
This commit is contained in:
timxx
2026-08-01 03:31:30 +08:00
parent e7a04ae104
commit 2df39dbd3e
1 changed files with 289 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Points_Based/PZdaili_interface.py → Points_Based/Proxy_iPzan.py
+289 -289
View File
@@ -1,290 +1,290 @@
# cron: 16 6 * * * # cron: 16 6 * * *
# new Env("ipzan代理白名单管理") # new Env("ipzan代理白名单管理")
import requests import requests
import os import os
import sys import sys
import time import time
from datetime import datetime from datetime import datetime
import warnings import warnings
from urllib3.exceptions import InsecureRequestWarning from urllib3.exceptions import InsecureRequestWarning
from Crypto.Cipher import AES from Crypto.Cipher import AES
from Crypto.Util.Padding import pad from Crypto.Util.Padding import pad
import json import json
# 禁用SSL警告 # 禁用SSL警告
warnings.filterwarnings("ignore", category=InsecureRequestWarning) warnings.filterwarnings("ignore", category=InsecureRequestWarning)
# ================= 配置区 ================= # ================= 配置区 =================
# 从单个环境变量中读取所有配置 # 从单个环境变量中读取所有配置
CONFIG_STR = os.getenv('API_KEY_PZ') CONFIG_STR = os.getenv('API_KEY_PZ')
if not CONFIG_STR: if not CONFIG_STR:
print("错误：未找到API_KEY_PZ环境变量") print("错误：未找到API_KEY_PZ环境变量")
sys.exit(1) sys.exit(1)
# 解析环境变量：USER_ID#LOGIN_PASSWORD#KEY#ORDER_NO#EXTRACT_KEY # 解析环境变量：USER_ID#LOGIN_PASSWORD#KEY#ORDER_NO#EXTRACT_KEY
try: try:
parts = CONFIG_STR.split('#') parts = CONFIG_STR.split('#')
if len(parts) < 5: if len(parts) < 5:
raise ValueError("环境变量格式不正确") raise ValueError("环境变量格式不正确")
USER_ID = parts[0] USER_ID = parts[0]
LOGIN_PASSWORD = parts[1] LOGIN_PASSWORD = parts[1]
API_KEY = parts[2] API_KEY = parts[2]
ORDER_NO = parts[3] ORDER_NO = parts[3]
EXTRACT_KEY = parts[4] EXTRACT_KEY = parts[4]
# 打印配置信息（调试用） # 打印配置信息（调试用）
print(f"USER_ID: {USER_ID}") print(f"USER_ID: {USER_ID}")
print(f"LOGIN_PASSWORD: {LOGIN_PASSWORD[:2]}****{LOGIN_PASSWORD[-2:]}") print(f"LOGIN_PASSWORD: {LOGIN_PASSWORD[:2]}****{LOGIN_PASSWORD[-2:]}")
print(f"API_KEY: {API_KEY[:2]}****{API_KEY[-2:]}") print(f"API_KEY: {API_KEY[:2]}****{API_KEY[-2:]}")
print(f"ORDER_NO: {ORDER_NO}") print(f"ORDER_NO: {ORDER_NO}")
print(f"EXTRACT_KEY: {EXTRACT_KEY[:2]}****{EXTRACT_KEY[-2:]}") print(f"EXTRACT_KEY: {EXTRACT_KEY[:2]}****{EXTRACT_KEY[-2:]}")
except Exception as e: except Exception as e:
print(f"解析环境变量失败：{str(e)}") print(f"解析环境变量失败：{str(e)}")
print("请确保环境变量格式为：USER_ID#LOGIN_PASSWORD#KEY#ORDER_NO#EXTRACT_KEY") print("请确保环境变量格式为：USER_ID#LOGIN_PASSWORD#KEY#ORDER_NO#EXTRACT_KEY")
sys.exit(1) sys.exit(1)
BASE_URL = "https://service.ipzan.com" BASE_URL = "https://service.ipzan.com"
IP_SERVICES = [ IP_SERVICES = [
{'url': 'http://ifconfig.me/ip', 'verify': False}, {'url': 'http://ifconfig.me/ip', 'verify': False},
{'url': 'http://ip.3322.net', 'verify': False}, {'url': 'http://ip.3322.net', 'verify': False},
{'url': 'http://myip.ipip.net', 'verify': False} {'url': 'http://myip.ipip.net', 'verify': False}
] ]
MAX_RETRIES = 3 MAX_RETRIES = 3
REQUEST_DELAY = 2 REQUEST_DELAY = 2
# ========================================== # ==========================================
def debug_log(msg, level="INFO"): def debug_log(msg, level="INFO"):
"""分级日志系统""" """分级日志系统"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
color_code = { color_code = {
"DEBUG": "\033[94m", # 蓝色 "DEBUG": "\033[94m", # 蓝色
"INFO": "\033[92m", # 绿色 "INFO": "\033[92m", # 绿色
"WARNING": "\033[93m", # 黄色 "WARNING": "\033[93m", # 黄色
"ERROR": "\033[91m", # 红色 "ERROR": "\033[91m", # 红色
"SUCCESS": "\033[92m", # 绿色（成功） "SUCCESS": "\033[92m", # 绿色（成功）
"SYSTEM": "\033[95m" # 紫色 "SYSTEM": "\033[95m" # 紫色
}.get(level, "\033[0m") }.get(level, "\033[0m")
print(f"{color_code}## [{timestamp}] [{level}] {msg}\033[0m") print(f"{color_code}## [{timestamp}] [{level}] {msg}\033[0m")
def get_public_ip(): def get_public_ip():
"""获取公网IP""" """获取公网IP"""
debug_log("正在获取公网IP...", "DEBUG") debug_log("正在获取公网IP...", "DEBUG")
for service in IP_SERVICES: for service in IP_SERVICES:
try: try:
response = requests.get(service['url'], timeout=5, verify=service['verify']) response = requests.get(service['url'], timeout=5, verify=service['verify'])
if response.status_code == 200: if response.status_code == 200:
ip = response.text.strip() ip = response.text.strip()
if validate_ip(ip): if validate_ip(ip):
debug_log(f"验证通过的公网IP：{ip}", "SUCCESS") debug_log(f"验证通过的公网IP：{ip}", "SUCCESS")
return ip return ip
except Exception as e: except Exception as e:
debug_log(f"{service['url']}查询失败：{str(e)}", "DEBUG") debug_log(f"{service['url']}查询失败：{str(e)}", "DEBUG")
debug_log("所有IP服务不可用", "ERROR") debug_log("所有IP服务不可用", "ERROR")
return None return None
def validate_ip(ip): def validate_ip(ip):
"""IP地址格式验证""" """IP地址格式验证"""
parts = ip.split('.') parts = ip.split('.')
return len(parts) == 4 and all(part.isdigit() and 0 <= int(part) < 256 for part in parts) return len(parts) == 4 and all(part.isdigit() and 0 <= int(part) < 256 for part in parts)
def generate_aes_signature(): def generate_aes_signature():
"""生成AES加密签名""" """生成AES加密签名"""
try: try:
# 获取当前时间戳（秒） # 获取当前时间戳（秒）
timestamp = str(int(time.time())) timestamp = str(int(time.time()))
# 构造签名内容：登录密码:套餐提取密匙:时间戳 # 构造签名内容：登录密码:套餐提取密匙:时间戳
data = f"{LOGIN_PASSWORD}:{EXTRACT_KEY}:{timestamp}" data = f"{LOGIN_PASSWORD}:{EXTRACT_KEY}:{timestamp}"
# 将API_KEY转换为16字节密钥 # 将API_KEY转换为16字节密钥
key = API_KEY.encode('utf-8') key = API_KEY.encode('utf-8')
# 确保密钥长度为16字节（AES-128） # 确保密钥长度为16字节（AES-128）
key = key.ljust(16, b'\0')[:16] key = key.ljust(16, b'\0')[:16]
# 创建AES加密器（ECB模式） # 创建AES加密器（ECB模式）
cipher = AES.new(key, AES.MODE_ECB) cipher = AES.new(key, AES.MODE_ECB)
# 对数据进行PKCS7填充 # 对数据进行PKCS7填充
padded_data = pad(data.encode('utf-8'), AES.block_size) padded_data = pad(data.encode('utf-8'), AES.block_size)
# 加密数据 # 加密数据
encrypted_data = cipher.encrypt(padded_data) encrypted_data = cipher.encrypt(padded_data)
# 返回十六进制字符串 # 返回十六进制字符串
return encrypted_data.hex() return encrypted_data.hex()
except Exception as e: except Exception as e:
debug_log(f"AES签名生成失败：{str(e)}", "ERROR") debug_log(f"AES签名生成失败：{str(e)}", "ERROR")
return None return None
def add_ip(ip): def add_ip(ip):
"""添加IP到白名单（使用AES签名）""" """添加IP到白名单（使用AES签名）"""
url = f"{BASE_URL}/whiteList-add" url = f"{BASE_URL}/whiteList-add"
# 生成AES签名 # 生成AES签名
sign = generate_aes_signature() sign = generate_aes_signature()
if not sign: if not sign:
debug_log("无法生成AES签名", "ERROR") debug_log("无法生成AES签名", "ERROR")
return False return False
params = { params = {
'no': ORDER_NO, 'no': ORDER_NO,
'sign': sign, 'sign': sign,
'ip': ip, 'ip': ip,
'replace': 1 # 启用自动替换功能 'replace': 1 # 启用自动替换功能
} }
debug_log(f"正在添加IP：{ip}", "INFO") debug_log(f"正在添加IP：{ip}", "INFO")
try: try:
response = requests.get(url, params=params, timeout=15) response = requests.get(url, params=params, timeout=15)
debug_log(f"添加接口原始响应：{response.text}", "DEBUG") debug_log(f"添加接口原始响应：{response.text}", "DEBUG")
# 尝试解析JSON响应 # 尝试解析JSON响应
try: try:
data = response.json() data = response.json()
except ValueError: except ValueError:
debug_log("响应不是有效的JSON格式", "ERROR") debug_log("响应不是有效的JSON格式", "ERROR")
return False return False
if data.get('code') == 0: if data.get('code') == 0:
debug_log(f"IP {ip} 添加成功", "SUCCESS") debug_log(f"IP {ip} 添加成功", "SUCCESS")
return True return True
error_msg = data.get('message') or '未知错误' error_msg = data.get('message') or '未知错误'
debug_log(f"添加失败：{error_msg}（状态码：{data.get('code')}）", "ERROR") debug_log(f"添加失败：{error_msg}（状态码：{data.get('code')}）", "ERROR")
except Exception as e: except Exception as e:
debug_log(f"请求添加接口失败：{str(e)}", "ERROR") debug_log(f"请求添加接口失败：{str(e)}", "ERROR")
return False return False
def verify_ip(ip): def verify_ip(ip):
"""校验IP是否在白名单中""" """校验IP是否在白名单中"""
url = f"{BASE_URL}/whiteList-verify" url = f"{BASE_URL}/whiteList-verify"
params = { params = {
'no': ORDER_NO, 'no': ORDER_NO,
'ip': ip 'ip': ip
} }
try: try:
response = requests.get(url, params=params, timeout=10) response = requests.get(url, params=params, timeout=10)
debug_log(f"校验接口原始响应：{response.text}", "DEBUG") debug_log(f"校验接口原始响应：{response.text}", "DEBUG")
# 尝试解析JSON响应 # 尝试解析JSON响应
try: try:
data = response.json() data = response.json()
except ValueError: except ValueError:
debug_log("响应不是有效的JSON格式", "ERROR") debug_log("响应不是有效的JSON格式", "ERROR")
return None return None
if data.get('code') == 0: if data.get('code') == 0:
# 根据实际接口返回结构调整 # 根据实际接口返回结构调整
if data.get('data') is True: if data.get('data') is True:
debug_log(f"IP {ip} 已在白名单中", "SUCCESS") debug_log(f"IP {ip} 已在白名单中", "SUCCESS")
return True return True
else: else:
debug_log(f"IP {ip} 不在白名单中", "INFO") debug_log(f"IP {ip} 不在白名单中", "INFO")
return False return False
error_msg = data.get('message') or '未知错误' error_msg = data.get('message') or '未知错误'
debug_log(f"校验失败：{error_msg}（状态码：{data.get('code')}）", "ERROR") debug_log(f"校验失败：{error_msg}（状态码：{data.get('code')}）", "ERROR")
except Exception as e: except Exception as e:
debug_log(f"请求校验接口失败：{str(e)}", "ERROR") debug_log(f"请求校验接口失败：{str(e)}", "ERROR")
return None return None
def get_balance(): def get_balance():
"""查询套餐余额和余量""" """查询套餐余额和余量"""
url = f"{BASE_URL}/userProduct-get" url = f"{BASE_URL}/userProduct-get"
params = { params = {
'no': ORDER_NO, 'no': ORDER_NO,
'userId': USER_ID # 保持参数名不变 'userId': USER_ID # 保持参数名不变
} }
try: try:
# 打印完整的请求URL以便调试 # 打印完整的请求URL以便调试
full_url = f"{url}?no={ORDER_NO}&userId={USER_ID}" full_url = f"{url}?no={ORDER_NO}&userId={USER_ID}"
debug_log(f"余额查询请求URL: {full_url}", "DEBUG") debug_log(f"余额查询请求URL: {full_url}", "DEBUG")
response = requests.get(url, params=params, timeout=10) response = requests.get(url, params=params, timeout=10)
debug_log(f"余额查询接口原始响应：{response.text}", "DEBUG") debug_log(f"余额查询接口原始响应：{response.text}", "DEBUG")
# 尝试解析JSON响应 # 尝试解析JSON响应
try: try:
data = response.json() data = response.json()
except ValueError: except ValueError:
debug_log("响应不是有效的JSON格式", "ERROR") debug_log("响应不是有效的JSON格式", "ERROR")
return None return None
if data.get('code') == 0: if data.get('code') == 0:
balance_info = data.get('data', {}) balance_info = data.get('data', {})
debug_log("余额查询成功", "SUCCESS") debug_log("余额查询成功", "SUCCESS")
return balance_info return balance_info
error_msg = data.get('message') or '未知错误' error_msg = data.get('message') or '未知错误'
debug_log(f"余额查询失败：{error_msg}（状态码：{data.get('code')}）", "ERROR") debug_log(f"余额查询失败：{error_msg}（状态码：{data.get('code')}）", "ERROR")
except Exception as e: except Exception as e:
debug_log(f"请求余额查询接口失败：{str(e)}", "ERROR") debug_log(f"请求余额查询接口失败：{str(e)}", "ERROR")
return None return None
def main(): def main():
debug_log("===== ipzan代理白名单管理 =====", "SYSTEM") debug_log("===== ipzan代理白名单管理 =====", "SYSTEM")
# 第一步：查询套餐余额（可选步骤） # 第一步：查询套餐余额（可选步骤）
debug_log("查询套餐余额...", "INFO") debug_log("查询套餐余额...", "INFO")
try: try:
balance_info = get_balance() balance_info = get_balance()
if balance_info: if balance_info:
# 格式化显示余额信息 # 格式化显示余额信息
balance_str = [] balance_str = []
if 'balance' in balance_info: if 'balance' in balance_info:
balance_str.append(f"钱包余额: {balance_info['balance']}") balance_str.append(f"钱包余额: {balance_info['balance']}")
if 'limitDaySurplus' in balance_info: if 'limitDaySurplus' in balance_info:
balance_str.append(f"今日余量: {balance_info['limitDaySurplus']}") balance_str.append(f"今日余量: {balance_info['limitDaySurplus']}")
if 'totalSurplus' in balance_info: if 'totalSurplus' in balance_info:
balance_str.append(f"套餐总量: {balance_info['totalSurplus']}") balance_str.append(f"套餐总量: {balance_info['totalSurplus']}")
if balance_str: if balance_str:
debug_log(" | ".join(balance_str), "INFO") debug_log(" | ".join(balance_str), "INFO")
else: else:
debug_log("余额查询失败，跳过余额显示", "WARNING") debug_log("余额查询失败，跳过余额显示", "WARNING")
except Exception as e: except Exception as e:
debug_log(f"余额查询过程中出错：{str(e)}", "ERROR") debug_log(f"余额查询过程中出错：{str(e)}", "ERROR")
# 第二步：获取当前公网IP # 第二步：获取当前公网IP
public_ip = get_public_ip() public_ip = get_public_ip()
if not public_ip: if not public_ip:
debug_log("无法获取有效公网IP", "FATAL") debug_log("无法获取有效公网IP", "FATAL")
sys.exit(1) sys.exit(1)
# 第三步：检查当前IP是否已在白名单 # 第三步：检查当前IP是否已在白名单
debug_log("检查当前IP状态...", "INFO") debug_log("检查当前IP状态...", "INFO")
ip_status = verify_ip(public_ip) ip_status = verify_ip(public_ip)
if ip_status is None: if ip_status is None:
debug_log("IP校验失败，请检查网络或服务状态", "FATAL") debug_log("IP校验失败，请检查网络或服务状态", "FATAL")
sys.exit(1) sys.exit(1)
if ip_status: if ip_status:
debug_log("当前IP已在白名单中，无需操作", "SUCCESS") debug_log("当前IP已在白名单中，无需操作", "SUCCESS")
sys.exit(0) sys.exit(0)
debug_log("当前IP不在白名单中，开始添加流程...", "WARNING") debug_log("当前IP不在白名单中，开始添加流程...", "WARNING")
# 第四步：添加当前公网IP（启用自动替换功能） # 第四步：添加当前公网IP（启用自动替换功能）
if not add_ip(public_ip): if not add_ip(public_ip):
debug_log("新IP添加失败", "FATAL") debug_log("新IP添加失败", "FATAL")
sys.exit(1) sys.exit(1)
# 第五步：最终验证 # 第五步：最终验证
debug_log("等待接口数据刷新...", "SYSTEM") debug_log("等待接口数据刷新...", "SYSTEM")
time.sleep(5) time.sleep(5)
final_status = verify_ip(public_ip) final_status = verify_ip(public_ip)
if final_status is True: if final_status is True:
debug_log(f"最终验证通过：{public_ip} 已生效", "SUCCESS") debug_log(f"最终验证通过：{public_ip} 已生效", "SUCCESS")
else: else:
debug_log(f"最终验证失败：请手动检查白名单状态", "FATAL") debug_log(f"最终验证失败：请手动检查白名单状态", "FATAL")
sys.exit(1) sys.exit(1)
debug_log("===== 操作成功完成 =====", "SYSTEM") debug_log("===== 操作成功完成 =====", "SYSTEM")
if __name__ == '__main__': if __name__ == '__main__':
main() main()