网赚：日进千刀: 天翼云盘任务脚本(2025年12月版)

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

1.购买服务器

阿里云：

服务器购买地址

https://t.aliyun.com/U/55RK8C

若失效，可用地址

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.代码如下

import osimport reimport rsaimport timeimport randomimport base64import requestsfrom urllib3.util.retry import Retryfrom requests.adapters import HTTPAdapterfrom datetime import datetime, timedelta, timezone
try:    import env    # windows 本地测试except ImportError:    pass

class TGA:    def __init__(self, token, chat_id, retry=2, timeout=5):        # 检测是否为空        if not token or not chat_id:            raise ValueError("Token and chat_id cannot be empty")
        self.token = token        self.chat_id = chat_id        self.retry = retry        self.timeout = timeout
    # 发文字    def send_text(self, text, parse_mode=None):        data = {            "chat_id": self.chat_id,            "text": text        }        if parse_mode:            data["parse_mode"] = parse_mode        return self._post("sendMessage", data=data)
    def send_markdown(self, text):        return self.send_text(text, "Markdown")

class Ecloud:    globals()["UA"] = 'Mozilla/5.0'    globals()["MODEL"] = "SM-G930K"    BI_RM = list("0123456789abcdefghijklmnopqrstuvwxyz")    B64MAP = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"    Referer = "https://m.cloud.189.cn/zhuanti/2016/sign/index.jsp"  # https://m.cloud.189.cn/zt/2024/grow-guide/index.html#/
    def __init__(self, username, password):        self.username = username        self.password = password
    @staticmethod    def retry_request(method, session, url, **kwargs):        max_retries = 3        delay = 3        for attempt in range(1, max_retries + 1):            try:                resp = session.request(method, url, **kwargs)                resp.raise_for_status()                return resp            except Exception as e:                print(f"[第 {attempt} 次请求失败] {method} {url} → {e}")                if attempt < max_retries:                    time.sleep(delay)                    delay += 2                else:                    raise        return None
    @staticmethod    def init_session_with_retry():        sess = requests.Session()        retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504],                        allowed_methods=["GET", "POST"])        adapter = HTTPAdapter(max_retries=retries)        sess.mount("http://", adapter)        sess.mount("https://", adapter)        return sess
    def int2char(self, a):        return self.BI_RM[a]
    def b64tohex(self, a):        d, e, c = "", 0, 0        for ch in a:            if ch != "=":                v = self.B64MAP.index(ch)                if e == 0:                    e = 1                    d += self.int2char(v >> 2)                    c = v & 3                elif e == 1:                    e = 2                    d += self.int2char((c << 2) | (v >> 4))                    c = v & 15                elif e == 2:                    e = 3                    d += self.int2char(c)                    d += self.int2char(v >> 2)                    c = v & 3                else:                    e = 0                    d += self.int2char((c << 2) | (v >> 4))                    d += self.int2char(v & 15)        if e == 1:            d += self.int2char(c << 2)        return d
    def rsa_encode(self, pubkey_str, text):        rsa_key = f"-----BEGIN PUBLIC KEY-----\n{pubkey_str}\n-----END PUBLIC KEY-----"        pubkey = rsa.PublicKey.load_pkcs1_openssl_pem(rsa_key.encode())        encrypted = rsa.encrypt(text.encode(), pubkey)        return self.b64tohex(base64.b64encode(encrypted).decode())
    def login_flow(self):        session = self.init_session_with_retry()        base_headers = {            'User-Agent': UA or "Mozilla 5.0",            'Accept-Encoding': 'gzip, deflate'        }        token_url = f"https://m.cloud.189.cn/udb/udb_login.jsp?pageId=1&pageKey=default&clientType=wap&redirectURL={self.Referer}"        r = self.retry_request("GET", session, token_url, headers=base_headers, timeout=10)        m = re.search(r"https?://[^\s'\"]+", r.text)        if not m:            raise Exception("登录跳转 URL 获取失败")        redirect = m.group()        r = self.retry_request("GET", session, redirect, headers=base_headers, timeout=10)        m2 = re.search(r"<a id=\"j-tab-login-link\"[^>]*href=\"([^\"]+)\"", r.text)        if not m2:            raise Exception("登录页面链接解析失败")        href = m2.group(1)        r = self.retry_request("GET", session, href, headers=base_headers, timeout=10)
        ct = re.findall(r"captchaToken' value='(.+?)'", r.text)[0]        lt = re.findall(r'lt = "(.+?)"', r.text)[0]        ret = re.findall(r"returnUrl= '(.+?)'", r.text)[0]        pid = re.findall(r'paramId = "(.+?)"', r.text)[0]        pubkey = re.findall(r'j_rsaKey" value="(\S+)"', r.text, re.M)[0]        session.headers.update({"lt": lt})
        user_enc = self.rsa_encode(pubkey, self.username)        pwd_enc = self.rsa_encode(pubkey, self.password)
        login_api = "https://open.e.189.cn/api/logbox/oauth2/loginSubmit.do"        data = {            "appKey": "cloud",            "accountType": "01",            "userName": f"{{RSA}}{user_enc}",            "password": f"{{RSA}}{pwd_enc}",            "validateCode": "",            "captchaToken": ct,            "returnUrl": ret,            "mailSuffix": "@189.cn",            "paramId": pid        }        login_headers = {            'User-Agent': UA or "Mozilla 5.0",            'Referer': 'https://open.e.189.cn/',            'Accept-Encoding': 'gzip, deflate'        }        r = self.retry_request("POST", session, login_api, data=data, headers=login_headers, timeout=10)        msg = r.json().get("msg", "无消息")        print("登录响应：", msg)        redirect_to = r.json().get("toUrl")        if not redirect_to:            raise Exception("登录跳转 URL 获取失败")        self.retry_request("GET", session, redirect_to, headers=base_headers, timeout=10)        return session
    def single_checkin(self):        try:            sess = self.login_flow()        except Exception as e:            print("登录重试结束仍失败：", e)            push = '❌ 错误，请查看运行日志！'            return push
        base_headers = {            'User-Agent': UA or "Mozilla 5.0",            'Referer': f"{self.Referer}",            'Host': 'm.cloud.189.cn',            'Accept-Encoding': 'gzip, deflate'        }        rand = str(int(time.time() * 1000))        sign_url = f'https://api.cloud.189.cn/mkt/userSign.action?rand={rand}&clientType=TELEANDROID&version=10.3.11&model={MODEL}'
        results = []        try:            r = self.retry_request("GET", sess, sign_url, headers=base_headers, timeout=10)            j = r.json()            bonus = j.get('netdiskBonus', 0)            if j.get('isSign') == "false":                msg = f"❌ 未签到，获得 {bonus}M 空间"            else:                msg = f"✅ 已签到，获得 {bonus}M 空间"            print(msg)            results.append(msg)        except Exception as e:            print("签到失败：", e)            push = '❌ 错误，请查看运行日志！'            return push
        # tasks = [        #     ("TASK_SIGNIN", "ACT_SIGNIN", ""),        #     ("TASK_SIGNIN_PHOTOS", "ACT_SIGNIN", ""),        #     ("TASK_2022_FLDFS_KJ", "ACT_SIGNIN", "链接3")        # ]        # for task, act, label in tasks:        #     try:        #         detail_url = (f'https://m.cloud.189.cn/v2/drawPrizeMarketDetails.action'        #                       f'?taskId={task}&activityId={act}')        #         r = self.retry_request("GET", sess, detail_url, headers=base_headers, timeout=10)        #         data = r.json()        #         if 'description' in data:        #             desc = data['description']        #             out = (f"{label}抽奖获得{desc}" if label else f"抽奖获得{desc}")        #             print(out)        #             results.append(out)        #     except Exception as e:        #         print(f"{label}抽奖失败：", e)
        print("天翼云盘签到结果：")        push = "\n".join(results)        return push

def pusher(msg):    PASS

def adduser(msg, uname, i):    if uname:        msg = f"🙍🏻‍♂️ 第{i}个账号" + mphone(uname) + "\n" + msg    return msg

def mphone(phone):    l = len(phone)    masked_phone = f"{phone[:3]}{"\\*" * (l - 7)}{phone[l - 4:]}"    return masked_phone

def main():    msg = ""    if (a := os.getenv("ACCOUNTS")) is None:        print("❌ 未设置 ACCOUNTS 环境变量")        pusher("❌ 未设置 ACCOUNTS 环境变量")        return    # 读取系统变量以 \n 或 && 分割变量    accounts = re.split('\n|&&', a)    print("✅ 检测到共：", len(accounts), "个天翼云盘账号\n")    # main    i = 0    while i < len(accounts):        username, password = accounts[i].replace(" ", "").split(";")        try:            assert username and password, "请检查账号密码是否填写正确"        except AssertionError:            push = '❌ 错误，请检查账号密码是否填写正确！'        else:            push = Ecloud(username, password).single_checkin()        push = adduser(push, username, i + 1)        msg += push + ("" if i + 1 == len(accounts) else "\n\n")        i += 1        if not i == len(accounts):            time.sleep(random.uniform(2, 5))    # print("\n推送预览：\n", msg)    pusher(msg)

if __name__ == "__main__":    UA = os.getenv("UA") or 'Mozilla/5.0'    MODEL = os.getenv("MODEL") or "SM-G930K"
    print("----------天翼云盘开始签到----------")    main()    print("----------天翼云盘签到完成----------")

解析

这是一个 天翼云盘（189 Cloud）多账号自动签到脚本，特点是：

从环境变量 ACCOUNTS 读取多账号（username;password，支持用 \n 或 && 分隔多组）
走 天翼云盘网页登录流程：先拿登录页参数/公钥 → RSA 加密账号密码 → 提交登录 → 跟随跳转拿到会话 Cookie
调用签到接口 mkt/userSign.action 完成签到，读取奖励（通常是网盘空间 netdiskBonus）
支持失败重试（requests + urllib3 Retry + 自定义 retry_request）
把结果通过Bot 推送（环境变量chat_id）
每个账号签到间隔随机等待，降低频率触发风控

主要方法

`Ecloud` 类（天翼云盘登录 + 签到核心）

1) `Ecloud.init_session_with_retry()`

创建 requests.Session() 并挂载 HTTPAdapter(max_retries=Retry(...))
用于自动处理 500/502/503/504 等临时错误

2) `Ecloud.retry_request(method, session, url, **kwargs)`

第二层重试：对每一次请求再做"最多3次、逐步延迟"的重试
这会叠加 Session 自带的 Retry，使网络不稳定时更抗打

3) `Ecloud.rsa_encode(pubkey_str, text)`

把网页返回的 RSA 公钥包装成 PEM
使用 rsa.encrypt() 加密明文
再把加密结果 base64 后转成"hex风格字符串"（用于天翼的 {RSA} 参数格式）

这个方法的意义：模拟网页登录提交时的加密账号密码。

4) `Ecloud.login_flow()`

这是整个脚本最核心的"登录流程"方法，做的事情概括为：

访问登录入口页拿到跳转地址
打开跳转页面，解析真正的登录页链接
从登录页 HTML 中提取关键字段：

captchaToken / lt / returnUrl / paramId / j_rsaKey(pubkey)

用 rsa_encode() 加密 username/password
POST 到 loginSubmit.do 提交登录
从响应里取 toUrl 并跟随跳转，最终拿到有效会话 Cookie
返回已登录的 session

5) `Ecloud.single_checkin()`

先调用 login_flow() 获取登录态 session
再访问签到接口：
https://api.cloud.189.cn/mkt/userSign.action?rand=...&clientType=...&model=...
解析返回 JSON：

isSign 判断已签到/未签到
netdiskBonus 获取奖励（空间多少 M）

拼装签到结果文本并返回（供后续统一推送）

脚本中注释掉的 tasks 抽奖逻辑，属于可扩展功能：原计划在签到后继续调用抽奖/任务接口。

函数（组织流程 + 多账号 + 推送）

1) `pusher(msg)`

如果设置了环境变量 TGA，则：

生成北京时间时间戳
把结果包装成 Markdown 文本
调 TGA.send_markdown() 推送

2) `mphone(phone)` / `adduser(msg, uname, i)`

mphone()：对用户名/手机号做脱敏显示（中间用 * 替换）
adduser()：在每个账号的结果前面加一行：第 i 个账号 + 脱敏账号

3) `main()`

读取 ACCOUNTS 环境变量，并用 \n 或 && 切分多账号
对每个账号：

拆出 username;password
调 Ecloud(username,password).single_checkin()
拼接结果
账号之间随机 sleep 2~5 秒

最终把汇总结果调用 pusher() 推送

4) `main` 入口

初始化 UA、MODEL（可用环境变量覆盖）
打印开始/结束日志并执行 main()

注意：

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

历史脚本txt文件获取>>

服务器搭建，人工服务咨询>>

【相关文章】

天翼云盘任务脚本(2025年9月版)

网赚：日进千刀

2025年12月24日星期三

天翼云盘任务脚本(2025年12月版)

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

主要方法

`Ecloud` 类（天翼云盘登录 + 签到核心）

1) `Ecloud.init_session_with_retry()`

2) `Ecloud.retry_request(method, session, url, **kwargs)`

3) `Ecloud.rsa_encode(pubkey_str, text)`

4) `Ecloud.login_flow()`

5) `Ecloud.single_checkin()`

函数（组织流程 + 多账号 + 推送）

1) `pusher(msg)`

2) `mphone(phone)` / `adduser(msg, uname, i)`

3) `main()`

4) `main` 入口

没有评论:

发表评论

天翼云盘任务脚本(2025年12月版)

2025年12月24日星期三

天翼云盘任务脚本(2025年12月版)

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

主要方法

Ecloud 类（天翼云盘登录 + 签到核心）

1) Ecloud.init_session_with_retry()

2) Ecloud.retry_request(method, session, url, **kwargs)

3) Ecloud.rsa_encode(pubkey_str, text)

4) Ecloud.login_flow()

5) Ecloud.single_checkin()

函数（组织流程 + 多账号 + 推送）

1) pusher(msg)

2) mphone(phone) / adduser(msg, uname, i)

3) main()

4) __main__ 入口

没有评论:

发表评论

天翼云盘任务脚本(2025年12月版)

`Ecloud` 类（天翼云盘登录 + 签到核心）

1) `Ecloud.init_session_with_retry()`

2) `Ecloud.retry_request(method, session, url, **kwargs)`

3) `Ecloud.rsa_encode(pubkey_str, text)`

4) `Ecloud.login_flow()`

5) `Ecloud.single_checkin()`

1) `pusher(msg)`

2) `mphone(phone)` / `adduser(msg, uname, i)`

3) `main()`

4) `main` 入口