-
Notifications
You must be signed in to change notification settings - Fork 322
/
Copy pathquanzhan.py
233 lines (178 loc) · 7.53 KB
/
quanzhan.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
"""
小程序搜 泉站大桶订水桶装水同城送水
变量 authcode # authorization #备注 （没有备注 也可以运行）
变量名 qztoken
项目 泉站订水
"""
import os
import requests
from datetime import datetime, timezone, timedelta
import json
import sys
import time
import random
from io import StringIO
enable_notification =1 # 控制是否启用通知的变量 0 不发送 1 发
# 只有在需要发送通知时才尝试导入notify模块
if enable_notification == 1:
try:
from notify import send
except ModuleNotFoundError:
print("警告：未找到notify.py模块。它不是一个依赖项，请勿错误安装。程序将退出。")
sys.exit(1)
#---------解--的简化0.2框架--------
# 配置参数
base_url = "https://microuser.quanzhan888.com" # 实际的基础URL
user_agent = "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/8519"
def get_beijing_date(): # 获取北京日期的函数
beijing_time = datetime.now(timezone(timedelta(hours=8)))
return beijing_time.date()
def timestamp_to_beijing_time(timestamp):
utc_zone = timezone.utc
beijing_zone = timezone(timedelta(hours=8))
utc_time = datetime.fromtimestamp(timestamp, utc_zone)
beijing_time = utc_time.astimezone(beijing_zone)
return beijing_time.strftime("%Y-%m-%d %H:%M:%S")
def get_env_variable(var_name):
value = os.getenv(var_name)
if value is None:
print(f'环境变量{var_name}未设置，请检查。')
return None
accounts = value.strip().split('\n') # 使用 \n 分割
num_accounts = len(accounts)
print(f'-----------本次账号运行数量：{num_accounts}-----------')
print(f'泉站大桶订水--QGh3amllamll ')
return accounts
#113.28824159502027
#23.103660007697727
def fz_hs(auth_code, authorization, user_agent, sign): #封装headers
return {
'Host': 'microuser.quanzhan888.com',
'Connection': 'keep-alive',
'Content-Length': '2',
'charset': 'utf-8',
'product': "shop",
'authcode': auth_code,
'authorization': authorization,
'user-agent': user_agent,
'sign': sign,
'Accept-Encoding': 'gzip,compress,br,deflate',
'platform': "wx",
'x-requested-with': 'xmlhttprequest',
'content-type': 'application/x-www-form-urlencoded',
}
def wdqbsj(auth_code, authorization): # 个人信息/钱包
url = f"{base_url}/user/get-wallet-info"
headers = fz_hs(auth_code, authorization, user_agent, "99914b932bd37a50b983c5e7c90ae93b")
data = json.dumps({}) # 发送空的JSON数据
#print(url)
try:
response = requests.post(url, headers=headers, data=data)
response.raise_for_status()
response_data = response.json()
#print("解析的JSON数据：", response_data)
# 判断code并提取所需数据
if response_data.get('code') == 0:
user_id = response_data['data']['wallet_info'].get('user_id')
total_balance = response_data['data']['wallet_info'].get('total_balance')
today_income = response_data['data']['wallet_info'].get('today_income')
#print(f"用户ID: {user_id}, 总余额: {total_balance}, 今日收入: {today_income}")
print(f"🆔: {user_id}, 总💸: {total_balance}, 今日: {today_income}")
# 判断今日收入是否大于0
if float(today_income) > 0:
print("今日已有收入，不需要签到")
#tj_sign(auth_code, authorization)#测试提交签到
else:
print("今日无收入，需要签到")
tj_sign(auth_code, authorization)
else:
print("响应代码不为0，完整响应体：", response_data)
except ValueError:
print("响应不是有效的JSON格式。")
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
def tj_sign(auth_code, authorization): # 提交签到
url = f"{base_url}/user/do-sign"
headers = fz_hs(auth_code, authorization, user_agent, "99914b932bd37a50b983c5e7c90ae93b")
data = json.dumps({}) # 发送空的JSON数据
try:
response = requests.post(url, headers=headers, data=data)
response.raise_for_status()
response_data = response.json()
#print("解析的JSON数据：", response_data)
# 提取所需数据并转换时间戳
if 'data' in response_data and len(response_data['data']) > 0:
for item in response_data['data']:
user_id = item.get('user_id')
sign_date = timestamp_to_beijing_time(item.get('sign_date'))
sign_time = timestamp_to_beijing_time(item.get('sign_time'))
#print(f"用户: {user_id}, 签名日期: {sign_date}, 签到时间: {sign_time}")
print(f" 签名日期: {sign_date}, 签到🎉: {sign_time}")
return response_data
except ValueError:
print("响应不是有效的JSON格式。")
return None
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
#------------通知开始-----------
class Tee:
def __init__(self, *files):
self.files = files
def write(self, obj):
for file in self.files:
file.write(obj)
file.flush() # 确保及时输出
def flush(self):
for file in self.files:
file.flush()
#------------通知结束-----------
def main():
string_io = StringIO()
original_stdout = sys.stdout
try:
sys.stdout = Tee(sys.stdout, string_io)
var_name = 'qztoken' # 环境变量名
accounts = get_env_variable(var_name)
if not accounts:
return
print(f'找到 {len(accounts)} 个账号的令牌。')
total_tokens = len(accounts)
for index, account in enumerate(accounts, start=1):
parts = account.split('#')
auth_code, authorization = parts[0], parts[1]
remark = None if len(parts) == 2 else parts[2] # 检查是否有备注
remark_info = f" --- 备注: {remark}" if remark else ""
print(f"------账号 {index}/{total_tokens}{remark_info} ------")
wdqbsj(auth_code, authorization) # 个人信息/钱包
# 暂停3到5秒
time.sleep(random.randint(3, 5))
finally:
sys.stdout = original_stdout
output_content = string_io.getvalue()
if enable_notification:
send("-泉站大桶订水-通知", output_content)
if __name__ == "__main__":
main()
"""
#本地测试用
os.environ['qztoken'] = '''
authcode # authorization
'''
def main():
var_name = 'qztoken' # 环境变量名
accounts = get_env_variable(var_name)
if not accounts:
return
print(f'找到 {len(accounts)} 个账号的令牌。')
total_tokens = len(accounts)
for index, account in enumerate(accounts, start=1):
parts = account.split('#')
auth_code, authorization = parts[0], parts[1]
remark = None if len(parts) == 2 else parts[2] # 检查是否有备注
remark_info = f" --- 备注: {remark}" if remark else ""
print(f"------账号 {index}/{total_tokens}{remark_info} ------")
wdqbsj(auth_code, authorization)# 个人信息/钱包
if __name__ == "__main__":
main()
"""