mirror of
https://github.com/wd210010/only_for_happly.git
synced 2026-11-06 04:36:44 +08:00
Create pinzan.py
This commit is contained in:
wd210010 authored and GitHub committed
1 parent
cffaa5e658
commit
76c7b4f302
1 file changed
+334
@@ -0,0 +1,334 @@
|
||||
#!/usr/bin/python3
|
||||
# -- coding: utf-8 --
|
||||
# -------------------------------
|
||||
# @Author : github@wd210010 https://github.com/wd210010/only_for_happly
|
||||
# @Time : 2025/12/24 9:23
|
||||
# -------------------------------
|
||||
# cron "0 0 8 * * *" script-path=xxx.py,tag=匹配cron用
|
||||
# const $ = new Env('品赞签到')
|
||||
# 变量名：Pzandaili ACCOUNT1#PASSWORD1@ACCOUNT2#PASSWORD2 或 & 分隔
|
||||
# 品赞注册链接：https://www.ipzan.com?pid=b8quopmio
|
||||
import warnings
|
||||
import os
|
||||
import random
|
||||
import base64
|
||||
import requests
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Dict, Any
|
||||
from contextlib import contextmanager
|
||||
|
||||
# -------------------------- 全局配置 --------------------------
|
||||
# 基础URL配置
|
||||
CONFIG_URLS = {
|
||||
"home": "https://www.ipzan.com?pid=b1sf3o4ao",
|
||||
"login": "https://service.ipzan.com/users-login",
|
||||
"receive": "https://service.ipzan.com/home/userWallet-receive"
|
||||
}
|
||||
|
||||
# 环境变量配置
|
||||
ENV_CONFIG = {
|
||||
"var_name": "Pzandaili",
|
||||
"separators": ["@", "&"], # 多账户分隔符
|
||||
"fixed_key": "QWERIPZAN1290QWER" # 加密固定密钥
|
||||
}
|
||||
|
||||
# 请求配置
|
||||
REQUEST_CONFIG = {
|
||||
"timeout": 15,
|
||||
"sleep_range": (2, 5), # 多账户延迟范围
|
||||
"retry_times": 1, # 接口重试次数
|
||||
"user_agents": [
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"
|
||||
]
|
||||
}
|
||||
|
||||
# 关闭SSL警告
|
||||
warnings.filterwarnings("ignore")
|
||||
|
||||
# -------------------------- 数据结构定义 --------------------------
|
||||
@dataclass
|
||||
class AccountInfo:
|
||||
"""账户信息数据类"""
|
||||
index: int
|
||||
account: str
|
||||
password: str
|
||||
|
||||
@dataclass
|
||||
class ProcessResult:
|
||||
"""账户处理结果数据类"""
|
||||
account_index: int
|
||||
account_masked: str
|
||||
status: str # success/failed/duplicate/error
|
||||
message: str
|
||||
raw_data: Optional[Any] = None
|
||||
|
||||
# -------------------------- 工具函数 --------------------------
|
||||
def get_random_user_agent() -> str:
|
||||
"""获取随机User-Agent"""
|
||||
return random.choice(REQUEST_CONFIG["user_agents"])
|
||||
|
||||
def get_base_headers() -> Dict[str, str]:
|
||||
"""获取基础请求头"""
|
||||
return {
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
|
||||
"Connection": "keep-alive",
|
||||
"Content-Type": "application/json;charset=UTF-8",
|
||||
"Origin": "https://ipzan.com",
|
||||
"Referer": "https://ipzan.com/",
|
||||
"User-Agent": get_random_user_agent()
|
||||
}
|
||||
|
||||
def mask_account(account: str) -> str:
|
||||
"""账号脱敏显示"""
|
||||
if len(account) > 7:
|
||||
return f"{account[:3]}****{account[-4:]}"
|
||||
return account[:2] + "****" + account[-1:] if len(account) > 3 else "****"
|
||||
|
||||
@contextmanager
|
||||
def request_session():
|
||||
"""请求会话上下文管理器，自动关闭会话"""
|
||||
session = requests.Session()
|
||||
try:
|
||||
session.headers.update(get_base_headers())
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def retry_on_failure(max_retries: int = 1):
|
||||
"""重试装饰器"""
|
||||
def decorator(func):
|
||||
def wrapper(*args, **kwargs):
|
||||
last_exception = None
|
||||
for attempt in range(max_retries + 1):
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except Exception as e:
|
||||
last_exception = e
|
||||
if attempt < max_retries:
|
||||
sleep_time = random.uniform(1, 3)
|
||||
print(f"⚠️ 尝试{attempt+1}失败，{sleep_time:.1f}秒后重试：{str(e)[:50]}")
|
||||
time.sleep(sleep_time)
|
||||
else:
|
||||
raise e
|
||||
raise last_exception
|
||||
return wrapper
|
||||
return decorator
|
||||
|
||||
# -------------------------- 核心加密逻辑 --------------------------
|
||||
def encrypt_account(phone: str, password: str) -> str:
|
||||
"""IPZAN专属加密逻辑（官网JS还原）"""
|
||||
# 基础加密
|
||||
plain_text = f"{phone}{ENV_CONFIG['fixed_key']}{password}"
|
||||
encoded_str = base64.b64encode(plain_text.encode("utf-8")).decode("utf-8")
|
||||
|
||||
# 生成400位混淆字符串
|
||||
random_hex = []
|
||||
for _ in range(80):
|
||||
hex_str = hex(int(random.random() * 10**16))[2:]
|
||||
random_hex.append(hex_str)
|
||||
random_hex = "".join(random_hex).ljust(400, "0")[:400]
|
||||
|
||||
# 分段拼接
|
||||
parts = [
|
||||
random_hex[:100], encoded_str[:8],
|
||||
random_hex[100:200], encoded_str[8:20],
|
||||
random_hex[200:300], encoded_str[20:],
|
||||
random_hex[300:400]
|
||||
]
|
||||
return "".join(parts)
|
||||
|
||||
# -------------------------- 账户加载 --------------------------
|
||||
def load_accounts_from_env() -> List[AccountInfo]:
|
||||
"""加载环境变量中的账户信息，返回结构化的账户列表"""
|
||||
env_value = os.getenv(ENV_CONFIG["var_name"])
|
||||
|
||||
if not env_value:
|
||||
raise ValueError(
|
||||
f"❌ 环境变量 {ENV_CONFIG['var_name']} 未设置！\n"
|
||||
f"📝 配置格式参考：\n"
|
||||
f" - 单账户：账号#密码\n"
|
||||
f" - 多账户：账号1#密码1@账号2#密码2（或&分隔）"
|
||||
)
|
||||
|
||||
# 解析多账户分隔符
|
||||
accounts_str_list = [env_value]
|
||||
for sep in ENV_CONFIG["separators"]:
|
||||
if sep in env_value:
|
||||
accounts_str_list = [s.strip() for s in env_value.split(sep) if s.strip()]
|
||||
break
|
||||
|
||||
# 解析每个账户信息
|
||||
accounts = []
|
||||
for idx, account_str in enumerate(accounts_str_list, 1):
|
||||
parts = account_str.split("#", 2)
|
||||
|
||||
# 格式校验（忽略推送链接部分）
|
||||
if len(parts) < 2:
|
||||
print(f"⚠️ 跳过格式错误账户{idx}：{account_str}（需为 账号#密码 格式）")
|
||||
continue
|
||||
|
||||
acc, pwd = parts[0].strip(), parts[1].strip()
|
||||
|
||||
if acc and pwd:
|
||||
accounts.append(AccountInfo(
|
||||
index=idx,
|
||||
account=acc,
|
||||
password=pwd
|
||||
))
|
||||
|
||||
if not accounts:
|
||||
raise ValueError("❌ 未加载到有效账户！请检查环境变量配置。")
|
||||
|
||||
return accounts
|
||||
|
||||
# -------------------------- 核心业务逻辑 --------------------------
|
||||
@retry_on_failure(max_retries=REQUEST_CONFIG["retry_times"])
|
||||
def login_account(session: requests.Session, account: str, password: str) -> str:
|
||||
"""账户登录，返回Token"""
|
||||
login_headers = get_base_headers()
|
||||
login_headers["Authorization"] = "Bearer null"
|
||||
|
||||
encrypted_account = encrypt_account(account, password)
|
||||
login_data = {"account": encrypted_account, "source": "ipzan-home-one"}
|
||||
|
||||
response = session.post(
|
||||
url=CONFIG_URLS["login"],
|
||||
headers=login_headers,
|
||||
json=login_data,
|
||||
timeout=REQUEST_CONFIG["timeout"]
|
||||
)
|
||||
response.raise_for_status()
|
||||
login_result = response.json()
|
||||
|
||||
if login_result.get("code") != 0:
|
||||
raise ValueError(f"登录失败：{login_result.get('message', '未知错误')}")
|
||||
|
||||
token = login_result.get("data", {}).get("token")
|
||||
if not token:
|
||||
raise ValueError("登录成功但未返回Token")
|
||||
|
||||
return token
|
||||
|
||||
@retry_on_failure(max_retries=REQUEST_CONFIG["retry_times"])
|
||||
def receive_ip(session: requests.Session, token: str) -> Dict[str, Any]:
|
||||
"""领取IP，返回领取结果"""
|
||||
receive_headers = get_base_headers()
|
||||
receive_headers["Authorization"] = f"Bearer {token}"
|
||||
|
||||
response = session.get(
|
||||
url=CONFIG_URLS["receive"],
|
||||
headers=receive_headers,
|
||||
timeout=REQUEST_CONFIG["timeout"]
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def process_single_account(account_info: AccountInfo) -> ProcessResult:
|
||||
"""处理单个账户的完整流程"""
|
||||
account_masked = mask_account(account_info.account)
|
||||
print(f"\n[{account_info.index}] 📱 开始处理账号：{account_masked}")
|
||||
|
||||
try:
|
||||
with request_session() as session:
|
||||
# 1. 登录
|
||||
token = login_account(session, account_info.account, account_info.password)
|
||||
print(" └─ ✅ 登录成功")
|
||||
|
||||
# 2. 领取IP
|
||||
receive_result = receive_ip(session, token)
|
||||
code = receive_result.get("code")
|
||||
msg = receive_result.get("message", "")
|
||||
data = receive_result.get("data", "")
|
||||
|
||||
# 3. 处理领取结果
|
||||
if code == 0:
|
||||
status = "success"
|
||||
message = f"领取成功：{data}"
|
||||
print(f" └─ 🎉 {message}")
|
||||
|
||||
elif code == -1 and "领取过" in str(msg):
|
||||
status = "duplicate"
|
||||
message = f"本周已领：{msg}"
|
||||
print(f" └─ ⚠️ {message}")
|
||||
|
||||
else:
|
||||
status = "failed"
|
||||
message = f"领取失败：{msg}"
|
||||
print(f" └─ ❌ {message}")
|
||||
|
||||
return ProcessResult(
|
||||
account_index=account_info.index,
|
||||
account_masked=account_masked,
|
||||
status=status,
|
||||
message=message,
|
||||
raw_data=receive_result
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"运行异常：{str(e)[:50]}"
|
||||
print(f" └─ ❌ {error_msg}")
|
||||
|
||||
return ProcessResult(
|
||||
account_index=account_info.index,
|
||||
account_masked=account_masked,
|
||||
status="error",
|
||||
message=error_msg
|
||||
)
|
||||
|
||||
# -------------------------- 主程序 --------------------------
|
||||
def main():
|
||||
"""主程序入口"""
|
||||
print("="*60)
|
||||
print("🎯 IPZAN 多账户自动领取工具（每周领IP）- 无推送版")
|
||||
print(f"🔗 注册链接：{CONFIG_URLS['home']}")
|
||||
print("="*60)
|
||||
|
||||
all_summaries = []
|
||||
|
||||
try:
|
||||
# 1. 加载账户
|
||||
accounts = load_accounts_from_env()
|
||||
print(f"\n📊 共加载 {len(accounts)} 个有效账户，开始执行任务...")
|
||||
|
||||
# 2. 处理每个账户
|
||||
for idx, account in enumerate(accounts):
|
||||
result = process_single_account(account)
|
||||
|
||||
# 构建汇总信息
|
||||
status_icon = {
|
||||
"success": "✅",
|
||||
"failed": "❌",
|
||||
"duplicate": "⚠️",
|
||||
"error": "❌"
|
||||
}.get(result.status, "❓")
|
||||
|
||||
summary = f"账号{result.account_index}：{result.account_masked}\n{status_icon} {result.message}"
|
||||
all_summaries.append(summary)
|
||||
|
||||
# 多账户延迟
|
||||
if idx < len(accounts) - 1:
|
||||
sleep_time = random.uniform(*REQUEST_CONFIG["sleep_range"])
|
||||
print(f"\n⏳ 多账户防并发，延迟 {sleep_time:.1f} 秒...")
|
||||
time.sleep(sleep_time)
|
||||
|
||||
except Exception as e:
|
||||
error_summary = f"❌ 程序全局错误：{str(e)[:100]}"
|
||||
print(f"\n{error_summary}")
|
||||
all_summaries.append(error_summary)
|
||||
|
||||
finally:
|
||||
# 输出最终结果
|
||||
print("\n" + "="*60)
|
||||
print("\n📝 执行结果汇总：")
|
||||
for summary in all_summaries:
|
||||
print(f" {summary}")
|
||||
|
||||
print("\n✅ 脚本执行完成！")
|
||||
print("="*60)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user