2025年10月11日星期六

小米钱包任务脚本

1.购买服务器阿里云：服务器购买地址https://t.aliyun.com/U/Bg6shY若失效，可用地址

1.购买服务器

阿里云：

服务器购买地址

https://t.aliyun.com/U/Bg6shY

若失效，可用地址

https://www.aliyun.com/daily-act/ecs/activity_selection?source=5176.29345612&userCode=49hts92d

腾讯云：

https://curl.qcloud.com/wJpWmSfU

若失效，可用地址

https://cloud.tencent.com/act/cps/redirect?redirect=2446&cps_key=ad201ee2ef3b771157f72ee5464b1fea&from=console

华为云

https://activity.huaweicloud.com/cps.html?fromacct=64b5cf7cc11b4840bb4ed2ea0b2f4468&utm_source=V1g3MDY4NTY=&utm_medium=cps&utm_campaign=201905

2.部署教程

2024年最新青龙面板跑脚本教程（一）持续更新中

3.代码如下

# cron: 0 12 * * *# const $ = new Env("小米钱包");
import osimport timeimport requestsimport urllib3from datetime import datetimefrom typing import Optional, Dict, Any, Union
# ==================== Bark 推送配置 ====================# Bark 推送地址（环境变量读取）BARK_PUSH = os.getenv("BARK_PUSH")
# 可以自定义参数，也可以留空CUSTOM_BARK_ICON = "Wallet.png"   # 自定义图标CUSTOM_BARK_GROUP = "小米钱包"              # 自定义分组PUSH_SWITCH = "1"    #推送开关，1开启，0关闭
# 定义全局变量BARK_ICON = CUSTOM_BARK_ICON or os.getenv("BARK_ICON", "")BARK_GROUP = CUSTOM_BARK_GROUP or os.getenv("BARK_GROUP", "")
# 覆盖环境变量os.environ["BARK_ICON"] = BARK_ICONos.environ["BARK_GROUP"] = BARK_GROUPos.environ["PUSH_SWITCH"] = PUSH_SWITCH
# =====================================================
import notifyurllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_ck_from_env():    ck_str = os.environ.get("xmqb", "")    accounts = []    if ck_str:        for item in ck_str.split("@"):            try:                passToken, userId = item.split("&")                accounts.append({'passToken': passToken.strip(), 'userId': userId.strip()})            except:                print(f"[Warning] 无法解析CK: {item}")    return accounts

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}")            return None        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.current_user_id = None  # 存储当前处理的用户ID        self.total_days = "未知"        self.today_records = []        self.error_info = ""
    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.error_info = f"获取任务列表失败：{response}"                print(self.error_info)                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.error_info = f'获取任务列表失败：{e}'            print(self.error_info)            return None
    def get_task(self, task_code):        try:            data = {                'activityCode': self.activity_code,                'taskCode': task_code,                'jrairstar_ph': '98lj8puDf9Tu/WwcyMpVyQ==',            }            response = self.rr.post(                'https://m.jr.airstarfinance.net/mp/api/generalActivity/getTask',                data=data,            )            if response and response['code'] != 0:                self.error_info = f'获取任务信息失败：{response}'                print(self.error_info)                return None            return response['value']['taskInfo']['userTaskId']        except Exception as e:            self.error_info = f'获取任务信息失败：{e}'            print(self.error_info)            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 response and response['code'] != 0:                self.error_info = f'完成任务失败：{response}'                print(self.error_info)                return None            return response['value']        except Exception as e:            self.error_info = f'完成任务失败：{e}'            print(self.error_info)            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 response and response['code'] != 0:                self.error_info = f'领取奖励失败：{response}'                print(self.error_info)        except Exception as e:            self.error_info = f'领取奖励失败：{e}'            print(self.error_info)
    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.error_info = f'获取兑换视频天数失败：{total_res}'                print(self.error_info)                return False            self.total_days = f"{int(total_res['value']) / 100:.2f}天" if total_res else "未知"
            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.error_info = f'查询任务完成记录失败：{response}'                print(self.error_info)                return False
            history_list = response['value']['data']            current_date = datetime.now().strftime("%Y-%m-%d")
            # 清空记录            self.today_records = []
            for a in history_list:                record_time = a['createTime']                record_date = record_time[:10]                if record_date == current_date:                    self.today_records.append({                        'createTime': record_time,                        'value': a['value']                    })
            return True        except Exception as e:            self.error_info = f'获取任务记录失败：{e}'            print(self.error_info)            return False
    def main(self):        if not self.queryUserJoinListAndQueryUserGoldRichSum():            return False        for i in range(2):            # 获取任务列表            tasks = self.get_task_list()            if not tasks:                return False
            task = tasks[0]            try:                t_id = task['generalActivityUrlInfo']['id']                self.t_id = t_id            except:                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(                t_id=t_id,                task_id=task_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)
            # 领取奖励            self.receive_award(                user_task_id=user_task_id            )
            time.sleep(2)
        # 重新获取最新记录        self.queryUserJoinListAndQueryUserGoldRichSum()        return True

def get_xiaomi_cookies(pass_token, user_id):    session = requests.Session()    login_url = 'https://account.xiaomi.com/pass/serviceLogin?callback=https%3A%2F%2Fapi.jr.airstarfinance.net%2Fsts%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:        error_msg = f"获取Cookie失败: {e}"        print(error_msg)        return None, error_msg

def generate_notification(account_id, rnl_instance):    """生成简化的通知消息"""    # 计算今日获得的总天数    today_total = sum(int(record["value"]) for record in rnl_instance.today_records) / 100
    msg = f"""    账号ID：{account_id}    当前兑换视频天数：{rnl_instance.total_days}    今日获得：+{today_total:.2f}天"""
    return msg

if __name__ == "__main__":
    ORIGINAL_COOKIES = get_ck_from_env()    if not ORIGINAL_COOKIES:        print(" 未获取到账号CK，请检查青龙环境变量 xmqb 是否配置正确！")        exit()
    # 构建简化通知消息    simplified_notification = "【小米钱包任务执行结果】\n"
    cookie_list = []    for account in ORIGINAL_COOKIES:        user_id = account['userId']        print(f"\n>>>>>>>>>> 正在处理账号 {user_id} <<<<<<<<<<")
        # 获取Cookie        cookie_result = get_xiaomi_cookies(account['passToken'], user_id)
        # 处理返回结果        if isinstance(cookie_result, tuple):            new_cookie, error = cookie_result        else:            new_cookie = cookie_result            error = None
        # 创建RNL实例并设置当前用户ID        rnl = RNL(new_cookie)        rnl.current_user_id = user_id
        if error:            rnl.error_info = error        else:            print(f"账号 {user_id} Cookie获取成功")            cookie_list.append(new_cookie)
            # 执行主程序            try:                rnl.main()            except Exception as e:                rnl.error_info = f"执行异常: {str(e)}"                print(rnl.error_info)
        # 生成当前账号的简化通知消息        account_notification = generate_notification(user_id, rnl)        simplified_notification += account_notification + "\n"
    # 判断是否推送    if PUSH_SWITCH == '1':        notify.send("小米钱包任务推送", simplified_notification)    else:        print("推送开关已关闭，不发送推送通知")

解析

该脚本的主要作用为：

  • 面向 小米钱包视频福利活动

    （activityCode=2211-videoWelfare） 的自动化脚本。

  • 从环境变量 xmqb 读取多个账号（passToken&userId，多账号用 @ 分隔），先到小米账号中心换取活动所需 Cookie，再：

    1. 查询已获"视频天数"和当日完成记录；

    2. 拉取"浏览组浏览任务"；

    3. 启动任务→等待→提交完成→抽奖领取；（循环两次，等价于"每天两次视频任务"）

    4. 重新查询今日记录与总天数；

    5. 汇总为简化通知（Bark/青龙 notify）。

主要方法

环境与通知

  • get_ck_from_env()：从环境变量 xmqb 解析账号列表（passToken&userId）。

  • 顶部 Bark/通知配置：通过 BARK_*、PUSH_SWITCH 控制是否推送，最后调用 notify.send 发送汇总。

请求封装

  • 类 RnlRequest：对 requests.Session 的轻封装。

    • update_cookies() / _parse_cookies() / dict_cookie_to_string()：设置与格式化 Cookie。

    • request() / get() / post()：统一发起请求、合并基础 UA 头、异常与 JSON 解析处理。

活动逻辑

  • 类 RNL（单账号一次执行）

    • GET /luckDraw：对已完成的 userTaskId 抽奖/领奖励。

    • GET /completeTask：提交完成（带一堆环境参数，如 app、deviceType、userExtra 等），成功返回 value（通常含 userTaskId）。

    • POST /getTask：若完成接口未直接返回 userTaskId，通过此接口补取 userTaskId。

    • POST /getTaskList：拉取任务清单，挑出"浏览组浏览任务"。

    • GET /queryUserGoldRichSum：获取累计可兑换视频天数（返回分值，代码换算成"天"）；

    • GET /queryUserJoinList：拉完成记录，筛出当天记录，填充 today_records。

    • 属性：活动码、RnlRequest 实例、今日记录 today_records、累计天数 total_days、错误信息等。

    • queryUserJoinListAndQueryUserGoldRichSum()

    • get_task_list()

    • get_task(task_code)

    • complete_task(task_id, t_id, brows_click_urlId)

    • receive_award(user_task_id)

    • main()（本账号主流程）

    1. 先查累计/今日记录；

    2. 循环 2 次：拿任务→等待 13s→completeTask→必要时 getTask→luckDraw；

    3. 再查一次累计/今日记录用于结果展示；返回成功/失败。

辅助

  • get_xiaomi_cookies(pass_token, user_id)
    使用传入的 passToken、userId 访问小米登录链接，借助会话 Cookie 组装活动域所需的
    cUserId 与 jrairstar_serviceToken，返回 "cUserId=...;jrairstar_serviceToken=..."。

  • generate_notification(account_id, rnl_instance)
    根据 RNL.today_records 汇总今日新增视频天数并与累计天数拼接成简报文本。

脚本入口

  • 读取账号 → 逐个账号：获取活动 Cookie → new RNL 并执行 main() → 累积简化通知 → notify.send 推送。

小提示

  • requests 使用了 verify=False（关闭 TLS 校验），如果不需要可去掉。

  • 活动接口以 code==0 代表成功，脚本已据此判断；若出现风控或返回结构变化，需要据日志调整。

  • 若要更稳，可给 get_task_list/complete_task/receive_award 增加简单重试。


注意：

本文部分变量已做脱敏处理，仅用于测试和学习研究，禁止用于商业用途，不能保证其合法性，准确性，完整性和有效性，请根据情况自行判断。技术层面需要提供帮助，可以通过打赏的方式进行探讨。


历史脚本txt文件获取>>
服务器搭建，人工服务咨询>>


时间：

没有评论:

发表评论

订阅： 博文评论 (Atom)

从0到$30K月利润：Pushcroll如何用「病毒式内容 + 自动化私域」实现爆发式增长

增长其实不是技巧，而是好奇心。\x0d\x0a你要不断问——为什么这条视频能火？为什么那条不能？\x0d\x0a当你能回答这个问题，你的增长就开始了。" 在短短几个月内，一款名为  Pushcroll  的应用靠一条"做俯卧撑才能刷短视频"的创意...