mirror/Ytong825-mao
1
0
Fork 0
mirror of https://github.com/Ytong825/mao.git synced 2025-12-18 07:44:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

删除 团币.py

Browse Source
This commit is contained in:
Ytong
2023-11-20 19:34:45 +08:00
parent 2717e2c58d
commit fff7ba2cb4
1 changed files with 0 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files

288
团币.py
View File

@@ -1,288 +0,0 @@
"""
time：2023.10.24
定时：一天至少3-5次
正常是每天700+币
new Env('美团小团币游戏中心');
抓包小程序或者app或者网页的token=Agxxxx 只要token后面的值
环境变量: 名称：bd_mttoken 值：Agxxxxxxxxxx
多账号新建变量或者用 & 分开
并发变量: bd_xtbbf = 1 默认不设置为1
更新日志：
10.24: 关闭授权，开源
9.26: 优化报错，并发变量
9.24: 新增账号并发运行
9.23: 新增每日获取小团币，异常重试
"""
import random
import base64
import os
import requests
import time
import string
from functools import partial
from user_agent import generate_user_agent
import threading
class Mttb:
def __init__(self, ck, num):
self.num = num
self.ck = ck
self.name = None
self.name = None
self.usid = None
self.actoken = None
self.xtb = None
self.wcxtb = None
self.ids = []
self.ids1 = []
self.id = None
self.tid = None
self.ua = generate_user_agent(os='android')
self.t_h = None
self.msg = ''
self.start = ''
self.end = ''
def main(self):
if self.login():
self.act()
self.cxtb()
if self.get_ids():
self.get_id()
def login(self):
try:
url = "https://open.meituan.com/user/v1/info/auditting?fields=auditAvatarUrl%2CauditUsername"
h = {
'Connection': 'keep-alive',
'Origin': 'https://mtaccount.meituan.com',
'User-Agent': self.ua,
'token': self.ck,
'Referer': 'https://mtaccount.meituan.com/user/',
'Accept-Encoding': 'gzip, deflate',
'Accept-Language': 'zh-CN,en-US;q=0.9',
'X-Requested-With': 'com.sankuai.meituan',
}
r = requests.get(url, headers=h)
if 'username' in r.text:
rj = r.json()
self.name = rj["user"]["username"]
self.usid = rj["user"]["id"]
xx = f'😶账号{self.num}\n🆔{self.name}\n'
self.start += xx
return True
else:
print(r.json())
except Exception as e:
print(f'登录异常：{e}')
exit(0)
def act(self):
try:
url = 'https://game.meituan.com/mgc/gamecenter/front/api/v1/login'
h = {
'Accept': 'application/json, text/plain, */*',
'Content-Length': '307',
'x-requested-with': 'XMLHttpRequest',
'User-Agent': self.ua,
'Content-Type': 'application/json;charset=UTF-8',
'cookie': f'token={self.ck}'
}
sing = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
data = {
"mtToken": self.ck,
"deviceUUID": '0000000000000A3467823460D436CAB51202F336236F6A167191373531985811',
"mtUserId": self.usid,
"idempotentString": sing
}
r = requests.post(url, headers=h, json=data)
if r.json()['data']['loginInfo']['accessToken'] is not None:
self.actoken = r.json()['data']['loginInfo']['accessToken']
else:
print(r.json())
except Exception as e:
print(f'获取token异常：{e}')
exit(0)
def cxtb(self):
try:
url = 'https://game.meituan.com/mgc/gamecenter/skuExchange/resource/counts?sceneId=3&gameId=10102'
self.t_h = {
'Accept': 'application/json, text/plain, */*',
'x-requested-with': 'XMLHttpRequest',
'User-Agent': self.ua,
'Content-Type': 'application/json;charset=UTF-8',
'mtgsig': '',
'actoken': self.actoken,
'mtoken': self.ck,
'cookie': f'token={self.ck}'
}
r = requests.get(url, headers=self.t_h)
rj = r.json()
if rj['msg'] == 'ok':
data = rj['data']
for d in data:
if self.xtb is not None:
self.wcxtb = d['count']
xx = f'💰当前小团币: {int(self.wcxtb)}({int(self.wcxtb) / 1000}元)\n'
self.end += xx
else:
self.xtb = d['count']
xx = f'💰小团币: {int(self.xtb)}({int(self.xtb) / 1000}元)\n'
self.start += xx
print(self.start)
except Exception as e:
print(f'🆔{self.name}>>>⚠️查询团币异常：{e}')
def get_ids(self):
try:
url = 'https://game.meituan.com/mgc/gamecenter/front/api/v1/mgcUser/task/queryMgcTaskInfo'
data = {
"externalStr": "",
"riskParams": {}
}
r = requests.post(url, headers=self.t_h, json=data)
rj = r.json()
if rj['msg'] == 'ok':
data_list = r.json()['data']['taskList']
for i in data_list:
self.ids.append(i['id'])
if self.ids:
random.shuffle(self.ids)
# print(self.ids)
return True
else:
print(f'{self.name}>>>获取任务失败！')
except Exception as e:
print(f'获取任务异常：{e}')
exit(0)
def get_id(self):
for idd in self.ids:
self.id = idd
if self.get_game():
self.post_id()
xx = f'😊账号{self.num}\n🆔{self.name}>>>🎉运行完成！\n'
self.end += xx
self.cxtb()
bchd = int(self.wcxtb) - int(self.xtb)
xx = f'🔔获取小团币: {bchd}\n'
self.end += xx
print(self.end)
def b64(self):
y_bytes = base64.b64encode(self.tid.encode('utf-8'))
y_bytes = y_bytes.decode('utf-8')
return y_bytes
def get_game(self):
try:
self.tid = f'mgc-gamecenter{self.id}'
self.tid = self.b64()
url = f'https://game.meituan.com/mgc/gamecenter/common/mtUser/mgcUser/task/finishV2?taskId={self.tid}'
r = requests.get(url, headers=self.t_h)
if r.status_code == 200:
if r.json()['msg'] == 'ok':
# print(f'{self.name}>>>{self.id} 领取任务成功！')
time.sleep(1)
return True
elif '完成过' in r.text:
# print(f'{self.name}>>>{self.id} 完成过领取任务成功！')
time.sleep(1)
return True
else:
print(f'🆔{self.name}>>>🌚任务状态: {r.text}')
else:
print(f'🆔{self.name}>>>请求错误: ', r.status_code)
except Exception as e:
print(f'🆔{self.name}>>>⚠️获取任务异常：{e}')
def post_id(self):
try:
url = 'https://game.meituan.com/mgc/gamecenter/front/api/v1/mgcUser/task/receiveMgcTaskReward?yodaReady=h5&csecplatform=4&csecversion=2.1.0&mtgsig={}'
data = {
"taskId": self.id,
"externalStr": "",
"riskParams": {}
}
r = requests.post(url, headers=self.t_h, json=data)
if r.status_code == 200:
if r.json()['msg'] == 'ok':
# print(f'{self.name}>>>{self.id},完成任务！\n')
time.sleep(1)
elif '异常' in r.text:
# print(f'{self.name}>>>{self.id},状态异常，任务不可领奖！\n')
time.sleep(1)
else:
print(f'{self.name}>>>{self.id},{r.text}\n')
time.sleep(1)
else:
print('请求错误!')
except Exception as e:
print(f'🆔{self.name}>>>⚠️完成任务异常：{e}')
if __name__ == '__main__':
print = partial(print, flush=True)
print('🔔当前版本V10.24\n🔔tg频道：https://t.me/dzr_byg')
token = os.environ.get("bd_mttoken")
if token is None:
print(f'⛔️未获取到ck变量：请检查变量是否填写')
exit(0)
if '&' in token:
tokens = token.split('&')
else:
tokens = [token]
bf = os.environ.get("bd_xtbbf")
if bf is None:
print(f'⛔️为设置并发变量，默认1')
bf = 2
print(f'✅获取到{len(tokens)}个账号')
print(f'🔔设置并发数: {bf}')
def run_account(tk, n):
run = Mttb(tk, n)
run.main()
threads = []
s_e = []
for i in range(len(tokens)):
a = i + 1
s_e.append(a)
t = threading.Thread(target=run_account, args=(tokens[i], a,))
threads.append(t)
if str(len(threads)) == str(bf):
print(f'==================⏳账号{s_e[0]}-{s_e[-1]}==================')
for t in threads:
t.start()
time.sleep(5)
print(f'⏳账号{s_e[0]}-{s_e[-1]}任务运行中！')
for t in threads:
t.join()
threads = []
s_e = []
time.sleep(5)
if threads == [] and s_e == []:
print(f'🔔全部账号运行完成！！！')
else:
print(f'==================账号{s_e[0]}-{s_e[-1]}==================')
for t in threads:
t.start()
time.sleep(3)
print(f'⏳账号{s_e[0]}-{s_e[-1]}任务运行中！')
for t in threads:
t.join()
threads = []
s_e = []
time.sleep(5)
print(f'🔔全部账号运行完成！！！')