mirror/XiaoGe-LiBai-yangmao
1
0
Fork 0
mirror of https://github.com/XiaoGe-LiBai/yangmao.git synced 2025-12-20 00:35:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

首次添加

Browse Source 
首次添加
This commit is contained in:
XiaoGe-LiBai
2023-08-11 22:48:08 +08:00
parent 088286d605
commit dbcf32c069
2 changed files with 1058 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

811
得物森林.py Normal file
View File

@@ -0,0 +1,811 @@
# !/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('得物森林')
# 抓包获取 x_auth_token
# 得物森林
# export dewu_x_auth_token='Bearer ey**&Bearer ey**',多账号使用换行或&
# 青龙拉取命令 ql raw https://raw.githubusercontent.com/q7q7q7q7q7q7q7/ziyou/main/%E5%BE%97%E7%89%A9%E6%A3%AE%E6%9E%97.py
# 第一个账号助力作者，其余账号依ck顺序助力
import os
import random
import re
import sys
import time
import requests
from urllib.parse import urlparse, parse_qs
X_AUTH_TOKEN = []
SHARE_CODE_LIST = []
AUTHOR_SHARE_CODE_LIST = []
# X_AUTH_TOKEN = ['Bearer eyJhbGciOi*******',
# 'Bearer eyJhbGciOi*******', ]
dewu_x_auth_token = os.getenv("dewu_x_auth_token")
if dewu_x_auth_token:
X_AUTH_TOKEN += dewu_x_auth_token.replace("&", "\n").split("\n")
# 下载作者的助力码
def download_author_share_code():
global AUTHOR_SHARE_CODE_LIST
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
# 获得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 # 最后浇水剩余不超过的克数
User_Agent = "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1062.0 Safari/536.3"
def __init__(self, x_auth_token, index, waterting_g=WATERTING_G, remaining_g=REMAINING_G, user_agent=User_Agent):
self.index = index
self.waterting_g = waterting_g # 每次浇水克数
self.remaining_g = remaining_g # 最后浇水剩余不超过的克数
self.session = requests.Session()
self.headers = {'SK': '', 'User-Agent': user_agent, 'x-auth-token': x_auth_token}
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()
# print(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
print(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()
# print(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()
# print(response_dict)
if response_dict.get('code') == 200:
print(f"签到成功！")
return
print(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()
# print(response_dict)
if response_dict.get('code') == 200:
# 暂时设置，看看礼盒是什么先
print(f"签到成功,获得{response_dict.get('data').get('Num')}g水滴")
return
print(f"签到失败！ {response_dict.get('msg')}")
# 领取气泡水滴
def receive_droplet_extra(self):
temporary_number = 0
countdown_time = 0
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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"获取气泡水滴信息失败! {response_dict}")
return
if response_dict.get('data').get('receivable') is True: # 判断是否能领取
if response_dict.get('data').get('dailyExtra'): # 第一次领取时
water_droplet_number = response_dict.get('data').get('dailyExtra').get('totalDroplet')
else: # 第二次领取时
water_droplet_number = response_dict.get('data').get('onlineExtra').get('totalDroplet')
# 如果二者不相等，说明浇水成功 但奖励有变化 继续浇水
if temporary_number != water_droplet_number: # 如果二者相等，说明浇水成功 但奖励没变化 不再浇水 直接领取
temporary_number = water_droplet_number
if water_droplet_number < 60:
print(f'当前气泡水滴{water_droplet_number}g，未满，开始浇水')
if self.waterting(): # 成功浇水 继续 否则 直接领取
time.sleep(0.5)
continue # 浇水成功后查询信息
print(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()
# print(response_dict)
if response_dict.get('code') != 200:
countdown_time += 60
if countdown_time > 60: # 已经等待过60s，任为领取成功，退出
print(f"领取气泡水滴失败! {response_dict}")
return
print(f'等待{countdown_time}秒后领取')
time.sleep(countdown_time)
continue
print(f"领取气泡水滴成功! 获得{response_dict.get('data').get('totalDroplet')}g水滴")
countdown_time = 0 # 领取成功，重置等待时间
else: # 今天不可领取了，退出
water_droplet_number = response_dict.get('data').get('dailyExtra').get('totalDroplet')
print(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()
# print(response_dict)
count = response_dict.get('data').get('dailyExtra').get('times')
if not count:
print(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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"领取木桶水滴失败! {response_dict}")
return
print(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()
# print(response_dict)
if response_dict.get('data').get('currentDroplet') == 100:
print(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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"获取助力码失败! {response_dict}")
return
keyword_desc = response_dict.get('data').get('keywordDesc').replace('\n', '')
print(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()
# print(response_dict)
droplet_number = response_dict.get('data').get('droplet')
return droplet_number
# 领取累计任务奖励
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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"领取累计任务奖励失败! {response_dict}")
return
print(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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"领取任务奖励失败! {response_dict}")
return
print(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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"领取浇水奖励失败! {response_dict}")
return
print(f"领取浇水奖励成功! 获得{response_dict.get('data').get('currentWateringReward').get('rewardNum')}g水滴")
# 领取等级奖励
def receive_level_reward(self):
while True:
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()
# print(response_dict)
if response_dict.get('code') != 200 or response_dict.get('data') is None:
print(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')
print(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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"浇水失败! {response_dict}")
return False
print(f"成功浇水{self.waterting_g}g")
if response_dict.get('data').get('nextWateringTimes') == 0:
print('开始领取浇水奖励')
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()
# print(response_dict)
if response_dict.get('code') != 200:
print(f"浇水失败! {response_dict}")
return False
print(f"成功浇水{self.waterting_g}g")
if response_dict.get('data').get('nextWateringTimes') == 0:
print('开始领取浇水奖励')
time.sleep(1)
self.receive_watering_reward()
return True
# 多次执行浇水，领取浇水奖励
def execute_receive_watering_reward(self):
while True:
url = 'https://app.dewu.com/hacking-tree/v1/tree/get_tree_info'
response = self.session.get(url, headers=self.headers)
response_dict = response.json()
# print(response_dict)
if response_dict.get('code') != 200:
print(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()
# print(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()
# print(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()
# print(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()
# print(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存在 转换为整数类型
if tasks_dict.get('isComplete') is True: # 可以直接领取奖励的
if task_name == '领40g水滴值' and not tasks_dict.get('receivable'): # 如果该值不存在，说明已经领过40g水滴了
continue
print(f'开始任务：{task_name}')
self.receive_task_reward(classify, task_id, task_type)
continue
print(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:
print('当前木桶水滴未达到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()
# print(response_dict)
if response_dict.get('code') == 200:
self.receive_task_reward(classify, task_id, task_type) # 领取奖励
continue
if task_name in ['去0元抽奖参与抽游戏皮肤', '参与1次上上签活动', '从桌面组件访问许愿树',
'去95分App逛潮奢尖货', '参与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': 0}
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 ['.*逛逛.*', '浏览.*15s']):
_json = {'taskId': task_id, 'taskType': task_type, 'btd': btd}
if self.task_commit_pre(_json):
print(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):
print(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):
print(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):
print(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
print(f'该任务暂时无法处理，请提交日志给作者！ {tasks_dict}')
# 执行累计任务
def execute_cumulative_task(self):
self.get_task_list() # 刷新任务列表
for task in self.cumulative_tasks_list:
if task.get('status') == 1:
print(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()
# print(response_dict)
if response_dict.get('data').get('isToday') is False: # 可领取
self.received_droplet_invest()
else:
print('今日已领取过水滴投资奖励了')
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()
# print(response_dict)
if response_dict.get('code') == 200 and response_dict.get('status') == 200:
print('水滴投资成功，水滴-100g')
return
if response_dict.get("msg") == '水滴不够了':
print(f'水滴投资失败，剩余水滴需超过100g，{response_dict.get("msg")}')
return
print(f'水滴投资出错！ {response_dict}')
return
else:
print('今日已经水滴投资过了！')
# 领取水滴投资
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()
# print(response_dict)
profit = response_dict.get('data').get('profit')
print(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()
# print(response_dict)
if response_dict.get('status') == 200:
keyword = response_dict.get('data').get('keyword')
keyword = re.findall('œ(.*?)œ ', keyword)
if keyword:
print(f'获取助力码成功 {keyword[0]}')
return keyword[0]
print(f'获取助力码失败！ {response_dict}')
# 助力
def help_user(self):
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()
invite_res = response_dict.get('data').get('inviteRes')
if any(re.match(pattern, invite_res) for pattern in ['助力成功', '助力失败，今日已助力过了']):
print(f'开始助力 {share_code}', end=' ')
print(invite_res)
return
time.sleep(random.randint(20, 30) / 10)
for share_code in SHARE_CODE_LIST:
print(f'开始助力 {share_code}', end=' ')
_json = {'keyword': share_code}
response = self.session.post(url, headers=self.headers, json=_json)
response_dict = response.json()
# print(response_dict)
invite_res = response_dict.get('data').get('inviteRes')
print(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()
# print(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')
print(f'获得{droplet}g水滴')
# print(response_dict)
continue
print(f'领取助力奖励出现未知错误！ {response_dict}')
return
print(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()
# print(response_dict)
if response_dict.get('data') is None:
return
reward_list = response_dict.get('data', {}).get('list')
if reward_list:
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:
print(f'获得{reward.get("num")}g水滴')
continue
print(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()
# print(response_dict)
data = response_dict.get('data')
if data is not None and data.get('isOk') is True:
print(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()
# print(response_dict)
if response_dict.get('data') is None:
print(f'今天已经完成过该任务了！')
return
if response_dict.get('data', {}).get('isReceived') is True:
print(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()
print(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()
# print(response_dict)
if response_dict.get('data') is None:
print('当前没有可以完成的任务！')
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()
# print(response_dict)
print(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()
# print(response_dict)
if response_dict.get('code') != 200:
print(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')
print(f"种树进度: {level}级 {user_watering_droplet}/{current_level_need_watering_droplet}")
def main(self):
character = '★★'
name, level = self.tree_info()
if name == '' and level == '':
return
print(f'目标：{name}')
print(f'剩余水滴：{self.get_droplet_number()}')
self.determine_whether_is_team_tree() # 判断是否是团队树
self.get_tree_planting_progress() # 获取种树进度
print(f'{character}开始签到')
self.droplet_check_in() # 签到
print(f'{character}开始领取气泡水滴')
self.receive_droplet_extra()
print(f'{character}开始完成每日任务')
self.execute_task()
print(f'{character}开始领取累计任务奖励')
self.execute_cumulative_task()
print(f'{character}开始领取木桶水滴')
self.judging_bucket_droplet()
print(f'{character}开始多次执行浇水，领取浇水奖励')
self.execute_receive_watering_reward()
print(f'{character}开始浇水充满气泡水滴')
self.waterting_droplet_extra()
print(f'{character}开始领取合种上线奖励')
self.receive_hybrid_online_reward()
print(f'{character}开始领取空中水滴')
self.receive_air_drop()
print(f'{character}开始进行水滴投资')
self.droplet_invest()
print(f'{character}开始点击8个商品获得水滴')
self.click_product()
print(f'{character}开始领取品牌特惠奖励')
self.receive_brand_specials()
print(f'{character}开始进行助力')
self.help_user()
print(f'{character}开始领取助力奖励')
self.receive_help_reward()
print(f'{character}开始领取等级奖励')
self.receive_level_reward()
print(f'{character}开始进行浇水直到少于{self.remaining_g}g')
self.waterting_until_less_than()
print(f'剩余水滴：{self.get_droplet_number()}')
time.sleep(1)
self.get_tree_planting_progress() # 获取种树进度
# 主程序
def main(ck_list):
if not ck_list:
print('没有获取到账号！')
return
try:
download_author_share_code()
except Exception as e:
if e:
pass
print(f'获取到{len(ck_list)}个账号！')
print('开始获取所有账号助力码')
for index, ck in enumerate(ck_list):
print(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):
print(f'*****第{index + 1}个账号*****')
DeWu(ck, index).main()
print('')
if __name__ == '__main__':
main(X_AUTH_TOKEN)
sys.exit()

247
腾讯视频签到领取.py Normal file
View File

@@ -0,0 +1,247 @@
import datetime
import requests
msg = []
QYWX_AM = '' # corpid,corpsecret,touser,agentid
push = 1 # 0 不推送 1推送
config = [
{
'name': '大号',#随意
'ck': '',
'ua': '',
# 不是主设备ua容易出现滑块
}
,
# {
# 'name': '',
# 'ck': '',
# 'ua': '',
# }
]
def tx(ck, ua, name):
global hai
url = 'https://vip.video.qq.com/rpc/trpc.new_task_system.task_system.TaskSystem/CheckIn?rpc_data=%7B%7D'
# 签到 https://vip.video.qq.com/rpc/trpc.new_task_system.task_system.TaskSystem/CheckIn?rpc_data={}
url0 = 'https://vip.video.qq.com/rpc/trpc.new_task_system.task_system.TaskSystem/ReadTaskList?rpc_data=%7B%22business_id%22:%221%22,%22platform%22:3%7D'
# 任务列表 https://vip.video.qq.com/rpc/trpc.new_task_system.task_system.TaskSystem/ReadTaskList?rpc_data={"business_id": "1", "platform": 3}
url2 = 'https://vip.video.qq.com/rpc/trpc.query_vipinfo.vipinfo.QueryVipInfo/GetVipUserInfoH5' # 查看会员等级
body2 = '{"geticon":1,"viptype":"svip","platform":7}'
headers = {
'User-Agent': ua,
'Referer': 'https://film.video.qq.com/x/vip-center/?aid=V0%24%241%3A0%242%3A7%243%3A8.7.30.25756%244%3A3%248%3A4%2412%3A&hidetitlebar=1&isDarkMode=0&uiType=REGULAR',
'Origin': 'https://film.video.qq.com',
'Host': 'vip.video.qq.com',
'Cookie': ck,
'Connection': 'keep-alive',
'Accept-Language': 'zh-cn',
'Accept-Encoding': 'gzip, deflate, br',
'Accept': 'application/json, text/plain, */*'
}
try:
resp = requests.get(url, headers=headers)
if resp.status_code == 200:
resp = resp.json()
print(f"账号:{name}签到{resp}")
msg.append('------------- 任务状态 -------------')
msg.append(f'开始执行账号:{name}')
if resp['ret'] == 0:
pass
# msg.append(f"VIP会员每日签到:(已签到，获得{resp['check_in_score']}点V力值)")
elif resp['ret'] == -2002:
# print(f"签到失败:今日已签到")
pass
# msg.append(f"签到失败:今日已签到")
else:
msg.append(f"签到异常:{resp}")
else:
print(f'账号:{name} 服务器响应错误:{resp.status_code}')
msg.append(f"账号:{name} CK已失效")
except Exception as e:
print(f'{name} 签到报错{e}')
try:
resp0 = requests.get(url0, headers=headers)
if resp0.status_code == 200:
resp0 = resp0.json()
# print(resp0)
for a in range(len(resp0['task_list'])):
# print(resp2['task_list'][a])
task_maintitle = resp0['task_list'][a]['task_maintitle'] # 任务标题 (比如VIP会员每日签到)
task_subtitle = resp0['task_list'][a]['task_subtitle'] # 完成任务v力值(比如已签到，获得5点V力值)
task_button_desc = resp0['task_list'][a]['task_button_desc'] # 任务完成/未完成(比如领取 去完成 已领取)
task_type = resp0['task_list'][a]['task_type'] # 常用任务(签到,绑定手机号，赠送福袋，看视频60min)
task_correlation_id = resp0['task_list'][a]['task_correlation_id'] # 任务id(看视频是1，福袋是12)
if task_type == '1':
if '已完成' in task_button_desc:
msg.append(f"{task_maintitle}({task_subtitle})")
# print(f"{task_maintitle}({task_subtitle})")
elif '去完成' in task_button_desc:
try:
url1 = f'https://vip.video.qq.com/rpc/trpc.new_task_system.task_system.TaskSystem/ProvideAward?rpc_data=%7B%22task_id%22:{task_correlation_id}%7D'
resp1 = requests.get(url1, headers=headers)
if resp1.status_code == 200:
resp1 = resp1.json()
print(f"账号:{name} {task_maintitle}:{resp1}")
if resp1['ret'] == 0:
# print(f"{task_maintitle}(已完成获得{resp1['provide_value']}V力值)")
msg.append(f"{task_maintitle}(已完成获得{resp1['provide_value']}点V力值)")
elif resp1['ret'] == -2003:
# msg.append(f'账号:{name} 看视频60min失败:未领取或已领取')
msg.append(f'{task_maintitle}-----(任务未完成)')
else:
# print(f"{task_maintitle}:{resp1}")
msg.append(f"{task_maintitle} 任务异常:{resp1}")
else:
print(f'{task_maintitle} 服务器响应错误:{resp1.status_code}')
# msg.append(f'账号:{name} {task_maintitle}:{resp1.status_code}')
except Exception as e:
print(f"{name} {task_maintitle}报错:{e}")
else:
# 领取
try:
url1 = f'https://vip.video.qq.com/rpc/trpc.new_task_system.task_system.TaskSystem/ProvideAward?rpc_data=%7B%22task_id%22:{task_correlation_id}%7D'
resp1 = requests.get(url1, headers=headers)
if resp1.status_code == 200:
resp1 = resp1.json()
print(f"账号:{name} {task_maintitle}:{resp1}")
if resp1['ret'] == 0:
# print(f"{task_maintitle}(已完成获得{resp1['provide_value']}V力值)")
msg.append(f"{task_maintitle}(已完成获得{resp1['provide_value']}点V力值)")
elif resp1['ret'] == -2003:
# msg.append(f'账号:{name} 看视频60min失败:未领取或已领取')
msg.append(f'{task_maintitle}-----(任务未完成)')
else:
# print(f"{task_maintitle}:{resp1}")
msg.append(f"{task_maintitle} 任务异常:{resp1}")
else:
print(f'{task_maintitle} 服务器响应错误:{resp1.status_code}')
# msg.append(f'账号:{name} {task_maintitle}:{resp1.status_code}')
except Exception as e:
print(f"{name} {task_maintitle}报错:{e}")
else:
pass
# msg.append(f"{task_maintitle}{task_button_desc}")
# # 其他任务标题
else:
# print(f'账号:{name} 服务器响应错误::{resp0.status_code}')
msg.append(f"账号:{name} CK已失效")
except Exception as e:
print(f'{name} {e}')
try:
resp2 = requests.post(url2, data=body2, headers=headers)
if resp2.status_code == 200:
resp2 = resp2.json()
# print(f"{name} {resp2}")
# print(f"当前等级:LV{resp2['level']} V力值:{resp2['score']}\n有效期至:{resp2['endTime']}")
if resp2['score'] >= 53600:
hai = f'您已是星光'
elif resp2['score'] >= 36800:
hai = f"还差{53600-resp2['score']}可升级"
elif resp2['score'] >= 16800:
hai = f"还差{36800 - resp2['score']}可升级"
else:
hai = f'不计算'
msg.append(f"会员等级:V{resp2['level']} V力值:{resp2['score']}({hai})\n\t\t有效期至:{resp2['endTime']}")
else:
print(f'服务器响应错误:{resp2.status_code}')
msg.append(f"账号:{name} CK已失效")
except Exception as e:
print(f'{name} {e}')
msg.append(f"-------- {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} --------\n")
def main():
if len(config) > 0:
for a in range(len(config)):
tx(config[a]['ck'], config[a]['ua'], config[a]['name'])
qy_push("腾讯视频签到通知", '\n'.join(msg))
else:
print(f'你的号{config}')
def qy_push(Content, Text):
if push == 1:
if not QYWX_AM:
print("QYWX_AM 未设置!!\n取消推送")
return
QYWX_AM_A = QYWX_AM.split(",")
try:
qy_url = f'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={QYWX_AM_A[0]}&corpsecret={QYWX_AM_A[1]}'
re_ss = requests.get(qy_url).json()
access_token = re_ss['access_token']
except Exception as e:
print(f'抛出异常:{e}')
return
qy_url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" + access_token
if len(QYWX_AM_A) == 4:
# print('文字推送')
print(f'\n~进入推送')
data = {
"touser": QYWX_AM_A[2],
"msgtype": "text",
"agentid": QYWX_AM_A[3],
"text": {
"content": f'{Content}\n{Text}\n通知时间:{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
},
}
re1 = requests.post(url=qy_url, json=data).json()
if re1['errcode'] == 0:
# print('推送标题:', Content)
print('推送内容:\n', Text)
print('server结果:推送成功🎉！')
else:
print(f"推送失败!{re1}")
elif len(QYWX_AM_A) == 5:
# print('图片推送')
print(f'\n~进入推送')
data = {
"touser": QYWX_AM_A[2],
"msgtype": "mpnews",
"agentid": QYWX_AM_A[3],
"mpnews": {
"articles": [
{
"title": Content,
"thumb_media_id": QYWX_AM_A[4],
"author": "摸鱼助手",
"content_source_url": "",
"content": Text.replace("\n", "<br/>"),
# "content": Text,
"digest": Text,
}
]
},
}
re1 = requests.post(url=qy_url, json=data).json()
if re1['errcode'] == 0:
# print('推送标题:', Content)
print('推送内容:\n', Text)
print('server结果:推送成功🎉！')
else:
print(f"推送失败!{re1}")
else:
print("QYWX_AM 设置错误!!\n取消推送")
else:
print(f'你选择了不推送!push ==1 推送哦\n\n{Content}\n{Text}')
if __name__ == '__main__':
main()