-
Notifications
You must be signed in to change notification settings - Fork 2
/
upd.py
151 lines (124 loc) · 5.13 KB
/
upd.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
import aiohttp
import asyncio
import json
import re
import argparse
from bs4 import BeautifulSoup
parser = argparse.ArgumentParser()
parser.add_argument("-v", required=True)
parser.add_argument("-s", required=True)
parser.add_argument("-u", required=True)
args = parser.parse_args()
version = args.v
source = args.s
u = args.u
async def send_request(json_data):
url = u + json_data
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.text()
soup = BeautifulSoup(data, "html.parser")
system_response = soup.find(
"div",
style="text-align:center;font-family:monospace;margin:50px auto 0;max-width:600px",
).text
print(f"Ответ от GAS: {system_response}")
async def check_url(session, url, platform_name):
try:
async with session.get(url) as response:
if response.status == 200:
if platform_name:
return response.url, platform_name
else:
return response.url
except aiohttp.ClientError:
pass
return None
async def pre_version(latest_urls):
message = f'Версия {version} {"отправлена" if latest_urls else "не найдена"}'
print(message)
if not latest_urls:
latest_urls = {"unknown": "unknown", "version": version}
latest_urls["version"] = version
latest_urls["source"] = source
req_ver = json.dumps(latest_urls, ensure_ascii=False, indent=2)
await send_request(req_ver)
def get_urls(find_url):
platform_urls = {}
version_pattern = re.compile(r"-([0-9]+)\.(exe|tbz)")
for url, platform in find_url:
match = version_pattern.search(str(url))
if match:
version_number = int(match.group(1))
if platform in platform_urls:
# Если найдено больше одного url для платформы, то выбираем с самой большой ревизией
current_version = int(
version_pattern.search(platform_urls[platform]).group(1)
)
if version_number > current_version:
platform_urls[platform] = str(url)
else:
platform_urls[platform] = str(url)
return platform_urls
async def main():
start_number = 0
before_enter = 1000
additional_searches = 10 # Максимальное кол-во шагов для дополнительного поиска
increment = 1000 # Размер шага
numbers = start_number
find_url = []
root_url = "https://upgrade.scdn.co/upgrade/client/"
platform_templates = {
"WIN32": "win32-x86/spotify_installer-{version}-{numbers}.exe",
"WIN64": "win32-x86_64/spotify_installer-{version}-{numbers}.exe",
"WIN-ARM64": "win32-arm64/spotify_installer-{version}-{numbers}.exe",
"OSX": "osx-x86_64/spotify-autoupdate-{version}-{numbers}.tbz",
"OSX-ARM64": "osx-arm64/spotify-autoupdate-{version}-{numbers}.tbz",
}
async with aiohttp.ClientSession() as session:
tasks = []
platform_names = list(platform_templates.keys())
for platform_name in platform_names:
numbers = start_number
while numbers <= before_enter:
url = root_url + platform_templates[platform_name].format(
version=version, numbers=numbers
)
tasks.append(check_url(session, url, platform_name))
numbers += 1
results = []
for task in asyncio.as_completed(tasks):
result = await task
if result is not None:
find_url.append(result)
# Дополнительный поиск
for i in range(additional_searches):
if len(get_urls(find_url)) < len(
platform_names
): # Если не найдены все платформы
start_number = before_enter + 1
before_enter += increment
tasks = []
for platform_name in platform_names:
numbers = start_number
while numbers <= before_enter:
url = root_url + platform_templates[platform_name].format(
version=version, numbers=numbers
)
tasks.append(check_url(session, url, platform_name))
numbers += 1
for task in asyncio.as_completed(tasks):
result = await task
if result is not None:
find_url.append(result)
latest_urls = get_urls(find_url)
if len(latest_urls) == len(platform_names):
await pre_version(latest_urls)
return
if find_url:
latest_urls = get_urls(find_url)
await pre_version(latest_urls)
else:
await pre_version(False)
asyncio.run(main())