Creepso/LoaderSpot
1
0
Fork 0
Code Issues Pull requests Projects Releases 10 Packages Wiki Activity Actions

Update LoaderSpot.py

Browse source 
- using head requests and shared session
- added conditional filtering for Win-x86 platform
- allowed user configuration of connection limits
- improved progress bar feedback
- minor changes and corrections
This commit is contained in:
amd64fox 2025-03-31 00:01:44 +03:00
parent e05e9b7b7c
commit b3f627d1c7
1 changed files with 160 additions and 93 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

253
LoaderSpot.py
View file

@ -6,7 +6,6 @@ from tqdm import tqdm
from typing import Optional, Tuple, List, Dict
from dataclasses import dataclass
from enum import Enum
import threading
class Platform(Enum):
@ -24,6 +23,27 @@ class SpotifyVersion:
end_number: int
def extract_base_version(version: str) -> str:
parts = version.split('.')
if len(parts) >= 3:
return f"{parts[0]}.{parts[1]}.{parts[2]}"
return version
def should_use_win_x86(version: str) -> bool:
base_version = extract_base_version(version)
try:
# Split by dot and compare parts
parts = base_version.split('.')
version_tuple = (int(parts[0]), int(parts[1]), int(parts[2]))
# Compare with version 1.2.53 using tuple comparison
return version_tuple <= (1, 2, 53)
except (ValueError, IndexError):
# If there's any error in parsing, default to include
return True
class UrlGenerator:
BASE_URL = "https://upgrade.scdn.co/upgrade/client/"
PLATFORM_PATHS = {
@ -43,48 +63,46 @@ async def check_url(
session: aiohttp.ClientSession, url: str, platform: Platform
) -> Optional[Tuple[str, Platform]]:
try:
async with session.get(url) as response:
async with session.head(url) as response:
if response.status == 200:
return str(response.url), platform
return url, platform
except aiohttp.ClientError:
pass
return None
async def fetch_versions_json() -> dict:
async with aiohttp.ClientSession() as session:
async with session.get(
"https://raw.githubusercontent.com/amd64fox/LoaderSpot/refs/heads/main/versions.json"
) as response:
if response.status == 200:
try:
return json.loads(await response.text())
except json.JSONDecodeError:
return {}
return {}
async def fetch_versions_json(session: aiohttp.ClientSession) -> dict:
async with session.get(
"https://raw.githubusercontent.com/amd64fox/LoaderSpot/refs/heads/main/versions.json"
) as response:
if response.status == 200:
try:
return json.loads(await response.text())
except json.JSONDecodeError:
return {}
return {}
async def submit_to_google_form(version: str, comment: str = "from LoaderSpot") -> None:
async def submit_to_google_form(session: aiohttp.ClientSession, version: str, comment: str = "from LoaderSpot") -> None:
form_url = "https://docs.google.com/forms/u/0/d/e/1FAIpQLSdqIxSjqt2PcjBlQzhvwqc4QckfWuq5qqWsrdpoTidQHsPGpw/formResponse"
data = {"entry.1104502920": version, "entry.1319854718": comment}
async with aiohttp.ClientSession() as session:
try:
await session.post(form_url, data=data)
except Exception:
pass
async def check_version_and_submit(version: str) -> None:
try:
versions_json = await fetch_versions_json()
await session.post(form_url, data=data)
except Exception:
pass
async def check_version_and_submit(session: aiohttp.ClientSession, version: str) -> None:
try:
versions_json = await fetch_versions_json(session)
version_exists = any(
ver_data.get("fullversion") == version
for ver_data in versions_json.values()
)
if not version_exists:
await submit_to_google_form(version)
await submit_to_google_form(session, version)
except Exception:
pass
@ -94,8 +112,12 @@ def validate_version(version: str) -> bool:
return bool(re.match(r"^\d+\.\d+\.\d+\.\d+\.g[0-9a-f]{8}$", version))
def validate_number(number: str, min_val: int, max_val: int) -> bool:
return number.isdigit() and min_val <= int(number) <= max_val
def validate_number(number: str) -> bool:
return number.isdigit()
def validate_range(start: int, end: int) -> bool:
return end >= start and (end - start) <= 20000
def get_valid_input(prompt: str, validator: callable, error_message: str) -> str:
@ -121,29 +143,52 @@ def calculate_total_urls(
async def search_installers(
version_info: SpotifyVersion, selected_platforms: List[Platform]
session: aiohttp.ClientSession, version_info: SpotifyVersion, selected_platforms: List[Platform]
) -> Dict[Platform, List[str]]:
async with aiohttp.ClientSession() as session:
tasks = []
total_urls = calculate_total_urls(
version_info.start_number, version_info.end_number, selected_platforms
)
tasks = []
total_urls = calculate_total_urls(
version_info.start_number, version_info.end_number, selected_platforms
)
for platform in selected_platforms:
for number in range(version_info.start_number, version_info.end_number + 1):
url = UrlGenerator.generate_url(platform, version_info.version, number)
tasks.append(check_url(session, url, platform))
for platform in selected_platforms:
for number in range(version_info.start_number, version_info.end_number + 1):
url = UrlGenerator.generate_url(platform, version_info.version, number)
tasks.append(check_url(session, url, platform))
results: Dict[Platform, List[str]] = {platform: [] for platform in Platform}
results: Dict[Platform, List[str]] = {platform: [] for platform in Platform}
with tqdm(total=total_urls) as pbar:
for task in asyncio.as_completed(tasks):
if result := await task:
url, platform = result
results[platform].append(url)
pbar.update(1)
avg_speed = [0, 0]
custom_bar_format = "{desc}: {percentage:3.0f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}"
with tqdm(total=total_urls, desc="Checking URLs",
bar_format=custom_bar_format + ", ? urls/sec]") as pbar:
def update_speed(rate):
nonlocal custom_bar_format
speed_str = f", {int(rate)} urls/sec]"
pbar.bar_format = custom_bar_format + speed_str
for task in asyncio.as_completed(tasks):
if result := await task:
url, platform = result
results[platform].append(url)
pbar.update(1)
rate = pbar.format_dict.get('rate', 0)
if rate > 0:
avg_speed[0] += rate
avg_speed[1] += 1
update_speed(rate)
if avg_speed[1] > 0:
avg_rate = avg_speed[0] / avg_speed[1]
final_speed_str = f", {int(avg_rate)} urls/sec (avg)]"
pbar.bar_format = custom_bar_format + final_speed_str
pbar.refresh()
return results
return results
def display_results(results: Dict[Platform, List[str]]) -> None:
@ -159,27 +204,33 @@ def display_results(results: Dict[Platform, List[str]]) -> None:
print("\nNothing found, consider increasing the search range")
def get_platform_choices() -> List[Platform]:
def get_platform_choices(version_spoti: str = "") -> List[Platform]:
print("\nSelect the link type for the search:")
for i, platform in enumerate(Platform, 1):
# Filter out WIN_X86 platform for versions > 1.2.53
available_platforms = list(Platform)
if version_spoti and not should_use_win_x86(version_spoti):
available_platforms = [p for p in Platform if p != Platform.WIN_X86]
for i, platform in enumerate(available_platforms, 1):
print(f"[{i}] {platform.value}")
print("[6] All platforms")
print(f"[{len(available_platforms) + 1}] All platforms")
while True:
choices = input("Enter the number(s): ").strip().split(",")
if not all(choice.strip() in "123456" for choice in choices):
print("Invalid input. Please enter numbers between 1 and 6")
if not all(choice.strip().isdigit() and 1 <= int(choice.strip()) <= len(available_platforms) + 1 for choice in choices):
print(f"Invalid input. Please enter numbers between 1 and {len(available_platforms) + 1}")
continue
if "6" in choices:
return list(Platform)
if str(len(available_platforms) + 1) in choices:
return available_platforms
selected = []
for choice in choices:
platform_index = int(choice.strip()) - 1
if 0 <= platform_index < len(Platform):
selected.append(list(Platform)[platform_index])
if 0 <= platform_index < len(available_platforms):
selected.append(available_platforms[platform_index])
if selected:
return selected
@ -187,55 +238,71 @@ def get_platform_choices() -> List[Platform]:
print("Please select at least one valid platform")
def get_max_connections() -> int:
while True:
max_connections = input("Maximum number of concurrent connections (default 100): ").strip()
if not max_connections:
return 100
if max_connections.isdigit() and int(max_connections) > 0:
return int(max_connections)
print("Please enter a valid positive number")
async def main(version_spoti: str = "") -> None:
if not version_spoti:
version_spoti = get_version_input()
# Increase connection limit
max_connections = get_max_connections()
connector = aiohttp.TCPConnector(limit=max_connections)
async with aiohttp.ClientSession(connector=connector) as session:
if not version_spoti:
version_spoti = get_version_input()
version_check_thread = threading.Thread(
target=lambda: asyncio.run(check_version_and_submit(version_spoti)), daemon=True
)
version_check_thread.start()
version_check_task = asyncio.create_task(check_version_and_submit(session, version_spoti))
start_number = int(
get_valid_input(
"Start search from: ",
lambda x: validate_number(x, 0, 20000),
"Number should be between 0 and 20000",
start_number = int(
get_valid_input(
"Start search from: ",
lambda x: validate_number(x),
"Please enter a valid number",
)
)
)
end_number = int(
get_valid_input(
"End search at: ",
lambda x: validate_number(x, start_number, 20000),
f"Number should be between {start_number} and 20000",
end_number = int(
get_valid_input(
"End search at: ",
lambda x: validate_number(x) and validate_range(start_number, int(x)),
f"Please enter a valid number that is at least {start_number} and no more than {start_number + 20000}",
)
)
)
version_info = SpotifyVersion(version_spoti, start_number, end_number)
selected_platforms = get_platform_choices()
version_info = SpotifyVersion(version_spoti, start_number, end_number)
selected_platforms = get_platform_choices(version_spoti)
print("\nSearching...\n")
results = await search_installers(version_info, selected_platforms)
display_results(results)
print("\nSearching...\n")
results = await search_installers(session, version_info, selected_platforms)
display_results(results)
print("\nChoose an option:")
print("[1] Perform the search with a new version")
print("[2] Perform the search again with the same version")
print("[3] Exit")
try:
await asyncio.wait_for(version_check_task, timeout=1.0)
except asyncio.TimeoutError:
pass
choice = input("Enter the number: ")
print("\nChoose an option:")
print("[1] Perform the search with a new version")
print("[2] Perform the search again with the same version")
print("[3] Exit")
if choice == "1":
print("\n")
await main()
elif choice == "2":
print("\n")
print(f"Search version: {version_spoti}")
await main(version_spoti)
elif choice == "3":
return
else:
print("Invalid choice. Exiting the program.")
choice = input("Enter the number: ")
if choice == "1":
print("\n")
await main()
elif choice == "2":
print("\n")
print(f"Search version: {version_spoti}")
await main(version_spoti)
elif choice == "3":
return
else:
print("Invalid choice. Exiting the program.")
if __name__ == "__main__":