mirror of
https://github.com/XiaoGe-LiBai/yangmao.git
synced 2025-12-17 03:58:13 +08:00
1000 lines
35 KiB
Python
1000 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
有赞商城自动签到脚本
|
||
|
||
【环境变量】
|
||
export YOUZAN_ACCOUNT=手机号#密码 (必填)
|
||
export CAPTCHA_SERVER=验证码服务器 (可选)
|
||
|
||
【店铺配置】
|
||
在 SHOPS 列表中添加: {"kdt_id": "店铺ID", "checkin_id": "签到ID"}
|
||
店铺名称会自动获取，无需手动填写
|
||
|
||
【定时任务】
|
||
cron: 0 8 * * * (每天早上8点)
|
||
|
||
【功能特性】
|
||
✅ 密码登录 + 自动验证码
|
||
✅ Token 自动缓存（7天有效）
|
||
✅ Token 失效自动重登
|
||
✅ 积分查询 + 商品查询
|
||
✅ 自动获取店铺名称
|
||
|
||
更新: 2025-12-10
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
import re
|
||
from datetime import datetime
|
||
from Crypto.Cipher import AES
|
||
from Crypto.Util.Padding import pad
|
||
import base64
|
||
import hashlib
|
||
|
||
# 修复 Windows 控制台编码问题
|
||
if sys.platform == 'win32':
|
||
import io
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
|
||
|
||
|
||
# ==================== Token 缓存配置 ====================
|
||
# Token 缓存文件路径（存储在脚本同目录下）
|
||
TOKEN_CACHE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".youzan_cache")
|
||
# =====================================================
|
||
|
||
|
||
# ==================== 店铺配置区域 ====================
|
||
# 在这里添加你需要签到的店铺信息
|
||
# 格式：{"kdt_id": "店铺ID", "checkin_id": "签到活动ID"}
|
||
SHOPS = [
|
||
{"kdt_id": "18739377", "checkin_id": "12063"},
|
||
{"kdt_id": "44877243", "checkin_id": "2162835"},
|
||
# {"kdt_id": "41067901", "checkin_id": "99"},
|
||
{"kdt_id": "129380009", "checkin_id": "3520910"},
|
||
# {"kdt_id": "109809208", "checkin_id": "2923467"},
|
||
{"kdt_id": "43958855", "checkin_id": "2910869"},
|
||
{"kdt_id": "42213767", "checkin_id": "2386563"},
|
||
{"kdt_id": "100464643", "checkin_id": "1597464"},
|
||
{"kdt_id": "117130552", "checkin_id": "3347128"},
|
||
{"kdt_id": "107786737", "checkin_id": "2299510"},
|
||
{"kdt_id": "44694253", "checkin_id": "18415"},
|
||
# {"kdt_id": "130177909", "checkin_id": "3549859"},
|
||
{"kdt_id": "93457151", "checkin_id": "1820214"},
|
||
{"kdt_id": "105036832", "checkin_id": "3997371"},
|
||
# {"kdt_id": "121810522", "checkin_id": "4804346"},
|
||
{"kdt_id": "146288343", "checkin_id": "4296415"},
|
||
{"kdt_id": "139827364", "checkin_id": "4806300"},
|
||
{"kdt_id": "16365465", "checkin_id": "13736"},
|
||
{"kdt_id": "77770507", "checkin_id": "9975"},
|
||
{"kdt_id": "43183730", "checkin_id": "24630"},
|
||
{"kdt_id": "96543670", "checkin_id": "3721624"},
|
||
{"kdt_id": "113745713", "checkin_id": "3800805"},
|
||
{"kdt_id": "179778907", "checkin_id": "5428477"},
|
||
{"kdt_id": "77093133", "checkin_id": "2947047"},
|
||
{"kdt_id": "44519665", "checkin_id": "2682116"},
|
||
{"kdt_id": "149536603", "checkin_id": "6287727"},
|
||
]
|
||
# =====================================================
|
||
|
||
|
||
class TokenCache:
|
||
"""Token 缓存管理类"""
|
||
|
||
def __init__(self, cache_dir: str = TOKEN_CACHE_DIR):
|
||
"""
|
||
初始化 Token 缓存管理器
|
||
:param cache_dir: 缓存目录路径
|
||
"""
|
||
self.cache_dir = cache_dir
|
||
# 确保缓存目录存在
|
||
if not os.path.exists(self.cache_dir):
|
||
os.makedirs(self.cache_dir)
|
||
|
||
@staticmethod
|
||
def _get_account_hash(mobile: str) -> str:
|
||
"""
|
||
生成账号的哈希值（用于文件名）
|
||
:param mobile: 手机号
|
||
:return: 哈希值
|
||
"""
|
||
return hashlib.md5(mobile.encode('utf-8')).hexdigest()
|
||
|
||
def get_cache_file(self, mobile: str) -> str:
|
||
"""
|
||
获取缓存文件路径
|
||
:param mobile: 手机号
|
||
:return: 缓存文件路径
|
||
"""
|
||
account_hash = self._get_account_hash(mobile)
|
||
return os.path.join(self.cache_dir, f"token_{account_hash}.json")
|
||
|
||
def save_token(self, mobile: str, token: str) -> bool:
|
||
"""
|
||
保存 Token 到缓存
|
||
:param mobile: 手机号
|
||
:param token: Token
|
||
:return: 是否保存成功
|
||
"""
|
||
try:
|
||
cache_file = self.get_cache_file(mobile)
|
||
cache_data = {
|
||
"mobile": mobile,
|
||
"token": token,
|
||
"timestamp": int(time.time())
|
||
}
|
||
|
||
with open(cache_file, 'w', encoding='utf-8') as f:
|
||
json.dump(cache_data, f, ensure_ascii=False, indent=2)
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f"⚠️ 保存 Token 缓存失败: {e}")
|
||
return False
|
||
|
||
def load_token(self, mobile: str) -> dict:
|
||
"""
|
||
从缓存加载 Token
|
||
:param mobile: 手机号
|
||
:return: Token 信息字典
|
||
"""
|
||
try:
|
||
cache_file = self.get_cache_file(mobile)
|
||
|
||
if not os.path.exists(cache_file):
|
||
return {
|
||
"success": False,
|
||
"message": "缓存文件不存在"
|
||
}
|
||
|
||
with open(cache_file, 'r', encoding='utf-8') as f:
|
||
cache_data = json.load(f)
|
||
|
||
# 检查缓存是否过期（7天）
|
||
cache_time = cache_data.get("timestamp", 0)
|
||
current_time = int(time.time())
|
||
if current_time - cache_time > 7 * 24 * 3600:
|
||
return {
|
||
"success": False,
|
||
"message": "缓存已过期（超过7天）"
|
||
}
|
||
|
||
return {
|
||
"success": True,
|
||
"token": cache_data.get("token"),
|
||
"mobile": cache_data.get("mobile"),
|
||
"cache_time": cache_time
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"读取缓存失败: {str(e)}"
|
||
}
|
||
|
||
def clear_token(self, mobile: str) -> bool:
|
||
"""
|
||
清除指定账号的 Token 缓存
|
||
:param mobile: 手机号
|
||
:return: 是否清除成功
|
||
"""
|
||
try:
|
||
cache_file = self.get_cache_file(mobile)
|
||
if os.path.exists(cache_file):
|
||
os.remove(cache_file)
|
||
return True
|
||
except Exception as e:
|
||
print(f"⚠️ 清除 Token 缓存失败: {e}")
|
||
return False
|
||
|
||
|
||
class YouzanAuth:
|
||
"""有赞账号认证类"""
|
||
|
||
# AES 加密配置（从 JS 脚本中提取）
|
||
AES_KEY = b"youzan.com._key_"
|
||
AES_IV = b"youzan.com.aesiv"
|
||
|
||
# 验证码服务器地址（从环境变量读取，默认使用 JS 脚本中的地址）
|
||
CAPTCHA_SERVER = os.getenv("CAPTCHA_SERVER", "https://captcha.fuckinghigh.eu.org")
|
||
|
||
def __init__(self):
|
||
self.session = requests.Session()
|
||
|
||
@staticmethod
|
||
def calculate_password_level(password: str) -> int:
|
||
"""
|
||
计算密码强度等级
|
||
:param password: 密码
|
||
:return: 密码等级 (-1 到 2)
|
||
"""
|
||
level = -1
|
||
if re.search(r'\d+', password):
|
||
level += 1
|
||
if re.search(r'[a-zA-Z]+', password):
|
||
level += 1
|
||
if re.search(r'[^A-Za-z0-9]+', password):
|
||
level += 1
|
||
return level
|
||
|
||
@staticmethod
|
||
def encrypt_password(password: str) -> str:
|
||
"""
|
||
使用 AES-CBC 加密密码
|
||
:param password: 明文密码
|
||
:return: 加密后的密码（Base64编码）
|
||
"""
|
||
cipher = AES.new(YouzanAuth.AES_KEY, AES.MODE_CBC, YouzanAuth.AES_IV)
|
||
padded_data = pad(password.encode('utf-8'), AES.block_size)
|
||
encrypted = cipher.encrypt(padded_data)
|
||
return base64.b64encode(encrypted).decode('utf-8')
|
||
|
||
def get_captcha_ticket(self, aid_encrypted: str) -> dict:
|
||
"""
|
||
从验证码服务器获取 ticket
|
||
:param aid_encrypted: 加密的 aid
|
||
:return: 验证码 ticket 信息
|
||
"""
|
||
try:
|
||
captcha_url = f"{self.CAPTCHA_SERVER}/captcha?aidEncrypted={aid_encrypted}"
|
||
|
||
print(f" 🔐 正在获取验证码...")
|
||
|
||
response = requests.get(captcha_url, timeout=30)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("ticket"):
|
||
print(f" ✅ 验证码获取成功")
|
||
return {
|
||
"success": True,
|
||
"ticket": result.get("ticket"),
|
||
"randstr": result.get("randstr"),
|
||
"verifyDuration": result.get("verifyDuration", 0),
|
||
"actionDuration": result.get("actionDuration", 0),
|
||
"sid": result.get("sid", "")
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": "验证码服务器未返回 ticket"
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"获取验证码失败: {str(e)}"
|
||
}
|
||
|
||
def login_with_password(self, mobile: str, password: str) -> dict:
|
||
"""
|
||
使用手机号和密码登录（支持验证码）
|
||
:param mobile: 手机号
|
||
:param password: 密码
|
||
:return: 登录结果
|
||
"""
|
||
url = "https://passport.youzan.com/api/login/password.json"
|
||
|
||
encrypted_password = self.encrypt_password(password)
|
||
password_level = self.calculate_password_level(password)
|
||
|
||
payload = {
|
||
"countryCode": "+86",
|
||
"mobile": mobile,
|
||
"password": encrypted_password,
|
||
"passwordLevel": password_level,
|
||
"passwordLength": len(password)
|
||
}
|
||
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 18_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.65(0x1800412e) NetType/WIFI Language/zh_CN',
|
||
'Content-Type': 'application/json',
|
||
'Accept': 'application/json',
|
||
}
|
||
|
||
try:
|
||
# 第一次登录尝试
|
||
response = self.session.post(url, json=payload, headers=headers, timeout=15)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
if result.get("code") == 40000310:
|
||
print(f" 🔐 检测到需要验证码，正在处理...")
|
||
decision_data = result.get("data", {}).get("decisionData", )
|
||
aid_encrypted = decision_data.get("aidEncrypted")
|
||
captcha_app_id = decision_data.get("captchaAppId")
|
||
|
||
if not aid_encrypted:
|
||
return {
|
||
"success": False,
|
||
"message": "无法获取验证码参数",
|
||
"mobile": mobile
|
||
}
|
||
|
||
# 调用验证码服务器
|
||
captcha_result = self.get_captcha_ticket(aid_encrypted)
|
||
|
||
if not captcha_result.get("success"):
|
||
return {
|
||
"success": False,
|
||
"message": captcha_result.get("message", "验证码获取失败"),
|
||
"mobile": mobile
|
||
}
|
||
|
||
print(f" 🔄 正在使用验证码重新登录...")
|
||
|
||
payload_with_captcha = {
|
||
"countryCode": "+86",
|
||
"mobile": mobile,
|
||
"password": encrypted_password,
|
||
"passwordLevel": password_level,
|
||
"passwordLength": len(password),
|
||
"behavior": {
|
||
"decisionCode": "TENCENT_LOW_RISK",
|
||
"verifyResult": {
|
||
"code": 0,
|
||
"data": {
|
||
"appid": captcha_app_id,
|
||
"ret": 0,
|
||
"ticket": captcha_result["ticket"],
|
||
"randstr": captcha_result["randstr"],
|
||
"verifyDuration": captcha_result["verifyDuration"],
|
||
"actionDuration": captcha_result["actionDuration"],
|
||
"sid": captcha_result["sid"]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
response = self.session.post(url, json=payload_with_captcha, headers=headers, timeout=15)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
# 处理登录结果
|
||
if result.get("code") == 0:
|
||
# 登录成功，从 Cookie 中提取 KDTSESSIONID
|
||
cookies = self.session.cookies.get_dict()
|
||
kdtsessionid = cookies.get("KDTSESSIONID", "")
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "登录成功",
|
||
"token": kdtsessionid,
|
||
"mobile": mobile,
|
||
"raw_response": result
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": result.get("msg", "登录失败"),
|
||
"code": result.get("code"),
|
||
"mobile": mobile
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"登录请求失败: {str(e)}",
|
||
"mobile": mobile
|
||
}
|
||
|
||
|
||
class YouzanCheckin:
|
||
"""有赞签到类"""
|
||
|
||
def __init__(self, token: str, session: requests.Session = None):
|
||
"""
|
||
初始化签到器
|
||
:param token: KDTSESSIONID
|
||
:param session: 可选的 requests.Session 对象（用于复用登录后的 session）
|
||
"""
|
||
self.token = token
|
||
self.session = session if session else requests.Session()
|
||
self.results = []
|
||
|
||
def get_shop_info(self, kdt_id: str) -> dict:
|
||
"""
|
||
获取店铺信息（包括店铺名称）
|
||
:param kdt_id: 店铺ID
|
||
:return: 店铺信息
|
||
"""
|
||
try:
|
||
url = f"https://h5.youzan.com/wscshopcore/extension/shop-info.json?kdt_id={kdt_id}"
|
||
|
||
extra_data = {
|
||
"is_weapp": 1,
|
||
"sid": self.token
|
||
}
|
||
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 18_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.65(0x1800412e) NetType/WIFI Language/zh_CN',
|
||
'extra-data': json.dumps(extra_data),
|
||
'Accept': 'application/json',
|
||
}
|
||
|
||
response = self.session.get(url, headers=headers, timeout=10)
|
||
result = response.json()
|
||
|
||
if result.get("code") == 0:
|
||
shop_data = result.get("data", {}).get("shop", {})
|
||
shop_name = shop_data.get("shopName", f"店铺{kdt_id}")
|
||
|
||
return {
|
||
"success": True,
|
||
"shop_name": shop_name
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"shop_name": f"店铺{kdt_id}"
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"shop_name": f"店铺{kdt_id}"
|
||
}
|
||
|
||
def get_points_info(self, kdt_id: str) -> dict:
|
||
"""
|
||
查询积分信息
|
||
:param kdt_id: 店铺ID
|
||
:return: 积分信息
|
||
"""
|
||
try:
|
||
# 构建查询参数
|
||
params = f"kdt_id={kdt_id}"
|
||
|
||
# 查询积分名称
|
||
points_name_url = f"https://h5.youzan.com/wscuser/membercenter/pointsName.json?{params}"
|
||
|
||
# 查询积分统计
|
||
stats_url = f"https://h5.youzan.com/wscuser/membercenter/stats.json?{params}"
|
||
|
||
extra_data = {
|
||
"is_weapp": 1,
|
||
"sid": self.token
|
||
}
|
||
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 18_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.65(0x1800412e) NetType/WIFI Language/zh_CN',
|
||
'extra-data': json.dumps(extra_data),
|
||
'Accept': 'application/json',
|
||
}
|
||
|
||
# 获取积分名称
|
||
name_response = self.session.get(points_name_url, headers=headers, timeout=10)
|
||
name_result = name_response.json()
|
||
|
||
# 获取积分统计
|
||
stats_response = self.session.get(stats_url, headers=headers, timeout=10)
|
||
stats_result = stats_response.json()
|
||
|
||
if name_result.get("code") == 0 and stats_result.get("code") == 0:
|
||
points_name = name_result.get("data", {}).get("pointsName", "积分")
|
||
points_value = stats_result.get("data", {}).get("stats", {}).get("points", 0)
|
||
|
||
return {
|
||
"success": True,
|
||
"points_name": points_name,
|
||
"points_value": points_value
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": "积分查询失败"
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"积分查询异常: {str(e)}"
|
||
}
|
||
|
||
def get_redeemable_goods(self, kdt_id: str) -> dict:
|
||
"""
|
||
查询可兑换商品列表
|
||
:param kdt_id: 店铺ID
|
||
:return: 商品列表
|
||
"""
|
||
try:
|
||
url = "https://h5.youzan.com/wscump/pointstore/listPointGoods.json"
|
||
|
||
params = {
|
||
"page_size": 1000,
|
||
"kdt_id": kdt_id
|
||
}
|
||
|
||
extra_data = {
|
||
"is_weapp": 1,
|
||
"sid": self.token
|
||
}
|
||
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 18_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.65(0x1800412e) NetType/WIFI Language/zh_CN',
|
||
'extra-data': json.dumps(extra_data),
|
||
'Accept': 'application/json',
|
||
}
|
||
|
||
response = self.session.get(url, headers=headers, params=params, timeout=15)
|
||
result = response.json()
|
||
|
||
if result.get("code") == 0:
|
||
# 注意：API 返回的数据结构有两种
|
||
# 1. data 是数组（直接包含商品列表）- 新版 API
|
||
# 2. data 是对象（包含 items 或 list 字段）- 旧版 API
|
||
data = result.get("data", [])
|
||
|
||
# 判断 data 是数组还是对象
|
||
if isinstance(data, list):
|
||
goods_list = data
|
||
else:
|
||
# 尝试从 items 或 list 字段获取
|
||
goods_list = data.get("items", data.get("list", []))
|
||
|
||
formatted_goods = []
|
||
for idx, item in enumerate(goods_list):
|
||
# 兼容多种数据结构
|
||
# 新版 API：直接从 item 获取字段
|
||
# 旧版 API：从嵌套对象中获取字段
|
||
|
||
# 尝试新版 API 字段（影视飓风等店铺）
|
||
name = item.get("goodsName")
|
||
stock = item.get("remainNum")
|
||
points = item.get("pointsPrice")
|
||
cash = item.get("remainPrice", 0)
|
||
|
||
# 如果新版字段不存在，尝试旧版 API 字段
|
||
if name is None:
|
||
# 优先检查优惠券类商品（couponGroupInfoDTO.groupName）
|
||
coupon_info = item.get("couponGroupInfoDTO", {})
|
||
name = coupon_info.get("groupName")
|
||
|
||
# 如果不是优惠券，尝试 generalGoodsInfoDTO.generalGoodsTitle
|
||
if name is None:
|
||
general_goods_info = item.get("generalGoodsInfoDTO", {})
|
||
name = general_goods_info.get("generalGoodsTitle")
|
||
|
||
# 如果还是没有，尝试 generalGoodsTitle 的其他变体
|
||
if name is None:
|
||
name = general_goods_info.get("title")
|
||
|
||
# 最后尝试 pointGoodsInfo.title
|
||
if name is None:
|
||
goods_info = item.get("pointGoodsInfo", {})
|
||
name = goods_info.get("title", "未知商品")
|
||
|
||
if stock is None:
|
||
stock_info = item.get("pointGoodsStockDTO", {})
|
||
stock = stock_info.get("availableStock", 0)
|
||
|
||
if points is None:
|
||
price_info = item.get("pointGoodsExchangePriceDTO", {})
|
||
points = price_info.get("points", 0)
|
||
if cash == 0:
|
||
cash = price_info.get("cash", 0)
|
||
|
||
formatted_goods.append({
|
||
"name": name if name else "未知商品",
|
||
"stock": stock if stock is not None else 0,
|
||
"points": points if points is not None else 0,
|
||
"cash": cash / 100 # 转换为元
|
||
})
|
||
|
||
return {
|
||
"success": True,
|
||
"goods": formatted_goods,
|
||
"total": len(formatted_goods)
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": result.get("msg", "商品查询失败")
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"message": f"商品查询异常: {str(e)}"
|
||
}
|
||
|
||
def checkin(self, shop_info: dict, show_points: bool = True, show_goods: bool = True) -> dict:
|
||
"""
|
||
执行签到
|
||
:param shop_info: 店铺信息字典
|
||
:param show_points: 是否显示积分信息
|
||
:param show_goods: 是否显示可兑换商品
|
||
:return: 签到结果
|
||
"""
|
||
kdt_id = shop_info.get("kdt_id")
|
||
checkin_id = shop_info.get("checkin_id")
|
||
|
||
# 检查配置完整性
|
||
if not kdt_id or not checkin_id or checkin_id == "未知":
|
||
return {
|
||
"success": False,
|
||
"name": shop_info.get("name", "未命名"),
|
||
"kdt_id": kdt_id,
|
||
"message": "店铺配置不完整,请补充 checkin_id"
|
||
}
|
||
|
||
# 如果没有提供店铺名称，自动获取
|
||
name = shop_info.get("name")
|
||
if not name:
|
||
shop_info_result = self.get_shop_info(kdt_id)
|
||
name = shop_info_result.get("shop_name", f"店铺{kdt_id}")
|
||
|
||
url = "https://h5.youzan.com/wscump/checkin/checkinV2.json"
|
||
|
||
params = {
|
||
"checkinId": checkin_id,
|
||
"kdt_id": kdt_id
|
||
}
|
||
|
||
# 构建 extra-data
|
||
extra_data = {
|
||
"is_weapp": 1,
|
||
"sid": self.token
|
||
}
|
||
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 18_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.65(0x1800412e) NetType/WIFI Language/zh_CN',
|
||
'extra-data': json.dumps(extra_data),
|
||
'Accept': 'application/json',
|
||
'Accept-Encoding': 'gzip, deflate, br',
|
||
'Accept-Language': 'zh-CN,zh;q=0.9',
|
||
}
|
||
|
||
try:
|
||
response = self.session.get(url, headers=headers, params=params, timeout=15)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
# 解析响应
|
||
code = result.get("code")
|
||
msg = result.get("msg", "未知状态")
|
||
data = result.get("data", {})
|
||
|
||
checkin_result = {
|
||
"name": name,
|
||
"kdt_id": kdt_id
|
||
}
|
||
|
||
if code == 0:
|
||
# 签到成功
|
||
continuous_days = data.get("continuousDays", 0)
|
||
today_reward = data.get("todayReward", "")
|
||
|
||
checkin_result.update({
|
||
"success": True,
|
||
"message": f"签到成功! 已连续签到 {continuous_days} 天",
|
||
"continuous_days": continuous_days,
|
||
"today_reward": today_reward,
|
||
"raw_response": result
|
||
})
|
||
elif code == -1:
|
||
# Token 失效
|
||
checkin_result.update({
|
||
"success": False,
|
||
"message": f"Token已失效: {msg}",
|
||
"need_refresh_token": True
|
||
})
|
||
else:
|
||
# 其他状态（比如已经签到过）
|
||
checkin_result.update({
|
||
"success": False,
|
||
"message": msg,
|
||
"raw_response": result
|
||
})
|
||
|
||
# 查询积分信息
|
||
if show_points:
|
||
time.sleep(0.5) # 避免请求过快
|
||
points_info = self.get_points_info(kdt_id)
|
||
checkin_result["points_info"] = points_info
|
||
|
||
# 查询可兑换商品
|
||
if show_goods:
|
||
time.sleep(0.5) # 避免请求过快
|
||
goods_info = self.get_redeemable_goods(kdt_id)
|
||
checkin_result["goods_info"] = goods_info
|
||
|
||
return checkin_result
|
||
|
||
except requests.exceptions.Timeout:
|
||
return {
|
||
"success": False,
|
||
"name": name,
|
||
"kdt_id": kdt_id,
|
||
"message": "请求超时,请检查网络连接"
|
||
}
|
||
except requests.exceptions.RequestException as e:
|
||
return {
|
||
"success": False,
|
||
"name": name,
|
||
"kdt_id": kdt_id,
|
||
"message": f"网络请求失败: {str(e)}"
|
||
}
|
||
except json.JSONDecodeError:
|
||
return {
|
||
"success": False,
|
||
"name": name,
|
||
"kdt_id": kdt_id,
|
||
"message": f"响应解析失败,原始响应: {response.text[:200]}"
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"name": name,
|
||
"kdt_id": kdt_id,
|
||
"message": f"未知错误: {str(e)}"
|
||
}
|
||
|
||
def batch_checkin(self, shops: list, show_points: bool = True, show_goods: bool = True) -> list:
|
||
"""
|
||
批量签到
|
||
:param shops: 店铺列表
|
||
:param show_points: 是否显示积分信息
|
||
:param show_goods: 是否显示可兑换商品
|
||
:return: 签到结果列表
|
||
"""
|
||
results = []
|
||
total = len(shops)
|
||
|
||
print(f"\n🎯 开始批量签到，共 {total} 个店铺\n")
|
||
|
||
for index, shop in enumerate(shops, 1):
|
||
kdt_id = shop.get("kdt_id")
|
||
|
||
result = self.checkin(shop, show_points, show_goods)
|
||
results.append(result)
|
||
|
||
checkin_id = shop.get("checkin_id")
|
||
print(f"🏪 店铺：{result['name']} id：checkinId={checkin_id}&kdt_id={kdt_id}")
|
||
|
||
status_emoji = "✅" if result.get("success") else "❌"
|
||
print(f"{status_emoji} 签到结果：{result['message']}")
|
||
|
||
if show_points and "points_info" in result:
|
||
points_info = result["points_info"]
|
||
if points_info.get("success"):
|
||
print(f"💰 拥有{points_info['points_name']}：{points_info['points_value']}")
|
||
|
||
if show_goods and "goods_info" in result:
|
||
goods_info = result["goods_info"]
|
||
if goods_info.get("success"):
|
||
goods_list = goods_info.get("goods", [])
|
||
points_name = result.get("points_info", {}).get("points_name", "积分")
|
||
|
||
if goods_list:
|
||
for goods in goods_list:
|
||
cash_str = f" + ¥{goods['cash']:.2f}" if goods['cash'] > 0 else ""
|
||
stock_emoji = "🟢" if goods['stock'] > 0 else "🔴"
|
||
print(f"{stock_emoji} 商品：{goods['name']} | 库存：{goods['stock']} | 需要{points_name}：{goods['points']}{cash_str}")
|
||
else:
|
||
print("📦 该店铺暂无可兑换商品")
|
||
else:
|
||
print(f"⚠️ 商品查询失败：{goods_info.get('message', '未知错误')}")
|
||
|
||
if index < total:
|
||
time.sleep(2)
|
||
|
||
print()
|
||
|
||
return results
|
||
|
||
|
||
def print_banner():
|
||
"""打印横幅"""
|
||
print("=" * 60)
|
||
print(" " * 15 + "🎉 有赞商城签到通杀")
|
||
print("=" * 60)
|
||
print(f"⏰ 执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print("=" * 60)
|
||
|
||
|
||
def send_notification(results: list, account_count: int = 1):
|
||
"""
|
||
发送通知 (适配青龙面板)
|
||
"""
|
||
try:
|
||
import notify
|
||
|
||
success_count = sum(1 for r in results if r.get("success"))
|
||
fail_count = len(results) - success_count
|
||
|
||
title = "有赞商城签到通知"
|
||
content = f"📊 签到完成\n\n"
|
||
|
||
if account_count > 1:
|
||
content += f"👥 账号数: {account_count}\n"
|
||
|
||
content += f"✅ 成功: {success_count} 个\n"
|
||
content += f"❌ 失败: {fail_count} 个\n\n"
|
||
|
||
# 多账号时按账号分组显示
|
||
if account_count > 1:
|
||
current_mobile = None
|
||
for r in results:
|
||
if r.get("mobile") != current_mobile:
|
||
current_mobile = r.get("mobile")
|
||
mobile_tail = r.get("mobile_tail", "****")
|
||
content += f"\n📱 账号尾号: {mobile_tail}\n"
|
||
|
||
status = "✅" if r.get("success") else "❌"
|
||
content += f"{status} {r['name']}: {r['message']}\n"
|
||
else:
|
||
# 单账号时直接显示
|
||
for r in results:
|
||
status = "✅" if r.get("success") else "❌"
|
||
content += f"{status} {r['name']}: {r['message']}\n"
|
||
|
||
notify.send(title, content)
|
||
except ImportError:
|
||
pass
|
||
except Exception as e:
|
||
print(f"⚠️ 通知发送失败: {e}")
|
||
|
||
|
||
def parse_accounts(account_str: str) -> list:
|
||
"""
|
||
解析账号配置字符串
|
||
:param account_str: 账号配置字符串，支持换行符或&分隔多账号
|
||
:return: 账号列表
|
||
"""
|
||
accounts = []
|
||
if not account_str:
|
||
return accounts
|
||
|
||
# 支持多账号，用换行符或 & 分隔
|
||
account_str = account_str.replace('&', '\n')
|
||
account_parts = account_str.split('\n')
|
||
|
||
for part in account_parts:
|
||
part = part.strip()
|
||
if '#' in part:
|
||
mobile, password = part.split('#', 1)
|
||
accounts.append({
|
||
"mobile": mobile.strip(),
|
||
"password": password.strip()
|
||
})
|
||
|
||
return accounts
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print_banner()
|
||
|
||
# 从环境变量读取账号配置
|
||
account_str = os.getenv("YOUZAN_ACCOUNT", "")
|
||
|
||
# 默认开启积分和商品查询
|
||
show_points = True
|
||
show_goods = True
|
||
|
||
if not account_str:
|
||
print("\n❌ 错误: 未找到账号配置")
|
||
print("\n📝 请在环境变量中配置:")
|
||
print(" 变量名: YOUZAN_ACCOUNT")
|
||
print(" 变量值: 手机号#密码")
|
||
print("\n💡 示例: 13800138000#yourpassword\n")
|
||
return
|
||
|
||
accounts = parse_accounts(account_str)
|
||
if not accounts:
|
||
print("❌ 错误: YOUZAN_ACCOUNT 格式不正确")
|
||
print("📝 正确格式: 手机号#密码 或 手机号1#密码1\\n手机号2#密码2\n")
|
||
return
|
||
|
||
print(f"\n📋 检测到 {len(accounts)} 个账号")
|
||
|
||
valid_shops = [s for s in SHOPS if s.get("checkin_id") and s["checkin_id"] != "未知"]
|
||
if not valid_shops:
|
||
print("\n❌ 错误: 没有可用的店铺配置")
|
||
print("\n📝 请在脚本中配置 SHOPS 列表，并确保填写了正确的 checkin_id\n")
|
||
return
|
||
|
||
all_results = []
|
||
|
||
# 外层循环：遍历所有账号
|
||
for idx, account in enumerate(accounts, 1):
|
||
mobile = account['mobile']
|
||
password = account['password']
|
||
mobile_tail = mobile[-4:] # 获取手机号后4位
|
||
|
||
if len(accounts) > 1:
|
||
print(f"\n{'='*60}")
|
||
print(f"🔄 账号 {idx}/{len(accounts)}: {mobile}")
|
||
print(f"{'='*60}")
|
||
else:
|
||
print(f"\n📱 账号: {mobile}")
|
||
|
||
token_cache = TokenCache()
|
||
token = None
|
||
auth = YouzanAuth()
|
||
|
||
cache_result = token_cache.load_token(mobile)
|
||
|
||
if cache_result.get("success"):
|
||
token = cache_result["token"]
|
||
cache_time = datetime.fromtimestamp(cache_result["cache_time"]).strftime('%Y-%m-%d %H:%M:%S')
|
||
print(f"✅ 使用缓存 Token（{cache_time}）")
|
||
|
||
# 即使有缓存，也要设置 Cookie 到 session
|
||
auth.session.cookies.set("KDTSESSIONID", token)
|
||
print()
|
||
else:
|
||
print(f"🔐 正在登录...\n")
|
||
|
||
login_result = auth.login_with_password(mobile, password)
|
||
|
||
if login_result["success"]:
|
||
token = login_result["token"]
|
||
print(f"✅ {login_result['message']}")
|
||
|
||
if token_cache.save_token(mobile, token):
|
||
print("💾 Token 已缓存\n")
|
||
else:
|
||
print("⚠️ Token 缓存失败\n")
|
||
else:
|
||
print(f"❌ 登录失败: {login_result['message']}\n")
|
||
continue
|
||
|
||
if not token:
|
||
print("\n❌ 错误: 无法获取有效的 Token\n")
|
||
continue
|
||
|
||
# 始终使用 auth.session，确保 Cookie 复用
|
||
checker = YouzanCheckin(token, session=auth.session)
|
||
results = checker.batch_checkin(valid_shops, show_points, show_goods)
|
||
|
||
token_expired = any(r.get("need_refresh_token") for r in results)
|
||
|
||
if token_expired:
|
||
print("\n⚠️ 检测到 Token 已失效，正在重新登录...\n")
|
||
|
||
token_cache.clear_token(mobile)
|
||
|
||
login_result = auth.login_with_password(mobile, password)
|
||
|
||
if login_result["success"]:
|
||
token = login_result["token"]
|
||
print("✅ 重新登录成功")
|
||
|
||
if token_cache.save_token(mobile, token):
|
||
print("💾 Token 已缓存\n")
|
||
|
||
print("🔄 重新签到中...\n")
|
||
checker = YouzanCheckin(token, session=auth.session)
|
||
results = checker.batch_checkin(valid_shops, show_points, show_goods)
|
||
else:
|
||
print(f"❌ 重新登录失败: {login_result['message']}\n")
|
||
|
||
# 为每个结果添加账号信息
|
||
for r in results:
|
||
r["mobile"] = mobile
|
||
r["mobile_tail"] = mobile_tail
|
||
|
||
all_results.extend(results)
|
||
|
||
# 发送通知
|
||
if all_results:
|
||
send_notification(all_results, len(accounts))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\n\n用户中断执行")
|
||
except Exception as e:
|
||
print(f"\n程序执行出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|