mirror of
https://github.com/CHERWING/CHERWIN_SCRIPTS.git
synced 2024-11-22 09:04:11 +08:00
427 lines
17 KiB
Python
427 lines
17 KiB
Python
# !/usr/bin/python3
|
||
# -- coding: utf-8 --
|
||
# -------------------------------
|
||
# @Author : cherwin
|
||
# -------------------------------
|
||
# cron "59 9 * 2-6 * *" script-path=xxx.py,tag=匹配cron用
|
||
# const $ = new Env('统一茄皇监控')
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import random
|
||
import string
|
||
import time
|
||
from datetime import datetime
|
||
from os import environ, path
|
||
|
||
import requests
|
||
from requests.packages.urllib3.exceptions import InsecureRequestWarning
|
||
|
||
# 禁用安全请求警告
|
||
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
|
||
|
||
if os.path.isfile('DEV_ENV.py'):
|
||
import DEV_ENV
|
||
|
||
if os.path.isfile('notify.py'):
|
||
from notify import send
|
||
print("加载通知服务成功！")
|
||
else:
|
||
print("加载通知服务失败!")
|
||
send_msg = ''
|
||
one_msg=''
|
||
def Log(cont=''):
|
||
global send_msg,one_msg
|
||
print(cont)
|
||
if cont:
|
||
one_msg += f'{cont}\n'
|
||
send_msg += f'{cont}\n'
|
||
|
||
USER_INFO = {}
|
||
|
||
class RUN:
|
||
def __init__(self, info,index):
|
||
global one_msg
|
||
one_msg = ''
|
||
split_info = info.split('@')
|
||
self.third_id = split_info[0]
|
||
self.wid = split_info[1]
|
||
len_split_info = len(split_info)
|
||
last_info = split_info[len_split_info - 1]
|
||
self.send_UID = None
|
||
if len_split_info > 0 and "UID_" in last_info:
|
||
self.send_UID = last_info
|
||
|
||
self.user_index = index + 1
|
||
print(f"\n---------开始执行第{self.user_index}个账号>>>>>")
|
||
self.s = requests.session()
|
||
self.s.verify = False
|
||
|
||
self.UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x6309080f) XWEB/8555'
|
||
|
||
self.headers = {
|
||
'User-Agent': self.UA,
|
||
'Sec-Fetch-Site': 'cross-site',
|
||
'Sec-Fetch-Mode': 'cors',
|
||
'Sec-Fetch-Dest': 'empty',
|
||
'Accept-Encoding': 'gzip, deflate, br',
|
||
'Accept-Language': 'zh-CN,zh;q=0.9',
|
||
'Origin': 'https://thekingoftomato.ioutu.cn',
|
||
'Referer': 'https://thekingoftomato.ioutu.cn/'
|
||
|
||
}
|
||
self.base_url = 'https://qiehuang-apig.xiaoyisz.com/qiehuangsecond/ga'
|
||
self.goodsid = None
|
||
self.Login_res = self.login()
|
||
|
||
def load_json(self):
|
||
try:
|
||
with open(f"INVITE_CODE/{ENV_NAME}_INVITE_CODE.json", 'r', encoding='utf-8') as file:
|
||
data = json.load(file)
|
||
return data
|
||
except FileNotFoundError:
|
||
print("未找到文件，返回空字典")
|
||
return {}
|
||
except Exception as e:
|
||
print(f"发生错误：{e}")
|
||
return {}
|
||
|
||
def make_request(self, url, method='post', headers={}, params={}):
|
||
if headers == {}:
|
||
headers = self.headers
|
||
if params == {}:
|
||
params = self.params
|
||
try:
|
||
if method.lower() == 'get':
|
||
response = requests.get(url, headers=headers, verify=False)
|
||
|
||
elif method.lower() == 'post':
|
||
response = requests.post(url, headers=headers, json=params, verify=False)
|
||
else:
|
||
raise ValueError("不支持的请求方法: " + method)
|
||
return response.json()
|
||
except requests.exceptions.RequestException as e:
|
||
print("请求异常：", e)
|
||
except ValueError as e:
|
||
print("值错误或不支持的请求方法：", e)
|
||
except Exception as e:
|
||
print("发生了未知错误：", e)
|
||
|
||
def gen_sign(self, parameters={}, body=None):
|
||
sorted_keys = sorted(parameters.keys())
|
||
parameter_strings = []
|
||
for key in sorted_keys:
|
||
if isinstance(parameters[key], dict):
|
||
parameter_strings.append(f"{key}={json.dumps(parameters[key])}")
|
||
else:
|
||
parameter_strings.append(f"{key}={parameters[key]}")
|
||
|
||
current_time = int(datetime.now().timestamp() * 1000)
|
||
secret_chars = list('BxzTx45uIGT25TTHIIBU2')
|
||
last_three_digits = str(current_time)[-3:]
|
||
for digit in last_three_digits:
|
||
secret_chars.insert(int(digit), digit)
|
||
|
||
secret = hashlib.md5(''.join(secret_chars).encode()).hexdigest()
|
||
nonce_str = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
|
||
|
||
sign_data = {
|
||
'client_id': 'game',
|
||
'nonstr': nonce_str,
|
||
'timestamp': current_time,
|
||
'body': json.dumps(body) if body else '',
|
||
'query': '&'.join(parameter_strings) if parameter_strings else '',
|
||
'secret': secret
|
||
}
|
||
|
||
sign_string = '|'.join([str(v) for v in sign_data.values()])
|
||
sign = hashlib.md5(sign_string.encode()).hexdigest().upper()
|
||
sign_header = {
|
||
'client_id': 'game',
|
||
'timestamp': str(current_time),
|
||
'nonstr': sign_data['nonstr'],
|
||
'sign': sign
|
||
}
|
||
self.headers.update(sign_header)
|
||
return self.headers
|
||
|
||
def login(self):
|
||
login_successful = False
|
||
try:
|
||
login_params = {
|
||
'thirdId': self.third_id,
|
||
'wid': self.wid
|
||
}
|
||
sign_header = self.gen_sign({}, login_params)
|
||
# print(self.headers)
|
||
# Hypothetically speaking, this is how you might perform a POST request in Python with the requests library.
|
||
response = self.s.post(f'{self.base_url}/public/api/login', json=login_params, headers=sign_header)
|
||
if response.status_code == 200:
|
||
response_data = response.json()
|
||
if response_data.get('code', -1) == 0:
|
||
auth = response_data['data']['token'] or ''
|
||
if auth:
|
||
login_successful = True
|
||
print(f'账号【{self.user_index}】登录成功')
|
||
Authorization = {'Authorization': auth}
|
||
self.headers.update(Authorization)
|
||
else:
|
||
print(f'账号【{self.user_index}】登录获取auth失败')
|
||
else:
|
||
print(f"登录获取auth失败[{response_data['code']}]: {response_data['message']}")
|
||
elif response.status_code == 403:
|
||
print('登录失败[403]: 黑IP了, 换个IP试试吧')
|
||
except Exception as e:
|
||
print(e)
|
||
finally:
|
||
return login_successful
|
||
|
||
def get_CapCode(self, slideImgInfo):
|
||
slidingImage = slideImgInfo.get('slidingImage', None)
|
||
backImage = slideImgInfo.get('backImage', None)
|
||
dddddocr_api = os.environ.get('OCR_API',False)
|
||
if not dddddocr_api:
|
||
print('未定义变量【OCR_API】\n取消验证码识别\n搭建方式：https://github.com/CHERWING/CHERWIN_SCRIPTS')
|
||
return False
|
||
if slidingImage and backImage:
|
||
data = {
|
||
"slidingImage": slidingImage,
|
||
"backImage": backImage
|
||
}
|
||
response = requests.post(f"{dddddocr_api}/capcode", data=json.dumps(data),headers={'Content-Type': 'application/json'})
|
||
print(response.json())
|
||
self.capcode = response.json().get('result','')
|
||
if self.capcode:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
def get_CapCode_local(self, slideImgInfo):
|
||
slidingImage = slideImgInfo.get('slidingImage', None)
|
||
backImage = slideImgInfo.get('backImage', None)
|
||
if slidingImage and backImage:
|
||
self.capcode =CHERWIN_TOOLS.CAPCODE(slidingImage,backImage)
|
||
if self.capcode:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
def checkUserCapCode(self):
|
||
print(f'提交验证码--->>>')
|
||
# try:
|
||
print(f'验证码：{self.capcode}')
|
||
params = {'xpos':self.capcode}
|
||
print(params)
|
||
sign_header = self.gen_sign(body=params)
|
||
url = f'{self.base_url}/checkUserCapCode'
|
||
response = self.s.post(url, headers=sign_header, json=params)
|
||
data = response.json()
|
||
code = data.get('code', -1)
|
||
if code == 0:
|
||
data = data.get('data', 0)
|
||
print(f"验证码正确，获取到[{data}]")
|
||
return True
|
||
else:
|
||
message = data.get('message', '')
|
||
print(f"验证码错误[{message}]")
|
||
return False
|
||
# except Exception as e:
|
||
# print(e)
|
||
|
||
def exchange_find(self):
|
||
print(f'获取兑换列表--->>>')
|
||
try:
|
||
sign_header = self.gen_sign()
|
||
url = f'{self.base_url}/exchange/find'
|
||
response = self.s.get(url, headers=sign_header)
|
||
data = response.json()
|
||
code = data.get('code', -1)
|
||
if code == 0:
|
||
data = data.get('data', {})[1]
|
||
num = data.get('num',0)
|
||
if num < 0 :
|
||
Log('兑换上限,跳过')
|
||
return False
|
||
# print(f"兑换列表：{data}")
|
||
exchangePrizeVoList = data.get('exchangePrizeVoList',None)
|
||
if exchangePrizeVoList:
|
||
for goods in exchangePrizeVoList:
|
||
name = goods.get('name','')
|
||
self.goodsid = goods.get('id','')
|
||
usableStock = goods.get('usableStock','')
|
||
if usableStock >0:
|
||
Log(f'ID：【{self.goodsid}】 【{name}】 当前剩余：{usableStock}【可兑换】')
|
||
self.sendMsg()
|
||
if send: send(f'{APP_NAME}挂机通知', send_msg + TIPS_HTML)
|
||
if TYQH_DHID and TYQH_DHID == self.goodsid:
|
||
self.exchange_reward(TYQH_DHID)
|
||
elif TYQH_DHID == '0':
|
||
self.exchange_reward(self.goodsid)
|
||
else:
|
||
continue
|
||
else:
|
||
print(f'ID：【{self.goodsid}】【{name}】 当前剩余：{usableStock}【不可兑换】')
|
||
else:
|
||
message = data.get('message', '')
|
||
print(f'{message}')
|
||
except Exception as e:
|
||
print(e)
|
||
|
||
def exchange_reward(self, reward_id):
|
||
print(f'尝试兑换--->>>')
|
||
try:
|
||
params = {'id': reward_id}
|
||
sign_header = self.gen_sign(params)
|
||
url = f'{self.base_url}/exchange/reward'
|
||
response = self.s.get(url, headers=sign_header, params=params)
|
||
data = response.json()
|
||
code = data.get('code', -1)
|
||
if code == 0:
|
||
data = data.get('data', {}).get('name') or ''
|
||
Log(f"兑换[{data}]成功")
|
||
elif code == 4000:
|
||
slideImgInfo = data.get('data', {}).get('slideImgInfo', None)
|
||
validateCount = data.get('data', {}).get('validateCount', None)
|
||
if slideImgInfo and validateCount:
|
||
print(f"兑换需要滑块验证")
|
||
if self.get_CapCode(slideImgInfo):
|
||
if self.checkUserCapCode():
|
||
self.exchange_reward(reward_id)
|
||
else:
|
||
print(f"兑换验证码上限")
|
||
else:
|
||
message = data.get('message', '')
|
||
print(f'兑换[id={reward_id}]: {message}')
|
||
except Exception as e:
|
||
print(e)
|
||
|
||
def userTask(self):
|
||
print('\n--------------- 开始日常任务 ---------------')
|
||
wait_time = random.randint(1000, 3000) / 1000.0 # 转换为秒
|
||
if not self.Login_res:
|
||
return False
|
||
self.exchange_find()
|
||
|
||
return True
|
||
|
||
|
||
def sendMsg(self, help=False):
|
||
if self.send_UID:
|
||
push_res = CHERWIN_TOOLS.wxpusher(self.send_UID, one_msg, APP_NAME, help)
|
||
print(push_res)
|
||
|
||
|
||
def down_file(filename, file_url):
|
||
print(f'开始下载：{filename}，下载地址：{file_url}')
|
||
try:
|
||
response = requests.get(file_url, verify=False, timeout=10)
|
||
response.raise_for_status()
|
||
with open(filename + '.tmp', 'wb') as f:
|
||
f.write(response.content)
|
||
print(f'【{filename}】下载完成！')
|
||
|
||
# 检查临时文件是否存在
|
||
temp_filename = filename + '.tmp'
|
||
if os.path.exists(temp_filename):
|
||
# 删除原有文件
|
||
if os.path.exists(filename):
|
||
os.remove(filename)
|
||
# 重命名临时文件
|
||
os.rename(temp_filename, filename)
|
||
print(f'【{filename}】重命名成功！')
|
||
return True
|
||
else:
|
||
print(f'【{filename}】临时文件不存在！')
|
||
return False
|
||
except Exception as e:
|
||
print(f'【{filename}】下载失败：{str(e)}')
|
||
return False
|
||
|
||
def import_Tools():
|
||
global CHERWIN_TOOLS,ENV, APP_INFO, TIPS, TIPS_HTML, AuthorCode
|
||
import CHERWIN_TOOLS
|
||
ENV, APP_INFO, TIPS, TIPS_HTML, AuthorCode = CHERWIN_TOOLS.main(APP_NAME, local_script_name, ENV_NAME,local_version)
|
||
import threading
|
||
|
||
def execute_task(infos, index):
|
||
run_result = RUN(infos, index).userTask()
|
||
if not run_result:
|
||
return
|
||
|
||
if __name__ == '__main__':
|
||
APP_NAME = '统一茄皇监控'
|
||
ENV_NAME = 'TYQH_JK'
|
||
print(f'''
|
||
✨✨✨ {APP_NAME}脚本 ✨✨✨
|
||
✨ 功能：
|
||
奖品监控
|
||
✨ 设置青龙变量：
|
||
export TYQH= '' 使用相同TYQH变量
|
||
export TYQH_DHID= '112' 设置兑换ID则开启自动兑换，设置为0遍历兑换全部可兑换商品
|
||
export OCR_API= 'http://localhost:3721'
|
||
✨ 由于青龙python版本问题无法直接使用dddocr需要自行搭建API，搭建方式：https://github.com/CHERWING/CHERWIN_OCR
|
||
✨ 如果你的环境可以安装dddocr库则可以替换代码内的【self.get_CapCode】为【self.get_CapCode_local】
|
||
export SCRIPT_UPDATE = 'False' 关闭脚本自动更新，默认开启
|
||
✨ 推荐定时：*/10 9:59 1-31 2-6 * * 2-6月每月1日9点55到10点1分，每10秒执行一次
|
||
✨ 第一个账号助力作者，其余互助
|
||
✨✨✨ @Author CHERWIN✨✨✨
|
||
''')
|
||
local_script_name = os.path.basename(__file__)
|
||
local_version = '2024.05.04'
|
||
if os.path.isfile('CHERWIN_TOOLS.py'):
|
||
import_Tools()
|
||
else:
|
||
if down_file('CHERWIN_TOOLS.py', 'https://py.cherwin.cn/CHERWIN_TOOLS.py'):
|
||
print('脚本依赖下载完成请重新运行脚本')
|
||
import_Tools()
|
||
else:
|
||
print('脚本依赖下载失败，请到https://py.cherwin.cn/CHERWIN_TOOLS.py下载最新版本依赖')
|
||
exit()
|
||
print(TIPS)
|
||
token = ''
|
||
ENV = os.environ.get('TYQH','')
|
||
TYQH_DHID = os.environ.get('TYQH_DHID',None)
|
||
if TYQH_DHID == '0':
|
||
Log(f'\n当前已设置TYQH_DHID变量，将自动遍历兑换商品')
|
||
elif TYQH_DHID:
|
||
Log(f'\n当前已设置TYQH_DHID变量，将自动兑换id【{TYQH_DHID}】商品')
|
||
else:
|
||
Log('\n未定义TYQH_DHID变量，取消自动兑换')
|
||
token = ENV if ENV else token
|
||
if not token:
|
||
print(f"未填写TYQH变量\n青龙可在环境变量设置TYQH 或者在本脚本文件上方将ck填入token =''")
|
||
exit()
|
||
tokens = CHERWIN_TOOLS.ENV_SPLIT(token)
|
||
# print(tokens)
|
||
if len(tokens) > 0:
|
||
print(f"\n>>>>>>>>>> 共获取到{len(tokens)} 个账号 <<<<<<<<<<")
|
||
# access_token = []
|
||
# threads = []
|
||
# for index, infos in enumerate(tokens):
|
||
# thread = threading.Thread(target=execute_task, args=(infos, index))
|
||
# threads.append(thread)
|
||
# thread.start()
|
||
#
|
||
# for thread in threads:
|
||
# thread.join()
|
||
while True:
|
||
current_time = time.localtime()
|
||
if (current_time.tm_hour == 9 and current_time.tm_min == 59 and current_time.tm_sec >= 59) or \
|
||
(current_time.tm_hour == 10 and current_time.tm_min == 1 and current_time.tm_sec <= 59):
|
||
if len(tokens) > 0:
|
||
print(f"\n>>>>>>>>>> 共获取到{len(tokens)} 个账号 <<<<<<<<<<")
|
||
access_token = []
|
||
threads = []
|
||
for index, infos in enumerate(tokens):
|
||
thread = threading.Thread(target=execute_task, args=(infos, index))
|
||
threads.append(thread)
|
||
thread.start()
|
||
|
||
for thread in threads:
|
||
thread.join()
|
||
elif current_time.tm_hour == 10 and current_time.tm_min > 5:
|
||
break
|
||
else:
|
||
time.sleep(1)
|
||
|