mirror of
https://github.com/3288588344/toulu.git
synced 2025-12-16 23:11:00 +08:00
866 lines
38 KiB
Python
866 lines
38 KiB
Python
#!/usr/bin/python3
|
||
# -- coding: utf-8 --
|
||
# ------------------------------
|
||
# cron "3 0 * * *" script-path=xxx.py,tag=匹配cron用
|
||
# const $ = new Env('联通和沃畅游密码登录');
|
||
# -------------------------------
|
||
|
||
|
||
|
||
#有问题联系3288588344
|
||
#频道：https://pd.qq.com/s/672fku8ge
|
||
|
||
|
||
|
||
"""
|
||
|
||
说明： 自动获取联通token_online、appid、ecs_token和access_token参数
|
||
|
||
注意：若是需要沃畅游的access_token参数，在首次使用时，联通app首页--5g新通信--联通畅游，点击个人中心，手动登录一遍
|
||
|
||
青龙环境变量：WoChangYouCK_PSW 手机号码&密码 账号和密码分别是联通app的账号和密码
|
||
|
||
|
||
下列变量都可选填，请按照自己的需求自行添加环境变量
|
||
联通app参数
|
||
变量名：IsChinaUnicomParam 格式：True或者False 是否生成token_online参数填入环境变量，是下面变量起作用的前提，默认False不生成填入
|
||
变量名：ChinaUnicomEnvName 格式：填指定生成的环境变量名，如chinaUnicomCookie，本脚本默认生成chinaUnicomCookie，用于生成跑联通脚本运行
|
||
变量名：ChinaUnicomParam_chatMoreStr 格式：生成的多账号使用的拼接连接符，默认使用&
|
||
变量名：ChinaUnicomParam_flag 格式：是否拼接appid，默认False不拼接，需要拼接请填入True
|
||
变量名：ChinaUnicomParam_chatStr 格式：appid和token_online之间的拼接符号，默认使用#
|
||
变量名：ChinaUnicomParam_appidIndex 格式：1或者2 拼接appid是在token_online前面还是后面，1代表前面，2代表后面,默认appid在前面
|
||
沃畅游参数
|
||
变量名：IsWoChangYouCK 格式：True或者False 是否生成沃畅游的access_token参数填入环境变量，默认False不生成填入
|
||
|
||
wxpusher推送(非必填)
|
||
青龙变量：WoChangYouCK_WXPUSHER_TOKEN wxpusher推送的token
|
||
青龙变量：WoChangYouCK_WXPUSHER_TOPIC_ID wxpusher推送的topicId(主题ID，非UID)
|
||
网址：https://wxpusher.zjiecode.com/admin/main/topics/list
|
||
|
||
有问题联系3288588344
|
||
频道：https://pd.qq.com/s/672fku8ge
|
||
|
||
"""
|
||
import requests,re
|
||
import json, os, random
|
||
import time
|
||
from datetime import datetime
|
||
from sys import stdout
|
||
import base64
|
||
response = requests.get("https://github.com/3288588344/toulu/blob/main/tl.txt")
|
||
response.encoding = 'utf-8'
|
||
txt = response.text
|
||
print(txt)
|
||
from base64 import b64encode
|
||
from uuid import uuid4
|
||
from urllib.parse import quote
|
||
try:
|
||
from Crypto.PublicKey.RSA import importKey, construct
|
||
from Crypto.Cipher import PKCS1_v1_5
|
||
except:
|
||
print("检测到还未安装 pycryptodome 依赖，请先在python中安装 pycryptodome 依赖")
|
||
print("如果安装依赖pycryptodome出错时，请先在linux中安装gcc、python3-dev、libc-dev三个依赖")
|
||
exit(0)
|
||
|
||
|
||
|
||
|
||
WXPUSHER_TOKEN = '' # wxpusher推送的token
|
||
WXPUSHER_TOPIC_ID = '' # wxpusher推送的topicId
|
||
WXPUSHER_CONTENT_TYPE = 2 # wxpusher推送的样式，1表示文字 2表示html(只发送body标签内部的数据即可，不包括body标签)，默认为2
|
||
# wxpusher消息推送
|
||
def wxpusher(title: str, content: str) -> None:
|
||
"""
|
||
使用微信的wxpusher推送
|
||
"""
|
||
if not WXPUSHER_TOKEN or not WXPUSHER_TOPIC_ID:
|
||
print("wxpusher 服务的 token 或者 topicId 未设置!!\n取消推送")
|
||
return
|
||
print("wxpusher 服务启动")
|
||
|
||
url = f"https://wxpusher.zjiecode.com/api/send/message"
|
||
headers = {"Content-Type": "application/json;charset=utf-8"}
|
||
contentType = 2
|
||
if not WXPUSHER_CONTENT_TYPE:
|
||
contentType = 2
|
||
else:
|
||
contentType = WXPUSHER_CONTENT_TYPE
|
||
if contentType == 2:
|
||
content = content.replace("\n", "<br/>")
|
||
data = {
|
||
"appToken":f"{WXPUSHER_TOKEN}",
|
||
"content":f"{content}",
|
||
"summary":f"{title}",
|
||
"contentType":contentType,
|
||
"topicIds":[
|
||
f'{WXPUSHER_TOPIC_ID}'
|
||
],
|
||
"verifyPay":False
|
||
}
|
||
response = requests.post(
|
||
url=url, data=json.dumps(data), headers=headers, timeout=15
|
||
).json()
|
||
|
||
if response["code"] == 1000:
|
||
print("wxpusher推送成功！")
|
||
else:
|
||
print("wxpusher推送失败！")
|
||
print(f"wxpusher推送出错响应内容：{response}" )
|
||
|
||
|
||
ql_auth_path = '/ql/data/config/auth.json'
|
||
ql_config_path = '/ql/data/config/config.sh'
|
||
#判断环境变量
|
||
flag = 'new'
|
||
if not os.path.exists(ql_auth_path):
|
||
ql_auth_path = '/ql/config/auth.json'
|
||
ql_config_path = '/ql/config/config.sh'
|
||
if not os.path.exists(ql_config_path):
|
||
ql_config_path = '/ql/config/config.js'
|
||
flag = 'old'
|
||
# ql_auth_path = r'D:\Docker\ql\config\auth.json'
|
||
ql_url = 'http://localhost:5600'
|
||
|
||
|
||
def __get_token() -> str or None:
|
||
with open(ql_auth_path, 'r', encoding='utf-8') as f:
|
||
j_data = json.load(f)
|
||
return j_data.get('token')
|
||
|
||
|
||
def __get__headers() -> dict:
|
||
headers = {
|
||
'Accept': 'application/json',
|
||
'Content-Type': 'application/json;charset=UTF-8',
|
||
'Authorization': 'Bearer ' + __get_token()
|
||
}
|
||
return headers
|
||
|
||
# 封装读取环境变量的方法
|
||
def get_cookie(key, default="", output=True):
|
||
def no_read():
|
||
if output:
|
||
print_now(f"未填写环境变量 {key} 请添加")
|
||
return default
|
||
return get_cookie_data(key) if get_cookie_data(key) else no_read()
|
||
|
||
#获取ck
|
||
def get_cookie_data(name):
|
||
ck_list = []
|
||
remarks_list = []
|
||
cookie = None
|
||
cookies = get_config_and_envs(name)
|
||
for ck in cookies:
|
||
data_temp = {}
|
||
if ck["name"] != name:
|
||
continue
|
||
if ck.get('status') == 0:
|
||
# ck_list.append(ck.get('value'))
|
||
# 直接添加CK
|
||
ck_list.append(ck)
|
||
# if len(ck_list) < 1:
|
||
# print('变量{}共配置{}条CK,请添加环境变量,或查看环境变量状态'.format(name,len(ck_list)))
|
||
return ck_list
|
||
|
||
# 修改print方法 避免某些环境下python执行print 不会去刷新缓存区导致信息第一时间不及时输出
|
||
def print_now(content):
|
||
print(content)
|
||
stdout.flush()
|
||
|
||
|
||
# 查询环境变量
|
||
def get_envs(name: str = None) -> list:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
if name is not None:
|
||
params['searchValue'] = name
|
||
res = requests.get(ql_url + '/api/envs', headers=__get__headers(), params=params)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return j_data['data']
|
||
return []
|
||
|
||
|
||
# 查询环境变量+config.sh变量
|
||
def get_config_and_envs(name: str = None) -> list:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
#返回的数据data
|
||
data = []
|
||
if name is not None:
|
||
params['searchValue'] = name
|
||
res = requests.get(ql_url + '/api/envs', headers=__get__headers(), params=params)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
data = j_data['data']
|
||
with open(ql_config_path, 'r', encoding='utf-8') as f:
|
||
while True:
|
||
# Get next line from file
|
||
line = f.readline()
|
||
# If line is empty then end of file reached
|
||
if not line :
|
||
break;
|
||
#print(line.strip())
|
||
exportinfo = line.strip().replace("\"","").replace("\'","")
|
||
#去除注释#行
|
||
rm_str_list = re.findall(r'^#(.+?)', exportinfo,re.DOTALL)
|
||
#print('rm_str_list数据：{}'.format(rm_str_list))
|
||
exportinfolist = []
|
||
if len(rm_str_list) == 1:
|
||
exportinfo = ""
|
||
#list_all = re.findall(r'export[ ](.+?)', exportinfo,re.DOTALL)
|
||
#print('exportinfo数据：{}'.format(exportinfo))
|
||
#以export分隔，字符前面新增标记作为数组0，数组1为后面需要的数据
|
||
list_all = ("标记"+exportinfo.replace(" ","").replace(" ","")).split("export")
|
||
#print('list_all数据：{}'.format(list_all))
|
||
if len(list_all) > 1:
|
||
#以=分割，查找需要的环境名字
|
||
tmp = list_all[1].split("=")
|
||
if len(tmp) > 1:
|
||
|
||
info = tmp[0]
|
||
if name in info:
|
||
#print('需要查询的环境数据：{}'.format(tmp))
|
||
data_tmp = []
|
||
data_json = {
|
||
'id': None,
|
||
'value': tmp[1],
|
||
'status': 0,
|
||
'name': name,
|
||
'remarks': "",
|
||
'position': None,
|
||
'timestamp': int(time.time()*1000),
|
||
'created': int(time.time()*1000)
|
||
}
|
||
if flag == 'old':
|
||
data_json = {
|
||
'_id': None,
|
||
'value': tmp[1],
|
||
'status': 0,
|
||
'name': name,
|
||
'remarks': "",
|
||
'position': None,
|
||
'timestamp': int(time.time()*1000),
|
||
'created': int(time.time()*1000)
|
||
}
|
||
#print('需要的数据：{}'.format(data_json))
|
||
data.append(data_json)
|
||
#print('第二次配置数据：{}'.format(data))
|
||
return data
|
||
|
||
|
||
# 新增环境变量
|
||
def post_envs(name: str, value: str, remarks: str = None) -> list:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
data = [{
|
||
'name': name,
|
||
'value': value
|
||
}]
|
||
if remarks is not None:
|
||
data[0]['remarks'] = remarks
|
||
res = requests.post(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return j_data['data']
|
||
return []
|
||
|
||
|
||
# 修改环境变量1，青龙2.11.0以下版本（不含2.11.0）
|
||
def put_envs_old(_id: str, name: str, value: str, remarks: str = None) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
|
||
data = {
|
||
'name': name,
|
||
'value': value,
|
||
'_id': _id
|
||
}
|
||
|
||
if remarks is not None:
|
||
data['remarks'] = remarks
|
||
res = requests.put(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|
||
|
||
|
||
# 修改环境变量2，青龙2.11.0以上版本（含2.11.0）
|
||
def put_envs_new(_id: int, name: str, value: str, remarks: str = None) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
|
||
data = {
|
||
'name': name,
|
||
'value': value,
|
||
'id': _id
|
||
}
|
||
|
||
if remarks is not None:
|
||
data['remarks'] = remarks
|
||
res = requests.put(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|
||
|
||
|
||
# 禁用环境变量
|
||
def disable_env(_id: str) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
data = [_id]
|
||
res = requests.put(ql_url + '/api/envs/disable', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|
||
|
||
|
||
# 启用环境变量
|
||
def enable_env(_id: str) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
data = [_id]
|
||
res = requests.put(ql_url + '/api/envs/enable', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|
||
|
||
# 删除环境变量
|
||
def delete_env(_id: str) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
data = [_id]
|
||
res = requests.delete(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|
||
|
||
|
||
|
||
|
||
def base64_encode(data):
|
||
message_bytes = data.encode('utf-8') # 将字符串转换为字节型
|
||
base64_data = base64.b64encode(message_bytes) #进行加密
|
||
# base64_data = base64.b64encode(data) # 进行加密
|
||
# print(base64_data,type(base64_data),len(base64_data))
|
||
base64_data = base64_data.decode('utf-8')
|
||
return base64_data
|
||
|
||
def base64_decode(data):
|
||
#base64_bytes = data.encode('utf-8')
|
||
message_bytes = base64.b64decode(data)
|
||
message = message_bytes.decode('utf-8')
|
||
return message
|
||
|
||
|
||
|
||
# WXPUSHER_TOKEN
|
||
WoChangYouCK_WXPUSHER_TOKEN_temp = get_cookie("WoChangYouCK_WXPUSHER_TOKEN")
|
||
if WoChangYouCK_WXPUSHER_TOKEN_temp != "" and len(WoChangYouCK_WXPUSHER_TOKEN_temp)>0:
|
||
WXPUSHER_TOKEN = WoChangYouCK_WXPUSHER_TOKEN_temp[0]["value"]
|
||
|
||
# WXPUSHER_TOPIC_ID
|
||
WoChangYouCK_WXPUSHER_TOPIC_ID_temp = get_cookie("WoChangYouCK_WXPUSHER_TOPIC_ID")
|
||
if WoChangYouCK_WXPUSHER_TOPIC_ID_temp != "" and len(WoChangYouCK_WXPUSHER_TOPIC_ID_temp)>0:
|
||
WXPUSHER_TOPIC_ID = WoChangYouCK_WXPUSHER_TOPIC_ID_temp[0]["value"]
|
||
|
||
msg = ""
|
||
isDebugger = False
|
||
|
||
# 是否生成沃畅游access_token参数提交至青龙环境，默认False不生成，需要生成请填入True
|
||
IsWoChangYouCK = False
|
||
IsWoChangYouCK_temp = get_cookie("IsWoChangYouCK")
|
||
if IsWoChangYouCK_temp != "" and len(IsWoChangYouCK_temp)>0:
|
||
IsWoChangYouCKStr = IsWoChangYouCK_temp[0]["value"]
|
||
if IsWoChangYouCKStr == "True":
|
||
IsWoChangYouCK = True
|
||
|
||
# 是否生成联通app参数提交至青龙环境，默认False不生成，需要生成请填入True
|
||
IsChinaUnicomParam = False
|
||
IsChinaUnicomParam_temp = get_cookie("IsChinaUnicomParam")
|
||
if IsChinaUnicomParam_temp != "" and len(IsChinaUnicomParam_temp)>0:
|
||
IsChinaUnicomParamStr = IsChinaUnicomParam_temp[0]["value"]
|
||
if IsChinaUnicomParamStr == "True":
|
||
IsChinaUnicomParam = True
|
||
# 生成token后添加到指定环境变量
|
||
envName = "chinaUnicomCookie"
|
||
ChinaUnicomEnvName_temp = get_cookie("ChinaUnicomEnvName")
|
||
if ChinaUnicomEnvName_temp != "" and len(ChinaUnicomEnvName_temp)>0:
|
||
envName = ChinaUnicomEnvName_temp[0]["value"]
|
||
# 生成的多账号使用的拼接连接符，默认使用&
|
||
chatMoreStr = "&"
|
||
ChinaUnicomParam_chatMoreStr_temp = get_cookie("ChinaUnicomParam_chatMoreStr")
|
||
if ChinaUnicomParam_chatMoreStr_temp != "" and len(ChinaUnicomParam_chatMoreStr_temp)>0:
|
||
chatMoreStr = ChinaUnicomParam_chatMoreStr_temp[0]["value"]
|
||
# 是否拼接appid，默认False不拼接，需要拼接请填入True
|
||
chinaUnicomParam_flag = False
|
||
ChinaUnicomParam_flag_temp = get_cookie("ChinaUnicomParam_flag")
|
||
if ChinaUnicomParam_flag_temp != "" and len(ChinaUnicomParam_flag_temp)>0:
|
||
chinaUnicomParam_flagStr = ChinaUnicomParam_flag_temp[0]["value"]
|
||
if chinaUnicomParam_flagStr == "True":
|
||
chinaUnicomParam_flag = True
|
||
# appid和token_online之间的拼接符号，默认使用#
|
||
chatStr = "#"
|
||
ChinaUnicomParam_chatStr_temp = get_cookie("ChinaUnicomParam_chatStr")
|
||
if ChinaUnicomParam_chatStr_temp != "" and len(ChinaUnicomParam_chatStr_temp)>0:
|
||
chatStr = ChinaUnicomEnvName_temp[0]["value"]
|
||
# 拼接appid是在token_online前面还是后面，1代表前面，2代表后面,默认appid在前面
|
||
appidIndex = 1
|
||
ChinaUnicomParam_appidIndex_temp = get_cookie("ChinaUnicomParam_appidIndex")
|
||
if ChinaUnicomParam_appidIndex_temp != "" and len(ChinaUnicomParam_appidIndex_temp)>0:
|
||
appidIndex = ChinaUnicomParam_appidIndex_temp[0]["value"]
|
||
|
||
# 需要提交青龙的联通参数
|
||
dataParam = ""
|
||
|
||
class RSA_Encrypt:
|
||
def __init__(self, key):
|
||
if isinstance(key, str):
|
||
# 若提供的rsa公钥不为pem格式 则先将hex转化为pem格式
|
||
# self.key = bytes.fromhex(key) if "PUBLIC KEY" not in key else key.encode()
|
||
self.key = self.public_key(key) if "PUBLIC KEY" not in key else key.encode()
|
||
else:
|
||
print("提供的公钥格式不正确")
|
||
|
||
def public_key(self, rsaExponent, rsaModulus=10001):
|
||
e = int(rsaExponent, 16)
|
||
n = int(rsaModulus, 16) # snipped for brevity
|
||
pubkey = construct((n, e)).export_key()
|
||
return pubkey
|
||
|
||
def encrypt(self, data, b64=False):
|
||
data = data.encode('utf-8')
|
||
length = len(data)
|
||
default_length = 117
|
||
pub_key = importKey(self.key)
|
||
cipher = PKCS1_v1_5.new(pub_key)
|
||
if length < default_length:
|
||
rsa_text = cipher.encrypt(data)
|
||
return b64encode(rsa_text).decode() if b64 else rsa_text.hex()
|
||
offset = 0
|
||
res = []
|
||
while length - offset > 0:
|
||
if length - offset > default_length:
|
||
res.append(cipher.encrypt(data[offset:offset + default_length]))
|
||
else:
|
||
res.append(cipher.encrypt(data[offset:]))
|
||
offset += default_length
|
||
byte_data = b''.join(res)
|
||
return b64encode(byte_data).decode() if b64 else byte_data.hex()
|
||
|
||
|
||
|
||
class UnicomLogin:
|
||
def __init__(self, phone: str, password: str):
|
||
self.rsa_key = "-----BEGIN PUBLIC KEY-----\nMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDc+CZK9bBA9IU+gZUOc6FUGu7y\nO9WpTNB0PzmgFBh96Mg1WrovD1oqZ+eIF4LjvxKXGOdI79JRdve9NPhQo07+uqGQ\ngE4imwNnRx7PFtCRryiIEcUoavuNtuRVoBAm6qdB0SrctgaqGfLgKvZHOnwTjyNq\njBUxzMeQlEC2czEMSwIDAQAB\n-----END PUBLIC KEY-----"
|
||
self.phone_num = phone.rstrip("\n")
|
||
self.password = password.rstrip("\n")
|
||
self.deviceId = uuid4().hex
|
||
self.appid = str(random.randint(0, 10))+"f"+str(random.randint(0, 10))+"af"+str(random.randint(0, 10))+str(random.randint(0, 10))+"ad"+str(random.randint(0, 10))+"912d306b5053abf90c7ebbb695887bc870ae0706d573c348539c26c5c0a878641fcc0d3e90acb9be1e6ef858a59af546f3c826988332376b7d18c8ea2398ee3a9c3db947e2471d32a49612"
|
||
self.access_token = ""
|
||
self.UA = "Mozilla/5.0 (Linux; Android 13; LE2100 Build/TP1A.220905.001; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/103.0.5060.129 Mobile Safari/537.36; unicom{version:android@11.0300,desmobile:"+self.phone_num+"};devicetype{deviceBrand:OnePlus,deviceModel:LE2100};{yw_code:}"
|
||
|
||
|
||
def login_unicom(self):
|
||
global dataParam
|
||
# print_now(self.phone_num+"---------"+self.password)
|
||
headers = {
|
||
'Host': 'm.client.10010.com',
|
||
'Accept': '*/*',
|
||
# 'User-Agent': 'ChinaUnicom.x CFNetwork iOS/15.0.1 unicom{version:iphone_c@11.0300}',
|
||
'User-Agent': self.UA,
|
||
'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
|
||
'Content-Type': 'application/x-www-form-urlencoded',
|
||
}
|
||
data = f"isFirstInstall=1&simCount=1&yw_code=&deviceOS=android13&mobile={quote(RSA_Encrypt(self.rsa_key).encrypt(self.phone_num, b64=True))}&netWay=Wifi&version=android%4011.0300&deviceId={self.deviceId}&password={quote(RSA_Encrypt(self.rsa_key).encrypt(self.password, b64=True))}&keyVersion=&provinceChanel=general&appId=ChinaunicomMobileBusiness&deviceModel=LE2100&androidId={uuid4().hex[8:24]}&deviceBrand=×tamp={datetime.today().__format__('%Y%m%d%H%M%S')}"
|
||
|
||
# data = {
|
||
# "version": "iphone_c@10.0700",
|
||
# "mobile": quote(RSA_Encrypt(self.rsa_key).encrypt(self.phone_num, b64=True)),
|
||
# "appId": self.appid,
|
||
# "deviceId": self.deviceId,
|
||
# "password": quote(RSA_Encrypt(self.rsa_key).encrypt(self.password, b64=True)),
|
||
# }
|
||
response = requests.post('https://m.client.10010.com/mobileService/login.htm', headers=headers,data=data)
|
||
data = response.json()
|
||
# print_now(f"{data}")
|
||
self.ecs_token = data.get("ecs_token")
|
||
if data.get("code") == '4':
|
||
print_now(f'账号【{self.phone_num}】账号密码错误，跳过')
|
||
return self.ecs_token
|
||
self.token_online = data.get("token_online")
|
||
if self.token_online == "" or self.token_online is None:
|
||
print_now(f'账号【{self.phone_num}】获取token_online失败，成功获取到【appid】：{self.appid}')
|
||
return self.ecs_token
|
||
else:
|
||
print_now(f'账号【{self.phone_num}】成功获取到【token_online】：{self.token_online}\n账号【{self.phone_num}】成功获取到【ecs_token】：{self.ecs_token}\n账号【{self.phone_num}】成功获取到【appid】：{self.appid}\n账号【{self.phone_num}】成功获取到【deviceId】：{self.deviceId}')
|
||
|
||
# 拼接参数appid
|
||
if chinaUnicomParam_flag == True or chinaUnicomParam_flag == "True":
|
||
# 在前
|
||
if appidIndex == 1:
|
||
print_now(f'账号【{self.phone_num}】生成appid+token为：\n{self.appId}{chatStr}{self.token_online}')
|
||
# 使用chatMoreStr拼接
|
||
dataParam = dataParam+chatMoreStr+self.appid+chatStr+self.token_online;
|
||
else:
|
||
# 在后
|
||
dataParam = dataParam+chatMoreStr+self.token_online+chatStr+self.appid;
|
||
else:
|
||
# 使用chatMoreStr拼接
|
||
dataParam = dataParam+chatMoreStr+self.token_online;
|
||
return self.ecs_token
|
||
|
||
|
||
# 方式一登录沃畅游
|
||
def get_wo_speed_ticket(self):
|
||
if self.ecs_token == "" or self.ecs_token is None:
|
||
return ""
|
||
|
||
cookies = {
|
||
'ecs_token': self.ecs_token,
|
||
}
|
||
|
||
headers = {
|
||
'Host': 'm.client.10010.com',
|
||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
|
||
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 15_0_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 unicom{version:iphone_c@10.0700}',
|
||
'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
|
||
'Referer': 'https://m.client.10010.com',
|
||
}
|
||
|
||
params = {
|
||
'to_url': 'https://web.wostore.cn/web/flowGame/index.html?channelId=GAMELTAPP_90006',
|
||
}
|
||
|
||
response = requests.get(
|
||
'https://m.client.10010.com/mobileService/openPlatform/openPlatLineNew.htm',
|
||
params=params,
|
||
cookies=cookies,
|
||
headers=headers,
|
||
allow_redirects=False
|
||
)
|
||
location = response.headers['Location']
|
||
return location[location.find("ticket=") + len('ticket='):location.rfind("&versio")]
|
||
|
||
|
||
def wo_speed_login_one(self):
|
||
if self.ticket == "" or self.ticket is None:
|
||
return ""
|
||
headers = {
|
||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
|
||
'Connection': 'keep-alive',
|
||
'Content-Type': 'application/json;charset=utf-8',
|
||
'Origin': 'https://web.wostore.cn',
|
||
'Referer': 'https://web.wostore.cn/',
|
||
'Sec-Fetch-Dest': 'empty',
|
||
'Sec-Fetch-Mode': 'cors',
|
||
'Sec-Fetch-Site': 'same-site',
|
||
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 Mobile/15E148 Safari/604.1 Edg/109.0.0.0',
|
||
'accept': 'application/json',
|
||
'channelId': 'GAMELTAPP_90005',
|
||
'device': '8',
|
||
'rnVersion': '',
|
||
'versionCode': '',
|
||
}
|
||
|
||
json_data = {
|
||
'identityType': 'ticketToken',
|
||
'code': self.ticket,
|
||
'uuid': '3cc1b0ff-ddb4-49dc-9e87-4d51811f7ea1',
|
||
}
|
||
|
||
response = requests.post('https://game.wostore.cn/api/app/user/v2/login', headers=headers, json=json_data)
|
||
d = response.json().get('data')
|
||
if not d or d == "":
|
||
print_now(f"可能是首次登录沃畅游。无法获取access_token，可先手动去联通app首页--5g新通信--联通畅游--个人中心，登录一下")
|
||
return ""
|
||
access_token = d.get('access_token',None)
|
||
if not access_token and access_token != "":
|
||
return access_token
|
||
else:
|
||
print_now(f"可能是首次登录沃畅游。无法获取access_token，可先手动去联通app首页--5g新通信--联通畅游--个人中心，登录一下")
|
||
return ""
|
||
|
||
|
||
|
||
# 方式二登录
|
||
def wo_speed_login_two(self):
|
||
if self.ecs_token == "" or self.ecs_token is None:
|
||
return ""
|
||
count = 0
|
||
while True:
|
||
if count < 3 and self.access_token == "":
|
||
count = self.login(count)
|
||
if self.access_token == "":
|
||
print_now(f"休眠3秒，再次请求获取access_token\n\n")
|
||
time.sleep(3)
|
||
else:
|
||
break
|
||
if self.access_token == "":
|
||
print_now(f"可能是首次登录沃畅游。无法获取access_token，可先手动去联通app首页--5g新通信--联通畅游，登录一下")
|
||
|
||
|
||
def login(self,count):
|
||
url = 'https://m.client.10010.com/mobileService/openPlatform/openPlatLineNew.htm?to_url=https://web.wostore.cn/web/flowGame/index.html?channelId=GAMELTAPP_90006&pushid=99'
|
||
headers = {
|
||
'Content-Type': 'application/json;charset=utf-8',
|
||
# "User-Agent": "Mozilla/5.0 (Linux; Android 13; LE2100 Build/TP1A.220905.001; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/103.0.5060.129 Mobile Safari/537.36; unicom{version:android@10.0600,desmobile:"+self.phone_num+"};devicetype{deviceBrand:OnePlus,deviceModel:LE2100};{yw_code:}",
|
||
"Cookie": "ecs_token=" + self.ecs_token
|
||
}
|
||
try:
|
||
response = requests.get(url, headers=headers,allow_redirects=False)
|
||
# print_now(f"【{time.strftime('%Y-%m-%d %H:%M:%S')}】 ---- 【{self.phone_num}】发送成功，响应：{response.headers}")
|
||
# 获取location
|
||
location = response.headers.get("location",None)
|
||
headers["User-Agent"] = "Mozilla/5.0 (Linux; Android 13; LE2100 Build/TP1A.220905.001; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/103.0.5060.129 Mobile Safari/537.36; unicom{version:android@10.0600,desmobile:"+self.phone_num+"};devicetype{deviceBrand:OnePlus,deviceModel:LE2100};{yw_code:}"
|
||
# 获取ticket、channelId
|
||
ticket = ""
|
||
channelId = ""
|
||
|
||
if location is not None and location != "":
|
||
key_value = location.split("?")[1].split("&")
|
||
for i in range(len(key_value)):
|
||
key_value_temp = key_value[i].split("=")
|
||
if key_value_temp[0] == "ticket":
|
||
ticket = key_value_temp[1]
|
||
if key_value_temp[0] == "channelId":
|
||
channelId = key_value_temp[1]
|
||
|
||
# 登录
|
||
url = 'https://web.wostore.cn/api/app//user/v2/login'
|
||
headers["channelId"] = channelId
|
||
data = {
|
||
'identityType': 'esToken',
|
||
'code': self.ecs_token,
|
||
'ticket': ticket,
|
||
'uuid': "3cc1b0ff-ddb4-49dc-9e87-4d51811f7ea1"
|
||
}
|
||
# print_now(f'location：{location}\ticket：{ticket}\nheaders:{headers}\ndata:{data}')
|
||
response = requests.post(url, headers=headers, data=json.dumps(data))
|
||
text = response.json()
|
||
# print_now(f"【{time.strftime('%Y-%m-%d %H:%M:%S')}】 ---- 【{self.phone_num}】 登录成功，响应：{text}\n")
|
||
self.access_token = ""
|
||
if text["code"] == 200:
|
||
data = text["data"]
|
||
access_token = data["access_token"]
|
||
# print_now(f'账号{self.phone_num}登录成功，成功获取access_token,额外需要使用时请复制保存: {access_token}\n')
|
||
if access_token is None or access_token == "":
|
||
self.access_token = ""
|
||
count += 1
|
||
else:
|
||
self.access_token = access_token
|
||
else:
|
||
print_now(f"【{time.strftime('%Y-%m-%d %H:%M:%S')}】 ---- 【{self.phone_num}】 登录请求响应：{text}\n")
|
||
self.access_token = ""
|
||
count += 1
|
||
except Exception as e:
|
||
count += 1
|
||
print_now(f"\n\n账号{self.phone_num}登录请求出现错误,出错响应内容：{e}\n\n正在第{count}次重试中。。。" )
|
||
self.access_token = ""
|
||
return count
|
||
|
||
|
||
|
||
def deal_data(self):
|
||
global msg
|
||
global IsWoChangYouCK
|
||
if self.access_token == "" or self.access_token is None:
|
||
print_now(f'账号【{self.phone_num}】获取access_token失败')
|
||
msg += f'账号【{self.phone_num}】获取access_token失败\n\n'
|
||
return ""
|
||
print_now(f'账号【{self.phone_num}】成功获取到【access_token】：{self.access_token}\n请复制保存使用')
|
||
# 是否填入
|
||
if IsWoChangYouCK == False or IsWoChangYouCK is None:
|
||
print_now(f'系统设置不写入access_token至青龙环境')
|
||
return ""
|
||
else:
|
||
print_now(f'系统设置写入access_token至青龙环境,开始写入。。。')
|
||
try:
|
||
# 获取沃畅游CK
|
||
cklist_temp = get_cookie("WoChangYouCK")
|
||
flag_temp = False
|
||
if len(cklist_temp)>0:
|
||
for i in range(len(cklist_temp)):
|
||
ck_temp = cklist_temp[i]
|
||
if ck_temp["remarks"] == phone:
|
||
flag_temp = True
|
||
put_flag = True
|
||
if flag == "old":
|
||
_id = ck_temp.get("_id",None)
|
||
if not _id:
|
||
_id = ck_temp["id"]
|
||
put_flag = put_envs_new(_id, ck_temp['name'], self.access_token, phone)
|
||
else:
|
||
put_flag = put_envs_old(_id, ck_temp['name'], self.access_token, phone)
|
||
# print("进入旧版本青龙禁用方法")
|
||
# disable_env(_id)
|
||
# delete_env(_id)
|
||
elif flag == "new":
|
||
put_flag = put_envs_new(ck_temp["id"], ck_temp['name'], self.access_token, phone)
|
||
# print("进入新版本青龙禁用方法")
|
||
# disable_env(ck_temp["id"])
|
||
# delete_env(ck_temp["id"])
|
||
if put_flag:
|
||
print_now(f"账号【{self.phone_num}】自动更新access_token至青龙环境：WoChangYouCK 备注为：{phone}")
|
||
msg += f"账号【{phone}】自动更新access_token至青龙环境：WoChangYouCK 备注为：{phone}\n\n"
|
||
else:
|
||
print_now(f"账号【{self.phone_num}】自动更新access_token至青龙环境：失败")
|
||
msg += f"账号【{phone}】自动更新access_token至青龙环境：失败\n\n"
|
||
if not flag_temp:
|
||
post_envs("WoChangYouCK", self.access_token, phone)
|
||
print_now(f"账号【{self.phone_num}】自动新增access_token至青龙环境：WoChangYouCK 备注为：{phone}")
|
||
msg += f"账号【{phone}】自动新增access_token至青龙环境：WoChangYouCK 备注为：{phone}\n\n"
|
||
except Exception as e:
|
||
print_now(f"【{time.strftime('%Y-%m-%d %H:%M:%S')}】 ---- 【{phone}】 登录失败，错误信息：{e}\n")
|
||
msg += f"【{time.strftime('%Y-%m-%d %H:%M:%S')}】 ---- 【{phone}】 登录失败，错误信息：{e}\n\n"
|
||
|
||
|
||
def main(self):
|
||
self.login_unicom()
|
||
# 方式一登录
|
||
# self.ticket = self.get_wo_speed_ticket()
|
||
# self.access_token = self.wo_speed_login_one()
|
||
# 方式二登录
|
||
self.wo_speed_login_two()
|
||
self.deal_data()
|
||
|
||
|
||
# 将生成的环境变量填入青龙环境中
|
||
def post_data_to_env():
|
||
global msg
|
||
global dataParam
|
||
# 是否填入
|
||
if IsChinaUnicomParam == False or IsChinaUnicomParam is None:
|
||
print_now(f'系统设置不写入token_online至青龙环境')
|
||
return ""
|
||
else:
|
||
print_now(f'系统设置写入token_online至青龙环境,开始写入。。。')
|
||
# 截取第一位拼接参数
|
||
if len(dataParam)>1:
|
||
dataParam = dataParam[1:]
|
||
else:
|
||
print_now(f'未获取到token_online参数')
|
||
return ""
|
||
try:
|
||
# 获取环境CK
|
||
cklist_temp = get_cookie(envName)
|
||
flag_temp = False
|
||
if len(cklist_temp)>0:
|
||
for i in range(len(cklist_temp)):
|
||
ck_temp = cklist_temp[i]
|
||
if ck_temp["remarks"] == "由“联通和沃畅游密码登录”脚本自动生成":
|
||
flag_temp = True
|
||
put_flag = True
|
||
if flag == "old":
|
||
_id = ck_temp.get("_id",None)
|
||
if not _id:
|
||
_id = ck_temp["id"]
|
||
put_flag = put_envs_new(_id, ck_temp['name'], dataParam, "由“联通和沃畅游密码登录”脚本自动生成")
|
||
else:
|
||
put_flag = put_envs_old(_id, ck_temp['name'], dataParam, "由“联通和沃畅游密码登录”脚本自动生成")
|
||
# print("进入旧版本青龙禁用方法")
|
||
# disable_env(_id)
|
||
# delete_env(_id)
|
||
elif flag == "new":
|
||
put_flag = put_envs_new(ck_temp["id"], ck_temp['name'], dataParam, "由“联通和沃畅游密码登录”脚本自动生成")
|
||
# print("进入新版本青龙禁用方法")
|
||
# disable_env(ck_temp["id"])
|
||
# delete_env(ck_temp["id"])
|
||
if put_flag:
|
||
print_now(f"已将全部正常执行生成联通参数的账号,自动更新至青龙环境：{envName} 由“联通和沃畅游密码登录”脚本自动生成")
|
||
msg += f"已将全部正常执行生成联通参数的账号,自动更新至青龙环境：{envName} 由“联通和沃畅游密码登录”脚本自动生成\n\n"
|
||
else:
|
||
print_now(f"全部正常执行生成联通参数的账号,自动更新至青龙环境：失败")
|
||
msg += f"全部正常执行生成联通参数的账号,自动更新至青龙环境：失败\n\n"
|
||
if not flag_temp:
|
||
post_envs(envName, dataParam, "由“联通和沃畅游密码登录”脚本自动生成")
|
||
print_now(f"已将全部正常执行生成联通参数的账号,自动新增至青龙环境：{envName} 备注为：由“联通和沃畅游密码登录”脚本自动生成")
|
||
msg += f"已将全部正常执行生成联通参数的账号,自动新增至青龙环境：{envName} 备注为：由“联通和沃畅游密码登录”脚本自动生成\n\n"
|
||
except Exception as e:
|
||
print_now(f"【{time.strftime('%Y-%m-%d %H:%M:%S')}】 ---- 联通和沃畅游密码登录脚本生成参数提交青龙失败，错误信息：{e}\n")
|
||
msg += f"【{time.strftime('%Y-%m-%d %H:%M:%S')}】 ---- 联通和沃畅游密码登录脚本生成参数提交青龙失败，错误信息：{e}\n\n"
|
||
|
||
|
||
def start(phone,password):
|
||
ul = UnicomLogin(phone,password)
|
||
ul.main()
|
||
|
||
if __name__ == "__main__":
|
||
l = []
|
||
ck_list = []
|
||
cklist = get_cookie("WoChangYouCK_PSW")
|
||
for i in range(len(cklist)):
|
||
|
||
#多账号以#分割开的ck
|
||
split1 = cklist[i]['value'].split("#")
|
||
#多账号以@分割开的ck
|
||
split2 = cklist[i]['value'].split("@")
|
||
#多账号以换行\n分割开的ck
|
||
split3 = cklist[i]['value'].split("\n")
|
||
remarks = cklist[i].get("remarks",None)
|
||
if len(split1)>1:
|
||
for j in range(len(split1)):
|
||
info = {}
|
||
info['value'] = split1[j]
|
||
info['remarks'] = split1[j].split("&")[0]
|
||
ck_list.append(info)
|
||
elif len(split2)>1:
|
||
for j in range(len(split2)):
|
||
info = {}
|
||
info['value'] = split2[j]
|
||
info['remarks'] = split2[j].split("&")[0]
|
||
ck_list.append(info)
|
||
elif len(split3)>1:
|
||
for j in range(len(split3)):
|
||
info = {}
|
||
info['value'] = split3[j]
|
||
info['remarks'] = split3[j].split("&")[0]
|
||
ck_list.append(info)
|
||
else:
|
||
if remarks is None or remarks == "":
|
||
cklist[i]['remarks'] = cklist[i]['value']
|
||
ck_list.append(cklist[i])
|
||
if len(ck_list)<1:
|
||
print_now('未添加CK,退出程序~')
|
||
exit(0)
|
||
|
||
|
||
for i in range(len(ck_list)):
|
||
ck = ck_list[i]
|
||
data = ck.get("value",None)
|
||
if data is None:
|
||
print_now("当前账号未填写 跳过\n")
|
||
continue
|
||
tmp_list = data.split("&")
|
||
if len(tmp_list)<2:
|
||
print_now("参数不齐 跳过\n")
|
||
continue
|
||
phone = tmp_list[0]
|
||
password = tmp_list[1]
|
||
print_now(f'开始执行第 {i+1} 个账号：{phone}')
|
||
start(phone,password)
|
||
#解决随机时间问题
|
||
ran_time = random.randint(3, 5)
|
||
if isDebugger == False and i != (len(ck_list)-1):
|
||
print_now(f"随机休眠{ran_time}秒，执行下一个账号操作\n\n")
|
||
time.sleep(ran_time)
|
||
else:
|
||
print_now("\n\n")
|
||
# 最后写入联通
|
||
post_data_to_env()
|
||
if WXPUSHER_TOKEN != "" and WXPUSHER_TOPIC_ID != "" and msg != "":
|
||
wxpusher("联通和沃畅游密码登录",msg)
|