一.青龙面板配置环境变量
1.配置通知渠道，企业微信应用或者飞书
2.名称为xmqb,值为抓包的两个值，中间用#号隔开
二.添加脚本
import os
import time
import requests
import urllib3
import json
from datetime import datetime
from typing import Optional, Dict, Any, Union
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class Notifier:
@staticmethod
def send(title: str, content: str):
"""通过青龙面板通知设置发送消息"""
notifier_type = os.environ.get("NOTIFY_TYPE", "").lower()
if not notifier_type:
print("未配置通知环境变量，无法发送通知")
return
try:
if "feishu" in notifier_type: # 飞书机器人通知
Notifier.feishu_notify(title, content)
elif "qywx" in notifier_type: # 企业微信应用通知
Notifier.qywx_notify(title, content)
else:
print(f"不支持的通知类型: {notifier_type}")
except Exception as e:
print(f"发送通知失败: {e}")
@staticmethod
def feishu_notify(title: str, content: str):
"""飞书机器人通知方式"""
fskey = os.environ.get("FSKEY")
if not fskey:
print("未配置FSKEY环境变量")
return
# 构建飞书Webhook URL
webhook_url = f"https://open.feishu.cn/open-apis/bot/v2/hook/{fskey}"
# 构建消息体（支持富文本格式）
payload = {
"msg_type": "post",
"content": {
"post": {
"zh_cn": {
"title": title,
"content": [
[{"tag": "text", "text": content}]
]
}
}
}
}
try:
response = requests.post(
webhook_url,
headers={"Content-Type": "application/json"},
data=json.dumps(payload),
timeout=10,
verify=False
)
if response.status_code != 200:
print(f"飞书通知发送失败: HTTP {response.status_code}")
else:
resp_data = response.json()
if resp_data.get("code") != 0:
print(f"飞书API返回错误: {resp_data.get('msg')}")
else:
print("✅ 飞书通知发送成功")
except Exception as e:
print(f"飞书通知异常: {str(e)}")
@staticmethod
def qywx_notify(title: str, content: str):
"""企业微信应用通知"""
qywx_am = os.environ.get("QYWX_AM")
if not qywx_am:
print("未配置QYWX_AM环境变量")
return
# 解析QYWX_AM格式: corpid,corpsecret,touser,agentid,素材类型
parts = qywx_am.split(',')
if len(parts) < 5:
print("QYWX_AM格式错误，应为: corpid,corpsecret,touser,agentid,素材类型")
return
corpid, corpsecret, touser, agentid, msg_type = parts[:5]
# 获取access_token
token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}"
try:
token_resp = requests.get(token_url, verify=False).json()
if token_resp.get('errcode') != 0:
print(f"获取企业微信access_token失败: {token_resp.get('errmsg')}")
return
access_token = token_resp.get('access_token')
except Exception as e:
print(f"获取企业微信access_token异常: {str(e)}")
return
# 根据素材类型构建消息体
send_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}"
payload = {
"touser": touser,
"agentid": agentid,
"msgtype": "text" if msg_type == "1" else "mpnews",
}
if msg_type == "1": # 文本消息
payload["text"] = {"content": f"{title}
{content}"}
else: # 图文消息(默认)
payload["mpnews"] = {
"articles": [
{
"title": title,
"thumb_media_id": "0", # 使用默认素材
"author": "系统通知",
"content": content.replace('
', '<br>'),
"digest": content[:100] # 摘要
}
]
}
# 图文消息需要设置safe参数
payload["safe"] = 0
try:
response = requests.post(
send_url,
json=payload,
verify=False
)
resp_data = response.json()
if resp_data.get('errcode') == 0:
print("✅ 企业微信通知发送成功")
else:
print(f"企业微信通知发送失败: {resp_data.get('errmsg')}")
except Exception as e:
print(f"企业微信通知异常: {str(e)}")
class RnlRequest:
def __init__(self, cookies: Union[str, dict]):
self.session = requests.Session()
self._base_headers = {
'Host': 'm.jr.airstarfinance.net',
'User-Agent': 'Mozilla/5.0 (Linux; U; Android 14; zh-CN; M2012K11AC Build/UKQ1.230804.001; AppBundle/com.mipay.wallet; AppVersionName/6.89.1.5275.2323; AppVersionCode/20577595; MiuiVersion/stable-V816.0.13.0.UMNCNXM; DeviceId/alioth; NetworkType/WIFI; mix_version; WebViewVersion/118.0.0.0) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Mobile Safari/537.36 XiaoMi/MiuiBrowser/4.3',
}
self.update_cookies(cookies)
def request(
self,
method: str,
url: str,
params: Optional[Dict[str, Any]] = None,
data: Optional[Union[Dict[str, Any], str, bytes]] = None,
json: Optional[Dict[str, Any]] = None,
**kwargs
) -> Optional[Dict[str, Any]]:
headers = {**self._base_headers, **kwargs.pop('headers', {})}
try:
resp = self.session.request(
verify=False,
method=method.upper(),
url=url,
params=params,
data=data,
json=json,
headers=headers,
**kwargs
)
resp.raise_for_status()
return resp.json()
except requests.RequestException as e:
print(f"[Request Error] {e}")
except ValueError as e:
print(f"[JSON Parse Error] {e}")
return None
def update_cookies(self, cookies: Union[str, dict]) -> None:
if cookies:
if isinstance(cookies, str):
dict_cookies = self._parse_cookies(cookies)
else:
dict_cookies = cookies
self.session.cookies.update(dict_cookies)
self._base_headers['Cookie'] = self.dict_cookie_to_string(dict_cookies)
@staticmethod
def _parse_cookies(cookies_str: str) -> Dict[str, str]:
return dict(
item.strip().split('=', 1)
for item in cookies_str.split(';')
if '=' in item
)
@staticmethod
def dict_cookie_to_string(cookie_dict):
cookie_list = []
for key, value in cookie_dict.items():
cookie_list.append(f"{key}={value}")
return "; ".join(cookie_list)
def get(self, url: str, params: Optional[Dict[str, Any]] = None, **kwargs) -> Optional[Dict[str, Any]]:
return self.request('GET', url, params=params, **kwargs)
def post(self, url: str, data: Optional[Union[Dict[str, Any], str, bytes]] = None,
json: Optional[Dict[str, Any]] = None, **kwargs) -> Optional[Dict[str, Any]]:
return self.request('POST', url, data=data, json=json, **kwargs)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.session.close()
class RNL:
def __init__(self, c):
self.t_id = None
self.options = {
"task_list": True,
"complete_task": True,
"receive_award": True,
"task_item": True,
"UserJoin": True,
}
self.activity_code = '2211-videoWelfare'
self.rr = RnlRequest(c)
self.logs = [] # 用于存储详细日志
def log(self, message: str):
"""记录详细日志"""
self.logs.append(message)
print(message)
def get_task_list(self):
data = {
'activityCode': self.activity_code,
}
try:
response = self.rr.post(
'https://m.jr.airstarfinance.net/mp/api/generalActivity/getTaskList',
data=data,
)
if response and response['code'] != 0:
self.log(f"获取任务列表失败: {response}")
return None
target_tasks = []
for task in response['value']['taskInfoList']:
if '浏览组浏览任务' in task['taskName']:
target_tasks.append(task)
return target_tasks
except Exception as e:
self.log(f'获取任务列表失败：{e}')
return None
def get_task(self, task_code):
try:
data = {
'activityCode': self.activity_code,
'taskCode': task_code,
}
response = self.rr.post(
'https://m.jr.airstarfinance.net/mp/api/generalActivity/getTask',
data=data,
)
if not response or response['code'] != 0:
self.log(f'获取任务信息失败：{response}')
return None
return response['value']['taskInfo']['userTaskId']
except Exception as e:
self.log(f'获取任务信息失败：{e}')
return None
def complete_task(self, task_id, t_id, brows_click_urlId):
try:
response = self.rr.get(
f'https://m.jr.airstarfinance.net/mp/api/generalActivity/completeTask?activityCode={self.activity_code}&app=com.mipay.wallet&isNfcPhone=true&channel=mipay_indexicon_TVcard&deviceType=2&system=1&visitEnvironment=2&userExtra=%7B%22platformType%22:1,%22com.miui.player%22:%224.27.0.4%22,%22com.miui.video%22:%22v2024090290(MiVideo-UN)%22,%22com.mipay.wallet%22:%226.83.0.5175.2256%22%7D&taskId={task_id}&browsTaskId={t_id}&browsClickUrlId={brows_click_urlId}&clickEntryType=undefined&festivalStatus=0',
)
if not response or response['code'] != 0:
self.log(f'完成任务失败：{response}')
return None
return response['value']
except Exception as e:
self.log(f'完成任务失败：{e}')
return None
def receive_award(self, user_task_id):
try:
response = self.rr.get(
f'https://m.jr.airstarfinance.net/mp/api/generalActivity/luckDraw?imei=&device=manet&appLimit=%7B%22com.qiyi.video%22:false,%22com.youku.phone%22:true,%22com.tencent.qqlive%22:true,%22com.hunantv.imgo.activity%22:true,%22com.cmcc.cmvideo%22:false,%22com.sankuai.meituan%22:true,%22com.anjuke.android.app%22:false,%22com.tal.abctimelibrary%22:false,%22com.lianjia.beike%22:false,%22com.kmxs.reader%22:true,%22com.jd.jrapp%22:false,%22com.smile.gifmaker%22:true,%22com.kuaishou.nebula%22:false%7D&activityCode={self.activity_code}&userTaskId={user_task_id}&app=com.mipay.wallet&isNfcPhone=true&channel=mipay_indexicon_TVcard&deviceType=2&system=1&visitEnvironment=2&userExtra=%7B%22platformType%22:1,%22com.miui.player%22:%224.27.0.4%22,%22com.miui.video%22:%22v2024090290(MiVideo-UN)%22,%22com.mipay.wallet%22:%226.83.0.5175.2256%22%7D'
)
if not response or response['code'] != 0:
self.log(f'领取奖励失败：{response}')
except Exception as e:
self.log(f'领取奖励失败：{e}')
def queryUserJoinListAndQueryUserGoldRichSum(self):
try:
total_res = self.rr.get('https://m.jr.airstarfinance.net/mp/api/generalActivity/queryUserGoldRichSum?app=com.mipay.wallet&deviceType=2&system=1&visitEnvironment=2&userExtra={"platformType":1,"com.miui.player":"4.27.0.4","com.miui.video":"v2024090290(MiVideo-UN)","com.mipay.wallet":"6.83.0.5175.2256"}&activityCode=2211-videoWelfare')
if not total_res or total_res['code'] != 0:
self.log(f'获取兑换视频天数失败：{total_res}')
return False
total = f"{int(total_res['value']) / 100:.2f}天" if total_res else "未知"
self.log(f"当前用户兑换视频天数：{total}")
response = self.rr.get(
f'https://m.jr.airstarfinance.net/mp/api/generalActivity/queryUserJoinList?&userExtra=%7B%22platformType%22:1,%22com.miui.player%22:%224.27.0.4%22,%22com.miui.video%22:%22v2024090290(MiVideo-UN)%22,%22com.mipay.wallet%22:%226.83.0.5175.2256%22%7D&activityCode={self.activity_code}&pageNum=1&pageSize=20',
)
if not response or response['code'] != 0:
self.log(f'查询任务完成记录失败：{response}')
return False
history_list = response['value']['data']
current_date = datetime.now().strftime("%Y-%m-%d")
self.log(f"-- {current_date} 当天任务记录 --")
for a in history_list:
record_time = a['createTime']
record_date = record_time[:10]
if record_date == current_date:
days = int(a['value']) / 100
self.log(f"{record_time} 领到视频会员，+{days:.2f}天")
return True
except Exception as e:
self.log(f'获取任务记录失败：{e}')
return False
def main(self):
self.logs = [] # 重置日志
if not self.queryUserJoinListAndQueryUserGoldRichSum():
return {"success": False, "logs": self.logs}
actual_success = False # 记录真实执行状态
for i in range(2):
# 获取任务列表
tasks = self.get_task_list()
if not tasks:
self.log("⚠️ 无法获取任务列表，跳过执行")
continue
task = tasks[0]
try:
t_id = task['generalActivityUrlInfo']['id']
self.t_id = t_id
except:
if not self.t_id:
self.log("⚠️ 无法获取t_id，跳过本次任务")
continue
t_id = self.t_id
task_id = task['taskId']
task_code = task['taskCode']
brows_click_url_id = task['generalActivityUrlInfo']['browsClickUrlId']
time.sleep(13)
# 完成任务（修正参数顺序）
user_task_id = self.complete_task(
task_id=task_id, # 修正参数顺序
t_id=t_id,
brows_click_urlId=brows_click_url_id,
)
time.sleep(2)
# 获取任务数据
if not user_task_id:
user_task_id = self.get_task(task_code=task_code)
time.sleep(2)
# 领取奖励
if user_task_id:
self.receive_award(user_task_id=user_task_id)
actual_success = True # 标记为执行成功
else:
self.log("⚠️ 无法获取user_task_id，跳过奖励领取")
time.sleep(2)
# 再次查询任务记录
self.queryUserJoinListAndQueryUserGoldRichSum()
return {
"success": actual_success,
"logs": self.logs
} # 返回执行状态和详细日志
def get_xiaomi_cookies(pass_token, user_id):
session = requests.Session()
login_url = 'https://account.xiaomi.com/pass/serviceLogin?callback=https://api.jr.airstarfinance.net/sts%3Fsign%3D1dbHuyAmee0NAZ2xsRw5vhdVQQ8%253D%26followup%3Dhttps%253A%252F%252Fm.jr.airstarfinance.net%252Fmp%252Fapi%252Flogin%253Ffrom%253Dmipay_indexicon_TVcard%2526deepLinkEnable%253Dfalse%2526requestUrl%253Dhttps%25253A%25252F%25252Fm.jr.airstarfinance.net%25252Fmp%25252Factivity%25252FvideoActivity%25253Ffrom%25253Dmipay_indexicon_TVcard%252526_noDarkMode%25253Dtrue%252526_transparentNaviBar%25253Dtrue%252526cUserId%25253Dusyxgr5xjumiQLUoAKTOgvi858Q%252526_statusBarHeight%25253D137&sid=jrairstar&_group=DEFAULT&_snsNone=true&_loginType=ticket'
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36 Edg/135.0.0.0',
'cookie': f'passToken={pass_token}; userId={user_id};'
}
try:
session.get(url=login_url, headers=headers, verify=False)
cookies = session.cookies.get_dict()
return f"cUserId={cookies.get('cUserId')};jrairstar_serviceToken={cookies.get('serviceToken')}"
except Exception as e:
print(f"获取Cookie失败: {e}")
return None
if __name__ == "__main__":
# 多账号配置区 ##################################
ORIGINAL_COOKIES = [
{ # 账号1
'passToken': 'V1:J7rrdw==',
'userId': '1235'
},
{ # 账号2
'passToken': 'V1:JC6LNGAp73majw==',
'userId': '23'
},
{ # 账号3
'passToken': 'V1:J7rAkbp4V1/w==',
'userId': '30859'
},
{ # 账号4
'passToken': 'V1:J7r==',
'userId': '2401'
}
# 可继续添加更多账号...
]
# 结束配置 ######################################
cookie_list = []
execution_results = {}
start_time = time.time()
date_str = datetime.now().strftime("%Y-%m-%d %H:%M")
# 处理每个账号的Cookie
for account in ORIGINAL_COOKIES:
user_id = account['userId']
print(f"
>>>>>>>>>> 正在处理账号 {user_id} <<<<<<<<<<")
execution_results[user_id] = {"success": False, "message": "未执行", "logs": []}
try:
new_cookie = get_xiaomi_cookies(account['passToken'], user_id)
if new_cookie:
cookie_list.append(new_cookie)
execution_results[user_id] = {"success": True, "message": "✅ Cookie获取成功", "logs": []}
print(f"账号 {user_id} Cookie获取成功")
else:
execution_results[user_id] = {"success": False, "message": "⚠️ Cookie获取失败", "logs": []}
print(f"⚠️ 账号 {user_id} Cookie获取失败，请检查配置")
except Exception as e:
execution_results[user_id] = {"success": False, "message": f"⚠️ Cookie获取异常: {str(e)}", "logs": []}
print(f"⚠️ 账号 {user_id} Cookie获取异常: {str(e)}")
print(f"
>>>>>>>>>> 共获取到{len(cookie_list)}个有效Cookie <<<<<<<<<<")
# 执行每个账号的任务
for index, c in enumerate(cookie_list):
user_id = ORIGINAL_COOKIES[index]['userId']
print(f"
--------- 开始执行第{index+1}个账号 ({user_id}) ---------")
try:
result = RNL(c).main()
actual_success = result["success"]
logs = result["logs"]
status_msg = "✅ 执行成功" if actual_success else "⚠️ 执行失败"
execution_results[user_id] = {
"success": actual_success,
"message": status_msg,
"logs": logs # 保存详细日志
}
print(f"第{index+1}个账号({user_id})执行结果: {'成功' if actual_success else '失败'}")
except Exception as e:
error_msg = f"⚠️ 执行异常: {str(e)}"
execution_results[user_id] = {"success": False, "message": error_msg, "logs": []}
print(f"⚠️ 第{index+1}个账号({user_id})执行异常: {str(e)}")
print(f"--------- 第{index+1}个账号({user_id})执行结束 ---------")
# 生成执行报告
success_count = sum(1 for res in execution_results.values() if res["success"])
failure_count = len(execution_results) - success_count
total_time = time.time() - start_time
# 构建报告内容 - 使用您期望的格式
report_content = [
f"🔥 小米钱包任务执行报告",
f"📅 执行时间: {date_str}",
f"⏱️ 总耗时: {total_time:.2f}秒",
f"✅ 成功: {success_count}个账号",
f"⚠️ 失败: {failure_count}个账号",
"",
"📝 账号详情:"
]
for i, (user_id, res) in enumerate(execution_results.items(), 1):
# 添加账号状态
report_content.append(f"{i}. {user_id}: {res['message']}")
# 添加该账号的详细日志
if res["logs"]:
report_content.extend(res["logs"])
report_content.append("") # 添加空行分隔账号
full_title = f"小米钱包任务执行结果 - {date_str}"
full_content = "
".join(report_content)
print("
" + "="*50)
print(full_content)
print("="*50)
# 发送通知
Notifier.send(full_title, full_content)三.抓包
访问：https://account.xiaomi.com/pass/serviceLogin?callback=https://api.jr.airstarfinance.net/sts%3Fsign%3D1dbHuyAmee0NAZ2xsRw5vhdVQQ8%253D%26followup%3Dhttps%253A%252F%252Fm.jr.airstarfinance.net%252Fmp%252Fapi%252Flogin%253Ffrom%253Dmipay_indexicon_TVcard%2526deepLinkEnable%253Dfalse%2526requestUrl%253Dhttps%25253A%25252F%25252Fm.jr.airstarfinance.net%25252Fmp%25252Factivity%25252FvideoActivity%25253Ffrom%25253Dmipay_indexicon_TVcard%252526_noDarkMode%25253Dtrue%252526_transparentNaviBar%25253Dtrue%252526cUserId%25253Dusyxgr5xjumiQLUoAKTOgvi858Q%252526_statusBarHeight%25253D137&sid=jrairstar&_group=DEFAULT&_snsNone=true&_loginType=ticket
输入手机号发送验证码，填写验证码，开启抓包，点击登录
登录成功后搜索https://account.xiaomi.com找Cookies中的passToken参数，注意这个参数结尾是==
第二个参数是找Cookies中的userId
推荐使用抓包工具：Fiddler
© 版权声明
文章版权归作者所有，未经允许请勿转载。
相关文章
暂无评论...
