青龙脚本之RoboMind (ROBO) 每日训练签到 v3.4

📅 2026/6/27 19:55:26
青龙脚本之RoboMind (ROBO) 每日训练签到 v3.4
脚本介绍这款脚本采用python编写后缀为py每天定时签到领取ROBO收益。达到一定数值之后可以直接提取到区块链钱包或者微信支付宝当然提取到区块链钱包更划算一点。使用脚本之前需要注册一个账号邮箱可以用任意一个邮箱。每日可收益180ROMO≈180*0.02U3.6U注册链接https://robomind.vip/register.html?inviteCodeLWVT4设置定时推荐* * 8,12,16 * * ?脚本内容#!/usr/bin/env python3 # -*- coding: utf-8 -*- RoboMind (ROBO) 每日训练签到青龙脚本 v3.4 注册链接https://robomind.vip/register.html?inviteCodeLWVT4 功能 1. 登录获取 Token 2. 每日训练签到核心 3. 领取任务奖励 4. 领取团队奖励 5. 查询账户资产 新增功能 1.推送执行结果到青龙面板通知 2.阶梯性增加收益签到一天增加1两天增加2 环境变量 export ROBO_ACCOUNTS邮箱#密码 export ROBO_ACCOUNTS邮箱1#密码1邮箱2#密码2 export ROBO_ACCOUNTS{email:xxx,password:xxx} export ROBO_NOTIFYtrue # 是否启用通知默认true import os import sys import json import time import traceback from datetime import datetime from urllib import request, error BASE_URL https://api.robomind.vip/api/v1 HEADERS { Content-Type: application/json, Accept: application/json, User-Agent: RoboMind/1.0.4 (Android; Flutter), } # 全局通知内容收集 NOTIFY_CONTENT [] NOTIFY_TITLE RoboMind 每日签到报告 def log(msg, levelINFO): ts datetime.now().strftime(%Y-%m-%d %H:%M:%S) print(f[{ts}] [{level}] {msg}) # 收集通知内容只收集非DEBUG级别 if level not in [DEBUG]: NOTIFY_CONTENT.append(f[{ts}] [{level}] {msg}) def safe_data(resp): 安全获取 data 字段兼容字符串 JSON d resp.get(data, {}) if isinstance(d, str): try: d json.loads(d) except: d {} return d if isinstance(d, dict) else {} def req(method, path, bodyNone, tokenNone): 发送 API 请求 url f{BASE_URL}{path} h dict(HEADERS) if token: h[Authorization] fBearer {token} b json.dumps(body).encode() if body else None req_obj request.Request(url, datab, headersh, methodmethod) try: with request.urlopen(req_obj, timeout30) as r: return json.loads(r.read().decode()) except error.HTTPError as e: try: return json.loads(e.read().decode()) except: return {code: e.code, message: str(e)} except Exception as e: return {code: -1, message: str(e)} def send_notify(title, content): 通过青龙面板的 notify.py 发送通知 title: 通知标题 content: 通知内容 try: # 方式1: 尝试导入 notify 模块 try: import notify log(找到 notify 模块尝试发送通知..., INFO) # 调用 notify.send 方法 if hasattr(notify, send): notify.send(title, content) log(通知发送成功 (notify.send), SUCCESS) return True elif hasattr(notify, send_msg): notify.send_msg(title, content) log(通知发送成功 (notify.send_msg), SUCCESS) return True elif hasattr(notify, main): notify.main(title, content) log(通知发送成功 (notify.main), SUCCESS) return True else: log(notify 模块没有找到合适的发送方法, WARNING) except ImportError: log(未找到 notify 模块尝试使用脚本文件, DEBUG) except Exception as e: log(f使用 notify 模块发送失败: {str(e)}, WARNING) # 方式2: 查找并执行 notify.py 脚本 notify_scripts [ /ql/scripts/notify.py, # 青龙默认路径 /ql/scripts/sendNotify.js, # 旧版青龙 JS版本 /ql/repo/scripts/notify.py, # 仓库脚本 /ql/data/scripts/notify.py, # 数据目录 ./notify.py, # 当前目录 ] for script_path in notify_scripts: if os.path.exists(script_path): log(f找到通知脚本: {script_path}, INFO) import subprocess if script_path.endswith(.py): cmd [sys.executable, script_path, title, content] elif script_path.endswith(.js): cmd [node, script_path, title, content] else: continue try: result subprocess.run( cmd, capture_outputTrue, textTrue, timeout30 ) if result.returncode 0: log(f通知发送成功 ({script_path}), SUCCESS) if result.stdout: log(f输出: {result.stdout[:200]}, DEBUG) return True else: log(f脚本执行失败: {result.stderr}, WARNING) except Exception as e: log(f执行脚本 {script_path} 失败: {str(e)}, WARNING) continue # 方式3: 写入本地文件最后的备用方案 log(所有通知方式失败写入本地文件, WARNING) return write_local_notification(title, content) except Exception as e: log(f发送通知异常: {str(e)}, ERROR) traceback.print_exc() return False def write_local_notification(title, content): 写入本地通知文件 try: notify_dir /ql/data/log if not os.path.exists(notify_dir): notify_dir ./logs if not os.path.exists(notify_dir): os.makedirs(notify_dir, exist_okTrue) timestamp datetime.now().strftime(%Y%m%d_%H%M%S) filename f{notify_dir}/robomind_notify_{timestamp}.txt with open(filename, w, encodingutf-8) as f: f.write(f标题: {title}\n) f.write(f时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n) f.write( * 50 \n) f.write(content) f.write(\n * 50 \n) log(f通知已写入本地文件: {filename}, SUCCESS) return True except Exception as e: log(f写入本地通知失败: {str(e)}, ERROR) return False class RoboAPI: def __init__(self): self.token None self.user {} def login(self, email, password): r req(POST, /auth/login, {email: email, password: password}) if r.get(code) 0: data safe_data(r) self.token data.get(token, ) self.user data.get(user, {}) if self.token: uid self.user.get(id, ?) code self.user.get(invite_code, ?) log(f登录成功 | 用户: {uid} | 邀请码: {code}, SUCCESS) return True log(f登录失败: {r.get(message, 未知错误)}, ERROR) return False def overview(self): r req(GET, /mining/overview, tokenself.token) return safe_data(r) if r.get(code) 0 else None def train(self): r req(POST, /mining/train, {}, tokenself.token) if r.get(code) 0: d safe_data(r) reward d.get(reward_robo, 0) release d.get(release_reward_robo, 0) log(f训练完成 | 训练奖励: {reward} ROBO | 释放奖励: {release} ROBO, SUCCESS) return True log(f训练失败: {r.get(message, ?)}, ERROR) return False def tasks(self): r req(GET, /mining/tasks, tokenself.token) if r.get(code) 0: data safe_data(r) return data.get(tasks, []) if isinstance(data, dict) else [] return [] def claim_task(self, code): r req(POST, f/mining/tasks/{code}/claim, {}, tokenself.token) if r.get(code) 0: log(f任务 [{code}] 领取成功, SUCCESS) return True msg r.get(message, ) if any(k in msg for k in [已领取, already, claimed]): log(f任务 [{code}] 已领取) else: log(f任务 [{code}] 不可领取: {msg}) return False def claim_team(self): r req(POST, /team/rewards/claim, {}, tokenself.token) if r.get(code) 0: d safe_data(r) total d.get(claimed_robo, 0) d.get(claimed_training_robo, 0) if total 0: log(f团队奖励: {total} ROBO, SUCCESS) else: log(暂无团队奖励) return True log(f团队奖励: {r.get(message, 无)}) return False def wallet(self): r req(GET, /wallet, tokenself.token) return safe_data(r) if r.get(code) 0 else None def robot(self): r req(GET, /robot/my, tokenself.token) return safe_data(r) if r.get(code) 0 else None def team(self): r req(GET, /team/overview, tokenself.token) return safe_data(r) if r.get(code) 0 else None def mask_email(email): if not in email: return email[:2] *** local, domain email.split(, 1) m local[:2] *** local[-1:] if len(local) 3 else local[:1] *** return f{m}{domain} def parse_accounts(raw): raw raw.strip() if not raw: return [] if raw.startswith({) or raw.startswith([): try: p json.loads(raw) return [p] if isinstance(p, dict) else p except: pass if # in raw: accounts [] for pair in raw.split(): pair pair.strip() if # in pair: parts pair.split(#, 1) accounts.append({email: parts[0].strip(), password: parts[1].strip()}) return accounts return [] def run(email, password): log( * 50) log(f账号: {mask_email(email)}) log( * 50) api RoboAPI() # 1. 登录 if not api.login(email, password): return False # 2. 训练状态 ov api.overview() if ov: gear ov.get(current_gear_name, ?) days ov.get(consecutive_training_days, 0) produce ov.get(daily_production_robo, 0) release ov.get(daily_release_robo, 0) pool ov.get(release_pool_robo, 0) avail ov.get(available_robo, 0) log(f档位: {gear} | 连续: {days}天 | 日产: {produce} | 日释放: {release}) log(f释放池: {pool} ROBO | 可用: {avail} ROBO) if ov.get(has_trained_today): log(今日已训练跳过) else: log(开始训练...) api.train() else: log(无法获取概览尝试直接训练...) api.train() # 3. 任务 log(- * 40) log([任务奖励]) tasks api.tasks() if tasks: for t in tasks: title t.get(title, ?) code t.get(code, ) completed t.get(completed, False) claimed t.get(claimed, False) claim_status t.get(claim_status, ) progress t.get(progress_text, ) reward t.get(reward_robo, 0) if claimed: log(f [{title}] 已领取) elif completed and claim_status claimable: log(f [{title}] 领取 {reward} ROBO...) api.claim_task(code) time.sleep(1) else: log(f [{title}] {progress} 未完成) else: log( 无任务) # 4. 团队奖励 log(- * 40) log([团队奖励]) api.claim_team() # 5. 资产 log(- * 40) log([资产汇总]) w api.wallet() if w: log(f 总资产: {w.get(total_assets_robo, 0)} ROBO) log(f 可用: {w.get(available_robo, 0)} | 释放池: {w.get(release_pool_robo, 0)}) log(f 今日训练: {w.get(today_training_robo, 0)} | 今日释放: {w.get(today_release_robo, 0)}) rb api.robot() if rb: s rb.get(summary, {}) log(f 机器人: {s.get(owned_count, 0)}/{s.get(total_count, 6)} 组件) tm api.team() if tm: level tm.get(level, {}) current level.get(current, {}) members tm.get(members, {}) rewards tm.get(rewards, {}) log(f 团队: {current.get(code, ?)} | L1: {members.get(first_level_count, 0)}人 | L2: {members.get(second_level_count, 0)}人) log(f 可领取: {rewards.get(claimable_robo, 0)} ROBO) log(完成, SUCCESS) return True def main(): global NOTIFY_TITLE log( * 50) log(RoboMind 每日签到) log( * 50) raw os.environ.get(ROBO_ACCOUNTS, ).strip() if not raw: log(未设置 ROBO_ACCOUNTS, ERROR) NOTIFY_TITLE ❌ RoboMind 签到失败 send_notify(NOTIFY_TITLE, 未配置账号请检查环境变量 ROBO_ACCOUNTS) sys.exit(1) accounts parse_accounts(raw) if not accounts: log(账号格式错误, ERROR) NOTIFY_TITLE ❌ RoboMind 签到失败 send_notify(NOTIFY_TITLE, 账号格式错误请检查环境变量 ROBO_ACCOUNTS) sys.exit(1) log(f共 {len(accounts)} 个账号) ok 0 account_results [] for i, acc in enumerate(accounts): try: e acc.get(email, ).strip() p acc.get(password, ).strip() if e and p: masked mask_email(e) log(f\n 开始处理账号 {i1}/{len(accounts)}: {masked}) # 创建账号专属日志收集 start_idx len(NOTIFY_CONTENT) success run(e, p) if success: ok 1 account_results.append(f✅ 账号 {i1}: {masked} - 成功) else: account_results.append(f❌ 账号 {i1}: {masked} - 失败) # 提取该账号的日志 account_logs NOTIFY_CONTENT[start_idx:] if account_logs: log(f账号 {masked} 日志已收集, DEBUG) except Exception as ex: log(f账号 {i1} 异常: {ex}, ERROR) traceback.print_exc() account_results.append(f❌ 账号 {i1}: 异常 - {str(ex)}) if i len(accounts) - 1: log(f等待3秒后处理下一个账号..., INFO) time.sleep(3) # 最终结果 log( * 50) result_msg f结果: {ok}/{len(accounts)} 成功 log(result_msg) log( * 50) # 构建通知标题 NOTIFY_TITLE f RoboMind 签到报告 - {result_msg} # 构建通知内容 content_lines [ f执行时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}, f执行结果: {result_msg}, , 账号详情:, ] content_lines.extend(account_results) content_lines.extend([ , 详细日志:, - * 40, ]) content_lines.extend(NOTIFY_CONTENT) # 限制日志长度避免通知内容过长 max_length 4000 # 大多数通知服务限制 final_content \n.join(content_lines) if len(final_content) max_length: # 截断日志部分 log_lines content_lines # 保留账号详情和日志的前面部分 header_lines content_lines[:len(account_results) 4] # 保留账号详情 log_part content_lines[len(account_results) 4:] # 日志部分 truncated_logs log_part[:80] # 保留前80行日志 final_content \n.join(header_lines) \n\n \n.join(truncated_logs) f\n\n... (日志共{len(log_part)}行截断显示前80行) # 发送通知 notify_enabled os.environ.get(ROBO_NOTIFY, true).lower() true if notify_enabled: log(正在发送通知..., INFO) log(f通知标题: {NOTIFY_TITLE[:50]}..., DEBUG) log(f通知内容长度: {len(final_content)} 字符, DEBUG) send_notify(NOTIFY_TITLE, final_content) else: log(通知已禁用 (ROBO_NOTIFYfalse), INFO) # 仍然写入本地日志 write_local_notification(NOTIFY_TITLE, final_content) if ok len(accounts): sys.exit(1) if __name__ __main__: main()