Skip to content

Fix_Universal use (Windows, Mac, Linux) #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Kryte wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix_Universal use (Windows, Mac, Linux) #1

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7a7b4cb
Fix_Universal use (Windows, Mac, Linux)
Kryte Feb 2, 2026
06138e3
Refactor account checking and remove unused code
Kryte Feb 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 22 additions & 130 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,17 +4,17 @@
import ctypes
import threading
import requests
import os

from pypresence import Presence
from concurrent.futures import ThreadPoolExecutor, as_completed
from functools import wraps
from logmagix import Logger, Home

# Caricamento configurazione
with open('input/config.toml') as f:
config = toml.load(f)

DEBUG = config['dev'].get('Debug', False)

log = Logger(style=2)

def debug(func_or_message, *args, **kwargs) -> callable:
Expand All @@ -30,14 +30,6 @@ def wrapper(*args, **kwargs):
if DEBUG:
log.debug(f"Debug: {func_or_message}")

def debug_response(response) -> None:
debug(response.headers)
try:
debug(response.text)
except:
debug(response.content)
debug(response.status_code)

class Miscellaneous:
def __init__(self):
self.valid_file_lock = threading.Lock()
Expand All @@ -61,7 +53,6 @@ def get_proxies(self) -> dict:
"http": f"http://{proxy_choice}",
"https": f"http://{proxy_choice}"
}
debug(f"Using proxy: {proxy_choice}")
return proxy_dict
except FileNotFoundError:
log.failure("Proxy file not found. Running in proxyless mode.")
Expand Down Expand Up @@ -105,8 +96,10 @@ def update_title(self, start_time) -> None:
current_total = self._total
title = f'Spotify Checker | Total: {current_total} | Time Elapsed: {elapsed_time}s'

sanitized_title = ''.join(c if c.isprintable() else '?' for c in title)
ctypes.windll.kernel32.SetConsoleTitleW(sanitized_title)
if hasattr(ctypes, 'windll'):
ctypes.windll.kernel32.SetConsoleTitleW(title)
else:
print(f'\33]0;{title}\a', end='', flush=True)
except Exception as e:
log.failure(f"Failed to update console title: {e}")

Expand All @@ -115,176 +108,75 @@ def increment_total(self):
self._total += 1
return self._total

class Status(threading.Thread):
def __init__(self):
threading.Thread.__init__(self, daemon=True)
self.total_accounts = 0
self.start_time = time.time()
self.running = True
self.RPC = None

def stop(self):
self.running = False
if self.RPC:
try:
self.RPC.close()
except:
pass

def run(self):
try:
client_id = "1345064270387216404"
self.RPC = Presence(client_id)
self.RPC.connect()
except Exception as e:
log.failure(f"Failed to connect RPC: {e}")
pass

while self.running:
try:

activity = {
"state": f"Generated: {self.total_accounts} accounts",
"details": "https://spotifyacc.mysellauth.com/",
"large_image": "spotify",
"large_text": "Spotify Account Checker",
"small_image": "logo",
"small_text": "discord.cyberious.xyz",
"start": int(self.start_time)
}

activity["buttons"] = [
{"label": "Discord Server", "url": "https://discord.cyberious.xyz"},
{"label": "Github", "url": "https://github.com/sexfrance"}
]

self.RPC.update(**activity)
time.sleep(15)
except Exception as e:
log.failure(f"RPC update error: {e}")
time.sleep(15)

class AccountChecker:
def __init__(self, proxy_dict: str = None):
self.session = requests.Session()
self.session.headers = {
'accept': '*/*',
'accept-encoding': 'gzip, deflate, br',
}

self.session.headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate, br'}
self.session.proxies = proxy_dict

def check(self, email: str) -> bool:
# Nota: URL di esempio, assicurati che l'API sia attiva
response = self.session.get(f'https://spclient.wg.spotify.com/signup/public/v1/account?email={email}&key=bff58e9698f40080ec4f9ad97a2f21e0&validate=1')

if response.status_code == 200:
if response.json()['status'] == 20:
return True
else:
return False
else:
log.failure(f"Failed to check {email[8]}... : {response.text}, {response.status_code}")

return response.json().get('status') == 20
return None


def check_account(email: str, password: str, Misc: Miscellaneous) -> bool:
max_retries = config['dev'].get('MaxRetries', 3)
retry_delay = 1
account_line = f"{email}:{password}"

for attempt in range(max_retries):
try:
proxies = Misc.get_proxies()
Account_Checker = AccountChecker(proxies)
verified = Account_Checker.check(email)
checker = AccountChecker(proxies)
verified = checker.check(email)

if verified is not None:
if verified:
with Misc.valid_file_lock:
with open("output/valid.txt", "a") as f:
f.write(f"{account_line}\n")
f.flush() # Force write to disk
log.success(f"Valid Account: {email[:8]}... | {password[:8]}...")
log.success(f"Valid: {email[:8]}...")
else:
with Misc.invalid_file_lock:
with open("output/invalid.txt", "a") as f:
f.write(f"{account_line}\n")
f.flush() # Force write to disk
log.failure(f"Invalid Account: {email[:8]}... | {password[:8]}...")
log.failure(f"Invalid: {email[:8]}...")

# Remove the account after successful verification
Misc.remove_account(account_line)
return True

except Exception as e:
if attempt < max_retries - 1:
log.warning(f"Attempt {attempt + 1}/{max_retries} failed for {email[:8]}... : {str(e)}")
time.sleep(retry_delay)
retry_delay *= 2
continue
else:
log.failure(f"All retries failed for {email[:8]}... : {str(e)}")

except:
time.sleep(1)
return False

def main() -> None:
try:
start_time = time.time()

# Initialize basic classes
Misc = Miscellaneous()
Banner = Home("Spotify Checker", align="center", credits="discord.cyberious.xyz")
Banner.display()

thread_count = config['dev'].get('Threads', 1)

# Start updating the title
title_updater = Misc.Title()
title_updater.start_title_updates(start_time)

# Initialize Status thread
status = Misc.Status()
status.start()

# Create output directory if it doesn't exist
import os
os.makedirs("output", exist_ok=True)

# Clear output files at start
open("output/valid.txt", "w").close()
open("output/invalid.txt", "w").close()

# Read all accounts from the file
with open("input/accounts.txt") as f:
accounts = []
for line in f:
line = line.strip()
if ':' in line:
parts = line.split(':')[:2] # Only take first two parts
if len(parts) == 2:
accounts.append(parts)
accounts = [line.strip().split(':')[:2] for line in f if ':' in line]

with ThreadPoolExecutor(max_workers=thread_count) as executor:
futures = []

for email, password in accounts: # Now we can safely unpack
futures.append(executor.submit(check_account, email, password, Misc))

futures = [executor.submit(check_account, acc[0], acc[1], Misc) for acc in accounts]
for future in as_completed(futures):
try:
if future.result():
total = title_updater.increment_total()
status.total_accounts = total
except Exception as e:
log.failure(f"Thread error: {e}")
if future.result():
title_updater.increment_total()

except KeyboardInterrupt:
log.info("Process interrupted by user. Exiting...")
status.stop()
log.info("Exiting...")
except Exception as e:
log.failure(f"An unexpected error occurred: {e}")
status.stop()
log.failure(f"Error: {e}")

if __name__ == "__main__":
main()
main()