mirror/smallfawn-QLScriptPublic
1
0
Fork 0
mirror of https://github.com/smallfawn/QLScriptPublic.git synced 2025-12-17 23:35:02 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
1d345255b5b6bcc0fed907d23421ec514be1ac58
smallfawn-QLScriptPublic/backup/dwnc.py
smallfawn c850c0453a 脱密
2025-01-25 16:42:28 +08:00

1003 lines
44 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# !/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()
Reference in New Issue View Git Blame Copy Permalink