diff --git a/backup/dwnc.py b/backup/dwnc.py deleted file mode 100644 index 74620d9..0000000 --- a/backup/dwnc.py +++ /dev/null @@ -1,1002 +0,0 @@ -# !/usr/bin/python3 -# -*- coding: utf-8 -*- -# @Time : 2023/7/6 12:24 -# @Author : ziyou -# ------------------------------- -# cron "1 8,10,12,15,18,22 * * *" script-path=xxx.py,tag=匹配cron用 -# const $ = new Env('得物森林') -# 抓包 https://app.dewu.com/hacking-tree/v1/user/init 获取 sk x_auth_token user_agent -# 得物森林 -# export dewu_x_auth_token='Bearer ey**&Bearer ey**',多账号使用换行或& -# export dewu_sk='9MFyPaKgdQl*******************' 任意一个账号的 sk -# export dewu_user_agent='*****pp/5.25.0******' 同一 sk 对应的 user_agent,其中需含有得物版本号 版本号需大于 5.24.5 -# user_agent示例: Mozilla/5.0 (Linux; U; Android 10; zh-cn; Mi 10 Pro Build/QKQ1.191117.002) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/77.0.3865.120 MQQBrowser/11.0 Mobile Safari/537.36 COVC/045429 Mobile Safari/537.36/duapp/5.24.5(android;13) -# user_agent示例: DUApp/5.25.0 (com.siwuai.duapp; build:5.25.0.120; iOS 15.6.0) Alamofire/5.3.0 -# 如需关闭助力功能设置 export dewu_help_signal='False' -# 青龙拉取命令 ql raw https://raw.githubusercontent.com/q7q7q7q7q7q7q7/ziyou/main/%E5%BE%97%E7%89%A9%E6%A3%AE%E6%9E%97.py -# 第一个账号助力作者,其余账号依ck顺序助力 -# https://t.me/q7q7q7q7q7q7q7_ziyou - - -import os -import random -import re -import sys -import time -import requests -from urllib.parse import urlparse, parse_qs - -ck_list = [] -share_code_list = [] -author_share_code_list = [] -HELP_SIGNAL = 'True' -SK = '' -USER_AGENT = '' -__version__ = '1.0.2' -all_print_list = [] # 用于记录所有 myprint 输出的字符串 - - -# 用于记录所有 print 输出的字符串,暂时实现 print 函数的sep和end -def myprint(*args, sep=' ', end='\n', **kwargs): - global all_print_list - output = "" - # 构建输出字符串 - for index, arg in enumerate(args): - if index == len(args) - 1: - output += str(arg) - continue - output += str(arg) + sep - output = output + end - all_print_list.append(output) - # 调用内置的 print 函数打印字符串 - print(*args, sep=sep, end=end, **kwargs) - - -# 发送通知消息 -def send_notification_message(title): - try: - from notify import send # 导入青龙通知文件 - - send(title, ''.join(all_print_list)) - except Exception as e: - if e: - print('发送通知消息失败!') - - -# 加载环境变量 -def get_env(): - global ck_list - global HELP_SIGNAL - global SK - global USER_AGENT - env_str = os.getenv("dewu_x_auth_token") - if env_str: - ck_list += env_str.replace("&", "\n").split("\n") - env_str = os.getenv("dewu_help_signal") - if env_str: - HELP_SIGNAL = env_str - env_str = os.getenv("dewu_sk") - if env_str: - SK = env_str.strip() - env_str = os.getenv("dewu_user_agent") - if env_str: - USER_AGENT = env_str.strip() - - -def get_version_from_github(): - latest_version = '获取失败' - username = "q7q7q7q7q7q7q7" - repo = "ziyou" - filepath = "得物森林.py" - url_list = [ - f'https://raw.fgit.cf/{username}/{repo}/main/{filepath}', - f"https://ghproxy.com/https://raw.githubusercontent.com/{username}/{repo}/main/{filepath}", - ] - for url in url_list: - try: - response = requests.get(url, timeout=(10, 10)) - if response.status_code == 200: - response_text = response.text - version_regex = r"^__version__\s*=\s*[\'\"]([^\'\"]*)[\'\"]" - version_match = re.search(version_regex, response_text, - re.MULTILINE) - if version_match is not None and __version__: - latest_version = version_match.group(1) - break - except Exception as e: - if e: - pass - myprint(f'现在运行的版本是:{__version__},最新版本:{latest_version}', - flush=True) - - -# 下载作者的助力码 -def download_author_share_code(): - global author_share_code_list - try: - response = requests.get('https://netcut.cn/p/d3436822ba03c0c3') - _list = re.findall(r'"note_content":"(.*?)"', response.text) - if _list: - share_code_list = _list[0].split(r'\n') - author_share_code_list += share_code_list - except Exception as e: - if e: - pass - - -# 获得url params 中键为key的值 -def get_url_key_value(url, key): - parsed_url = urlparse(url) - query_params = parse_qs(parsed_url.query) - _dict = {k: v[0] if len(v) == 1 else v for k, v in query_params.items()} - key_value = _dict.get(key) - return key_value - - -class DeWu: - WATERTING_G: int = 40 # 每次浇水克数 - REMAINING_G: int = 1800 # 最后浇水剩余不超过的克数 - - def __init__(self, x_auth_token, index, waterting_g=WATERTING_G, - remaining_g=REMAINING_G): - self.index = index - self.waterting_g = waterting_g # 每次浇水克数 - self.remaining_g = remaining_g # 最后浇水剩余不超过的克数 - self.session = requests.Session() - app_version = re.findall(r'pp/([0-9]+\.[0-9]+\.[0-9]+)', USER_AGENT)[0] - self.headers = {'appVersion': app_version, - 'User-Agent': USER_AGENT, - 'x-auth-token': x_auth_token, - 'uuid': '0000000000000000', - 'SK': SK, } - self.tree_id = 0 # 树的id - self.tasks_completed_number = 0 # 任务完成数 - self.cumulative_tasks_list = [] # 累计计任务列表 - self.tasks_dict_list = [] # 任务字典列表 - self.is_team_tree = False # 是否是团队树 - - # 种树奖品 - def tree_info(self): - url = 'https://app.dewu.com/hacking-tree/v1/user/target/info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200: - name = response_dict.get('data').get('name') - level = response_dict.get('data').get('level') - return name, level - myprint(response_dict.get('msg')) - return '', '' - - # 判断是否是团队树 - def determine_whether_is_team_tree(self): - url = 'https://app.dewu.com/hacking-tree/v1/team/info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('data').get('show') is True and response_dict.get( - 'data').get('teamTreeId'): - self.is_team_tree = True - - # 领潮金币签到 - def check_in(self): - url = 'https://app.dewu.com/hacking-game-center/v1/sign/sign' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200: - myprint(f"签到成功!") - return - myprint(f"签到失败! {response_dict.get('msg')}") - - # 水滴7天签到 - def droplet_check_in(self): - url = 'https://app.dewu.com/hacking-tree/v1/sign/sign_in' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200: - # 暂时设置,看看礼盒是什么先 - myprint(f"签到成功,获得{response_dict.get('data').get('Num')}g水滴") - return - myprint(f"签到失败! {response_dict.get('msg')}") - - # 领取气泡水滴 - def receive_droplet_extra(self): - temporary_number = -1 # 用于判断两次浇水,奖励是否有变化 - countdown_time = 0 - receive_signal = False - for _ in range(50): - url = 'https://app.dewu.com/hacking-tree/v1/droplet-extra/info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"获取气泡水滴信息失败! {response_dict}") - return - data = response_dict.get('data') - receivable = data.get('receivable') - if receivable is True: # 判断今天是否可领取 - if data.get('dailyExtra'): # 第一次领取时 - water_droplet_number = data.get('dailyExtra').get( - 'totalDroplet') - else: # 第二次领取时 - water_droplet_number = data.get('onlineExtra').get( - 'totalDroplet') - # 如果二者相等,说明浇水成功 但奖励没变化 不再浇水 直接领取 或者 接受到直接领取信号 - if temporary_number == water_droplet_number or receive_signal: - myprint(f"当前可领取气泡水滴{water_droplet_number}g") - url = 'https://app.dewu.com/hacking-tree/v1/droplet-extra/receive' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - countdown_time += 60 - if countdown_time > 60: # 已经等待过60s,仍未领取成功,退出 - myprint(f"领取气泡水滴失败! {response_dict}") - return - myprint(f'等待{countdown_time}秒后领取') - time.sleep(countdown_time) - continue - myprint( - f"领取气泡水滴成功! 获得{response_dict.get('data').get('totalDroplet')}g水滴") - countdown_time = 0 # 领取成功,重置等待时间 - continue - temporary_number = water_droplet_number - myprint(f'当前气泡水滴{water_droplet_number}g,未满,开始浇水') - if not self.waterting(): # 浇水失败 - receive_signal = True # 给出直接领取信号 - time.sleep(0.5) - continue # 浇水成功后查询信息 - # 今天不可领取了,退出 - water_droplet_number = response_dict.get('data').get( - 'dailyExtra').get('totalDroplet') - myprint( - f"{response_dict.get('data').get('dailyExtra').get('popTitle')}," - f"已经积攒{water_droplet_number}g水滴!") - return - - # 浇水充满气泡水滴 - def waterting_droplet_extra(self): - while True: - url = 'https://app.dewu.com/hacking-tree/v1/droplet-extra/info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - count = response_dict.get('data').get('dailyExtra').get('times') - if not count: - myprint( - f"气泡水滴已充满,明日可领取{response_dict.get('data').get('dailyExtra').get('totalDroplet')}g") - return - for _ in range(count): - if not self.waterting(): # 无法浇水时退出 - return - time.sleep(0.5) - - # 领取木桶水滴,200秒满一次,每天领取3次 - def receive_bucket_droplet(self): - url = 'https://app.dewu.com/hacking-tree/v1/droplet/get_generate_droplet' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"领取木桶水滴失败! {response_dict}") - return - myprint( - f"领取木桶水滴成功! 获得{response_dict.get('data').get('droplet')}g水滴") - - # 判断木桶水滴是否可以领取 - def judging_bucket_droplet(self): - url = 'https://app.dewu.com/hacking-tree/v1/droplet/generate_info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('data').get('currentDroplet') == 100: - myprint( - f"今天已领取木桶水滴{response_dict.get('data').get('getTimes')}次") - self.receive_bucket_droplet() - return True - return False - - # 获取助力码 - def get_shared_code(self): - url = 'https://app.dewu.com/hacking-tree/v1/keyword/gen' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"获取助力码失败! {response_dict}") - return - keyword_desc = response_dict.get('data').get('keywordDesc').replace( - '\n', '') - myprint(f"获取助力码成功! {keyword_desc}") - - # 获得当前水滴数 - def get_droplet_number(self): - url = 'https://app.dewu.com/hacking-tree/v1/user/init' - _json = {"keyword": ""} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - data = response_dict.get('data') - if data: - droplet_number = data.get('droplet') - return droplet_number - myprint(f'获得当前水滴数出错 {response_dict}') - return '获取失败' - - # 领取累计任务奖励 - def receive_cumulative_tasks_reward(self, condition): - url = 'https://app.dewu.com/hacking-tree/v1/task/extra' - _json = {'condition': condition} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"领取累计任务奖励失败! {response_dict}") - return - myprint( - f"领取累计任务奖励成功! 获得{response_dict.get('data').get('num')}g水滴") - - # 领取任务奖励 - def receive_task_reward(self, classify, task_id, task_type): - time.sleep(0.2) - url = 'https://app.dewu.com/hacking-tree/v1/task/receive' - if task_type in [251, ]: - _json = {'classify': classify, 'taskId': task_id, 'completeFlag': 1} - else: - _json = {'classify': classify, 'taskId': task_id} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"领取任务奖励失败! {response_dict}") - return - myprint( - f"领取任务奖励成功! 获得{response_dict.get('data').get('num')}g水滴") - - # 领取浇水奖励 - def receive_watering_reward(self): - url = 'https://app.dewu.com/hacking-tree/v1/tree/get_watering_reward' - _json = {'promote': ''} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"领取浇水奖励失败! {response_dict}") - return - myprint( - f"领取浇水奖励成功! 获得{response_dict.get('data').get('currentWateringReward').get('rewardNum')}g水滴") - - # 领取等级奖励 - def receive_level_reward(self): - for _ in range(20): - url = 'https://app.dewu.com/hacking-tree/v1/tree/get_level_reward' - _json = {'promote': ''} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200 or response_dict.get( - 'data') is None: - myprint(f"领取等级奖励失败! {response_dict.get('msg')}") - return - level = response_dict.get('data').get('levelReward').get( - 'showLevel') - 1 - reward_num = response_dict.get('data').get( - 'currentLevelReward').get('rewardNum') - myprint(f"领取{level}级奖励成功! 获得{reward_num}g水滴") - if response_dict.get('data').get('levelReward').get( - 'isComplete') is False: - return - time.sleep(1) - - # 浇水 - def waterting(self): - if self.is_team_tree is True: # 如果是团队树,使用团队浇水 - return self.team_waterting() - url = 'https://app.dewu.com/hacking-tree/v1/tree/watering' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"浇水失败! {response_dict}") - return False - myprint(f"成功浇水{self.waterting_g}g") - if response_dict.get('data').get('nextWateringTimes') == 0: - myprint('开始领取浇水奖励') - time.sleep(1) - self.receive_watering_reward() - return True - - # 团队浇水 - def team_waterting(self): - url = 'https://app.dewu.com/hacking-tree/v1/team/tree/watering' - _json = {"teamTreeId": self.tree_id} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"浇水失败! {response_dict}") - return False - myprint(f"成功浇水{self.waterting_g}g") - if response_dict.get('data').get('nextWateringTimes') == 0: - myprint('开始领取浇水奖励') - time.sleep(1) - self.receive_watering_reward() - return True - - # 多次执行浇水,领取浇水奖励 - def execute_receive_watering_reward(self): - for _ in range(20): - url = 'https://app.dewu.com/hacking-tree/v1/tree/get_tree_info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"获取种树进度失败! {response_dict}") - return - count = response_dict.get('data').get( - 'nextWateringTimes') # 获取浇水奖励还需要的浇水次数 - if response_dict.get('data').get( - 'wateringReward') is None or count <= 0: # 没有奖励时退出 - return - for _ in range(count): - if not self.waterting(): # 无法浇水时退出 - return - time.sleep(0.5) - - # 浇水直到少于 指定克数 - def waterting_until_less_than(self): - droplet_number = self.get_droplet_number() - if droplet_number > self.remaining_g: - count = int((droplet_number - self.remaining_g) / self.waterting_g) - for _ in range(count + 1): - if not self.waterting(): # 无法浇水时退出 - return - time.sleep(0.5) - - # 提交任务完成状态 - def submit_task_completion_status(self, _json): - url = 'https://app.dewu.com/hacking-task/v1/task/commit' - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200: - return True - return False - - # 获取任务列表 - def get_task_list(self): - url = 'https://app.dewu.com/hacking-tree/v1/task/list' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200: - self.tasks_completed_number = response_dict.get('data').get( - 'userStep') # 任务完成数量 - self.cumulative_tasks_list = response_dict.get('data').get( - 'extraAwardList') # 累计任务列表 - self.tasks_dict_list = response_dict.get('data').get( - 'taskList') # 任务列表 - return True - - # 水滴大放送任务步骤1 - def task_obtain(self, task_id, task_type): - url = 'https://app.dewu.com/hacking-task/v1/task/obtain' - _json = {'taskId': task_id, 'taskType': task_type} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200 and response_dict.get( - 'status') == 200: - return True - return False - - # 浏览任务开始 且等待16s TaskType有变化 浏览15s会场会变成16 - def task_commit_pre(self, _json): - url = 'https://app.dewu.com/hacking-task/v1/task/pre_commit' - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200 and response_dict.get( - 'status') == 200: - return True - return False - - # 执行任务 - def execute_task(self): - self.get_task_list() # 刷新任务列表 - for tasks_dict in self.tasks_dict_list: - if tasks_dict.get('isReceiveReward') is True: # 今天不能进行操作了,跳过 - continue - if tasks_dict.get('rewardCount') >= 3000: # 获取水滴超过3000的,需要下单,跳过 - continue - # 'taskId' 任务ID - # 'taskName' 任务名字 - # 'isComplete' 是否未完成 - # 'isReceiveReward' 完成后是否领取奖励 - # 'taskType'任务类型 - # 'rewardCount' 完成任务所获得的奖励水滴 - # 'isObtain' 是否完成任务前置要求 - # 'jumpUrl' 是否完成任务前置要求 - classify = tasks_dict.get('classify') - task_id = tasks_dict.get('taskId') - task_type = tasks_dict.get('taskType') - task_name = tasks_dict.get('taskName') - btd = get_url_key_value(tasks_dict.get('jumpUrl'), 'btd') - btd = int(btd) if btd else btd # 如果bid存在 转换为整数类型 - spu_id = get_url_key_value(tasks_dict.get('jumpUrl'), 'spuId') - spu_id = int(spu_id) if spu_id else spu_id # 如果spuId存在 转换为整数类型 - - if tasks_dict.get('isComplete') is True: # 可以直接领取奖励的 - if task_name == '领40g水滴值' and not tasks_dict.get( - 'receivable'): # 如果该值不存在,说明已经领过40g水滴了 - continue - myprint(f'开始任务:{task_name}') - self.receive_task_reward(classify, task_id, task_type) - continue - - myprint(f'★开始任务:{task_name}') - if task_name == '完成一次签到': # 签到 - self.check_in() - data = {'taskId': tasks_dict['taskId'], - 'taskType': str(tasks_dict['taskType'])} - if self.submit_task_completion_status(data): - self.receive_task_reward(classify, task_id, - task_type) # 领取奖励 - continue - - if task_name == '领40g水滴值': # 每天8点/12点/18点/22点 领40g水滴 - self.receive_task_reward(classify, task_id, task_type) # 领取奖励 - continue - - if task_name == '收集一次水滴生产': - if self.judging_bucket_droplet(): - self.receive_task_reward(classify, task_id, - task_type) # 领取奖励 - else: - myprint('当前木桶水滴未达到100g,下次来完成任务吧!') - continue - - if task_name == '浏览【我】的右上角星愿森林入口': - _json = _json = {"action": task_id} - url = 'https://app.dewu.com/hacking-tree/v1/user/report_action' - response = self.session.post(url, headers=self.headers, - json=_json) # 提交完成状态 - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200: - self.receive_task_reward(classify, task_id, - task_type) # 领取奖励 - continue - - if any(re.match(pattern, task_name) for pattern in - ['参与1次上上签活动', '从桌面组件访问许愿树', - '参与1次拆盲盒', '去.*']): - _json = _json = {'taskId': task_id, 'taskType': str(task_type)} - self.submit_task_completion_status(_json) # 提交完成状态 - self.receive_task_reward(classify, task_id, task_type) # 领取奖励 - continue - - if any(re.match(pattern, task_name) for pattern in ['.*收藏.*']): - _json = _json = {'taskId': task_id, 'taskType': str(task_type), - 'btd': btd, 'spuId': spu_id} - self.submit_task_completion_status(_json) # 提交完成状态 - self.receive_task_reward(classify, task_id, task_type) # 领取奖励 - continue - - if any(re.match(pattern, task_name) for pattern in - ['.*订阅.*', '.*逛一逛.*', '逛逛.*活动']): - _json = _json = {'taskId': task_id, 'taskType': str(task_type), - 'btd': btd} - self.submit_task_completion_status(_json) # 提交完成状态 - self.receive_task_reward(classify, task_id, task_type) # 领取奖励 - continue - - if any(re.match(pattern, task_name) for pattern in - ['.*逛逛.*', '浏览.*s']): - _json = {'taskId': task_id, 'taskType': task_type, 'btd': btd} - if self.task_commit_pre(_json): - myprint(f'等待16秒') - time.sleep(16) - _json = {'taskId': task_id, 'taskType': str(task_type), - 'activityType': None, 'activityId': None, - 'taskSetId': None, 'venueCode': None, - 'venueUnitStyle': None, 'taskScene': None, - 'btd': btd} - self.submit_task_completion_status(_json) # 提交完成状态 - self.receive_task_reward(classify, task_id, - task_type) # 领取奖励 - continue - - if any(re.match(pattern, task_name) for pattern in ['.*晒图.*']): - _json = {'taskId': task_id, 'taskType': task_type} - if self.task_commit_pre(_json): - myprint(f'等待16秒') - time.sleep(16) - _json = {'taskId': task_id, 'taskType': str(task_type), - 'activityType': None, 'activityId': None, - 'taskSetId': None, 'venueCode': None, - 'venueUnitStyle': None, 'taskScene': None} - self.submit_task_completion_status(_json) # 提交完成状态 - self.receive_task_reward(classify, task_id, - task_type) # 领取奖励 - continue - - if task_name == '完成五次浇灌': - count = tasks_dict.get('total') - tasks_dict.get( - 'curStep') # 还需要浇水的次数=要浇水的次数-以浇水的次数 - if self.get_droplet_number() < (count * self.waterting_g): - myprint(f'当前水滴不足以完成任务,跳过') - continue - for _ in range(count): - time.sleep(0.5) - if not self.waterting(): # 无法浇水时退出 - break - else: - _json = {'taskId': tasks_dict['taskId'], - 'taskType': str(tasks_dict['taskType'])} - if self.submit_task_completion_status(_json): - self.receive_task_reward(classify, task_id, - task_type) # 领取奖励 - continue - - if any(re.match(pattern, task_name) for pattern in - ['.*专场', '.*水滴大放送']): - if self.task_obtain(task_id, task_type): - _json = {'taskId': task_id, 'taskType': 16} - if self.task_commit_pre(_json): - myprint(f'等待16秒') - time.sleep(16) - _json = {'taskId': task_id, 'taskType': str(task_type)} - self.submit_task_completion_status(_json) # 提交完成状态 - self.receive_task_reward(classify, task_id, - task_type) # 领取奖励 - continue - myprint(f'该任务暂时无法处理,请提交日志给作者! {tasks_dict}') - - # 执行累计任务 - def execute_cumulative_task(self): - self.get_task_list() # 刷新任务列表 - for task in self.cumulative_tasks_list: - if task.get('status') == 1: - myprint(f'开始领取累计任务数达{task.get("condition")}个的奖励') - self.receive_cumulative_tasks_reward(task.get('condition')) - time.sleep(1) - - # 水滴投资 - def droplet_invest(self): - url = 'https://app.dewu.com/hacking-tree/v1/invest/info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('data').get('isToday') is False: # 可领取 - self.received_droplet_invest() - else: - myprint('今日已领取过水滴投资奖励了') - time.sleep(2) - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - if response_dict.get('data').get('triggered') is True: # 可投资 - url = 'https://app.dewu.com/hacking-tree/v1/invest/commit' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') == 200 and response_dict.get( - 'status') == 200: - myprint('水滴投资成功,水滴-100g') - return - if response_dict.get("msg") == '水滴不够了': - myprint( - f'水滴投资失败,剩余水滴需超过100g,{response_dict.get("msg")}') - return - myprint(f'水滴投资出错! {response_dict}') - return - else: - myprint('今日已经水滴投资过了!') - - # 领取水滴投资 - def received_droplet_invest(self): - url = 'https://app.dewu.com/hacking-tree/v1/invest/receive' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - profit = response_dict.get('data').get('profit') - myprint(f"领取水滴投资成功! 获得{profit}g水滴") - - # 获取助力码 - def get_share_code(self) -> str: - url = 'https://app.dewu.com/hacking-tree/v1/keyword/gen' - response = self.session.post(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('status') == 200: - keyword = response_dict.get('data').get('keyword') - keyword = re.findall('œ(.*?)œ ', keyword) - if keyword: - myprint(f'获取助力码成功 {keyword[0]}') - return keyword[0] - myprint(f'获取助力码失败! {response_dict}') - - # 助力 - def help_user(self): - if HELP_SIGNAL == 'False': - myprint('助力功能已设置为关闭') - return - url = 'https://app.dewu.com/hacking-tree/v1/user/init' - if self.index == 0: - for share_code in author_share_code_list: - _json = {'keyword': share_code} - response = self.session.post(url, headers=self.headers, - json=_json) - response_dict = response.json() - data = response_dict.get('data') - if not data: - continue - invite_res = data.get('inviteRes') - if any(re.match(pattern, invite_res) for pattern in - ['助力成功', '助力失败,今日已助力过了']): - myprint(f'开始助力 {share_code}', end=' ') - myprint(invite_res) - return - time.sleep(random.randint(20, 30) / 10) - for share_code in share_code_list: - myprint(f'开始助力 {share_code}', end=' ') - _json = {'keyword': share_code} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - data = response_dict.get('data') - if not data: - print(f'助力异常 {response_dict}') - continue - invite_res = data.get('inviteRes') - myprint(invite_res) - if any(re.match(pattern, invite_res) for pattern in - ['助力成功', '助力失败,今日已助力过了']): - return - time.sleep(random.randint(20, 30) / 10) - return - - # 领取助力奖励 - def receive_help_reward(self): - url = 'https://app.dewu.com/hacking-tree/v1/invite/list' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('status') == 200: - reward_list = response_dict.get('data').get('list') - if not reward_list: - return - for reward in reward_list: - if reward.get('status') != 0: # 为0时才可以领取 - continue - invitee_user_id = reward.get('inviteeUserId') - url = 'https://app.dewu.com/hacking-tree/v1/invite/reward' - _json = {'inviteeUserId': invitee_user_id} - response = self.session.post(url, headers=self.headers, - json=_json) - response_dict = response.json() - if response_dict.get('status') == 200: - droplet = response_dict.get('data').get('droplet') - myprint(f'获得{droplet}g水滴') - # myprint(response_dict) - continue - myprint(f'领取助力奖励出现未知错误! {response_dict}') - return - myprint(f'获取助力列表出现未知错误! {response_dict}') - return - - # 领取合种上线奖励 - def receive_hybrid_online_reward(self): - url = f'https://app.dewu.com/hacking-tree/v1/team/sign/list?teamTreeId={self.tree_id}' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('data') is None: - return - reward_list = response_dict.get('data', {}).get('list') - if not reward_list: - return - for reward in reward_list: - # 如果任务完成但是未领取 - if reward.get('isComplete') is True and reward.get( - 'isReceive') is False: - url = 'https://app.dewu.com/hacking-tree/v1/team/sign/receive' - _json = {"teamTreeId": self.tree_id, "day": reward.get('day')} - response = self.session.post(url, headers=self.headers, - json=_json) - response_dict = response.json() - if response_dict.get('data').get('isOk') is True: - myprint(f'获得{reward.get("num")}g水滴') - continue - myprint(f'领取合种上线奖励出现未知错误! {response_dict}') - return - - # 领取空中水滴 - def receive_air_drop(self): - for _ in range(2): - url = 'https://app.dewu.com/hacking-tree/v1/droplet/air_drop_receive' - _json = {"clickCount": 20, "time": int(time.time())} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - data = response_dict.get('data') - if data is not None and data.get('isOk') is True: - myprint(f'获得{response_dict.get("data").get("droplet")}g水滴') - time.sleep(1) - continue - break - - # 点击8个商品获得水滴 - def click_product(self): - product_list = [{"spuId": 3030863, "timestamp": 1690790735382, - "sign": "2889b16b3077c5719288d105a14ffa1e"}, - {"spuId": 4673547, "timestamp": 1690790691956, - "sign": "cc3cc95253d29a03fc6e79bfe2200143"}, - {"spuId": 1502607, "timestamp": 1690791565022, - "sign": "04951eac012785ccb2600703a92c037b"}, - {"spuId": 2960612, "timestamp": 1690791593097, - "sign": "fb667d45bc3950a7beb6e3fa0fc05089"}, - {"spuId": 3143593, "timestamp": 1690791613243, - "sign": "82b9fda61be79f7b8833087508d6abe2"}, - {"spuId": 3067054, "timestamp": 1690791639606, - "sign": "2808f3c7cf2ededea17d3f70a2dc565d"}, - {"spuId": 4448037, "timestamp": 1690791663078, - "sign": "335bc519ee9183c086beb009adf93738"}, - {"spuId": 3237561, "timestamp": 1690791692553, - "sign": "5c113b9203a510b7068b3cd0f6b7c25e"}, - {"spuId": 3938180, "timestamp": 1690792014889, - "sign": "3841c0272443dcbbab0bcb21c94c6262"}, ] - for product in product_list: - url = 'https://app.dewu.com/hacking-tree/v1/product/spu' - _json = product - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('data') is None: - myprint(f'今天已经完成过该任务了!') - return - if response_dict.get('data', {}).get('isReceived') is True: - myprint( - f'获得{response_dict.get("data").get("dropLetReward")}g水滴') - return - time.sleep(1) - - # 领取发现水滴 - def receive_discover_droplet(self): - while True: - url = 'https://app.dewu.com/hacking-tree/v1/product/task/seek-receive' - _json = {"sign": "9888433e6d10b514e5b5be4305d123f0", - "timestamp": int(time.time() * 1000)} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - myprint(response_dict) - pass - - # 领取品牌特惠奖励 - def receive_brand_specials(self): - url = 'https://app.dewu.com/hacking-ad/v1/activity/list?bizId=tree' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('data') is None: - myprint('当前没有可以完成的任务!') - return - if response_dict.get('data').get('list') is None: - myprint('当前没有可以完成的任务!') - return - ad_list = response_dict.get('data').get('list') - for ad in ad_list: - if ad.get('isReceived') is True: - continue - aid = ad.get('id') - url = 'https://app.dewu.com/hacking-ad/v1/activity/receive' - _json = {"bizId": "tree", "aid": aid} - response = self.session.post(url, headers=self.headers, json=_json) - response_dict = response.json() - # myprint(response_dict) - myprint(f'获得{response_dict.get("data").get("award")}g水滴') - time.sleep(1) - - # 获取种树进度 - def get_tree_planting_progress(self): - url = 'https://app.dewu.com/hacking-tree/v1/tree/get_tree_info' - response = self.session.get(url, headers=self.headers) - response_dict = response.json() - # myprint(response_dict) - if response_dict.get('code') != 200: - myprint(f"获取种树进度失败! {response_dict}") - return - self.tree_id = response_dict.get('data').get('treeId') - level = response_dict.get('data').get('level') - current_level_need_watering_droplet = response_dict.get('data').get( - 'currentLevelNeedWateringDroplet') - user_watering_droplet = response_dict.get('data').get( - 'userWateringDroplet') - myprint( - f"种树进度: {level}级 {user_watering_droplet}/{current_level_need_watering_droplet}") - - def main(self): - character = '★★' - name, level = self.tree_info() - droplet_number = self.get_droplet_number() - if not (name and level and droplet_number): - myprint("请求数据异常!") - return - myprint(f'目标:{name}') - myprint(f'剩余水滴:{droplet_number}') - self.determine_whether_is_team_tree() # 判断是否是团队树 - self.get_tree_planting_progress() # 获取种树进度 - myprint(f'{character}开始签到') - self.droplet_check_in() # 签到 - myprint(f'{character}开始领取气泡水滴') - self.receive_droplet_extra() - myprint(f'{character}开始完成每日任务') - self.execute_task() - myprint(f'{character}开始领取累计任务奖励') - self.execute_cumulative_task() - myprint(f'{character}开始领取木桶水滴') - self.judging_bucket_droplet() - myprint(f'{character}开始多次执行浇水,领取浇水奖励') - self.execute_receive_watering_reward() - myprint(f'{character}开始浇水充满气泡水滴') - self.waterting_droplet_extra() - myprint(f'{character}开始领取合种上线奖励') - self.receive_hybrid_online_reward() - myprint(f'{character}开始领取空中水滴') - self.receive_air_drop() - myprint(f'{character}开始进行水滴投资') - self.droplet_invest() - myprint(f'{character}开始点击8个商品获得水滴') - self.click_product() - myprint(f'{character}开始领取品牌特惠奖励') - self.receive_brand_specials() - myprint(f'{character}开始进行助力') - self.help_user() - myprint(f'{character}开始领取助力奖励') - self.receive_help_reward() - myprint(f'{character}开始领取等级奖励') - self.receive_level_reward() - myprint(f'{character}开始进行浇水直到少于{self.remaining_g}g') - self.waterting_until_less_than() - myprint(f'剩余水滴:{self.get_droplet_number()}') - time.sleep(1) - self.get_tree_planting_progress() # 获取种树进度 - - -# 主程序 -def main(): - get_version_from_github() - get_env() - if not ck_list: - myprint('没有获取到账号!') - return - if not SK: - myprint('dewu_sk 未填写!') - return - if not USER_AGENT: - myprint('dewu_user_agent 未填写!') - return - _list = re.findall(r'pp/([0-9]+\.[0-9]+\.[0-9]+)', USER_AGENT) - if not _list: - myprint('dewu_user_agent 中无法匹配到得物app版本号,重新抓个试试吧') - return - download_author_share_code() - myprint(f'获取到{len(ck_list)}个账号!') - if HELP_SIGNAL == 'True': - myprint('开始获取所有账号助力码') - for index, ck in enumerate(ck_list): - myprint(f'第{index + 1}个账号:', end='') - share_code_list.append(DeWu(ck, index).get_share_code()) - time.sleep(0.5) - for index, ck in enumerate(ck_list): - myprint(f'*****第{index + 1}个账号*****') - DeWu(ck, index).main() - myprint('') - send_notification_message(title='得物森林') # 发送通知消息 - - -if __name__ == '__main__': - main() - sys.exit() diff --git a/dwnc.js b/得物.js similarity index 100% rename from dwnc.js rename to 得物.js