import re
import os
import sys
import signal
import json
import time
import uuid
import base64
import random
import string
import logging
import threading
import queue
import urllib.parse
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse
import atexit
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed, wait, FIRST_COMPLETED
from concurrent.futures import TimeoutError as FuturesTimeoutError
# Force UTF-8 on stdout/stderr so unicode glyphs (✓ ✗ ⚠ ·) survive on Windows cp1254 consoles
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from colorama import Fore, Style, init as colorama_init
try:
from curl_cffi import requests as crequests
_HAS_CURL_CFFI = True
except Exception:
_HAS_CURL_CFFI = False
try:
import primp
_HAS_PRIMP = True
except Exception:
_HAS_PRIMP = False
# ============================================================
# CONSTANTS
# ============================================================
# ---- HTTP backend ----
HTTP_BACKEND = "curl_cffi" # "curl_cffi" | "primp"
ALLOW_BACKEND_FALLBACK = True
# Her yeni session için TLS fingerprint havuzdan rastgele seçilir. Spotify
# tek bir fingerprint ile çok hesap açılınca tıkandığı için (Chrome ~100
# hesapta bloke), profil çeşitliliği başarı oranını yüksek tutuyor.
IMPERSONATE_PROFILES_PRIMP = [
# Chrome
"chrome_145", "chrome_146", "chrome_147", "chrome_148",
# Edge
"edge_146", "edge_147", "edge_148",
# Opera
"opera_129", "opera_130", "opera_131",
# Safari
"safari_26", "safari_26.3",
# Firefox
"firefox_146", "firefox_147", "firefox_148",
]
IMPERSONATE_PROFILES_CURL_CFFI = [
# Chrome
"chrome142", "chrome145", "chrome146",
"chrome_148_windows_10", "chrome_148_linux_unknown", "chrome_148_macos_26.0",
"chrome_147_windows_10", "chrome_147_macos_26.0",
# Edge
"edge_147_windows_10", "edge_147_macos_26.0",
# Opera
"opera_130_windows_10", "opear_130_macos_26.0",
# Safari
"safari260", "safari260_ios", "safari2601",
"safari_26.0.1_macos_26.0", "safari_26.0_ios_26.0",
# Firefox
"firefox144", "firefox147",
"firefox_149_linux_unknown", "firefox_149_windows_10", "firefox_145_macos_26.0",
# Tor
"tor145",
]
# Impersonate seçim stratejisi:
# True → HER session'da farklı profil (tekrarsız shuffled-deck): havuz karıştırılır,
# birer birer dağıtılır, havuz bitince yeniden karıştırılır. Ardışık 25 hesabın
# aynı fingerprint'i paylaşması engellenir (fingerprint çeşitliliği maks.).
# False → eski davranış: 1 random profil DEFAULT_THREADS(=25) hesap boyunca sabit,
# grup dolunca yeniden randomize.
IMPERSONATE_HER_SESSION = False
# ---- Proxy (single strings) ----
PROXY = "http://USER452249-zone-custom:x3dC0S3BDN94@global.rotgb.711proxy.com:10000"
STICKY_PROXY = "http://USER452249-zone-custom-session-10593177-sessTime-5-sessAuto-1:x3dC0S3BDN94@global.rotgb.711proxy.com:10000"
NO_PROXY = False
# HTTP session (Spotify account create akışı) hangi proxy'yi kullansın?
# "sticky" → STICKY_PROXY (default — sticky session ID'li, IP rotasyonu yok)
# "regular" → PROXY (sticky session ID'siz, provider havuzundan döner)
# Captcha solver'lar (v2 + v3) her durumda STICKY_PROXY kullanmaya devam eder
# — bu seçim sadece Spotify'a giden HTTP isteklerini etkiler.
ACCOUNT_PROXY_TYPE = "sticky"
# ---- v6_proxy (payload `proxy` alanı için, her hesapta yeni random port) ----
# V6_PROXY:random_port formatında string üretilir; payload'taki `proxy` alanı
# bu üretilen değeri kullanır (account'un Spotify'a bağlanırken kullandığı
# STICKY_PROXY ile farklı bir değer).
V6_PROXY = "194.31.64.219" #194.1.184.88
PROXY_PORT_MIN = 3000
PROXY_PORT_MAX = 7999
# ---- proxy2 (payload `proxy2` alanı) ----
# PROXY2_API_KULLAN=True ise boot'ta bir kez API'den çekilir, program kapanana
# kadar aynı değer kullanılır. Fetch, geçici ağ sorunlarına karşı
# PROXY2_INIT_MAX_RETRIES kez denenir; hepsi tükenirse program durur (fatal).
# False ise hesap'ın kendi proxy'si (scheme stripped) proxy2 olarak yazılır.
PROXY2_API_KULLAN = True
PROXY2_API_URL = "http://global.rotgbapi.711proxy.com:8089/gen?zone=custom&ptype=1&count=1&proto=http&stype=text&split=%5Cr%5Cn&sessType=rotating"
PROXY2_HTTP_TIMEOUT = 10
PROXY2_INIT_MAX_RETRIES = 4 # boot fetch'i — geçici hata/boş yanıtta bu kadar dene
# ---- Output ----
# Tek hedef seçimi:
# "local" → JSON dosyası verimliHesaplar/ klasörüne kaydedilir
# "remote" → login_token'dan sp_dc çıkarılır, COOKIE_UPLOAD_URL'e yüklenir
# (yerele kayıt yapılmaz)
OUTPUT_MODE = "remote"
OUTPUT_DIR = "verimliHesaplar"
# ---- API Keys ----
TWO_CAPTCHA_KEY = "2bf4231b6321cc17b4079c9cf321a8de"
CAPMONSTER_API_KEY = "b0a0bc9b20a0096ee5daf14a6a8caf9b"
CAPSOLVER_API_KEY = "CAP-F94FF3D97CA9E3494072D1865A292F2B2BA56890AD16F0F9E090801C92BEBD6D"
SOLVECAPTCHA_API_KEY = "ab0bf82e7b248a3f49c60ebb2223e2d6"
# ---- Solver selection ----
# V3 provider seçenekleri: "inline" | "2captcha" | "capmonster" | "capsolver" | "solvecaptcha"
# "inline" = orijinal davranış (Google reCAPTCHA Enterprise'a doğrudan
# Solver class ile simüle istek — hiçbir provider key gerektirmez).
# V2 provider seçenekleri:
# "2captcha" | "capmonster" | "capsolver" | "solvecaptcha"
# → tek sağlayıcı (ücretli)
# "inline" → Google'a doğrudan, key
# gerektirmez AMA sadece IP/fingerprint güveni yüksekse token verir.
# Spotify challenge'a düşmüş bir IP zaten Google için şüpheli — pratikte
# başarı oranı düşük (%10-30 temiz residential proxy ile, ~%0 datacenter).
# "inline_then_2captcha" / "inline_then_capmonster" / "inline_then_capsolver" /
# "inline_then_solvecaptcha"
# → önce inline (bedava) dene, fail ederse paid provider'a düş.
# Başarırsa para tasarrufu, başaramazsa fark etmez.
V2_SOLVER = "capsolver" # checkbox captcha (challenge dalı)
V3_SOLVER = "inline" # invisible captcha (account/create)
# CHALLENGE_ATLA: True → hesap açarken challenge (v2 captcha) gelirse ÇÖZME,
# o denemeyi bırak ve yeni hesaba geç (captcha maliyeti/süresinden tasarruf).
# Hem web hem mobil akışta geçerli. Ücretli mail açıksa bırakılan denemenin
# maili finally'de otomatik iptal/iade edilir. False → challenge çözülür (mevcut).
CHALLENGE_ATLA = False
# Session cookie'lerini provider'ın createTask payload'ında `cookies` field'ı
# olarak gönder. Format provider'a göre değişir:
# - 2captcha / Capmonster : `name=value; name=value` (HTTP Cookie header tarzı)
# - solvecaptcha : `KEY1:Value1;KEY2:Value2;` (`_cookies_for` çevirir)
# - Capsolver : cookies field'ı yok, bayrak True olsa da iletilmez
# False → hiçbir provider'a gönderilmez.
CAPTCHA_SEND_COOKIES = True
# ---- SIGNUP MODE (web | mobile) ----
# "web" → mevcut akış (Next.js signup sayfası + JSON account/create + v3 ön-captcha)
# "mobile" → proto2.py akışı (clienttoken + protobuf account/create; v3 yok, doğrudan v2 challenge)
SIGNUP_MODE = "mobile"
# Mobil akış ayarları (yalnızca SIGNUP_MODE == "mobile" iken kullanılır)
# GETCLIENT_VARIANT: clienttoken hangi profille alınır
# "android" = NativeAndroidData + sabit attestation (orijinal mobil yakalama)
# "ios" = NativeIOSData (best-effort)
# "desktop" = NativeWindowsData (elle protobuf; dış bağımlılık yok)
# "web" = open.spotify.com (JSON js_sdk_data)
GETCLIENT_VARIANT = "ios"
# SIGNUP_PLATFORM: signup protobuf'u hangi mobil platform olarak gider ("android" | "ios")
# ÖNEMLİ: GETCLIENT_VARIANT ile AYNI olmalı — clienttoken hangi platforma
# issue edildiyse signup da o platform olmalı. Aksi halde create isteğinde
# Client-Token'ın client_id'si ile X-Client-Id/UA/app_version çelişir (403 flag).
SIGNUP_PLATFORM = "ios"
# MOBILE_DEVICE_POOL_KULLAN: True → her hesap için IOS_DEVICE_POOL/ANDROID_DEVICE_POOL'dan
# tutarlı bir cihaz (model + os) seçilir; clienttoken body/UA + signup UA hep aynı cihazı
# kullanır. False → eski hardcoded tek cihaz (iPhone10,3 / SM-N935F).
MOBILE_DEVICE_POOL_KULLAN = True
# MOBILE_APP_VERSION_POOL_KULLAN: True → her hesap için IOS_APP_VERSIONS/ANDROID_APP_VERSIONS'dan
# bir app sürümü seçilir (client_id SABİT, sadece sürüm çeşitlenir). clienttoken client_version,
# signup app_version/header ve UA hep aynı sürümü kullanır. False → profil default'u.
MOBILE_APP_VERSION_POOL_KULLAN = True
# SYNC_SIGNUP_CLIENT_ID: signup X-Client-Id, clienttoken client_id'siyle aynı mı (True)
# yoksa signup platformunun kendi id'si mi (False — orijinal yakalamaya sadık)
SYNC_SIGNUP_CLIENT_ID = False
# SIGNUP_IMPERSONATE: ARTIK KULLANILMIYOR — mobil session de web ile aynı
# IMPERSONATE_PROFILES_CURL_CFFI havuzundan _pick_impersonate ile rotasyon yapıyor.
# (Geriye uyumluluk için bırakıldı; referans/elle override gerekirse kullanılabilir.)
SIGNUP_IMPERSONATE = "chrome142"
# USE_PROTOBUF_SESSION_ID: invoke/complete için session_id kaynağı (deneme sırası)
# True → önce protobuf signup yanıtındaki session_id
# False → önce challenge sayfası __NEXT_DATA__.sessionId
USE_PROTOBUF_SESSION_ID = True
# SHOW_BODY: mobil signup protobuf ağacını konsola dök (debug). Normalde kapalı.
SHOW_BODY = False
# ---- Profile data (opt-in) ----
USE_PROFILE_DATA = True # True → veritabanlari/isimveCinsiyetEN.json kullan
PROFILE_DATA_FILE = "veritabanlari/isimveCinsiyetEN.json"
# ---- Cookie upload (OUTPUT_MODE == "remote" ile aktif) ----
COOKIE_UPLOAD_URL = "http://spotify.masterswork.pro/sistem/spotifyCookieYaz.php"
SP_DC_AUTH_URL = "https://www.spotify.com/api/signup/authenticate"
SP_DC_CSRF_URL = "https://www.spotify.com/us/signup"
SP_DC_UPLOAD_TIMEOUT = 10
# ---- Infinite mode ----
INFINITE_MODE = True # True → cookie_control_url polling ile batch-batch çalış
COOKIE_CONTROL_URL = "http://spotify.masterswork.pro/sistem/spotifyCookieKontrol.php?kontrol=true"
INFINITE_CHECK_INTERVAL = 3 # seconds between checks
INFINITE_BATCH_SIZE = 25 # accounts per batch
# ---- HTTP retry (lock-protected, 2^n + jitter) ----
HTTP_AUTO_RETRIES = 3
HTTP_RETRY_MAX_BACKOFF = 10
# ---- Threading ----
DEFAULT_THREADS = 25
INTER_SUBMIT_DELAY = 0.10 # saniye — ardışık worker submit'leri arası taban gecikme (0 = burst)
INTER_SUBMIT_JITTER = 0.15 # saniye — submit gecikmesine eklenen rastgele [0, jitter]
WORKER_START_JITTER = 0.50 # saniye — her worker başında rastgele [0, jitter] bekleme (burst kırma)
STATS_REPORT_INTERVAL = 30
# Bir worker'ın bir hesabı bitirmesi için tavan süre (saniye). Bu süreyi
# aşan future "abandoned" sayılır: STATS.record_abandoned() çağrılır, yenisi
# submit edilir. Rate-watchdog abandoned'ı rate hesabında fail gibi sayar.
# Asıl thread Python limit'i nedeniyle öldürülemez — arka planda leak olabilir
# (proxy/captcha provider iyileşince eninde sonunda biter veya GC toplar).
# Amaç pool'u sonsuza dek tıkamamak.
PER_ACCOUNT_TIMEOUT = 180
# ---- Rate watchdog ----
# Her RATE_WATCHDOG_INTERVAL saniyede son interval'daki hesap üretim başarı
# oranı kontrol edilir. Eğer son interval'da:
# - RATE_MIN_SAMPLES'tan az örnek varsa → kontrol atlanır (anlamlı sayı yok)
# - başarı oranı RATE_LOW_THRESHOLD'un altındaysa → RATE_LOW_PAUSE süresince
# tüm worker'lar pause edilir (yeni hesap denemesi başlamaz).
# Pause aktifken yeni örnek üretilmediği için bir sonraki watchdog kontrolü
# atlanır — döngüsel pause tuzağı yok. Pause bitince sıfır sayaçla taze ölçüm.
RATE_WATCHDOG_INTERVAL = 300 # 5 dakika (saniye) — kontrol periyodu
RATE_LOW_THRESHOLD = 10.0 # % — bunun altı düşük rate sayılır
RATE_LOW_PAUSE = 600 # 10 dakika (saniye) — düşük rate'te bekleme
RATE_MIN_SAMPLES = 10 # rate hesabı için son interval'da min örnek
# (< 10: 1-2 fail %50 hata marjı yapar, anlamsız)
# ---- Logging ----
LOG_LEVEL = logging.INFO
LOG_TO_FILE = False
LOG_FILE_NAME = "run.log"
# ---- Locale ----
# URL'ler, header'lar ve reCAPTCHA dili TR pazarıyla tutarlı.
# UYARI: Önceki yorum "tr-TR accept-language ile TCF build boş dönüyor" diyordu;
# eğer bu sorun hâlâ geçerliyse create akışı kırılabilir. Sorun çıkarsa header'ları
# en-US'a geri alın (LOCALE_LANG_HEADER = "en-US,en;q=0.9").
LOCALE = "tr-tr"
LOCALE_LANG_HEADER = "tr-TR,tr;q=0.9"
LOCALE_LANG_SHORT = "tr"
TIMEZONE_LABEL = "GMT%2B0300+(Turkey+Time)"
# ---- TCF / eupubconsent-v2 ----
# OneTrust geolocation API artık consent string dönmüyor (sadece geo). Cookie
# tarayıcıda JS SDK tarafından client-side üretiliyor — Python ile alınamıyor.
# Spotify cookie'yi zorunlu tutmuyor, akış cookie'siz de çalışıyor. True yaparsan
# eski dead-code yolunu aktive edersin (her hesapta 1 HTML + 1 OneTrust call
# atar, dönüş None olur, eupubconsent-v2 set edilmez).
TCF_ENABLED = False
# ---- User Agent ----
USER_AGENT = "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Mobile Safari/537.36"
USER_AGENT_SHORT = "Mozilla/5.0"
SEC_CH_UA = '"Not-A.Brand";v="24", "Chromium";v="146"'
SEC_CH_UA_PLATFORM = '"Android"'
SEC_CH_UA_MOBILE = "?1"
# True → her session açılışında o session'ın impersonate profiline göre UA
# string'i taslaktan üretilir; sec-ch-ua* header'ları sadece "chrome*" profili
# seçildiyse gönderilir (Firefox/Safari/Opera/Edge profillerinde sec-ch-*
# tamamen düşürülür çünkü gerçek tarayıcı da göndermez/farklı gönderir).
# False → yukarıdaki sabit USER_AGENT + SEC_CH_* kullanılır (orijinal davranış).
DYNAMIC_USER_AGENT = True
SDK_VERSION = "4.0.3"
APP_VERSION = "1.0.0"
SIGNUP_APP_NAME = "www-signup2"
# ---- Email / Account generation ----
# rand_email() — USE_PROFILE_DATA=False iken random domain seçilir
EMAIL_DOMAINS = ["gmail.com", "outlook.com", "yahoo.com", "protonmail.com"]
# ProfileDataManager (USE_PROFILE_DATA=True veya JSON fallback) — sabit domain
PROFILE_EMAIL_DOMAIN = "gmail.com" #gmail.com
EMAIL_NAME_LEN = 8
EMAIL_NUM_MIN = 10
EMAIL_NUM_MAX = 9999
DOB_YEAR_MIN = 1978 # dar aralıkta kümelenmesin diye geniş (hepsi 18+)
DOB_YEAR_MAX = 2006
NAME_LEN = 6
NAME_NUM_MIN = 10
NAME_NUM_MAX = 999
# ---- Mail kaynağı ----
# Hesabın email'i üç kaynaktan biri olabilir:
# "random" → profile/rand_email() çıktısı aynen kullanılır (gerçek olmayan,
# var olmayan bir kutu — emailChallenge gelirse çözülemez).
# "ucretli" → anymessage.shop'tan order edilen mail (local-part'ı BİZ
# seçemeyiz, onlar veriyor; bkz. ucretli_mail_al()).
# "ucretsiz" → tempmail.com.tr'de local-part'ı biz seçerek açılan ücretsiz
# mailbox (bkz. ucretsiz_mail_al()) — iptal/iade mekanizması yok.
# USE_PROFILE_DATA (yukarıda) bu seçimden bağımsız: "random" ve "ucretsiz"
# kaynaklarında display_name/dob/gender/local-part üretiminde etkili olmaya
# devam eder; "ucretli"de email zaten anymessage tarafından dayatıldığı için
# sadece display_name/dob/gender'ı etkiler.
MAIL_SOURCE = "ucretsiz"
# ---- Ücretli mail (anymessage.shop) ----
# MAIL_SOURCE="ucretli" iken her hesap için harici mail API'sinden bir mail
# order edilir; dönen email `mail` alanına, order id ise `mailSifre` alanına
# yazılır. Mail onayı YAPILMAZ — sadece bakiye kontrol + order + kayıt.
# Bakiye yoksa/order/network hatasında random üretilen maile fallback
# yapılır (akış kesilmez).
UCRETLI_MAIL = (MAIL_SOURCE == "ucretli") # ucretli_mail_al() bu bayrağı okur
UCRETLI_MAIL_TOKEN = "CExidOceqifaW7HopYS8BumOcYM8VAG1" # anymessage.shop token
UCRETLI_MAIL_BASE_URL = "https://api.anymessage.shop"
UCRETLI_MAIL_SITE = "spotify.com" # $site parametresi
UCRETLI_MAIL_DOMAIN = "outlook.com,hotmail.com" # $domain — order edilecek mail domaini (kendine göre ayarla)
UCRETLI_MAIL_TIMEOUT = 15
UCRETLI_MAIL_MIN_BALANCE = 0.0 # bu değerin ÜSTÜnde bakiye varsa order edilir
# ---- Ücretsiz mail (tempmail.com.tr) ----
# MAIL_SOURCE="ucretsiz" iken local-part'ı biz seçeriz (fallback email'in
# @ öncesi kısmı — random veya profile'dan geleni, USE_PROFILE_DATA neyi
# üretmişse), domain sabit. Laravel CSRF akışı: GET / → csrf-token + cookie'ler,
# POST /get_messages ("warm-up" — session'da mailbox satırı oluşturur, dönen
# değer önemsiz), POST /change (asıl mailbox'ı set eder — warm-up ÇAĞRILMADAN
# önce denenirse sessizce "return" döner, hiçbir şey değişmez — empirik olarak
# doğrulandı), sonraki POST /get_messages'lar gelen kutuyu okur.
FREE_MAIL_BASE_URL = "https://tempmail.com.tr"
FREE_MAIL_DOMAIN = "misket.net" # sabit domain
FREE_MAIL_TIMEOUT = 15
# emailChallenge (challenge.spotify.com'un __NEXT_DATA__ üzerinden istediği
# e-posta doğrulama kodu) gelince kodu bu aralıklarla aktif MAIL_SOURCE'un
# gelen kutusundan poll ederiz (bkz. wait_for_email_code()).
EMAIL_CODE_POLL_INTERVAL = 5 # saniye — getmessage çağrıları arası bekleme
EMAIL_CODE_POLL_TIMEOUT = 90 # saniye — bu sürede kod gelmezse pes edilir
# ---- Spotify URLs ----
SPOTIFY_ORIGIN = "https://www.spotify.com"
SPOTIFY_REFERER = "https://www.spotify.com/"
SIGNUP_URL = f"https://www.spotify.com/{LOCALE}/signup"
SIGNUP_LANDING_ENC = f"https%3A%2F%2Fwww.spotify.com%2F{LOCALE}%2Fsignup"
OPEN_SPOTIFY_LANDING_REF = "https%3A%2F%2Fopen.spotify.com%2F"
SIGNUP_REDIRECT_PATH = "/api/signup/redirect"
SPCLIENT_VALIDATE_URL = "https://spclient.wg.spotify.com/signup/public/v2/account/validate"
SPCLIENT_CREATE_URL = "https://spclient.wg.spotify.com/signup/public/v2/account/create"
SPCLIENT_COMPLETE_URL = "https://spclient.wg.spotify.com/signup/public/v2/account/complete-creation"
EMAIL_DOSSIER_URL = "https://spclient.wg.spotify.com/email-dossier/generate-dossier"
APRESOLVE_URL = "https://apresolve.spotify.com/?type=spclient"
GABO_FALLBACK_URL = "https://gew4-spclient.spotify.com/gabo-receiver-service/public/v3/events"
GABO_PATH = "gabo-receiver-service/public/v3/events"
NEXT_DATA_URL_TPL = "https://www.spotify.com/_next/data/{build_id}/{locale}/signup.json?locale={locale}"
SIGNUP_CHALLENGE_CB_TPL = "https://www.spotify.com/signup/challenge?flow_ctx={flow_ctx}&ubi={ubi}&locale={locale}"
CHALLENGE_BASE = "https://challenge.spotify.com"
CHALLENGE_GET_SESSION_URL = "https://challenge.spotify.com/api/v1/get-session"
CHALLENGE_INVOKE_URL = "https://challenge.spotify.com/api/v1/invoke-challenge-command"
CHALLENGE_FALLBACK_URL_TPL = "https://challenge.spotify.com/c/{ch_session}"
# ---- OneTrust ----
ONETRUST_GEO_URL_TPL = "https://geolocation.onetrust.com/cookieconsentpub/v1/geo/location/datasetinfo?domain={dsid}"
# ---- Google reCAPTCHA ----
GOOGLE_BASE = "https://www.google.com"
GOOGLE_RECAPTCHA_ENTERPRISE = "https://www.google.com/recaptcha/enterprise"
GOOGLE_RECAPTCHA_DEFAULT_API2 = "https://www.google.com/recaptcha/api2"
GOOGLE_STATIC_V = "i7X0JrnYWy9Y_5EYdoFM79kV"
GOOGLE_STATIC_RENDER_URL_TPL = "https://www.google.com/recaptcha/{static_v}/recaptcha__en.js"
RECAPTCHA_HL = "tr"
RECAPTCHA_CB_VALUE = "phia893uehwb"
# ---- Solver shared HTTP retry ----
# SHARED_RETRY_CLIENT (urllib3 Retry) için: 3 retry, 2^n backoff.
SOLVER_HTTP_RETRY_TOTAL = 3
SOLVER_BACKOFF = 1
# ---- 2captcha ----
TWOCAPTCHA_CREATE_TASK_URL = "https://api.2captcha.com/createTask"
TWOCAPTCHA_GET_RESULT_URL = "https://api.2captcha.com/getTaskResult"
TWOCAPTCHA_GET_BALANCE_URL = "https://api.2captcha.com/getBalance"
TWOCAPTCHA_POLL_INTERVAL = 3
TWOCAPTCHA_MAX_POLLS = 60
# ---- Capmonster ----
CAPMONSTER_BASE_URL = "https://api.capmonster.cloud/"
CAPMONSTER_POLL_INTERVAL = 3
CAPMONSTER_MAX_POLLS = 60
# ---- Capsolver ----
CAPSOLVER_BASE_URL = "https://api.capsolver.com/"
CAPSOLVER_POLL_INTERVAL = 3
CAPSOLVER_MAX_POLLS = 60
# ---- Solvecaptcha ----
# Legacy 2captcha API uyumlu (in.php / res.php endpoint'leri). 2captcha.com'dan
# AYRI bir provider — kendi API key'i gerektirir. cookies formatı farklı:
# `KEY1:Value1;KEY2:Value2;` (colon-semicolon), HTTP Cookie header tarzında değil.
SOLVECAPTCHA_BASE_URL = "https://api.solvecaptcha.com"
SOLVECAPTCHA_IN_URL = f"{SOLVECAPTCHA_BASE_URL}/in.php"
SOLVECAPTCHA_RES_URL = f"{SOLVECAPTCHA_BASE_URL}/res.php"
SOLVECAPTCHA_POLL_INTERVAL = 3
SOLVECAPTCHA_MAX_POLLS = 60
# ---- Captcha actions ----
SIGNUP_SUBMIT_ACTION = "website/signup/submit_email"
SIGNUP_LOAD_ACTION = "website/signup/load"
# ---- Timeouts ----
SESSION_TIMEOUT = 30
CREATE_TIMEOUT = 60
CHALLENGE_TIMEOUT = 30
SUBPROCESS_TIMEOUT = 30
# ---- Retry counts ----
CREATE_MAX_RETRIES = 3
CREATE_MAX_ATTEMPTS = 3
GET_SESSION_MAX_RETRIES = 3
RETRY_SLEEP = 2
# ---- Cookies ----
COOKIE_PREFERRED_ORDER = [
"sp_new", "sp_landing", "sp_m", "sp_landingref",
"OptanonAlertBoxClosed", "sp_t", "OptanonConsent",
"__Host-sp_csrf_sid",
]
SPOTIFY_COOKIE_DOMAIN = ".spotify.com"
# ---- Misc ----
GREETING_REFERER = "https://www.google.com/"
COLORAMA_AUTORESET = True
# ============================================================
# BACKEND RESOLUTION
# ============================================================
def _resolve_backend():
primary = HTTP_BACKEND
if primary == "curl_cffi" and _HAS_CURL_CFFI:
return "curl_cffi"
if primary == "primp" and _HAS_PRIMP:
return "primp"
if not ALLOW_BACKEND_FALLBACK:
raise RuntimeError(f"requested backend '{primary}' unavailable and fallback disabled")
if _HAS_CURL_CFFI:
return "curl_cffi"
if _HAS_PRIMP:
return "primp"
raise RuntimeError("no HTTP backend available; install curl_cffi or primp")
ACTIVE_BACKEND = _resolve_backend()
# ============================================================
# LOGGING (diagnostic)
# ============================================================
colorama_init(autoreset=COLORAMA_AUTORESET)
def _ensure_output_dir():
Path(OUTPUT_DIR).mkdir(parents=True, exist_ok=True)
_ensure_output_dir()
_root_logger = logging.getLogger("spotifyHesapAcici")
_root_logger.setLevel(LOG_LEVEL)
_root_logger.handlers.clear()
_root_logger.propagate = False
class _ColorFormatter(logging.Formatter):
COLORS = {
"DEBUG": Fore.WHITE,
"INFO": Fore.CYAN,
"WARNING": Fore.YELLOW,
"ERROR": Fore.RED,
"CRITICAL": Fore.MAGENTA,
}
def format(self, record):
color = self.COLORS.get(record.levelname, "")
ts = datetime.fromtimestamp(record.created).strftime("%H:%M:%S")
return f"{color}{ts} {record.levelname[0]} {record.name}{Style.RESET_ALL} {record.getMessage()}"
_stream = logging.StreamHandler()
_stream.setFormatter(_ColorFormatter())
_root_logger.addHandler(_stream)
_file_handler = None
if LOG_TO_FILE:
_file_handler = logging.FileHandler(Path(OUTPUT_DIR) / LOG_FILE_NAME, encoding="utf-8")
_file_handler.setFormatter(logging.Formatter(
"%(asctime)s %(levelname)s %(name)s [%(threadName)s] %(message)s"
))
_root_logger.addHandler(_file_handler)
logger = _root_logger
# File-only mirror logger for StepLogger lines: the per-account UI prints to
# stdout itself, but we want the same lines preserved in run.log without
# duplicating them on the terminal.
_step_file_logger = logging.getLogger("spotifyHesapAcici.step")
_step_file_logger.setLevel(LOG_LEVEL)
_step_file_logger.handlers.clear()
_step_file_logger.propagate = False
if _file_handler is not None:
_step_file_logger.addHandler(_file_handler)
# ============================================================
# STEP LOGGER — per-event coloured lines
# ============================================================
_print_lock = threading.Lock() # yalnızca _states sözlüğünü korur (begin/_end)
_step_local = threading.local()
# Konsol çıktısı kuyruk tabanlı: worker thread'ler bloklayıcı terminal I/O'da
# (print+flush) beklemek yerine satırı kuyruğa atar; tek bir daemon printer
# thread'i sırayla basar. 50-100 thread'de _emit başına print lock contention'ı
# ve I/O bloklaması ortadan kalkar. Dosya logger'ı (_step_file_logger) hâlâ
# senkron yazıldığı için kapanışta kuyrukta kalan satırlar kaybolsa bile log
# dosyası eksiksizdir.
_log_queue = queue.Queue(maxsize=10000)
def _log_printer_loop():
while True:
line = _log_queue.get()
try:
print(line, flush=True)
except Exception:
pass
finally:
_log_queue.task_done()
_log_printer_thread = threading.Thread(
target=_log_printer_loop, name="log-printer", daemon=True
)
_log_printer_thread.start()
def _flush_log_queue():
# Kapanışta kuyruktaki son satırları bas (daemon printer atexit'te hâlâ canlı).
try:
_log_queue.join()
except Exception:
pass
atexit.register(_flush_log_queue)
class StepLogger:
"""Emit one coloured line per event.
Colours:
white = normal/info (no prefix)
green = success ending (✓)
red = failure ending (✗)
yellow = warning (⚠)
Thread-local context: ``begin(tid, email)`` stores tid+t0 on the calling
thread, then ``info/warn/success/fail`` use that implicit context, so deep
helper functions can log without threading the tid argument through.
Each line also goes to the project file logger so the run log mirrors what
the operator sees on screen.
"""
def __init__(self):
self._states = {}
def begin(self, tid, email):
with _print_lock:
self._states[tid] = {"email": email, "t0": time.time()}
_step_local.tid = tid
_step_local.t0 = time.time()
_step_local.email = email
self._emit(tid, f"start · {email}", Fore.WHITE, level="INFO")
def info(self, msg, tid=None):
tid = tid if tid is not None else getattr(_step_local, "tid", None)
if tid is None:
return
self._emit(tid, msg, Fore.WHITE, level="INFO")
def warn(self, msg, tid=None):
tid = tid if tid is not None else getattr(_step_local, "tid", None)
if tid is None:
return
self._emit(tid, "⚠ " + msg, Fore.YELLOW, level="WARNING")
def success(self, msg, tid=None):
tid = tid if tid is not None else getattr(_step_local, "tid", None)
if tid is None:
return
self._emit(tid, "✓ " + msg, Fore.GREEN, level="INFO")
self._end(tid)
def fail(self, reason, detail=None, tid=None):
tid = tid if tid is not None else getattr(_step_local, "tid", None)
if tid is None:
return
msg = "✗ " + reason
if detail:
msg += f" — {detail}"
self._emit(tid, msg, Fore.RED, level="ERROR")
try:
STATS.record_fail_reason(reason)
except Exception:
pass
self._end(tid)
def _end(self, tid):
with _print_lock:
self._states.pop(tid, None)
if getattr(_step_local, "tid", None) == tid:
_step_local.tid = None
_step_local.t0 = None
_step_local.email = None
def _emit(self, tid, text, color, level="INFO"):
st = self._states.get(tid)
t0 = st["t0"] if st else getattr(_step_local, "t0", None) or time.time()
ts = datetime.now().strftime("%H:%M:%S")
elapsed = time.time() - t0
line_core = f"{ts} [{tid:>3}] +{elapsed:5.1f}s · {text}"
# Bloklamadan kuyruğa at; kuyruk doluysa (printer geri kalmış) son çare
# senkron bas ki satır kaybolmasın.
try:
_log_queue.put_nowait(f"{color}{line_core}{Style.RESET_ALL}")
except queue.Full:
print(f"{color}{line_core}{Style.RESET_ALL}", flush=True)
try:
_step_file_logger.log(getattr(logging, level, logging.INFO), f"[{tid}] {text}")
except Exception:
pass
LOG = StepLogger()
# ============================================================
# STATS
# ============================================================
class Stats:
"""Thread-safe sayaç deposu. Tüm ekleyiciler lock-protected."""
def __init__(self):
self._lock = threading.Lock()
self.ok = 0
self.fail = 0
self.abandoned = 0 # PER_ACCOUNT_TIMEOUT aşan worker'lar
self.fail_reasons = {} # reason -> count
self.cap_v3_ok = 0
self.cap_v3_fail = 0
self.cap_v2_ok = 0
self.cap_v2_fail = 0
self.email_challenge_ok = 0 # emailChallenge · kod başarıyla alındı (captcha maliyeti yok)
self.email_challenge_fail = 0 # emailChallenge · kod alınamadı/mailSifre yok/timeout
self.challenge_count = 0 # challenge dalına düşen hesaplar
self.upload_ok = 0 # remote modda başarılı upload
self.upload_fail = 0
self.total_duration = 0.0 # başarılı hesapların süre toplamı
self.start_time = time.time()
def record(self, success, duration=0.0):
with self._lock:
if success:
self.ok += 1
self.total_duration += duration
else:
self.fail += 1
def record_abandoned(self):
with self._lock:
self.abandoned += 1
def record_fail_reason(self, reason):
with self._lock:
self.fail_reasons[reason] = self.fail_reasons.get(reason, 0) + 1
def record_captcha(self, kind, ok):
with self._lock:
if kind == "v3":
if ok:
self.cap_v3_ok += 1
else:
self.cap_v3_fail += 1
elif kind == "v2":
if ok:
self.cap_v2_ok += 1
else:
self.cap_v2_fail += 1
def record_challenge(self):
with self._lock:
self.challenge_count += 1
def record_email_challenge(self, ok):
with self._lock:
if ok:
self.email_challenge_ok += 1
else:
self.email_challenge_fail += 1
def record_upload(self, ok):
with self._lock:
if ok:
self.upload_ok += 1
else:
self.upload_fail += 1
def snapshot(self):
with self._lock:
elapsed = time.time() - self.start_time
total = self.ok + self.fail
cap_v3_total = self.cap_v3_ok + self.cap_v3_fail
cap_v2_total = self.cap_v2_ok + self.cap_v2_fail
avg_duration = (self.total_duration / self.ok) if self.ok else 0
return {
"ok": self.ok,
"fail": self.fail,
"abandoned": self.abandoned,
"total": total,
"elapsed_s": round(elapsed, 1),
"rate_per_min": round(total / elapsed * 60, 2) if elapsed > 0 else 0,
"success_rate": round(self.ok / total * 100, 1) if total else 0,
"avg_duration_s": round(avg_duration, 2),
"cap_v3_ok": self.cap_v3_ok,
"cap_v3_fail": self.cap_v3_fail,
"cap_v3_rate": round(self.cap_v3_ok / cap_v3_total * 100, 1) if cap_v3_total else 0,
"cap_v2_ok": self.cap_v2_ok,
"cap_v2_fail": self.cap_v2_fail,
"cap_v2_rate": round(self.cap_v2_ok / cap_v2_total * 100, 1) if cap_v2_total else 0,
"email_challenge_ok": self.email_challenge_ok,
"email_challenge_fail": self.email_challenge_fail,
"challenge_count": self.challenge_count,
"upload_ok": self.upload_ok,
"upload_fail": self.upload_fail,
"fail_reasons": dict(self.fail_reasons),
}
STATS = Stats()
_shutdown_event = threading.Event()
# Rate-watchdog state. _pause_until = epoch timestamp; now < bu → pause aktif.
_pause_until = 0.0
_last_check_ok = 0
_last_check_fail = 0
_last_check_abandoned = 0
def _stats_reporter():
if not STATS_REPORT_INTERVAL:
return
while not _shutdown_event.wait(STATS_REPORT_INTERVAL):
s = STATS.snapshot()
logger.info(
f"stats ok={s['ok']} fail={s['fail']} rate={s['rate_per_min']}/min "
f"success={s['success_rate']}% elapsed={s['elapsed_s']}s"
)
def _rate_watchdog():
"""Her RATE_WATCHDOG_INTERVAL'da son interval'daki rate'i kontrol et.
Rolling-interval mantığı: son kontrolden bu yana üretilen ok/fail farkı
alınır. Cumulative değil — bu sayede pause sırasında 0 örnek olduğundan
bir sonraki kontrol atlanır, pause re-trigger döngüsü oluşmaz.
"""
global _pause_until, _last_check_ok, _last_check_fail, _last_check_abandoned
while not _shutdown_event.wait(RATE_WATCHDOG_INTERVAL):
s = STATS.snapshot()
new_ok = s["ok"] - _last_check_ok
new_fail = s["fail"] - _last_check_fail
new_abandoned = s["abandoned"] - _last_check_abandoned
# Abandoned'ı fail sayıyoruz: rate hesabında denominator'a katılır, böylece
# stuck pool durumu da "düşük rate" tetikleyebilir.
new_total = new_ok + new_fail + new_abandoned
# Pencereyi her tick'te sıfırla — "az örnek" dalında da güncellenmezse
# sonraki interval birden çok interval'ı kapsar ve rate hesabı bozulur.
_last_check_ok = s["ok"]
_last_check_fail = s["fail"]
_last_check_abandoned = s["abandoned"]
if new_total < RATE_MIN_SAMPLES:
logger.info(
f"rate-watchdog · az örnek (yeni:{new_total} < {RATE_MIN_SAMPLES}) — pause kontrolü atlandı"
)
continue
rate = (new_ok / new_total) * 100.0
if rate < RATE_LOW_THRESHOLD:
_pause_until = time.time() + RATE_LOW_PAUSE
logger.warning(
f"rate-watchdog · DÜŞÜK rate {rate:.1f}% "
f"(yeni:{new_ok}/{new_total}, abandoned:{new_abandoned}, eşik:{RATE_LOW_THRESHOLD}%) "
f"→ {RATE_LOW_PAUSE//60}dk pause"
)
else:
logger.info(
f"rate-watchdog · OK rate {rate:.1f}% (yeni:{new_ok}/{new_total}, abandoned:{new_abandoned})"
)
def _wait_if_paused():
"""Worker başlangıcında çağrılır. Pause aktifse kalan süreyi bekler."""
now = time.time()
if now >= _pause_until:
return
remaining = _pause_until - now
# Per-worker jitter: pause bitince 50-100 worker aynı anda Spotify'a
# yüklenip rate'i tekrar düşürmesin (thundering herd). Her worker rastgele
# birkaç saniye daha bekler → kademeli devreye giriş.
jitter = random.uniform(0, min(30.0, DEFAULT_THREADS * 0.3))
# Watchdog zaten warning bastı; worker tarafı debug seviyesinde kalsın
logger.debug(f"worker pause: {int(remaining)}s (+{jitter:.1f}s jitter)")
_shutdown_event.wait(remaining + jitter)
# ============================================================
# UTILS
# ============================================================
def rand_str(length=8):
return ''.join(random.choices(string.ascii_lowercase, k=length))
def rand_email():
name = rand_str(EMAIL_NAME_LEN) + str(random.randint(EMAIL_NUM_MIN, EMAIL_NUM_MAX))
return f"{name}@{random.choice(EMAIL_DOMAINS)}"
def rand_pass():
return _vary_password()
def rand_dob():
y, m, d = _vary_dob()
return f"{y}-{m:02d}-{d:02d}"
def rand_name():
return rand_str(NAME_LEN) + str(random.randint(NAME_NUM_MIN, NAME_NUM_MAX))
# ---- Gerçekçilik/çeşitlilik yardımcıları (display_name / password / dob) ----
# Tekdüze desenleri kırmak için: her hesapta farklı format/uzunluk/dağılım.
_PASS_SPECIALS = "!@#$%^&*_-.?"
def _vary_display_name(isim, soyisim):
"""Gerçekçi display_name format çeşitliliği — tekdüze 'İsim Soyisim' yerine."""
i, s = isim or "", soyisim or ""
if not s:
return i or rand_str(6)
r = random.random()
if r < 0.45: return f"{i} {s}" # İsim Soyisim (en yaygın)
if r < 0.60: return i # sadece isim
if r < 0.72: return f"{i.lower()} {s.lower()}" # küçük harf
if r < 0.82: return f"{i} {s[:1].upper()}" # İsim S
if r < 0.90: return f"{i}{s}" # İsimSoyisim
if r < 0.96: return f"{i} {s}{random.randint(1, 99)}" # İsim Soyisim42
return i.lower() # isim (küçük)
def _vary_password(isim=None, soyisim=None):
"""Spotify-geçerli (upper+lower+digit+special, len>=10) + gerçekçi çeşitlilik."""
kind = random.randint(0, 3)
if kind == 0 and isim:
pw = f"{isim.capitalize()}{random.choice(_PASS_SPECIALS)}{random.randint(100, 99999)}"
elif kind == 1 and isim and soyisim:
pw = f"{isim.capitalize()}{soyisim.lower()}{random.choice(_PASS_SPECIALS)}{random.randint(10, 999)}"
elif kind == 2:
n = random.randint(10, 16)
chars = [random.choice(string.ascii_uppercase), random.choice(string.ascii_lowercase),
random.choice(string.digits), random.choice(_PASS_SPECIALS)]
pool = string.ascii_letters + string.digits + _PASS_SPECIALS
chars += [random.choice(pool) for _ in range(n - 4)]
random.shuffle(chars)
pw = "".join(chars)
else:
parts = [(isim or rand_str(6)).capitalize(), str(random.randint(1000, 999999)),
random.choice(_PASS_SPECIALS)]
random.shuffle(parts)
pw = "".join(parts)
# Spotify kural güvencesi: eksik karakter sınıfı varsa tamamla, min 10 char.
if not any(c.isupper() for c in pw): pw += random.choice(string.ascii_uppercase)
if not any(c.islower() for c in pw): pw += random.choice(string.ascii_lowercase)
if not any(c.isdigit() for c in pw): pw += str(random.randint(0, 9))
if not any(c in _PASS_SPECIALS for c in pw): pw += random.choice(_PASS_SPECIALS)
if len(pw) < 10: pw += str(random.randint(1000, 9999))
return pw
def _vary_dob():
"""Geniş + ay-duyarlı (gün 1-31) doğum tarihi → (year, month, day)."""
year = random.randint(DOB_YEAR_MIN, DOB_YEAR_MAX)
month = random.randint(1, 12)
if month == 2:
max_day = 29 if (year % 4 == 0 and (year % 100 != 0 or year % 400 == 0)) else 28
elif month in (4, 6, 9, 11):
max_day = 30
else:
max_day = 31
return year, month, random.randint(1, max_day)
def _enc_varint(val):
res = []
while val > 0:
b = val & 0x7f
val >>= 7
if val > 0:
b |= 0x80
res.append(b)
return bytes(res) if res else bytes([0])
def make_ubi(page_inst, app_inst, flow_id=None):
ts_ms = int(time.time() * 1000)
ts_var = _enc_varint(ts_ms)
proto = bytes([0x08, 0x02])
proto += bytes([0x10]) + ts_var
proto += bytes([0x1a, 36]) + page_inst.encode()
proto += bytes([0x22, 36]) + app_inst.encode()
proto += bytes([0x3a, 36]) + app_inst.encode()
proto += bytes([0x42, 16]) + b"user_interaction"
if flow_id:
proto += bytes([0x4a, 36]) + flow_id.encode()
proto += bytes([0x50, 0x00])
return base64.b64encode(proto).decode()
# ============================================================
# DYNAMIC USER-AGENT (per-session, derived from impersonate profile)
# ============================================================
#
# Session açılışında seçilen impersonate profil ismi parse edilip o tarayıcı
# için UA template'i doldurulur ve thread-local'a yazılır. headers() çağrıları
# thread-local'dan okur; DYNAMIC_USER_AGENT=False ise thread-local boş kalır
# ve sabit USER_AGENT kullanılır.
#
# sec-ch-ua / sec-ch-ua-platform / sec-ch-ua-mobile sadece Chrome family
# (chrome*) profilinde gönderilir — Firefox/Safari gerçek tarayıcısı da bu
# header'ları yollamaz, Edge/Opera kullanıcı isteği gereği listeden hariç.
_dyn_ua_local = threading.local()
_IMPERSONATE_RE = re.compile(r"^(chrome|edge|safari|firefox|opera)_?(\d+(?:\.\d+)?)", re.IGNORECASE)
# Impersonate grup rotasyonu: her hesapta yeniden random seçmek yerine
# `DEFAULT_THREADS` adet hesap boyunca aynı TLS fingerprint kullanılır;
# grup dolduğunda yeniden randomize edilir. Bir "tarayıcı sürümü" tutarlılığı
# için kısa pencerede aynı profil paylaşılır, uzun vadede pool gezilir.
_impersonate_lock = threading.Lock()
_impersonate_counter = 0
_current_impersonate = None
_impersonate_deck = [] # IMPERSONATE_HER_SESSION için tekrarsız dağıtım destesi
_impersonate_deck_pool = None # destenin hangi havuzdan üretildiği (kimlik kontrolü)
def _pick_impersonate(pool):
"""IMPERSONATE_HER_SESSION=True → her çağrıda farklı profil (tekrarsız shuffled-deck;
havuz bitince yeniden karıştırılır). False → `DEFAULT_THREADS` adet hesap boyunca
aynı random profil, grup dolunca yeniden random."""
global _impersonate_counter, _current_impersonate
global _impersonate_deck, _impersonate_deck_pool
with _impersonate_lock:
if IMPERSONATE_HER_SESSION:
# Deste boşsa veya havuz değiştiyse yeniden kur + karıştır.
if not _impersonate_deck or _impersonate_deck_pool is not pool:
_impersonate_deck = list(pool)
random.shuffle(_impersonate_deck)
_impersonate_deck_pool = pool
return _impersonate_deck.pop()
# Eski davranış: grup-bazlı random (DEFAULT_THREADS hesapta bir seçim).
if _current_impersonate is None or _impersonate_counter % DEFAULT_THREADS == 0:
_current_impersonate = random.choice(pool)
_impersonate_counter += 1
return _current_impersonate
def _parse_impersonate(profile):
"""Parse impersonate profil ismini (browser, major_version, platform) tuple'ına çevir.
Örnekler:
"chrome146" → ("chrome", "146", "windows")
"chrome_146_macos_26.0" → ("chrome", "146", "macos")
"chrome_146_windows_10" → ("chrome", "146", "windows")
"safari_26.3" → ("safari", "26", "macos")
"safari184_ios" → ("safari", "18", "ios")
"safari2601" → ("safari", "26", "macos")
"firefox_149_linux_unknown" → ("firefox", "149", "linux")
"edge_138_windows_10" → ("edge", "138", "windows")
"opera_130_macos_26.0" → ("opera", "130", "macos")
"""
if not profile:
return None, None, None
p = profile.lower()
m = _IMPERSONATE_RE.match(p)
if not m:
return None, None, None
browser = m.group(1)
raw_ver = m.group(2)
# Safari condensed forms (no `_` ayrıştırıcı, ardı ardına rakam): 184=18.4,
# 260=26.0, 2601=26.0.1. Bunlarda ilk 2 rakam major versiyon.
safari_offset = len("safari")
safari_separator = profile[safari_offset:safari_offset+1] if browser == "safari" and len(profile) > safari_offset else ""
if browser == "safari" and safari_separator != "_" and "." not in raw_ver and len(raw_ver) >= 3:
version = raw_ver[:2]
elif "." in raw_ver:
version = raw_ver.split(".", 1)[0]
else:
version = raw_ver
# Platform tespiti — suffix anahtar kelimelerine bak.
if "ios" in p:
platform = "ios"
elif "android" in p:
platform = "android"
elif "macos" in p or "_mac" in p:
platform = "macos"
elif "linux" in p:
platform = "linux"
elif "windows" in p:
platform = "windows"
elif browser == "safari":
platform = "macos" # Safari default → macOS
else:
platform = "windows"
return browser, version, platform
def _build_dynamic_ua(browser, version, platform):
"""Browser/version/platform üçlüsünden UA string'i üret."""
if not browser or not version:
return None
v = version
if browser == "chrome":
if platform == "macos":
return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36"
if platform == "linux":
return f"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36"
if platform == "android":
return f"Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Mobile Safari/537.36"
return f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36"
if browser == "edge":
if platform == "macos":
return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36 Edg/{v}.0.0.0"
return f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v}.0.0.0 Safari/537.36 Edg/{v}.0.0.0"
if browser == "firefox":
if platform == "macos":
return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:{v}.0) Gecko/20100101 Firefox/{v}.0"
if platform == "linux":
return f"Mozilla/5.0 (X11; Linux x86_64; rv:{v}.0) Gecko/20100101 Firefox/{v}.0"
return f"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:{v}.0) Gecko/20100101 Firefox/{v}.0"
if browser == "safari":
if platform == "ios":
return f"Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/{v}.0 Mobile/15E148 Safari/604.1"
return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/{v}.0 Safari/605.1.15"
if browser == "opera":
# Opera Chromium tabanlı. Opera 126-130 için Chromium = Opera + 16
# (resmi Opera changelog'undan doğrulandı, 2025-Q4 / 2026-Q1 sürümleri).
# Bilinen sürümler tabloda; gerisi formülle (eski sürümlerde offset farklı
# olabileceğinden tablo dışı sonuç tahmin sayılır).
OPERA_TO_CHROME = {
"126": "142", "127": "143", "128": "144",
"129": "145", "130": "146",
}
if v in OPERA_TO_CHROME:
chrome_v = OPERA_TO_CHROME[v]
else:
try:
chrome_v = str(int(v) + 16)
except ValueError:
chrome_v = v
if platform == "macos":
return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{chrome_v}.0.0.0 Safari/537.36 OPR/{v}.0.0.0"
return f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{chrome_v}.0.0.0 Safari/537.36 OPR/{v}.0.0.0"
return None
def _build_dynamic_secch(browser, version, platform):
"""Sadece Chrome family için sec-ch-ua / sec-ch-ua-platform / sec-ch-ua-mobile
üçlüsü döner. Diğer browser'larda (None, None, None) → header'lar omit edilir.
"""
if browser != "chrome":
return None, None, None
plat_map = {
"windows": '"Windows"',
"macos": '"macOS"',
"linux": '"Linux"',
"ios": '"iOS"',
"android": '"Android"',
}
ch_platform = plat_map.get(platform, '"Windows"')
ch_mobile = "?1" if platform in ("android", "ios") else "?0"
ch_ua = f'"Not-A.Brand";v="24", "Chromium";v="{version}", "Google Chrome";v="{version}"'
return ch_ua, ch_platform, ch_mobile
def current_ua():
"""Şu anki aktif UA — DYNAMIC_USER_AGENT=True ise session başlangıcında
set edilen thread-local, değilse sabit USER_AGENT. Header'larda doğrudan
USER_AGENT yerine bunu kullan ki tüm Spotify-facing isteklerde tutarlı
fingerprint olsun.
"""
return getattr(_dyn_ua_local, "ua", None) or USER_AGENT
def current_impersonate():
"""Şu anki thread'in aktif TLS impersonate profili (`chrome146`,
`firefox_149_windows_10` vb.). Session açılışında set edilir, yoksa None.
`_session_with_retries` ve `visit_challenge` Solver/captcha çağrılarının
da aynı TLS fingerprint'i ile çıkması için bunu okur.
"""
return getattr(_dyn_ua_local, "impersonate", None)
def current_sec_ch_headers():
"""sec-ch-ua / sec-ch-ua-platform / sec-ch-ua-mobile için dict döner.
Chrome dışı dinamik UA'da boş dict — caller bunu **{}` ile merge edince
sec-ch-* header'ları hiç set edilmez.
"""
if not getattr(_dyn_ua_local, "emit_sec_ch", True):
return {}
return {
"sec-ch-ua-platform": getattr(_dyn_ua_local, "sec_ch_platform", None) or SEC_CH_UA_PLATFORM,
"sec-ch-ua": getattr(_dyn_ua_local, "sec_ch_ua", None) or SEC_CH_UA,
"sec-ch-ua-mobile": getattr(_dyn_ua_local, "sec_ch_mobile", None) or SEC_CH_UA_MOBILE,
}
def _activate_dynamic_ua(impersonate):
"""Session açılışında çağrılır. DYNAMIC_USER_AGENT=False ise UA/sec-ch
thread-local'ı temizlenir (header'lar sabit USER_AGENT'a düşer). Aktif
impersonate profili DYNAMIC bayrağından bağımsız olarak her zaman saklanır
— Solver class ve visit_challenge gibi TLS-impersonate ihtiyacı olan kod
parçaları `current_impersonate()` ile okur.
"""
_dyn_ua_local.impersonate = impersonate
_dyn_ua_local.ua = None
_dyn_ua_local.sec_ch_ua = None
_dyn_ua_local.sec_ch_platform = None
_dyn_ua_local.sec_ch_mobile = None
_dyn_ua_local.emit_sec_ch = True # True = sec-ch header'larını gönder
if not DYNAMIC_USER_AGENT:
return
browser, version, platform = _parse_impersonate(impersonate)
ua = _build_dynamic_ua(browser, version, platform)
if not ua:
return
_dyn_ua_local.ua = ua
if browser == "chrome":
ch_ua, ch_platform, ch_mobile = _build_dynamic_secch(browser, version, platform)
_dyn_ua_local.sec_ch_ua = ch_ua
_dyn_ua_local.sec_ch_platform = ch_platform
_dyn_ua_local.sec_ch_mobile = ch_mobile
_dyn_ua_local.emit_sec_ch = True
else:
# Chrome dışı → sec-ch-* header'ları gerçek tarayıcı da göndermez,
# gönderirsek tutarsız fingerprint olur.
_dyn_ua_local.emit_sec_ch = False
def headers(session_id=None, user_agent=None):
dyn_ua = getattr(_dyn_ua_local, "ua", None)
dyn_sec_ch_ua = getattr(_dyn_ua_local, "sec_ch_ua", None)
dyn_sec_ch_platform = getattr(_dyn_ua_local, "sec_ch_platform", None)
dyn_sec_ch_mobile = getattr(_dyn_ua_local, "sec_ch_mobile", None)
emit_sec_ch = getattr(_dyn_ua_local, "emit_sec_ch", True)
h = {
"user-agent": user_agent or dyn_ua or USER_AGENT,
"content-type": "application/json",
"accept": "*/*",
"origin": SPOTIFY_ORIGIN,
"sec-fetch-site": "same-site",
"sec-fetch-mode": "cors",
"sec-fetch-dest": "empty",
"referer": SPOTIFY_REFERER,
}
if emit_sec_ch:
h["sec-ch-ua-platform"] = dyn_sec_ch_platform or SEC_CH_UA_PLATFORM
h["sec-ch-ua"] = dyn_sec_ch_ua or SEC_CH_UA
h["sec-ch-ua-mobile"] = dyn_sec_ch_mobile or SEC_CH_UA_MOBILE
return h
def _format_session_cookies(session):
"""Serialize session cookies as `name=value; name=value` string.
Used for 2captcha/Capmonster `cookies` field — both providers accept the
same Cookie-header format. solvecaptcha aynı çıktıyı alıp `_cookies_for`
içinde kendi `KEY:Value;` formatına çevirir. Capsolver cookies field'ını
desteklemiyor, çağrılsa bile yok sayılır. Returns None when session has no
cookies, so the solver can omit the field rather than send an empty string.
"""
if session is None:
return None
try:
items = list(session.cookies.items())
except Exception:
return None
if not items:
return None
return "; ".join(f"{k}={v}" for k, v in items)
# ============================================================
# v6 PROXY (payload `proxy`) + proxy2 (payload `proxy2`)
# ============================================================
# Boot zamanında `_init_proxy2()` ile bir kez doldurulur; program kapanana
# kadar bu değer kullanılır. PROXY2_API_KULLAN=False ise None kalır ve
# proxy2_al() hesap proxy'sine düşer.
_proxy2_cache = None
def _build_v6_proxy():
"""Her hesap için yeni: V6_PROXY:random_port format string."""
return f"{V6_PROXY}:{random.randint(PROXY_PORT_MIN, PROXY_PORT_MAX)}"
def _strip_proxy_scheme(proxy_url):
"""`http://user:pass@host:80` → `user:pass@host:80`.
Kaynak (`spotifyHesapAcici_request_test.py`) formülü:
self.proxy['server'].split("://")[-1].split("/")[0]
"""
if not proxy_url:
return ""
return proxy_url.split("://")[-1].split("/")[0]
def _init_proxy2():
"""Boot'ta bir kez proxy2 API'sini çağır, cache'le.
PROXY2_API_KULLAN=True iken API isteği/boş yanıt PROXY2_INIT_MAX_RETRIES
kez (exponential backoff ile) denenir — geçici bir ağ hiccup'ı artık tüm
botu açılışta öldürmüyor. Tüm denemeler tükenirse program fatal exit yapar
(kullanıcı isteği: kalıcı/konfigürasyon hatasında sessizce devam etmek
yerine açıkça durmak). False ise no-op.
"""
global _proxy2_cache
if not PROXY2_API_KULLAN:
logger.info("proxy2 · API kullanılmıyor (hesap proxy'si fallback)")
return
last_err = None
for attempt in range(PROXY2_INIT_MAX_RETRIES):
logger.info(f"proxy2 · fetching from {PROXY2_API_URL} (attempt {attempt+1}/{PROXY2_INIT_MAX_RETRIES})")
try:
r = SHARED_RETRY_CLIENT.get(PROXY2_API_URL, timeout=PROXY2_HTTP_TIMEOUT)
except Exception as e:
last_err = f"API isteği başarısız ({e})"
else:
val = (getattr(r, "text", "") or "").strip()
if val:
_proxy2_cache = val
logger.info(f"proxy2 · cached: {val}")
return
last_err = f"API boş yanıt verdi (status={r.status_code})"
if attempt < PROXY2_INIT_MAX_RETRIES - 1:
delay = min(2 ** attempt + random.uniform(0, 1), HTTP_RETRY_MAX_BACKOFF)
logger.warning(f"proxy2 · {last_err} — {delay:.1f}s sonra retry {attempt+1}/{PROXY2_INIT_MAX_RETRIES}")
time.sleep(delay)
logger.error(f"proxy2 · fatal: {PROXY2_INIT_MAX_RETRIES} denemenin hepsi başarısız ({last_err})")
raise SystemExit(1)
def proxy2_al(hesap_proxy=""):
"""payload `proxy2` alanı için değer üret.
PROXY2_API_KULLAN=False → hesap_proxy (scheme stripped) direkt döner.
PROXY2_API_KULLAN=True → boot'ta cache'lenmiş API değeri döner.
"""
if not PROXY2_API_KULLAN:
return hesap_proxy or ""
return _proxy2_cache or ""
# ============================================================
# COOKIE / SESSION COMPAT LAYER
# ============================================================
def _split_set_cookie_header(set_cookie_value):
if not set_cookie_value:
return []
parts = []
start = 0
i = 0
in_expires = False
s = set_cookie_value
while i < len(s):
ch = s[i]
if s[i:i+8].lower() == 'expires=':
in_expires = True
i += 8
continue
if in_expires:
if ch == ';':
in_expires = False
else:
if ch == ',':
parts.append(s[start:i].strip())
start = i + 1
i += 1
tail = s[start:].strip()
if tail:
parts.append(tail)
return [p for p in parts if p]
class _PrimpCookieObj:
__slots__ = ("name", "value", "domain", "path")
def __init__(self, name, value, domain="", path="/"):
self.name = name
self.value = value
self.domain = domain
self.path = path
class _CookiesView:
"""Uniform cookies facade over either backend.
Supports: .get, .set, .items, .keys, .clear, .jar, iter, dict()
"""
def __init__(self, backend, storage):
self._backend = backend
self._s = storage
def get(self, name, default=None):
if self._backend == "curl_cffi":
v = self._s.get(name)
return v if v is not None else default
return self._s.get(name, default)
def set(self, name, value, domain=None):
if self._backend == "curl_cffi":
try:
if domain:
self._s.set(name, value, domain=domain)
else:
self._s.set(name, value)
except TypeError:
self._s.set(name, value)
else:
self._s[name] = value
def items(self):
return list(self._s.items())
def keys(self):
return list(self._s.keys())
def __iter__(self):
return iter(self.items())
def clear(self, domain=None, path=None, name=None):
if self._backend == "curl_cffi":
if name:
try:
self._s.delete(name)
except Exception:
pass
else:
self._s.clear()
else:
if name:
self._s.pop(name, None)
else:
self._s.clear()
@property
def jar(self):
if self._backend == "curl_cffi":
return list(self._s.jar)
return [_PrimpCookieObj(n, v) for n, v in self._s.items()]
# ============================================================
# HTTP SESSIONS
# ============================================================
class CurlCffiSession:
"""curl_cffi tabanlı session — PrimpSession ile bire bir aynı manuel cookie
yönetimi yapar.
curl_cffi'nin native cookie jar'ı domain matching uyguladığı için
`cookies.set(name, value)` (domain'siz) çağırılan cookie'ler sonraki
isteklerde otomatik gönderilmiyordu — orijinal akış primp wrapper'da flat
dict üstüne kuruluydu ve domain göz ardı ediliyordu. Aşağıdaki uygulama
primp davranışını birebir taklit ediyor: cookie değerleri kendi dict'imizde
saklanıyor, her istekten önce manuel `cookie:` header üretiliyor,
response'taki Set-Cookie elle parse edilip dict'e yazılıyor.
"""
backend = "curl_cffi"
def __init__(self, proxy=None):
self._proxy = proxy
# TLS fingerprint — DEFAULT_THREADS boyutunda gruplar halinde rotasyon
self.impersonate = _pick_impersonate(IMPERSONATE_PROFILES_CURL_CFFI)
_activate_dynamic_ua(self.impersonate)
self.client = crequests.Session(impersonate=self.impersonate)
if proxy:
self.client.proxies = {"http": proxy, "https": proxy}
self._cookies = {}
self.timeout = SESSION_TIMEOUT
@property
def proxies(self):
return {"https": self._proxy, "http": self._proxy} if self._proxy else None
@property
def cookies(self):
# "primp" backend = flat-dict facade; aynı semantik curl_cffi için de
# geçerli (domain'siz, host bağımsız).
return _CookiesView("primp", self._cookies)
def _update_cookies(self, resp_headers):
set_cookie = resp_headers.get("set-cookie")
if not set_cookie:
return
for cookie_str in _split_set_cookie_header(set_cookie):
name_val = cookie_str.split(";", 1)[0].strip()
if "=" in name_val:
n, v = name_val.split("=", 1)
self._cookies[n.strip()] = v.strip()
def _cookie_header(self):
if not self._cookies:
return None
parts = []
for nm in COOKIE_PREFERRED_ORDER:
if nm in self._cookies:
parts.append(f"{nm}={self._cookies[nm]}")
for n, v in self._cookies.items():
if n not in COOKIE_PREFERRED_ORDER:
parts.append(f"{n}={v}")
return "; ".join(parts)
def _merge_cookie(self, hdrs):
# PrimpSession ile bire bir aynı: jar'daki cookie'ler otomatik
# `cookie:` header'ı olarak eklenir. Kullanıcı zaten cookie header
# verirse ikisi concatenate edilir — günümüzde manuel cookie geçen
# yer kalmadı, ama emniyet için concatenation davranışı korunuyor.
h = dict(hdrs) if hdrs else {}
ck = self._cookie_header()
if ck:
existing = h.get("cookie") or h.get("Cookie")
h["cookie"] = (existing + "; " + ck) if existing else ck
return h
def get(self, url, headers=None, timeout=None, allow_redirects=True):
h = self._merge_cookie(headers)
r = self.client.get(url, headers=h, timeout=timeout or self.timeout,
allow_redirects=allow_redirects)
self._update_cookies(r.headers)
return r
def post(self, url, headers=None, data=None, json=None, timeout=None, allow_redirects=True):
# JSON'u manuel serialize edip bytes data olarak gönderiyoruz —
# PrimpSession'ın `content=body` davranışıyla aynı.
h = self._merge_cookie(headers)
body = data
if json is not None:
import json as _j
body = _j.dumps(json).encode()
r = self.client.post(url, headers=h, data=body,
timeout=timeout or self.timeout,
allow_redirects=allow_redirects)
self._update_cookies(r.headers)
return r
def close(self):
try:
self.client.close()
except Exception:
pass
class PrimpSession:
backend = "primp"
def __init__(self, proxy=None):
self._proxy = proxy
# TLS fingerprint — DEFAULT_THREADS boyutunda gruplar halinde rotasyon
self.impersonate = _pick_impersonate(IMPERSONATE_PROFILES_PRIMP)
_activate_dynamic_ua(self.impersonate)
self.client = primp.Client(impersonate=self.impersonate, proxy=proxy)
self._cookies = {}
self.timeout = SESSION_TIMEOUT
@property
def proxies(self):
return {"https": self._proxy, "http": self._proxy} if self._proxy else None
@property
def cookies(self):
return _CookiesView("primp", self._cookies)
def _update_cookies(self, resp_headers):
set_cookie = resp_headers.get("set-cookie")
if not set_cookie:
return
for cookie_str in _split_set_cookie_header(set_cookie):
name_val = cookie_str.split(";", 1)[0].strip()
if "=" in name_val:
n, v = name_val.split("=", 1)
self._cookies[n.strip()] = v.strip()
def _cookie_header(self):
if not self._cookies:
return None
parts = []
for nm in COOKIE_PREFERRED_ORDER:
if nm in self._cookies:
parts.append(f"{nm}={self._cookies[nm]}")
for n, v in self._cookies.items():
if n not in COOKIE_PREFERRED_ORDER:
parts.append(f"{n}={v}")
return "; ".join(parts)
def _merge_cookie(self, hdrs):
h = dict(hdrs) if hdrs else {}
ck = self._cookie_header()
if ck:
existing = h.get("cookie") or h.get("Cookie")
h["cookie"] = (existing + "; " + ck) if existing else ck
return h
def _sync_primp_cookies(self, url):
try:
pc = self.client.get_cookies(url)
if pc:
for n, v in pc.items():
if n not in self._cookies:
self._cookies[n] = v
except Exception:
pass
def get(self, url, headers=None, timeout=None, allow_redirects=True):
h = self._merge_cookie(headers)
r = self.client.get(url, headers=h, timeout=timeout or self.timeout)
self._update_cookies(r.headers)
self._sync_primp_cookies(url)
return r
def post(self, url, headers=None, data=None, json=None, timeout=None, allow_redirects=True):
h = self._merge_cookie(headers)
body = data
if json is not None:
import json as _j
body = _j.dumps(json).encode()
r = self.client.post(url, headers=h, content=body, timeout=timeout or self.timeout)
self._update_cookies(r.headers)
self._sync_primp_cookies(url)
return r
def close(self):
try:
self.client.close()
except Exception:
pass
def new_session(proxy=None):
if ACTIVE_BACKEND == "curl_cffi":
return CurlCffiSession(proxy)
return PrimpSession(proxy)
def cookie_header(session, preferred=None):
if preferred is None:
preferred = COOKIE_PREFERRED_ORDER
cookies = {}
for c in session.cookies.jar:
cookies[c.name] = c.value
parts = []
for nm in preferred:
if nm in cookies:
parts.append(f"{nm}={cookies[nm]}")
for n, v in cookies.items():
if n not in preferred:
parts.append(f"{n}={v}")
return "; ".join(parts)
# ============================================================
# CONSENT
# ============================================================
def consent_add(session):
ts = int(time.time())
ts_ms = int(time.time() * 1000)
session.cookies.set("sp_new", "1", domain=SPOTIFY_COOKIE_DOMAIN)
# sp_landing = signup'a düştüğümüz URL, sp_landingref = referrer (open.spotify.com).
# Spotify bu ikisinden "kullanıcı hangi yüzeyden / hangi kampanyadan geldi"yi
# türetiyor; eksik ya da uyumsuz olursa create endpoint'i bazen reddediyor.
session.cookies.set("sp_landing", SIGNUP_LANDING_ENC, domain=SPOTIFY_COOKIE_DOMAIN)
session.cookies.set("sp_landingref", OPEN_SPOTIFY_LANDING_REF, domain=SPOTIFY_COOKIE_DOMAIN)
session.cookies.set("sp_m", LOCALE, domain=SPOTIFY_COOKIE_DOMAIN)
consent_ts = time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())
session.cookies.set("OptanonAlertBoxClosed", consent_ts, domain=SPOTIFY_COOKIE_DOMAIN)
if TCF_ENABLED:
LOG.info("consent · building TCF eupubconsent-v2")
eupub = _build_tcf(session, ts)
if eupub:
LOG.info(f"consent · TCF built len={len(eupub)}")
session.cookies.set("eupubconsent-v2", eupub, domain=SPOTIFY_COOKIE_DOMAIN)
else:
# OneTrust script bazı bölgelerde sayfaya gömülmüyor — bu durumda
# `eupubconsent-v2` cookie atlanır, akış kesilmez (orijinal davranış).
LOG.info("consent · TCF skipped (no OneTrust script on page)")
t = time.localtime()
datestamp = (
f"{['Mon','Tue','Wed','Thu','Fri','Sat','Sun'][t.tm_wday]}+"
f"{['Jan','Feb','Mar','Apr','May','Jun','Jul','Aug','Sep','Oct','Nov','Dec'][t.tm_mon-1]}+"
f"{t.tm_mday:02d}+{t.tm_year}+{t.tm_hour:02d}%3A{t.tm_min:02d}%3A{t.tm_sec:02d}+{TIMEZONE_LABEL}"
)
cid = str(uuid.uuid4())
session.cookies.set(
"OptanonConsent",
(
f"isGpcEnabled=0&datestamp={datestamp}&version=202601.2.0&browserGpcFlag=0"
f"&isIABGlobal=false&hosts=&consentId={cid}&interactionCount=0&isAnonUser=1"
f"&prevHadToken=0&landingPath={SIGNUP_LANDING_ENC}"
f"&groups=s00%3A1%2Cf00%3A0%2Cm00%3A0%2Ct00%3A0%2Cf11%3A0%2CID01%3A0%2Ci00%3A0"
f"%2CV2STACK3%3A0%2CV2STACK11%3A0%2CV2STACK20%3A0%2Cm03%3A1%2Cm02%3A0%2Ct02%3A0"
f"&intType=1&crTime={ts_ms}"
),
domain=SPOTIFY_COOKIE_DOMAIN,
)
if not session.cookies.get("_gcl_au"):
session.cookies.set("_gcl_au", f"1.1.{random.randint(100000000, 999999999)}.{ts}", domain=SPOTIFY_COOKIE_DOMAIN)
if not session.cookies.get("_cs_mk_ga"):
session.cookies.set("_cs_mk_ga", f"{random.random():.16f}_{ts_ms}", domain=SPOTIFY_COOKIE_DOMAIN)
if not session.cookies.get("_scid"):
scid = "_" + uuid.uuid4().hex
session.cookies.set("_scid", scid, domain=SPOTIFY_COOKIE_DOMAIN)
session.cookies.set("_scid_r", scid, domain=SPOTIFY_COOKIE_DOMAIN)
if not session.cookies.get("_ga"):
session.cookies.set("_ga", f"GA1.1.{random.randint(100000000, 999999999)}.{ts}", domain=SPOTIFY_COOKIE_DOMAIN)
def _build_tcf(session, ga_ts):
try:
r = session.get(SIGNUP_URL, headers={
"user-agent": current_ua(),
"accept": "text/html",
})
m = re.search(r'data-domain-script="([^"]+)"', r.text)
if not m:
return None
dsid = m.group(1)
except Exception as e:
logger.debug(f"_build_tcf signup-fetch failed: {e}")
return None
try:
r = session.get(ONETRUST_GEO_URL_TPL.format(dsid=dsid), headers={
"user-agent": USER_AGENT_SHORT, "accept": "application/json",
})
if r.status_code != 200:
return None
geo = r.json()
except Exception as e:
logger.debug(f"_build_tcf onetrust failed: {e}")
return None
cs = geo.get("consentData", {}).get("consentString")
if not cs:
return None
try:
# TCF v2 consent string formatı: `core.disclosed_vendors.publisher_tc...`
# nokta-ayrılmış base64url segmentleri. Sadece core segment'i yeniden yazıyoruz.
parts = cs.split('.')
# base64url → standart base64 (Python b64decode için)
core_s = parts[0].replace('-', '+').replace('_', '/')
pad = 4 - len(core_s) % 4
if pad != 4:
core_s += '=' * pad
core_raw = base64.b64decode(core_s)
# Core segment'in bit layout'u (TCF v2 spec):
# bits 0-5 : version (6 bit)
# bits 6-41 : Created timestamp (36 bit, 1/10 saniye epoch)
# bits 42-77 : LastUpdated timestamp (36 bit, aynı format)
# Her iki timestamp'i de "şimdi"ye eşitleyip cookie'yi taze gösteriyoruz —
# Spotify "eski consent" görürse re-prompt tetikleyebilir.
core_bits = list(''.join(format(b, '08b') for b in core_raw))
new_ds = format(int(ga_ts * 10), '036b')
for i in range(36):
core_bits[6 + i] = new_ds[i] # Created
core_bits[42 + i] = new_ds[i] # LastUpdated
core_new = ''.join(core_bits)
# Byte hizalama için sonu 0 ile doldur (8'in katı yap)
while len(core_new) % 8 != 0:
core_new += '0'
core_new_raw = bytearray(int(core_new[i:i+8], 2) for i in range(0, len(core_new), 8))
parts[0] = base64.urlsafe_b64encode(bytes(core_new_raw)).decode().rstrip('=')
return '.'.join(parts)
except Exception as e:
logger.debug(f"_build_tcf decode failed: {e}")
return None
# ============================================================
# EVENTS (gabo tracking)
# ============================================================
def gabo(session, evts, inst_id):
url = _resolve_gabo_url(session)
payload = {"suppress_persist": False, "events": evts}
try:
r = session.post(url, headers=headers(), json=payload)
return r.status_code == 200
except Exception as e:
logger.debug(f"gabo post failed: {e}")
return False
def _resolve_gabo_url(session):
try:
r = session.get(APRESOLVE_URL, headers={
"accept": "application/json",
"user-agent": current_ua(),
})
if r.status_code == 200:
data = r.json()
hosts = data.get("spclient", [])
if hosts:
host = hosts[0].replace(":443", "")
return f"https://{host}/{GABO_PATH}"
except Exception as e:
logger.debug(f"_resolve_gabo_url failed: {e}")
return GABO_FALLBACK_URL
def _ctx(inst_id):
try:
cid_val = base64.b64encode(uuid.UUID(inst_id).bytes).decode()
except ValueError:
cid_val = base64.b64encode(inst_id.encode()).decode()[:22] + "=="
return {
"context_application": {"version": APP_VERSION},
"context_client_id": {"value": cid_val},
"context_correlation_id": {"value": inst_id},
"context_sdk": {"type": "javascript", "version_name": SDK_VERSION},
"context_time": {"timestamp": int(time.time() * 1000)},
"context_user_agent": {"value": current_ua()},
}
def _seq():
return base64.b64encode(bytes([random.randint(0, 255) for _ in range(16)])).decode()
def flow_corr(session, flow_id, inst_id):
gabo(session, [{
"event_name": "FlowIdSpTCorrelationNonAuth",
"fragments": {**_ctx(inst_id), "message": {
"flow_id": flow_id,
"sp_t": inst_id,
"sp_t_absence_reason": None,
}},
"sequence_id": _seq(),
"sequence_number": 1,
}], inst_id)
def signup_pre(session, flow_id, inst_id, page_inst, app_inst, page_pw, page_ui, page_terms):
gabo(session, [{
"event_name": "SignupWebEventNonAuth",
"fragments": {**_ctx(inst_id), "message": {
"action": "page-view",
"payload": "email-form",
"destination_url": SIGNUP_REDIRECT_PATH,
"flow_id": flow_id,
"sp_t": inst_id,
}},
"sequence_id": _seq(),
"sequence_number": 1,
}], inst_id)
flow_corr(session, flow_id, inst_id)
metric_event(session, inst_id, "recaptcha_load")
def signup_post(session, flow_id, inst_id, page_inst, app_inst, page_pw, page_ui, page_terms):
gabo(session, [{
"event_name": "UbiProd1PageTransitionNonAuth",
"fragments": {**_ctx(inst_id), "message": {
"page_instance_id": page_pw,
"page_id": "signup/password",
"entity_uri": "spotify:page:signup/password",
"navigational_root": None,
"from_page_instance_id": page_inst,
"from_page_id": "signup",
"from_entity_uri": "spotify:page:signup",
"from_navigational_root": None,
"navigation_reason": "unknown",
"app_instance_id": app_inst,
"app_visit_index": 0,
"flow_id": flow_id,
}},
"sequence_id": _seq(),
"sequence_number": 1,
}], inst_id)
metric_event(session, inst_id, "new_signup_email_attempt", locale=LOCALE)
def metric_event(session, inst_id, what, locale=None, metric_type="counter"):
tags = {
"app": SIGNUP_APP_NAME,
"application": SIGNUP_APP_NAME,
"environment": "browser",
}
if locale:
tags["locale"] = locale
tags["market"] = locale
tags["origin"] = ""
if what == "new_signup_registration_attempt":
tags["type"] = "adaptive_email"
gabo(session, [{
"event_name": "SemanticMetricClientNonAuth",
"fragments": {
"context_sdk": {"type": "javascript", "version_name": SDK_VERSION},
"context_time": {"timestamp": int(time.time() * 1000)},
"context_user_agent": {"value": current_ua()},
"message": {
"component_id": SIGNUP_APP_NAME,
"key": SIGNUP_APP_NAME,
"metric_type": metric_type,
"tags": tags,
"value": 1,
"what": what,
},
},
"sequence_id": _seq(),
"sequence_number": 1,
}], inst_id)
def registration_attempt(session, inst_id):
metric_event(session, inst_id, "new_signup_registration_attempt")
# ============================================================
# HTTP RETRY HELPER (2^n + jitter, lock-protected, atexit cleanup)
# ============================================================
class RetryingClient:
"""Wraps a `requests.Session` with retry + exponential backoff.
Behaviour:
- `auto_retries = HTTP_AUTO_RETRIES` total attempts (default 3).
- Backoff: `min(2**attempt + random.uniform(0,1), HTTP_RETRY_MAX_BACKOFF)` seconds.
- Requests run CONCURRENTLY: `requests.Session` + the urllib3 connection pool
are thread-safe for issuing requests, so we do NOT serialize the request
path (an old RLock there throttled all HTTP down to one in-flight request
at a time and defeated `pool_maxsize`). The session is made stateless
(cookies are not persisted) to avoid cross-thread cookie bleed.
- A lock guards only `close()` so an `atexit` close can't race in-flight use.
- On interpreter exit, the underlying session is closed via `atexit`.
"""
__slots__ = ("_client", "_close_lock", "_auto_retries", "_closed")
def __init__(self, auto_retries=None, pool_size=100):
import atexit
from http.cookiejar import DefaultCookiePolicy
self._client = requests.Session()
# Stateless: reject all cookies so concurrent threads sharing this
# session never read/write a shared cookie jar (these endpoints —
# solver/proxy/upload APIs — are stateless, so this is behaviour-safe).
self._client.cookies.set_policy(DefaultCookiePolicy(allowed_domains=[]))
adapter = HTTPAdapter(
pool_connections=pool_size,
pool_maxsize=pool_size,
max_retries=0,
)
self._client.mount("http://", adapter)
self._client.mount("https://", adapter)
self._close_lock = threading.Lock()
self._auto_retries = (auto_retries if auto_retries is not None else HTTP_AUTO_RETRIES) + 1
self._closed = False
atexit.register(self._safe_close)
def _safe_close(self):
with self._close_lock:
if not self._closed:
try:
self._client.close()
except Exception:
pass
self._closed = True
def request(self, method, url, **kw):
kw.setdefault("timeout", SESSION_TIMEOUT)
last_err = None
for attempt in range(self._auto_retries):
try:
# No lock here: concurrent requests share the thread-safe
# connection pool, which is the whole point of pool_maxsize.
return self._client.request(method.upper(), url, **kw)
except Exception as e:
last_err = e
if attempt < self._auto_retries - 1:
delay = min(2 ** attempt + random.uniform(0, 1), HTTP_RETRY_MAX_BACKOFF)
time.sleep(delay)
raise RuntimeError(f"HTTP request failed after {self._auto_retries} attempts: {last_err}")
def get(self, url, **kw):
return self.request("GET", url, **kw)
def post(self, url, **kw):
return self.request("POST", url, **kw)
# A single shared RetryingClient instance for solver/upload/helper calls.
# Pool sized to the thread count so 50-100 concurrent threads don't block on a
# full connection pool ("connection pool is full" urllib3 warning).
SHARED_RETRY_CLIENT = RetryingClient(
auto_retries=HTTP_AUTO_RETRIES,
pool_size=max(100, DEFAULT_THREADS * 2),
)
# ============================================================
# PROFILE DATA MANAGER (opt-in)
# ============================================================
class ProfileDataManager:
"""Thread-safe profile generator.
If USE_PROFILE_DATA is true AND the JSON file exists, names/genders come
from the file (pairs of entries form first+last name). Otherwise falls back
to random alphanumeric generation. The Turkish keys (`isim`, `soyisim`,
`dogumGunu`, ...) are kept so the upload payload matches the PHP endpoint
contract on the server side.
"""
def __init__(self, json_file_path=None, enabled=False):
self._lock = threading.RLock()
self._data = []
self._idx = 0
self._use_file = False
if enabled and json_file_path:
try:
path = Path(json_file_path)
if path.exists():
with open(path, "r", encoding="utf-8") as f:
self._data = json.load(f)
random.shuffle(self._data)
self._use_file = True
logger.info(f"profile · loaded {len(self._data)} entries from {json_file_path}")
else:
logger.warning(f"profile · file not found: {json_file_path} → random fallback")
except Exception as e:
logger.warning(f"profile · load failed ({e}) → random fallback")
else:
logger.info("profile · using random generation")
def _random_profile(self):
isim = rand_str(8)
soyisim = rand_str(8)
rastgele_alfa = string.ascii_lowercase + string.digits
mail = "".join(random.sample(rastgele_alfa, 10)) + "@" + PROFILE_EMAIL_DOMAIN
sifre = _vary_password(isim, soyisim)
dogum_yil, dogum_ay, dogum_gun = _vary_dob()
gender = random.randint(1, 2)
return {
"isim": isim,
"soyisim": soyisim,
"dogumGunu": f"{dogum_gun:02d}",
"dogumAyi": f"{dogum_ay:02d}",
"dogumYili": str(dogum_yil),
"mail": mail,
"cinsiyet": gender,
"sifre": sifre,
"display_name": _vary_display_name(isim, soyisim),
"birthdate": f"{dogum_yil}-{dogum_ay:02d}-{dogum_gun:02d}",
}
def get(self):
with self._lock:
if not self._use_file or not self._data:
return self._random_profile()
if self._idx >= len(self._data):
logger.warning("profile · pool exhausted, switching to random")
self._use_file = False
return self._random_profile()
p1 = self._data[self._idx % len(self._data)]
p2 = self._data[(self._idx + 1) % len(self._data)]
self._idx += 2
isim = p1.get("isim", rand_str(8))
soyisim = p2.get("isim", rand_str(8))
cinsiyet = 1 if p2.get("cinsiyet") == "E" else 2
rastgele_alfa = string.ascii_lowercase + string.digits
mail = "".join(random.sample(rastgele_alfa, 10)) + "@" + PROFILE_EMAIL_DOMAIN
sifre = _vary_password(isim, soyisim)
dogum_yil, dogum_ay, dogum_gun = _vary_dob()
return {
"isim": isim,
"soyisim": soyisim,
"dogumGunu": f"{dogum_gun:02d}",
"dogumAyi": f"{dogum_ay:02d}",
"dogumYili": str(dogum_yil),
"mail": mail,
"cinsiyet": cinsiyet,
"sifre": sifre,
"display_name": _vary_display_name(isim, soyisim),
"birthdate": f"{dogum_yil}-{dogum_ay:02d}-{dogum_gun:02d}",
}
PROFILES = ProfileDataManager(PROFILE_DATA_FILE, enabled=USE_PROFILE_DATA)
# ============================================================
# CAPTCHA SOLVER PROVIDERS (Capmonster / Capsolver)
# ============================================================
class _BaseProviderSolver:
"""Common skeleton for Capmonster / Capsolver."""
BASE_URL = ""
NAME = ""
def __init__(self, api_key, retries, http=None):
self.api_key = api_key
self.retries = retries
self.http = http or SHARED_RETRY_CLIENT
def _post_json(self, endpoint, payload):
payload = dict(payload)
payload["clientKey"] = self.api_key
url = self.BASE_URL + endpoint
# No per-solver lock: self.http (RetryingClient) is already thread-safe
# and concurrent. A second lock here serialized every createTask/poll
# across all threads sharing this solver instance.
r = self.http.post(url, json=payload)
try:
return r.json()
except Exception:
return {"errorId": -1, "errorCode": "BadJSON", "errorDescription": r.text[:200]}
def get_balance(self):
d = self._post_json("getBalance", {})
if d.get("errorId", 0) != 0:
raise RuntimeError(f"{self.NAME} balance err: {d.get('errorCode')} {d.get('errorDescription')}")
return d.get("balance")
def _build_task(self, url, site_key, action, task_kind, proxy=None):
raise NotImplementedError
def solve(self, url, site_key, action, task_kind, user_agent=USER_AGENT,
enterprise_payload=None, on_poll=None, proxy=None, cookies=None):
"""Returns (token, solution_user_agent, task_id, provider) 4-tuple on
success. Returns None on createTask fail (task_id yok).
Returns (None, None, task_id, provider) on poll error or timeout —
timeout durumunda `_report_bad` çağrılır.
`solution_user_agent` is `solution.userAgent` from the provider
response if present; caller should use it for the downstream request
that consumes the token (Spotify ties UA→token validity).
`cookies` is `"name=value; name=value"` string. Only forwarded to
Capmonster; Capsolver task schema doesn't accept a cookies field.
"""
task = self._build_task(url, site_key, action, task_kind, proxy=proxy)
is_v2 = task_kind == "v2"
# userAgent: provider task şemalarında sadece v2'de kabul ediliyor.
# v3 task'larında hiçbir sağlayıcı UA field'ı tanımıyor → atla.
# v2'de de Capsolver kendi UA'sını seçiyor → sadece Capmonster v2.
if user_agent and is_v2 and self.NAME == "capmonster" and "userAgent" not in task:
task["userAgent"] = user_agent
# enterprisePayload:
# v2 — Capmonster + Capsolver kabul ediyor.
# v3 — sadece Capsolver kabul ediyor (Capmonster RecaptchaV3EnterpriseTask
# şemasında enterprisePayload yok).
if enterprise_payload and "enterprisePayload" not in task:
if is_v2 or self.NAME == "capsolver":
if isinstance(enterprise_payload, dict) and enterprise_payload.get("s"):
task["enterprisePayload"] = {"s": enterprise_payload["s"]}
else:
task["enterprisePayload"] = enterprise_payload
# cookies: sadece Capmonster v2 task şemasında var; v3'lerde yok,
# Capsolver'ın hiçbir task tipinde yok.
if cookies and is_v2 and self.NAME == "capmonster" and "cookies" not in task:
task["cookies"] = cookies
d = self._post_json("createTask", {"task": task})
if d.get("errorId", 0) != 0:
LOG.warn(f"{self.NAME} {task_kind} · createTask err {d.get('errorCode')}: {d.get('errorDescription')}")
return None
task_id = d.get("taskId")
LOG.info(f"{self.NAME} {task_kind} · taskId={task_id}")
for attempt in range(self.retries):
# Interruptible sleep: shutdown breaks the poll immediately instead
# of leaving the thread blocked for up to retries*interval seconds.
if _shutdown_event.wait(CAPMONSTER_POLL_INTERVAL if self.NAME == "capmonster" else CAPSOLVER_POLL_INTERVAL):
LOG.warn(f"{self.NAME} {task_kind} · shutdown — poll iptal")
return None, None, task_id, self.NAME
if on_poll:
try:
on_poll()
except Exception:
pass
r = self._post_json("getTaskResult", {"taskId": task_id})
if r.get("errorId", 0) != 0:
LOG.warn(f"{self.NAME} {task_kind} · poll err {r.get('errorCode')}: {r.get('errorDescription')}")
return None, None, task_id, self.NAME
status = r.get("status")
if status == "ready":
sol = r.get("solution", {})
token = sol.get("gRecaptchaResponse") or sol.get("token")
sol_ua = sol.get("userAgent")
if token:
return token, sol_ua, task_id, self.NAME
LOG.warn(f"{self.NAME} {task_kind} · ready but no token")
return None, None, task_id, self.NAME
if attempt and attempt % 15 == 0:
LOG.info(f"{self.NAME} {task_kind} · still processing (attempt {attempt+1})")
LOG.warn(f"{self.NAME} {task_kind} · timeout after {self.retries} polls")
_report_bad(self.NAME, task_id, task_kind)
return None, None, task_id, self.NAME
class CapmonsterSolver(_BaseProviderSolver):
BASE_URL = CAPMONSTER_BASE_URL
NAME = "capmonster"
def _build_task(self, url, site_key, action, task_kind, proxy=None):
# Capmonster resmi docs ile birebir uyumlu task şemaları:
# v2: `RecaptchaV2EnterpriseTask` (proxy varsa proxy fields ile)
# v3: `RecaptchaV3EnterpriseTask` (proxy desteklemiyor, minScore=0.7)
# `isInvisible` resmi örneklerde yok — kaldırıldı.
# `pageAction` v2'de opsiyonel ama Spotify "challenge" string gönderiyor.
# `minScore` v3'te kritik — Spotify yüksek skorlu token bekliyor.
if task_kind == "v2":
task = {
"type": "RecaptchaV2EnterpriseTask",
"websiteURL": url,
"websiteKey": site_key,
"pageAction": action,
}
if proxy:
parsed = urlparse(proxy)
task["proxyType"] = "socks5" if parsed.scheme == "socks5" else "http"
task["proxyAddress"] = parsed.hostname
task["proxyPort"] = int(parsed.port) if parsed.port else 80
if parsed.username:
task["proxyLogin"] = parsed.username
if parsed.password:
task["proxyPassword"] = parsed.password
return task
return {
"type": "RecaptchaV3EnterpriseTask",
"websiteURL": url,
"websiteKey": site_key,
"pageAction": action,
"minScore": 0.7,
}
def _capsolver_proxy_string(proxy_url):
"""Standard proxy URL → Capsolver single-string format.
`http://user:pass@host:port` → `http:host:port:user:pass`
`socks5://user:pass@host:port` → `socks5:host:port:user:pass`
"""
if not proxy_url:
return ""
parsed = urlparse(proxy_url)
scheme = "socks5" if parsed.scheme == "socks5" else "http"
host = parsed.hostname or ""
port = parsed.port or ""
user = parsed.username or ""
pwd = parsed.password or ""
return f"{scheme}:{host}:{port}:{user}:{pwd}"
class CapsolverSolver(_BaseProviderSolver):
BASE_URL = CAPSOLVER_BASE_URL
NAME = "capsolver"
def _build_task(self, url, site_key, action, task_kind, proxy=None):
# Capsolver resmi docs ile birebir uyumlu task şemaları:
# v2 + proxy → `ReCaptchaV2EnterpriseTask` + `proxy: "http:..."`
# v2 proxysiz → `ReCaptchaV2EnterpriseTaskProxyLess`
# v3 + proxy → `ReCaptchaV3EnterpriseTask` + `proxy: "http:..."`
# v3 proxysiz → `ReCaptchaV3EnterpriseTaskProxyLess`
# Capsolver tek-string proxy formatı kullanıyor (capmonster'ın 5
# ayrı field'ından farklı).
if task_kind == "v2":
task = {
"type": "ReCaptchaV2EnterpriseTask" if proxy else "ReCaptchaV2EnterpriseTaskProxyLess",
"websiteURL": url,
"websiteKey": site_key,
"pageAction": action,
}
if proxy:
task["proxy"] = _capsolver_proxy_string(proxy)
return task
task = {
"type": "ReCaptchaV3EnterpriseTask" if proxy else "ReCaptchaV3EnterpriseTaskProxyLess",
"websiteURL": url,
"websiteKey": site_key,
"pageAction": action,
}
if proxy:
task["proxy"] = _capsolver_proxy_string(proxy)
return task
def _get_provider(name):
"""Lazily build the requested provider; cache singletons by name."""
cache = _get_provider._cache
if name in cache:
return cache[name]
inst = None
if name == "capmonster":
if not CAPMONSTER_API_KEY:
LOG.warn("capmonster · API key empty")
else:
inst = CapmonsterSolver(CAPMONSTER_API_KEY, retries=CAPMONSTER_MAX_POLLS)
elif name == "capsolver":
if not CAPSOLVER_API_KEY:
LOG.warn("capsolver · API key empty")
else:
inst = CapsolverSolver(CAPSOLVER_API_KEY, retries=CAPSOLVER_MAX_POLLS)
cache[name] = inst
return inst
_get_provider._cache = {}
def _get_2captcha_balance():
"""2captcha hesap bakiyesini USD olarak döner; key/network sorununda None."""
if not TWO_CAPTCHA_KEY or TWO_CAPTCHA_KEY == "put ur 2cap api key here":
return None
try:
r = SHARED_RETRY_CLIENT.post(
TWOCAPTCHA_GET_BALANCE_URL,
json={"clientKey": TWO_CAPTCHA_KEY},
timeout=10,
)
d = r.json()
if d.get("errorId", 0) != 0:
logger.debug(f"2captcha balance err: {d.get('errorCode')} {d.get('errorDescription')}")
return None
return d.get("balance")
except Exception as e:
logger.debug(f"2captcha balance fetch failed: {e}")
return None
def _get_solvecaptcha_balance():
"""solvecaptcha hesap bakiyesini USD olarak döner; key/network sorununda None."""
if not SOLVECAPTCHA_API_KEY or SOLVECAPTCHA_API_KEY == "XXX":
return None
try:
r = SHARED_RETRY_CLIENT.get(
SOLVECAPTCHA_RES_URL,
params={"key": SOLVECAPTCHA_API_KEY, "action": "getbalance", "json": 1},
timeout=10,
)
d = r.json()
if d.get("status") != 1:
logger.debug(f"solvecaptcha balance err: {d.get('request')}")
return None
return d.get("request")
except Exception as e:
logger.debug(f"solvecaptcha balance fetch failed: {e}")
return None
# ============================================================
# ÜCRETLİ MAIL (anymessage.shop) — bakiye + order helper'ları
# ============================================================
def _ucretli_mail_bakiye():
"""anymessage.shop bakiyesini float döner; key yok / hata → None."""
if not UCRETLI_MAIL_TOKEN or UCRETLI_MAIL_TOKEN == "XXX":
return None
try:
r = SHARED_RETRY_CLIENT.get(
f"{UCRETLI_MAIL_BASE_URL}/user/balance",
params={"token": UCRETLI_MAIL_TOKEN},
timeout=UCRETLI_MAIL_TIMEOUT,
)
d = r.json()
if d.get("status") != "success":
logger.debug(f"ucretli mail balance err: {d.get('value')}")
return None
return float(d.get("balance", 0))
except Exception as e:
logger.debug(f"ucretli mail balance fetch failed: {e}")
return None
def _ucretli_mail_order():
"""Yeni mail order eder. Başarılı → (email, id_str); aksi halde (None, None)."""
if not UCRETLI_MAIL_TOKEN or UCRETLI_MAIL_TOKEN == "XXX":
return None, None
try:
r = SHARED_RETRY_CLIENT.get(
f"{UCRETLI_MAIL_BASE_URL}/email/order",
params={"token": UCRETLI_MAIL_TOKEN,
"site": UCRETLI_MAIL_SITE,
"domain": UCRETLI_MAIL_DOMAIN},
timeout=UCRETLI_MAIL_TIMEOUT,
)
d = r.json()
if d.get("status") != "success":
logger.debug(f"ucretli mail order err: {d.get('value')}")
return None, None
return d.get("email"), str(d.get("id", ""))
except Exception as e:
logger.debug(f"ucretli mail order failed: {e}")
return None, None
def ucretli_mail_al(profile, fallback_email, tid=None):
"""UCRETLI_MAIL açıksa: bakiye kontrol → order → profile['mailSifre'] = id,
profile['mail'] = order email; order edilen email'i döner.
Kapalı / bakiye yetersiz / order-network hatası → fallback_email döner ve
profile['mailSifre'] HİÇ set edilmez (akış kesilmez).
`tid` — çağıran taraf (generate_mobile/generate) LOG.begin()'den ÖNCE
çağırıyor; StepLogger tid yoksa sessizce basmıyor, idx'i caller'dan
açıkça alıp geçiyoruz ki bu fonksiyonun logları görünür olsun."""
if not UCRETLI_MAIL:
return fallback_email
bakiye = _ucretli_mail_bakiye()
if bakiye is None or bakiye <= UCRETLI_MAIL_MIN_BALANCE:
LOG.warn(f"ucretli mail · bakiye yetersiz ({bakiye}) → random maile fallback", tid=tid)
return fallback_email
email, mail_id = _ucretli_mail_order()
if not email:
LOG.warn("ucretli mail · order başarısız → random maile fallback", tid=tid)
return fallback_email
profile["mail"] = email
profile["mailSifre"] = mail_id # yalnızca başarılı order'da set edilir
LOG.info(f"ucretli mail · order ok email={email} id={mail_id} bakiye={bakiye}", tid=tid)
return email
def _ucretli_mail_iptal(mail_id):
"""Order edilen maili iptal/iade eder (best-effort, hata fırlatmaz).
Hesap açılamazsa veya mail alındıktan sonraki herhangi bir olumsuzlukta
bakiye iadesi için çağrılır. mail_id boşsa (mail order edilmemişse) no-op."""
if not mail_id or not UCRETLI_MAIL_TOKEN or UCRETLI_MAIL_TOKEN == "XXX":
return
try:
r = SHARED_RETRY_CLIENT.get(
f"{UCRETLI_MAIL_BASE_URL}/email/cancel",
params={"token": UCRETLI_MAIL_TOKEN, "id": mail_id},
timeout=UCRETLI_MAIL_TIMEOUT,
)
d = r.json()
if d.get("status") == "success":
LOG.info(f"ucretli mail · iptal/iade edildi id={mail_id}")
else:
logger.debug(f"ucretli mail cancel err id={mail_id}: {d.get('value')}")
except Exception as e:
logger.debug(f"ucretli mail cancel failed id={mail_id}: {e}")
def _ucretli_mail_getmessage(mail_id):
"""anymessage.shop'tan gelen maili çeker.
Dönüş: (html, terminal).
- html: mesaj geldiyse ham HTML, gelmediyse None.
- terminal: True ise tekrar denemenin anlamı yok (token hatası / iptal edilmiş
aktivasyon / aktivasyon yok); "henüz gelmedi" (wait message) terminal DEĞİLDİR."""
if not UCRETLI_MAIL_TOKEN or UCRETLI_MAIL_TOKEN == "XXX" or not mail_id:
return None, True
try:
r = SHARED_RETRY_CLIENT.get(
f"{UCRETLI_MAIL_BASE_URL}/email/getmessage",
params={"token": UCRETLI_MAIL_TOKEN, "id": mail_id},
timeout=UCRETLI_MAIL_TIMEOUT,
)
d = r.json()
if d.get("status") != "success":
reason = d.get("value")
if reason == "wait message":
return None, False
logger.debug(f"ucretli mail getmessage err id={mail_id}: {reason}")
return None, True
return d.get("message"), False
except Exception as e:
logger.debug(f"ucretli mail getmessage failed id={mail_id}: {e}")
return None, False
# ============================================================
# ÜCRETSİZ MAIL (tempmail.com.tr) — local-part'ı biz seçebildiğimiz ücretsiz
# sağlayıcı. Laravel CSRF + cookie session akışı: GET / → csrf-token +
# cookie'ler → POST /get_messages ("warm-up"; session'da bir mailbox satırı
# oluşturur, dönen değer önemsiz) → POST /change (asıl mailbox'ı set eder).
#
# KRİTİK (empirik olarak doğrulandı): warm-up çağrısı ATLANIRSA /change
# sessizce "return" (6 byte, düz metin, JSON DEĞİL) döner ve HİÇBİR ŞEY
# DEĞİŞMEZ — session'da henüz güncellenecek bir mailbox satırı yoktur.
# Bu adım atlanınca /get_messages hep sunucunun kendi atadığı rastgele bir
# mailbox'ı gösterir, bizim name/domain'imiz sessizce yok sayılır. Proxy veya
# TLS/HTTP2 fingerprint taklidi (curl_cffi) GEREKMİYOR — düz `requests` ile
# de çalışıyor, tek şart doğru istek sırası.
#
# İptal/iade YOK (ücretsiz) — hesap başarısız olsa da yapılacak bir şey yok.
# ============================================================
def _free_mail_csrf(html):
"""Ana sayfadaki değerini çeker."""
m = re.search(r' - ..." biçiminde devam eden
kısım) — dil/yerel ayardan bağımsız çalışır. Diğerleri yalnızca fallback."""
if not html:
return None
m = re.search(r'display:\s*none[^>]*>\s*(\d{4,8})\s*-\s*', html)
if not m:
m = re.search(r'\s*(\d{4,8})\s*])\s*(\d{4,8})\s*-\s*\S', html)
if not m:
m = re.search(rf'(? 2:
parts = self.url.split("/")
self.url = "https://" + parts[2]
url_port = self.url + ":443"
return base64.b64encode(url_port.encode("utf-8")).decode("utf-8").replace("=", ".")
def _reload_params(self, reCaptchaToken):
return {
"v": self.v,
"co": self._co_string(),
"reason": "q",
"size": self.size,
"hl": self.hl,
"k": self.key,
"c": reCaptchaToken,
"chr": "%5B61%2C36%2C84%5D",
"vh": "13599012192",
"bg": "",
"sa": self.action or "action",
}
def _get_recaptcha_token(self):
anchor_params = {
"ar": "1",
"k": self.key,
"co": self._co_string(),
"hl": self.hl,
"v": self.v,
"sa": self.action or "action",
"size": self.size,
"cb": self.callback_value,
}
self.headers["referer"] = self.url + "/"
anchor = self.session.get(
f"{self.api}/anchor",
params=anchor_params,
cookies=self.cookies,
headers=self.headers,
).text
self.anchor_url = self.api + "/anchor"
for k, v in anchor_params.items():
self.anchor_url += f"&{k}={v}"
self.headers["referer"] = self.anchor_url
return anchor.split('recaptcha-token" value="')[1].split('">')[0]
def token(self):
recaptcha_token = self._get_recaptcha_token()
self.headers.pop("content-type", None)
reload = self.session.post(
f"{self.api}/reload?k={self.key}",
data=self._reload_params(recaptcha_token),
cookies=self.cookies,
headers=self.headers,
).text
rresp_token = reload.split('"rresp","')[1].split('"')[0]
self._solver_time = format(time.time() - self._solver_time, ".3f")
return rresp_token
# ============================================================
# CAPTCHA WRAPPERS — dispatch to selected provider per task kind
# ============================================================
def _solve_v3_inline(site_key, page_url, action, proxy):
"""Original inline solver (no provider API).
Returns (token, None, None, "inline") on success — inline solver has no
solution.userAgent or task_id to surface (Solver class kullanılan UA =
current_ua(), zaten session ile aynı; task_id de yok çünkü provider yok).
Fail durumunda None döner.
"""
try:
s = Solver(siteKey=site_key, siteUrl=page_url,
proxy=proxy, action=action)
if "enterprise" not in s.api:
s.api = GOOGLE_RECAPTCHA_ENTERPRISE
token = s.token()
if token and len(token) > 20:
return token, None, None, "inline"
return None
except Exception as e:
LOG.warn(f"inline v3 failed: {e}")
return None
def _solve_v2_inline(site_key, page_url, proxy, action="challenge"):
"""Aynı Solver akışını v2 checkbox için kullan.
v2'de Google sadece IP/fingerprint güveni yüksekse token verir; düşük
güvende resim challenge'ı döner ve parser boş bulur. Spotify challenge
dalı IP'yi zaten bayrakladığı için pratikte başarı oranı düşük — ama
bedava deneme zararsız (fail ederse paid solver'a düşülebilir).
Returns (token, None, None, "inline") on success — task_id yok (provider yok).
"""
try:
s = Solver(siteKey=site_key, siteUrl=page_url,
proxy=proxy, action=action, size="normal")
if "enterprise" not in s.api:
s.api = GOOGLE_RECAPTCHA_ENTERPRISE
token = s.token()
if token and len(token) > 20:
return token, None, None, "inline"
return None
except Exception as e:
LOG.warn(f"inline v2 failed: {e}")
return None
def _solve_v2_twocaptcha(site_key, page_url, proxy, enterprise_payload, site_key_type, on_poll, cookies=None):
"""v2 via 2captcha API. Returns (token, ua, task_id, "2captcha") on success,
None on createTask fail, (None, None, task_id, "2captcha") on timeout/poll fail.
Timeout durumunda internal `_report_bad` çağırır (provider'a refund/quality
feedback için).
"""
# Spotify her zaman enterprise reCAPTCHA kullandığı için non-enterprise
# dalı kaldırıldı. `site_key_type` parametresi imza uyumu için duruyor.
_ = site_key_type
# Dinamik UA aktifse session impersonate'i ile aynı UA gönderilir; böylece
# 2captcha solution.userAgent ve sonraki Spotify create call'u tutarlı olur.
win_ua = current_ua()
task = {
"type": "RecaptchaV2EnterpriseTask",
"websiteURL": page_url,
"websiteKey": site_key,
"pageAction": "challenge",
}
if enterprise_payload and enterprise_payload.get("s"):
task["enterprisePayload"] = {"s": enterprise_payload["s"]}
parsed = urlparse(proxy)
task["proxyType"] = "socks5" if parsed.scheme == "socks5" else "http"
task["proxyAddress"] = parsed.hostname
task["proxyPort"] = str(parsed.port)
if parsed.username:
task["proxyLogin"] = parsed.username
if parsed.password:
task["proxyPassword"] = parsed.password
task["userAgent"] = win_ua
if cookies:
task["cookies"] = cookies
# 2captcha API çağrısı proxy'siz: clientKey ile auth ediliyor, IP'mize
# bakmıyor. Proxy bilgisi `task` payload'ında zaten var.
try:
r = SHARED_RETRY_CLIENT.post(TWOCAPTCHA_CREATE_TASK_URL,
json={"clientKey": TWO_CAPTCHA_KEY, "task": task},
timeout=SESSION_TIMEOUT)
data = r.json()
except Exception as e:
LOG.warn(f"2captcha v2 · createTask network error: {e}")
return None
if data.get("errorId", 0) != 0:
LOG.warn(f"2captcha v2 · createTask err {data.get('errorCode')}: {data.get('errorDescription')}")
return None
task_id = data["taskId"]
LOG.info(f"2captcha v2 · taskId={task_id}")
for _ in range(TWOCAPTCHA_MAX_POLLS):
if _shutdown_event.wait(TWOCAPTCHA_POLL_INTERVAL):
break # shutdown — bekleme döngüsünü kır
if on_poll:
try:
on_poll()
except Exception:
pass
try:
res = SHARED_RETRY_CLIENT.post(TWOCAPTCHA_GET_RESULT_URL,
json={"clientKey": TWO_CAPTCHA_KEY, "taskId": task_id},
timeout=SESSION_TIMEOUT)
result = res.json()
if result.get("errorId", 0) != 0:
LOG.warn(f"2captcha v2 · poll err {result.get('errorCode')}: {result.get('errorDescription')}")
# Provider task'ı kötü kabul etti — reportbad anlamsız.
return None, None, task_id, "2captcha"
if result.get("status") == "ready":
sol = result.get("solution", {})
return sol.get("gRecaptchaResponse"), sol.get("userAgent"), task_id, "2captcha"
except Exception as e:
logger.debug(f"2captcha v2 poll err: {e}")
continue
LOG.warn(f"2captcha v2 · timeout after {TWOCAPTCHA_MAX_POLLS} polls")
_report_bad("2captcha", task_id, "v2")
return None, None, task_id, "2captcha"
def _solve_v3_twocaptcha(site_key, page_url, action, proxy, cookies=None):
"""v3 via 2captcha API (RecaptchaV3TaskProxyless + isEnterprise=true).
Spotify reCAPTCHA Enterprise kullandığı için `isEnterprise: True` flag'i
zorunlu — 2captcha bunu görmezse standart v3 modunda çözer ve Google
Enterprise endpoint token'ı reddedebilir.
`cookies` parametresi imza uyumu için duruyor ama 2captcha v3 task
şemasında cookies field'ı yok — yutuyoruz.
Returns (token, ua, task_id, "2captcha") on success, None on createTask fail,
(None, None, task_id, "2captcha") on timeout (reportbad çağrılır).
"""
_ = cookies
task = {
"type": "RecaptchaV3TaskProxyless",
"websiteURL": page_url,
"websiteKey": site_key,
"pageAction": action,
"minScore": 0.7,
"isEnterprise": True,
}
try:
r = SHARED_RETRY_CLIENT.post(TWOCAPTCHA_CREATE_TASK_URL,
json={"clientKey": TWO_CAPTCHA_KEY, "task": task},
timeout=SESSION_TIMEOUT)
data = r.json()
except Exception as e:
LOG.warn(f"2captcha v3 · createTask network error: {e}")
return None
if data.get("errorId", 0) != 0:
LOG.warn(f"2captcha v3 · createTask err {data.get('errorCode')}: {data.get('errorDescription')}")
return None
task_id = data["taskId"]
LOG.info(f"2captcha v3 · taskId={task_id}")
for _ in range(TWOCAPTCHA_MAX_POLLS):
if _shutdown_event.wait(TWOCAPTCHA_POLL_INTERVAL):
break # shutdown — bekleme döngüsünü kır
try:
res = SHARED_RETRY_CLIENT.post(TWOCAPTCHA_GET_RESULT_URL,
json={"clientKey": TWO_CAPTCHA_KEY, "taskId": task_id},
timeout=SESSION_TIMEOUT)
result = res.json()
if result.get("errorId", 0) != 0:
LOG.warn(f"2captcha v3 · poll err {result.get('errorCode')}: {result.get('errorDescription')}")
return None, None, task_id, "2captcha"
if result.get("status") == "ready":
sol = result.get("solution", {})
return sol.get("gRecaptchaResponse"), sol.get("userAgent"), task_id, "2captcha"
except Exception:
continue
LOG.warn(f"2captcha v3 · timeout after {TWOCAPTCHA_MAX_POLLS} polls")
_report_bad("2captcha", task_id, "v3")
return None, None, task_id, "2captcha"
def _solve_v2_solvecaptcha(site_key, page_url, proxy, enterprise_payload, on_poll, cookies=None):
"""v2 enterprise via solvecaptcha API (legacy in.php / res.php).
Returns (token, None, task_id, "solvecaptcha") on success — solvecaptcha
legacy API'si `solution.userAgent` dönmüyor, UA bizim gönderdiğimiz oluyor.
Timeout'ta `_report_bad` çağrılır.
cookies parametresi solvecaptcha-specific format'ta gelmeli
(`KEY1:Value1;KEY2:Value2;`) — caller `solve_v2` içindeki `_cookies_for`
helper'ı bu dönüşümü yapıyor. HTTP Cookie header tarzı format DEĞİL.
"""
win_ua = current_ua()
data = {
"key": SOLVECAPTCHA_API_KEY,
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": page_url,
"json": 1,
"enterprise": 1,
"version": "v2",
"action": "challenge",
"userAgent": win_ua,
}
if enterprise_payload and enterprise_payload.get("s"):
data["data-s"] = enterprise_payload["s"]
if cookies:
data["cookies"] = cookies
data.update(_solvecaptcha_proxy_fields(proxy))
# solvecaptcha API çağrısı proxy'siz: key ile auth, proxy task field'ında.
try:
r = SHARED_RETRY_CLIENT.post(SOLVECAPTCHA_IN_URL, data=data, timeout=SESSION_TIMEOUT)
resp = r.json()
except Exception as e:
LOG.warn(f"solvecaptcha v2 · in.php network error: {e}")
return None
if resp.get("status") != 1:
LOG.warn(f"solvecaptcha v2 · in.php err: {resp.get('request')}")
return None
task_id = resp["request"]
LOG.info(f"solvecaptcha v2 · taskId={task_id}")
for _ in range(SOLVECAPTCHA_MAX_POLLS):
if _shutdown_event.wait(SOLVECAPTCHA_POLL_INTERVAL):
break # shutdown — bekleme döngüsünü kır
if on_poll:
try:
on_poll()
except Exception:
pass
try:
res = SHARED_RETRY_CLIENT.get(SOLVECAPTCHA_RES_URL, params={
"key": SOLVECAPTCHA_API_KEY,
"action": "get",
"id": task_id,
"json": 1,
}, timeout=SESSION_TIMEOUT)
result = res.json()
if result.get("status") == 1:
token = result.get("request")
if token and len(token) > 20:
return token, None, task_id, "solvecaptcha"
return None, None, task_id, "solvecaptcha"
req_val = result.get("request", "")
if req_val and req_val != "CAPCHA_NOT_READY":
LOG.warn(f"solvecaptcha v2 · poll err: {req_val}")
return None, None, task_id, "solvecaptcha"
except Exception as e:
logger.debug(f"solvecaptcha v2 poll err: {e}")
continue
LOG.warn(f"solvecaptcha v2 · timeout after {SOLVECAPTCHA_MAX_POLLS} polls")
_report_bad("solvecaptcha", task_id, "v2")
return None, None, task_id, "solvecaptcha"
def _solve_v3_solvecaptcha(site_key, page_url, action, proxy, cookies=None):
"""v3 enterprise via solvecaptcha API (legacy in.php / res.php).
Proxy field'ı v3'te gönderilmiyor (2captcha v3 ile aynı konvansiyon —
proxyless task). cookies field'ı v3'te anlamsız, gönderilmiyor.
Returns (token, None, task_id, "solvecaptcha") on success.
"""
_ = cookies, proxy # v3'te kullanılmıyor
win_ua = current_ua()
data = {
"key": SOLVECAPTCHA_API_KEY,
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": page_url,
"json": 1,
"enterprise": 1,
"version": "v3",
"action": action,
"min_score": 0.7,
"userAgent": win_ua,
}
try:
r = SHARED_RETRY_CLIENT.post(SOLVECAPTCHA_IN_URL, data=data, timeout=SESSION_TIMEOUT)
resp = r.json()
except Exception as e:
LOG.warn(f"solvecaptcha v3 · in.php network error: {e}")
return None
if resp.get("status") != 1:
LOG.warn(f"solvecaptcha v3 · in.php err: {resp.get('request')}")
return None
task_id = resp["request"]
LOG.info(f"solvecaptcha v3 · taskId={task_id}")
for _ in range(SOLVECAPTCHA_MAX_POLLS):
if _shutdown_event.wait(SOLVECAPTCHA_POLL_INTERVAL):
break # shutdown — bekleme döngüsünü kır
try:
res = SHARED_RETRY_CLIENT.get(SOLVECAPTCHA_RES_URL, params={
"key": SOLVECAPTCHA_API_KEY,
"action": "get",
"id": task_id,
"json": 1,
}, timeout=SESSION_TIMEOUT)
result = res.json()
if result.get("status") == 1:
token = result.get("request")
if token and len(token) > 20:
return token, None, task_id, "solvecaptcha"
return None, None, task_id, "solvecaptcha"
req_val = result.get("request", "")
if req_val and req_val != "CAPCHA_NOT_READY":
LOG.warn(f"solvecaptcha v3 · poll err: {req_val}")
return None, None, task_id, "solvecaptcha"
except Exception:
continue
LOG.warn(f"solvecaptcha v3 · timeout after {SOLVECAPTCHA_MAX_POLLS} polls")
_report_bad("solvecaptcha", task_id, "v3")
return None, None, task_id, "solvecaptcha"
def solve_v3(site_key, page_url, action=SIGNUP_LOAD_ACTION, proxy=None, skip_direct=False,
enterprise_payload=None, cookies=None):
"""Dispatch v3 captcha to V3_SOLVER provider.
NOTE: orijinal main.py akışı her zaman "inline" yolunu kullanıyordu (Google
reCAPTCHA Enterprise endpoint'ine doğrudan simüle istek — Solver class).
Bu yol provider API key gerektirmiyor ve Spotify tarafında en güvenilir
skor üretir. Başka provider'lar opsiyonel — özellikle proxy kullanan ve
yüksek hacimli senaryolar için.
Returns (token, ua, task_id, provider) 4-tuple on success, None on failure.
`task_id` provider'a sonradan _report_bad çağırmak için caller'a verilir;
inline solver'da None'dır. `provider` her zaman dolu — "inline" dahil.
"""
provider = (V3_SOLVER or "inline").lower()
LOG.info(f"captcha v3 · provider={provider}")
if provider == "inline":
return _solve_v3_inline(site_key, page_url, action, proxy)
if provider == "capmonster":
s = _get_provider("capmonster")
if s:
return s.solve(page_url, site_key, action, "v3",
user_agent=current_ua(), enterprise_payload=enterprise_payload,
cookies=cookies)
LOG.warn("capmonster missing → fallback inline")
return _solve_v3_inline(site_key, page_url, action, proxy)
if provider == "capsolver":
s = _get_provider("capsolver")
if s:
# Capsolver v3 proxy desteği var → Spotify IP matching için geç
return s.solve(page_url, site_key, action, "v3",
user_agent=current_ua(), enterprise_payload=enterprise_payload,
proxy=proxy, cookies=cookies)
LOG.warn("capsolver missing → fallback inline")
return _solve_v3_inline(site_key, page_url, action, proxy)
if provider == "2captcha":
if TWO_CAPTCHA_KEY and TWO_CAPTCHA_KEY != "put ur 2cap api key here":
return _solve_v3_twocaptcha(site_key, page_url, action, proxy, cookies=cookies)
LOG.warn("2captcha key missing → fallback inline")
return _solve_v3_inline(site_key, page_url, action, proxy)
if provider == "solvecaptcha":
if SOLVECAPTCHA_API_KEY and SOLVECAPTCHA_API_KEY != "XXX":
return _solve_v3_solvecaptcha(site_key, page_url, action, proxy, cookies=cookies)
LOG.warn("solvecaptcha key missing → fallback inline")
return _solve_v3_inline(site_key, page_url, action, proxy)
# inline-first chain modes — bedava deneme, fail ederse paid provider.
# Önceden sadece "inline_then_solvecaptcha" implemente ediliydi; diğer üçü
# (2captcha/capmonster/capsolver) docstring'de belgeliydi ama burada
# eşleşmediği için sessizce plain "inline"a düşüyordu — artık hepsi çalışıyor.
if provider in ("inline_then_2captcha", "inline_then_capmonster",
"inline_then_capsolver", "inline_then_solvecaptcha"):
LOG.info("captcha v3 · trying inline first (free attempt)")
res = _solve_v3_inline(site_key, page_url, action, proxy)
if res:
LOG.info("captcha v3 · inline succeeded — no paid fallback needed")
return res
LOG.info(f"captcha v3 · inline failed → falling back to {provider.split('_then_')[1]}")
if provider == "inline_then_2captcha":
if TWO_CAPTCHA_KEY and TWO_CAPTCHA_KEY != "put ur 2cap api key here":
return _solve_v3_twocaptcha(site_key, page_url, action, proxy, cookies=cookies)
LOG.warn("2captcha key missing → giving up")
return None
if provider == "inline_then_capmonster":
s = _get_provider("capmonster")
if s:
return s.solve(page_url, site_key, action, "v3",
user_agent=current_ua(), enterprise_payload=enterprise_payload,
cookies=cookies)
LOG.warn("capmonster missing → giving up")
return None
if provider == "inline_then_capsolver":
s = _get_provider("capsolver")
if s:
return s.solve(page_url, site_key, action, "v3",
user_agent=current_ua(), enterprise_payload=enterprise_payload,
proxy=proxy, cookies=cookies)
LOG.warn("capsolver missing → giving up")
return None
# inline_then_solvecaptcha
if SOLVECAPTCHA_API_KEY and SOLVECAPTCHA_API_KEY != "XXX":
return _solve_v3_solvecaptcha(site_key, page_url, action, proxy, cookies=cookies)
LOG.warn("solvecaptcha key missing → giving up")
return None
return _solve_v3_inline(site_key, page_url, action, proxy)
def solve_v2(site_key, page_url, proxy=None, enterprise_payload=None,
is_invisible=False, site_key_type=None, skip_direct=False, on_poll=None,
cookies=None):
"""Dispatch v2 captcha to V2_SOLVER provider.
Provider seçenekleri:
"2captcha" → 2captcha API (default, %95+ başarı, ücretli)
"capmonster" → Capmonster API (ücretli)
"capsolver" → Capsolver API (ücretli)
"solvecaptcha" → solvecaptcha API (legacy in.php/res.php, ücretli)
"inline" → Google'a doğrudan, key gerekmez
(sadece IP/fingerprint güveni yüksekse,
Spotify challenge'ında genellikle FAIL)
"inline_then_2captcha" → önce inline dene, fail ederse 2captcha
"inline_then_capmonster" / "inline_then_capsolver" /
"inline_then_solvecaptcha" → benzer fallback
Returns (token, user_agent, task_id, provider) 4-tuple on success.
`task_id` caller'ın sonradan _report_bad çağırması için lazım; inline
solver'da None'dır. Failure'da (None, None, task_id_or_None, provider).
"""
provider = (V2_SOLVER or "").lower()
LOG.info(f"captcha v2 · provider={provider}")
# cookies parametresi caller tarafında HTTP Cookie header tarzı geliyor
# (`name=value; name=value`). solvecaptcha kendine has formata
# (`KEY1:Value1;KEY2:Value2;`) çevrilmesi gerekiyor — caller hangi
# provider'a gittiğini bilmediği için burada dönüşüm yapıyoruz.
def _cookies_for(target_provider):
if not cookies or target_provider != "solvecaptcha":
return cookies
# `a=1; b=2; ` → `a:1;b:2;`
pairs = [p.strip() for p in cookies.split(";") if p.strip()]
out = []
for p in pairs:
if "=" in p:
k, v = p.split("=", 1)
out.append(f"{k.strip()}:{v.strip()}")
return ";".join(out) + ";" if out else None
def _via_2captcha():
if not TWO_CAPTCHA_KEY or TWO_CAPTCHA_KEY == "put ur 2cap api key here":
LOG.warn("captcha v2 · 2captcha key missing")
return None
return _solve_v2_twocaptcha(site_key, page_url, proxy, enterprise_payload,
site_key_type, on_poll, cookies=_cookies_for("2captcha"))
def _via_capmonster():
s = _get_provider("capmonster")
if not s:
return None
# proxy ile çağırıyoruz → Capmonster v2 enterprise+proxy task tipi
# (Spotify IP matching için captcha bizim sticky'mizden çözülüyor görünür).
return s.solve(page_url, site_key, "challenge", "v2",
user_agent=current_ua(), enterprise_payload=enterprise_payload,
on_poll=on_poll, proxy=proxy, cookies=_cookies_for("capmonster"))
def _via_capsolver():
s = _get_provider("capsolver")
if not s:
return None
# Capsolver v2 enterprise+proxy task tipi — Spotify IP matching için
return s.solve(page_url, site_key, "challenge", "v2",
user_agent=current_ua(), enterprise_payload=enterprise_payload,
on_poll=on_poll, proxy=proxy, cookies=_cookies_for("capsolver"))
def _via_solvecaptcha():
if not SOLVECAPTCHA_API_KEY or SOLVECAPTCHA_API_KEY == "XXX":
LOG.warn("captcha v2 · solvecaptcha key missing")
return None
return _solve_v2_solvecaptcha(site_key, page_url, proxy, enterprise_payload,
on_poll, cookies=_cookies_for("solvecaptcha"))
def _finalize(res):
# Provider fonksiyonu None döndüyse (createTask başarısız) caller'a da
# None propagate edilir ki task_id'yi olmayan bir provider'a reportbad
# atmasın. (None, None, None, "") şeklinde 4-tuple uydurmuyoruz —
# caller "if not result" kontrolü yapıyor.
if not res:
return None, None, None, ""
# res zaten 4-tuple
return res
# Pure single-provider modes
if provider == "inline":
return _finalize(_solve_v2_inline(site_key, page_url, proxy))
if provider == "capmonster":
res = _via_capmonster()
if res:
return _finalize(res)
LOG.warn("capmonster missing/failed → fallback 2captcha")
return _finalize(_via_2captcha())
if provider == "capsolver":
res = _via_capsolver()
if res:
return _finalize(res)
LOG.warn("capsolver missing/failed → fallback 2captcha")
return _finalize(_via_2captcha())
if provider == "solvecaptcha":
res = _via_solvecaptcha()
if res:
return _finalize(res)
LOG.warn("solvecaptcha missing/failed → fallback 2captcha")
return _finalize(_via_2captcha())
# inline-first chain modes — bedava deneme, fail ederse paid provider
if provider in ("inline_then_2captcha", "inline_then_capmonster",
"inline_then_capsolver", "inline_then_solvecaptcha"):
LOG.info("captcha v2 · trying inline first (free attempt)")
res = _solve_v2_inline(site_key, page_url, proxy)
if res:
LOG.info("captcha v2 · inline succeeded — no paid fallback needed")
return _finalize(res)
LOG.info(f"captcha v2 · inline failed → falling back to {provider.split('_then_')[1]}")
if provider == "inline_then_2captcha":
return _finalize(_via_2captcha())
if provider == "inline_then_capmonster":
return _finalize(_via_capmonster() or _via_2captcha())
if provider == "inline_then_capsolver":
return _finalize(_via_capsolver() or _via_2captcha())
# inline_then_solvecaptcha
return _finalize(_via_solvecaptcha() or _via_2captcha())
# Default / unknown → 2captcha
return _finalize(_via_2captcha())
# ============================================================
# STEPS
# ============================================================
def _extract_next_data(html):
m = re.search(r']*id="__NEXT_DATA__"[^>]*>(.*?)', html)
if not m:
m = re.search(r'__NEXT_DATA__\s*=\s*({.*?})\s*;?\s*', html)
if not m:
return {}
try:
return json.loads(m.group(1))
except Exception as e:
logger.debug(f"_extract_next_data parse failed: {e}")
return {}
# ---- step0_init cache (modül-global, thread-safe) ----
# Empirik doğrulandı: aşağıdaki alanlar aynı IP'den farklı session/UA/tarayıcıda
# birebir aynı dönüyor — modül-global tek slot cache güvenli. Bunlar dışında
# kalan oturum-bağlı alanlar (flow_id, sp_t, page_inst, app_inst) cache'lenmez,
# her çağrıda taze üretilir/fetch edilir.
#
# Spotify deploy yaparsa build_id değişir; eski build_id ile `get_flow_id`
# 404 dönerse `_init_cache_invalidate()` çağrılır → bir sonraki step0_init
# taze fetch yapar.
INIT_CACHE_TTL = 300 # 5 dk; modüler sürümün TTL'i
_init_cache = None
_init_cache_time = 0.0
_init_cache_lock = threading.Lock()
def _init_cache_invalidate():
"""Spotify deploy yapınca build_id eskir; bu çağrı cache'i flush eder."""
global _init_cache, _init_cache_time
with _init_cache_lock:
if _init_cache is not None:
logger.warning("step0_init · cache invalidated")
_init_cache = None
_init_cache_time = 0.0
def step0_init(session, force_refresh=False):
"""Spotify signup page'inden __NEXT_DATA__ çıkar.
Sabit alanlar (build_id, api_key, recaptcha_*_key, creation_point,
campaign_referrer, referrer) modül-global cache'lenir; oturum-bağlı
alanlar (page_inst, app_inst, sp_t, flow_id) her çağrıda taze.
`force_refresh=True` ile cache bypass edilir (deploy sonrası recovery).
Otomatik recovery: cache'deki build_id ile yapılan `get_flow_id` çağrısı
404 dönerse `_init_cache_invalidate()` çağrılır → bir sonraki step0_init
taze HTML fetch ile yeni build_id'yi alır.
"""
global _init_cache, _init_cache_time
page_inst = str(uuid.uuid4())
app_inst = str(uuid.uuid4())
# Cache hit yolu: sabit alanları cache'den, sp_t/flow_id'yi taze döndür.
if not force_refresh:
with _init_cache_lock:
now = time.time()
cache_valid = (
_init_cache is not None
and (now - _init_cache_time) < INIT_CACHE_TTL
)
cached = dict(_init_cache) if cache_valid else None
if cached is not None:
LOG.info(f"init · cache hit (age={int(time.time() - _init_cache_time)}s)")
# flow_id taze fetch — build_id sabit kaldığı sürece /_next/data endpoint'i
# geçerli. 404 dönerse get_flow_id cache'i invalidate eder.
flow_id = get_flow_id(session, cached["build_id"])
return {
**cached,
"flow_id": flow_id,
"sp_t": None,
"page_inst": page_inst,
"app_inst": app_inst,
}
# Cache miss / force_refresh — signup HTML fetch
h_tr = {
**headers(),
"upgrade-insecure-requests": "1",
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
"sec-fetch-site": "cross-site",
"sec-fetch-mode": "navigate",
"sec-fetch-user": "?1",
"sec-fetch-dest": "document",
"referer": GREETING_REFERER,
"accept-encoding": "gzip, deflate, br, zstd",
"accept-language": LOCALE_LANG_HEADER,
}
try:
r = session.get(SIGNUP_URL, headers=h_tr)
except Exception as e:
LOG.warn(f"step0_init request failed: {e}")
logger.debug(f"step0_init request failed: {e}")
return None
body_size = len(getattr(r, "text", "") or "")
LOG.info(f"init · status={r.status_code} size={body_size}B (cache miss)")
if r.status_code != 200:
return None
nd = _extract_next_data(r.text)
props = nd.get("props", {}).get("pageProps", {})
build_id = nd.get("buildId")
flow_id = props.get("flowId")
keys = props.get("keys", {})
headers_prop = props.get("headers", {})
parameters = props.get("parameters", {})
# Cache'lenebilir sabit alanlar
cacheable = {
"build_id": build_id,
"api_key": keys.get("signupServiceAppKey"),
"recaptcha_scoring_key": keys.get("recaptchaScoringKey"),
"recaptcha_checkbox_key": keys.get("recaptchaCheckboxKey"),
"creation_point": parameters.get("creationPoint", SIGNUP_URL),
"campaign_referrer": headers_prop.get("campaignReferrer", ""),
"referrer": headers_prop.get("referrer", ""),
}
# build_id null gelirse cache yazma — eksik veriyi cache'lemek bir
# sonraki get_flow_id çağrısını sürekli fail eder.
if cacheable["build_id"]:
with _init_cache_lock:
_init_cache = cacheable
_init_cache_time = time.time()
return {
**cacheable,
"flow_id": flow_id,
"sp_t": headers_prop.get("spT"),
"page_inst": page_inst,
"app_inst": app_inst,
}
def get_flow_id(session, build_id):
if not build_id:
return None
url = NEXT_DATA_URL_TPL.format(build_id=build_id, locale=LOCALE)
h = {
**headers(),
"accept": "*/*",
"sec-fetch-site": "same-origin",
"sec-fetch-mode": "cors",
"sec-fetch-dest": "empty",
"referer": SIGNUP_URL,
"accept-encoding": "gzip, deflate, br, zstd",
"accept-language": LOCALE_LANG_HEADER,
"x-nextjs-data": "1",
}
try:
r = session.get(url, headers=h)
LOG.info(f"flow · GET data.json status={r.status_code}")
if r.status_code == 200:
fid = r.json().get("pageProps", {}).get("flowId")
return fid
# 404 → Spotify deploy yapmış, cached build_id eskimiş. Cache'i flush
# et ki bir sonraki step0_init çağrısı taze HTML fetch yapsın.
if r.status_code == 404:
_init_cache_invalidate()
except Exception as e:
LOG.warn(f"get_flow_id failed: {e}")
logger.debug(f"get_flow_id failed: {e}")
return None
def step1_validate(session, email, password, inst_id, api_key, creation_point):
tracking = {
"creation_flow": "",
"creation_point": creation_point,
"referrer": "",
"origin_vertical": "",
"origin_surface": "",
}
client_info = {
"api_key": api_key,
"app_version": "v2",
"capabilities": [1],
"installation_id": inst_id,
"platform": "www",
"client_id": "",
}
try:
r = session.post(SPCLIENT_VALIDATE_URL, headers=headers(), json={
"fields": [{"field": "FIELD_EMAIL", "value": email}],
"client_info": client_info,
"tracking": tracking,
})
except Exception as e:
LOG.warn(f"validate email request failed: {e}")
return False
LOG.info(f"validate · email status={r.status_code}")
if r.status_code == 200:
data = r.json()
if data.get("fields"):
for f in data["fields"]:
if f.get("field") == "FIELD_EMAIL" and f.get("error"):
LOG.warn(f"validate · email error={f.get('error')}")
return False
else:
return False
try:
r = session.post(SPCLIENT_VALIDATE_URL, headers=headers(), json={
"fields": [{"field": "FIELD_PASSWORD", "value": password}],
"client_info": client_info,
"tracking": tracking,
})
except Exception as e:
LOG.warn(f"validate password request failed: {e}")
return False
LOG.info(f"validate · password status={r.status_code}")
if r.status_code != 200:
return False
data = r.json()
if data.get("fields"):
for f in data["fields"]:
if f.get("field") == "FIELD_PASSWORD" and f.get("error"):
LOG.warn(f"validate · password error={f.get('error')}")
return False
return True
def step2_create(session, email, password, display_name, dob, inst_id, token,
flow_id, page_inst, app_inst, api_key, creation_point,
gender=1, callback_uri=None, callback_timestamp=None,
user_agent=None):
"""Account create call.
`callback_uri` / `callback_timestamp` are stamped once by `generate()` so
the later complete-creation request uses the **same** flow_ctx timestamp
in its Referer header — Spotify ties session continuation to that token.
`user_agent` — when the captcha solver returned a `solution.userAgent`,
pass it here so the create call's UA matches the UA Google bound the
token to (Enterprise tokens often UA-bound).
"""
if callback_timestamp is None:
callback_timestamp = int(time.time())
flow_ctx = f"{flow_id}:{callback_timestamp}"
ubi = make_ubi(page_inst, app_inst, flow_id)
cb_uri = callback_uri or SIGNUP_CHALLENGE_CB_TPL.format(
flow_ctx=urllib.parse.quote(flow_ctx),
ubi=urllib.parse.quote(ubi),
locale=LOCALE,
)
payload = {
"account_details": {
"birthdate": dob,
"consent_flags": {"eula_agreed": True, "send_email": True, "third_party_email": True},
"display_name": display_name,
"gender": gender,
"email_and_password_identifier": {"email": email, "password": password},
},
"callback_uri": cb_uri,
"client_info": {
"api_key": api_key,
"app_version": "v2",
"capabilities": [1],
"installation_id": inst_id,
"platform": "www",
"client_id": "",
},
"tracking": {
"creation_flow": "",
"creation_point": creation_point,
"referrer": "",
"origin_vertical": "",
"origin_surface": "",
},
"recaptcha_token": token,
"submission_id": str(uuid.uuid4()),
"flow_id": flow_id,
}
# tmp_s pattern kaldırıldı — ana session ile POST atılıyor.
# Cookie'ler session'ın _merge_cookie'si tarafından otomatik ekleniyor.
h = {
**headers(user_agent=user_agent),
"accept-language": LOCALE_LANG_HEADER,
"accept-encoding": "gzip, deflate, br, zstd",
"cache-control": "no-cache",
"pragma": "no-cache",
"referer": SIGNUP_URL,
}
r = None
for attempt in range(CREATE_MAX_RETRIES):
try:
r = session.post(SPCLIENT_CREATE_URL, headers=h, json=payload, timeout=CREATE_TIMEOUT)
break
except Exception as e:
if attempt < CREATE_MAX_RETRIES - 1:
LOG.warn(f"create · retry {attempt+1}/{CREATE_MAX_RETRIES}: {e}")
time.sleep(RETRY_SLEEP)
else:
LOG.warn(f"create · exhausted: {e}")
raise
if r is None or r.status_code != 200:
code = r.status_code if r is not None else None
body = None
try:
body = r.text[:300] if r is not None else None
except Exception:
pass
LOG.warn(f"create · http {code} body={(body or '')[:120]}")
return {"status": "error", "code": code, "error": body, "cookies": dict(session.cookies.items())}
LOG.info(f"create · http 200 body={len(r.text)}B")
data = r.json()
if data.get("success"):
return {
"status": "success",
"username": data["success"].get("username"),
"login_token": data["success"].get("login_token"),
"cookies": dict(session.cookies.items()),
}
if data.get("challenge"):
ch = data["challenge"]
ch_session = ch.get("session_id")
ch_id = ch.get("challenge_id")
ch_url = ""
ch_ubi = ""
if ch_session:
gs_h = {
"content-type": "application/json",
"origin": SPOTIFY_ORIGIN,
"referer": SPOTIFY_REFERER,
"sec-fetch-site": "cross-site",
"sec-fetch-mode": "cors",
"sec-fetch-dest": "empty",
}
for attempt in range(GET_SESSION_MAX_RETRIES):
try:
LOG.info(f"get-session · POST attempt {attempt+1}")
r2 = session.post(CHALLENGE_GET_SESSION_URL, headers=gs_h, json=ch)
LOG.info(f"get-session · status={r2.status_code}")
if r2.status_code == 200:
sd = r2.json()
ip = sd.get("in_progress", {})
cd = ip.get("challenge_details", {})
wl = cd.get("web_challenge_launcher", {})
ch_url = wl.get("url", "")
ch_ubi = ch.get("ubi", "")
if not ch_id and ch_url:
p = urllib.parse.urlparse(ch_url)
parts = p.path.split("/")
if len(parts) >= 4:
ch_id = parts[3]
break
# Eskiden burada koşulsuz `break` vardı — bu yüzden non-200
# yanıtlarda GET_SESSION_MAX_RETRIES hiç devreye girmiyordu
# (yalnızca network exception'larında retry yapılıyordu).
# Artık HTTP hatası da son denemeye kadar retry ediliyor.
if attempt < GET_SESSION_MAX_RETRIES - 1:
LOG.warn(f"get-session · http {r2.status_code} — retry {attempt+1}/{GET_SESSION_MAX_RETRIES}")
time.sleep(RETRY_SLEEP)
else:
LOG.warn(f"get-session · http {r2.status_code} (exhausted)")
except Exception as e:
if attempt < GET_SESSION_MAX_RETRIES - 1:
LOG.warn(f"get-session retry {attempt+1}: {e}")
time.sleep(RETRY_SLEEP)
else:
LOG.warn(f"get-session hatası (exhausted): {e}")
if not ch_id and ch_session:
try:
LOG.info("get-session · fallback HTML scrape")
r3 = session.get(
CHALLENGE_FALLBACK_URL_TPL.format(ch_session=ch_session),
headers={**headers(), "accept": "text/html,application/xhtml+xml"},
)
if r3.status_code == 200:
m = re.search(r'challenge_id["\']?\s*[:=]\s*["\']([^"\']+)', r3.text)
if m:
ch_id = m.group(1)
LOG.info(f"get-session · fallback found challenge_id={_short(ch_id, 12)}")
except Exception as e:
LOG.warn(f"challenge fallback failed: {e}")
return {
"status": "challenge",
"session_id": ch_session,
"challenge_id": ch_id,
"challenge_url": ch_url,
"ubi": ch_ubi,
"cookies": dict(session.cookies.items()),
}
return None
def visit_challenge(session, challenge_url):
"""Challenge sayfasını ziyaret edip redirect zincirinde Spotify'ın set
ettiği cookie'leri jar'a aktar.
Eski sürüm `curl.exe` subprocess'i ile çalışıyordu (Windows native curl,
TLS impersonate yok). Artık doğrudan session.get() — main session ile
aynı TLS fingerprint + impersonate kullanılır, redirect zincirindeki
ara-Set-Cookie'ler de client'ın internal cookie jar'ından senkronize
edilir (sadece final response'un Set-Cookie'sini görmek yerine).
"""
m = re.search(r'/c/([^/]+)/', challenge_url)
if m:
session.cookies.set("sp_csid", m.group(1))
h = {
"user-agent": current_ua(),
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"accept-language": LOCALE_LANG_HEADER,
"cache-control": "no-cache",
"pragma": "no-cache",
"referer": SPOTIFY_REFERER,
"sec-fetch-site": "same-site",
"sec-fetch-mode": "navigate",
"sec-fetch-dest": "document",
**current_sec_ch_headers(),
}
try:
r = session.get(challenge_url, headers=h, allow_redirects=True,
timeout=SUBPROCESS_TIMEOUT)
# Redirect zincirinde set edilen cookie'leri internal jar'dan al —
# session._update_cookies sadece final response'u görüyor.
applied = 0
try:
client_jar = getattr(session.client, "cookies", None)
if client_jar is not None:
if hasattr(client_jar, "items"):
iter_items = list(client_jar.items())
else:
iter_items = [(c.name, c.value) for c in client_jar]
for n, v in iter_items:
if n not in session._cookies:
session._cookies[n] = v
applied += 1
except Exception as e:
logger.debug(f"visit_challenge cookie sync failed: {e}")
status = getattr(r, "status_code", "?")
LOG.info(f"visit challenge · status={status} +{applied} cookies")
parsed = urllib.parse.urlparse(challenge_url)
path_only = f"{CHALLENGE_BASE}{parsed.path}"
session.cookies.set("sp_landing", urllib.parse.quote(path_only, safe=""))
return True
except Exception as e:
LOG.warn(f"visit_challenge failed: {e}")
return False
def invoke_challenge(session, session_id, challenge_id, recaptcha_token, challenge_url=None,
user_agent=None):
referer = challenge_url if challenge_url else f"{CHALLENGE_BASE}/"
h = {
**headers(user_agent=user_agent),
"accept": "application/json",
"content-type": "application/json",
"origin": CHALLENGE_BASE,
"referer": referer,
"cache-control": "no-cache",
"pragma": "no-cache",
"sec-fetch-site": "same-origin",
"sec-fetch-mode": "cors",
"sec-fetch-dest": "empty",
}
payload = {
"session_id": session_id,
"challenge_id": challenge_id,
"recaptcha_challenge_v1": {"solve": {"recaptcha_token": recaptcha_token}},
}
try:
r = session.post(CHALLENGE_INVOKE_URL, headers=h, json=payload, timeout=CHALLENGE_TIMEOUT)
if r.status_code == 200:
return r.json()
except Exception as e:
logger.debug(f"invoke_challenge failed: {e}")
return None
def invoke_email_challenge(session, session_id, challenge_id, code, challenge_url=None,
user_agent=None):
"""reCAPTCHA yerine e-posta doğrulama kodu istenen challenge'lar için
(`emailChallenge` __NEXT_DATA__'da) invoke-challenge-command çağrısı."""
referer = challenge_url if challenge_url else f"{CHALLENGE_BASE}/"
h = {
**headers(user_agent=user_agent),
"accept": "application/json",
"content-type": "application/json",
"origin": CHALLENGE_BASE,
"referer": referer,
"cache-control": "no-cache",
"pragma": "no-cache",
"sec-fetch-site": "same-origin",
"sec-fetch-mode": "cors",
"sec-fetch-dest": "empty",
}
payload = {
"session_id": session_id,
"challenge_id": challenge_id,
"email_challenge_v1": {"solve": {"code": code}},
}
try:
r = session.post(CHALLENGE_INVOKE_URL, headers=h, json=payload, timeout=CHALLENGE_TIMEOUT)
if r.status_code == 200:
return r.json()
except Exception as e:
logger.debug(f"invoke_email_challenge failed: {e}")
return None
def complete_creation(session, session_id, referer=None):
"""Complete the account creation after challenge solved.
Spotify backend validates that the referer matches the original signup
flow_ctx; pass the same `callback_uri` produced in step2_create.
"""
h = {
**headers(),
"content-type": "application/json",
"accept": "application/json",
"origin": SPOTIFY_ORIGIN,
"referer": referer or SIGNUP_URL,
}
try:
r = session.post(SPCLIENT_COMPLETE_URL, headers=h, json={"session_id": session_id},
timeout=CHALLENGE_TIMEOUT)
LOG.info(f"complete creation · status={r.status_code}")
if r.status_code == 200:
data = r.json()
if data.get("success"):
return {
"status": "success",
"username": data["success"].get("username"),
"login_token": data["success"].get("login_token"),
"cookies": dict(session.cookies.items()),
}
if data.get("challenge"):
return {"status": "challenge", "cookies": dict(session.cookies.items())}
except Exception as e:
LOG.warn(f"complete_creation failed: {e}")
return None
# ============================================================
# sp_dc EXTRACTION + COOKIE UPLOAD
# ============================================================
_SP_DC_COUNTRIES = ["us", "pt", "de", "fr", "it", "tr"]
def get_sp_dc(login_token):
"""Convert a fresh `login_token` into the long-lived `sp_dc` cookie.
Uses an isolated requests.Session (no proxy) — Spotify's authenticate
endpoint is sensitive to IP rotation here. Returns `(sp_dc, csrf_token)`
or `(None, None)` on failure.
"""
if not login_token:
LOG.warn("sp_dc · no login_token")
return None, None
try:
sess = requests.Session()
csrf_headers = {
"authority": "www.spotify.com",
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"accept-language": LOCALE_LANG_SHORT,
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"',
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"sec-fetch-site": "none",
"sec-fetch-user": "?1",
"upgrade-insecure-requests": "1",
}
LOG.info("sp_dc · fetching CSRF token")
csrf_resp = sess.get(SP_DC_CSRF_URL, headers=csrf_headers, timeout=SP_DC_UPLOAD_TIMEOUT)
if csrf_resp.status_code != 200:
LOG.warn(f"sp_dc · CSRF fetch status={csrf_resp.status_code}")
return None, None
try:
csrf_token = csrf_resp.text.split("csrfToken")[1].split('"')[2]
except Exception:
LOG.warn("sp_dc · CSRF parse failed")
return None, None
LOG.info(f"sp_dc · CSRF={_short(csrf_token, 16)}")
auth_headers = {
"authority": "www.spotify.com",
"accept": "*/*",
"accept-language": LOCALE_LANG_SHORT,
"content-type": "application/x-www-form-urlencoded",
"origin": "https://www.spotify.com",
"referer": SP_DC_CSRF_URL,
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"',
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
"x-csrf-token": csrf_token,
}
LOG.info("sp_dc · POST authenticate")
auth_resp = sess.post(
SP_DC_AUTH_URL,
headers=auth_headers,
data=f"splot={login_token}",
timeout=SP_DC_UPLOAD_TIMEOUT,
)
set_cookie = auth_resp.headers.get("set-cookie", "")
if not set_cookie:
LOG.warn("sp_dc · no set-cookie in response")
return None, None
m = re.findall(r"sp_dc=(.*?);", set_cookie, re.DOTALL)
if not m:
LOG.warn("sp_dc · cookie not found in set-cookie")
return None, None
sp_dc = m[0]
LOG.info(f"sp_dc · extracted {_short(sp_dc, 20)}")
return sp_dc, csrf_token
except Exception as e:
LOG.warn(f"sp_dc extraction failed: {e}")
return None, None
def check_cookie_control(control_url):
"""Polling endpoint for INFINITE_MODE; returns True if response body == b'true'."""
if not control_url:
return True
try:
r = SHARED_RETRY_CLIENT.get(control_url, timeout=SP_DC_UPLOAD_TIMEOUT)
return r.status_code == 200 and r.content == b"true"
except Exception as e:
LOG.warn(f"cookie-control check failed: {e}")
return False
# ============================================================
# ACCOUNT PERSISTENCE
# ============================================================
_save_lock = threading.Lock()
def _safe_filename(s):
return re.sub(r'[^A-Za-z0-9._-]+', '_', s or "")
def _filename_for(account):
"""Local JSON dosya adı: `{username}.json`.
Spotify bazen `spotify:user:XXX` formatında döner — bu durumda prefix
atılır. Username yoksa email'e geri düşer.
"""
u = (account.get("username") or "").strip()
if u.startswith("spotify:user:"):
u = u[len("spotify:user:"):]
safe = _safe_filename(u) if u else _safe_filename(account.get("email", "unknown"))
return f"{safe}.json"
def _build_account_payload(account, profile, sp_dc, csrf_token):
"""Canonical Turkish-keyed account JSON — hem local hem remote için kullanılır.
Local JSON dosyasının içeriği ile remote endpoint'e gönderilen base64-encoded
JSON birebir aynıdır. PHP `spotifyCookieYaz.php` kontratıyla uyumlu.
NOT: `proxy` ve `proxy2` payload alanları **sadece kayıt metaverisidir** —
Spotify'a bağlanırken kullanılan gerçek STICKY_PROXY ile veya başka bir
yerde kullanılan gerçek proxy'lerle ilişkili değildir.
- `proxy` → her hesap için yeni `V6_PROXY:random_port`
- `proxy2` → boot'ta cache'lenmiş API değeri (veya PROXY2_API_KULLAN=False
ise hesabın gerçek proxy'sinin scheme-stripped hali)
"""
zaman = int(time.time())
pd = profile or {}
try:
dogum_yili = int(pd.get("dogumYili", 1990))
except (TypeError, ValueError):
dogum_yili = 1990
# proxy2 için fallback: PROXY2_API_KULLAN=False iken hesap'ın gerçek
# proxy'si (NO_PROXY ise boş) scheme-stripped olarak kullanılır.
hesap_proxy_str = "" if NO_PROXY else _strip_proxy_scheme(STICKY_PROXY)
payload = {
"isim": pd.get("isim", "") or account.get("display_name", ""),
"soyisim": pd.get("soyisim", ""),
"dogumGunu": pd.get("dogumGunu", "01"),
"dogumAyi": pd.get("dogumAyi", "01"),
"dogumYili": dogum_yili,
"mail": account.get("email", ""),
"cinsiyet": pd.get("cinsiyet", 1),
"ulke": random.choice(_SP_DC_COUNTRIES),
"sifre": account.get("password", ""),
"proxy": _build_v6_proxy(),
"proxy2": proxy2_al(hesap_proxy_str),
"acilisZamani": zaman,
"csrfToken": csrf_token or "",
"kullaniciAdi": account.get("username", "") or "",
"girisKodu": account.get("login_token", "") or "",
"kontrolZamani": zaman,
"sifresiSifirlandi": 0,
"sp_dc": sp_dc or "",
"cookieKontrol": True,
"erisimKoduKontrol": False,
"profilResmi": 0,
"python": True,
"impersonation": account.get("impersonation", "") or "",
"ua": account.get("ua", "") or "",
}
# mailSifre yalnızca ücretli mail başarıyla order edilmişse eklenir
# (fallback / UCRETLI_MAIL=False → profile'da anahtar hiç oluşmaz).
mail_sifre = pd.get("mailSifre", "")
if mail_sifre:
payload["mailSifre"] = mail_sifre
return payload
def _save_payload_locally(payload, account):
"""Canonical payload'u `verimliHesaplar/{username}.json` olarak yazar."""
_ensure_output_dir()
fpath = Path(OUTPUT_DIR) / _filename_for(account)
with _save_lock:
with open(fpath, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, default=str)
return str(fpath)
def _upload_payload(payload):
"""Canonical payload'u base64-encoded JSON olarak uzak endpoint'e GET ile gönder."""
if not COOKIE_UPLOAD_URL:
LOG.warn("upload · COOKIE_UPLOAD_URL boş")
return False
try:
body = json.dumps(payload, separators=(",", ":"))
b64 = base64.b64encode(body.encode("ascii")).decode("ascii")
LOG.info(f"upload · POST → {COOKIE_UPLOAD_URL[:50]}...")
r = SHARED_RETRY_CLIENT.get(f"{COOKIE_UPLOAD_URL}?cookie={b64}",
timeout=SP_DC_UPLOAD_TIMEOUT)
ok = r.status_code == 200
LOG.info(f"upload · status={r.status_code} ok={ok}")
return ok
except Exception as e:
LOG.warn(f"upload failed: {e}")
return False
# ============================================================
# MAIN FLOW
# ============================================================
def _short(val, n=10):
if val is None:
return "None"
s = str(val)
return s if len(s) <= n else s[:n] + "..."
def _persist(account, profile):
"""Tek "sink" — hesap başarılı olduktan sonra ne yapılacaksa burda yapılır.
Her iki modda da:
1. login_token'dan sp_dc çıkar (her zaman)
2. Canonical Turkish-keyed payload'u oluştur (her zaman)
Sonra OUTPUT_MODE'a göre:
- "local" → `verimliHesaplar/{username}.json` olarak kaydet
- "remote" → base64-encoded JSON olarak uzak endpoint'e gönder
Local ve remote payload'lar birebir aynı içerikte.
"""
# 1. sp_dc her durumda çıkarılır (canonical payload bunu içerir)
LOG.info("sp_dc · extracting from login_token")
sp_dc, csrf_token = get_sp_dc(account.get("login_token"))
if sp_dc:
account["sp_dc"] = sp_dc
account["csrf_token"] = csrf_token
LOG.info(f"sp_dc · ok (csrf={_short(csrf_token, 12)})")
else:
LOG.warn("sp_dc · extraction failed (payload sp_dc alanı boş kalacak)")
# 2. Canonical payload — local + remote için identical
payload = _build_account_payload(account, profile, sp_dc or "", csrf_token or "")
# 3. Hedefe yaz
if OUTPUT_MODE == "remote":
ok = _upload_payload(payload)
account["uploaded"] = ok
STATS.record_upload(ok)
return "remote://uploaded" if ok else "remote://failed"
fp = _save_payload_locally(payload, account)
return fp
# ============================================================
# MOBILE / PROTOBUF AKIŞI (proto2.py'den entegre)
# clienttoken (+attestation +hashcash) → protobuf account/create →
# challenge (siteKey __NEXT_DATA__'dan) → solve_v2 → invoke → complete.
# Web akışındaki captcha/persist/profile/stats altyapısını yeniden kullanır.
# ============================================================
import copy as _copy
import gzip as _gzip
import secrets as _secrets
try:
import brotli as _brotli
_HAS_BROTLI = True
except Exception:
_HAS_BROTLI = False
# librespot dış bağımlılığı KALDIRILDI: tüm clienttoken body'leri (android/ios/windows)
# kendi build_proto encoder'ımızla kuruluyor — windows çıktısı librespot ile byte-byte aynı.
# ---- protobuf varint + parser + encoder ----
def read_varint(data, i):
shift = 0
result = 0
while True:
b = data[i]
result |= ((b & 0x7F) << shift)
i += 1
if not (b & 0x80):
break
shift += 7
return result, i
def encode_varint(value):
out = b""
while True:
b = value & 0x7F
value >>= 7
if value:
out += bytes([b | 0x80])
else:
out += bytes([b])
break
return out
def is_proto(data):
return len(data) >= 1 and any(b < 32 for b in data)
def parse_proto(data, i=0):
res = []
while i < len(data):
try:
key, i = read_varint(data, i)
except Exception:
break
field = key >> 3
wire = key & 7
if wire == 0:
val, i = read_varint(data, i)
res.append({"field": field, "type": "varint", "value": val})
elif wire == 2:
length, i = read_varint(data, i)
raw = data[i:i + length]
i += length
if is_proto(raw):
nested = parse_proto(raw, 0)
if nested:
res.append({"field": field, "type": "nested", "value": nested, "_raw": raw.hex()})
else:
res.append({"field": field, "type": "bytes", "value": raw.hex()})
else:
try:
res.append({"field": field, "type": "string", "value": raw.decode("utf-8")})
except Exception:
res.append({"field": field, "type": "bytes", "value": raw.hex()})
else:
break
return res
def decode_response(r):
data = r.content
encoding = (r.headers.get("Content-Encoding", "") or "").lower()
if not encoding:
return data
try:
if "gzip" in encoding:
return _gzip.decompress(data)
if "br" in encoding and _HAS_BROTLI:
return _brotli.decompress(data)
except Exception:
pass
return data
def encode_string(field, value):
data = value if isinstance(value, bytes) else value.encode("utf-8")
return encode_varint((field << 3) | 2) + encode_varint(len(data)) + data
def encode_nested(field, items):
inner = b""
for it in items:
if it["type"] == "string":
inner += encode_string(it["field"], it["value"])
elif it["type"] == "varint":
inner += encode_varint((it["field"] << 3) | 0) + encode_varint(it["value"])
elif it["type"] == "nested":
inner += encode_nested(it["field"], it["value"])
elif it["type"] == "bytes":
raw = bytes.fromhex(it["value"])
inner += encode_varint((it["field"] << 3) | 2) + encode_varint(len(raw)) + raw
return encode_varint((field << 3) | 2) + encode_varint(len(inner)) + inner
def build_proto(json_data):
out = b""
for item in json_data:
if item["type"] == "string":
out += encode_string(item["field"], item["value"])
elif item["type"] == "varint":
out += encode_varint((item["field"] << 3) | 0) + encode_varint(item["value"])
elif item["type"] == "nested":
out += encode_nested(item["field"], item["value"])
elif item["type"] == "bytes":
raw = bytes.fromhex(item["value"])
out += encode_varint((item["field"] << 3) | 2) + encode_varint(len(raw)) + raw
return out
def _node_by_path(tree, path):
items = tree
node = None
for depth, f in enumerate(path):
node = next((it for it in items if it["field"] == f), None)
if node is None:
raise KeyError(f"field {f} bulunamadı (yol: {path})")
if depth < len(path) - 1:
items = node["value"]
return node
def proto_get(tree, *path):
return _node_by_path(tree, path)["value"]
def proto_set(tree, value, *path):
_node_by_path(tree, path)["value"] = value
def _find(items, field):
return next((x for x in items if x["field"] == field), None)
def dump_tree(tree, _prefix=()):
"""Protobuf ağacının tüm field yollarını + değerlerini düz liste halinde yazdırır (debug)."""
for it in tree:
path = _prefix + (it["field"],)
dotted = ".".join(str(p) for p in path)
if it["type"] == "nested":
print(f"{dotted:<14} (nested)")
dump_tree(it["value"], path)
else:
print(f"{dotted:<14} {it['type']:<7} = {it['value']!r}")
# ---- signup protobuf template + platform profilleri ----
SIGNUP_TEMPLATE = [
{"field": 1, "type": "string", "value": "https://auth-callback.spotify.com/r/android/music/signup"},
{"field": 2, "type": "nested", "value": [
{"field": 1, "type": "string", "value": "Qewadsq"},
{"field": 2, "type": "string", "value": "2005-06-15"},
{"field": 3, "type": "varint", "value": 2},
{"field": 4, "type": "nested", "value": [
{"field": 1, "type": "nested", "value": [{"field": 1, "type": "varint", "value": 1}]},
{"field": 3, "type": "string", "value": ""},
{"field": 4, "type": "string", "value": ""},
{"field": 5, "type": "string", "value": ""},
]},
{"field": 101, "type": "nested", "value": [
{"field": 1, "type": "string", "value": "3qewadsq@gmail.com"},
{"field": 2, "type": "string", "value": "Test*123123"},
]},
]},
{"field": 3, "type": "nested", "value": [
{"field": 1, "type": "string", "value": "142b583129b2df829de3656f9eb484e6"},
{"field": 2, "type": "string", "value": "Android-ARM"},
{"field": 3, "type": "string", "value": "8.7.32.1554"},
{"field": 4, "type": "bytes", "value": "01"},
{"field": 5, "type": "string", "value": "68d1f41e9d60971f"},
]},
{"field": 4, "type": "nested", "value": [
{"field": 1, "type": "string", "value": "client_mobile"},
]},
]
SIGNUP_PROFILES = {
"android": {
"callback_url": "https://auth-callback.spotify.com/r/android/music/signup",
"key": "142b583129b2df829de3656f9eb484e6",
"os": "Android-ARM",
"app_version": "8.7.32.1554",
"install_len": 8,
"x_client_id": "9a8d2f0ce77a4e248bb71fefcb557637",
"headers": {
"User-Agent": "Spotify/8.7.32.1554 Android/30 (SM-N935F)",
"Spotify-App-Version": "8.7.32.1554",
"App-Platform": "Android",
},
},
"ios": {
"callback_url": "https://auth-callback.spotify.com/r/ios/music/signup",
"key": "bff58e9698f40080ec4f9ad97a2f21e0",
"os": "iOS-ARM",
"app_version": "8.8.54",
"install_len": 16,
"x_client_id": "58bd3c95768941ea9eb4350aaa033eb3",
"headers": {
"User-Agent": "Spotify/8.8.54iOS/16.0.2(iPhone10,3)",
"Spotify-App-Version": "8.8.54.544",
"App-Platform": "iOS",
},
},
}
def build_signup_json(platform, email, password, install_id,
display_name=None, dob=None, gender=None, app_version=None):
sp = SIGNUP_PROFILES[platform]
appv = app_version["marketing"] if app_version else sp["app_version"] # client_id SABİT, sürüm seçili
t = _copy.deepcopy(SIGNUP_TEMPLATE)
proto_set(t, sp["callback_url"], 1)
if display_name is not None:
proto_set(t, display_name, 2, 1) # field 2.1 display_name
if dob is not None:
proto_set(t, dob, 2, 2) # field 2.2 birthdate "YYYY-MM-DD"
if gender is not None:
proto_set(t, int(gender), 2, 3) # field 2.3 gender varint (1=erkek, 2=kadın)
proto_set(t, email, 2, 101, 1)
proto_set(t, password, 2, 101, 2)
proto_set(t, sp["key"], 3, 1)
proto_set(t, sp["os"], 3, 2)
proto_set(t, appv, 3, 3)
proto_set(t, install_id, 3, 5)
return t
# ---- clienttoken profilleri + attestation + bodyler ----
GETCLIENT_PROFILES = {
"web": {
"client_id": "d8a5ed958d274c2e8ee717e6a4b0971d",
"client_version": "1.2.52.442.g6c6a4fd0",
"impersonate": "chrome142",
"headers": {
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
"Origin": "https://open.spotify.com",
"Referer": "https://open.spotify.com/",
},
},
"android": {
"client_id": "9a8d2f0ce77a4e248bb71fefcb557637",
"client_version": "8.7.32.1554",
"impersonate": "chrome142",
"headers": {
"Accept": "application/x-protobuf",
"Cache-Control": "no-cache, no-store, max-age=0",
"User-Agent": "Spotify/8.7.32.1554 Android/30 (SM-N935F)",
"Content-Type": "application/x-protobuf",
"Accept-Encoding": "gzip, deflate, br",
},
},
"ios": {
"client_id": "58bd3c95768941ea9eb4350aaa033eb3",
"client_version": "8.8.54.544",
"impersonate": "chrome142",
"headers": {
"Accept": "application/x-protobuf",
"Cache-Control": "no-cache, no-store, max-age=0",
"User-Agent": "Spotify/8.8.54iOS/16.0.2(iPhone10,3)",
"Content-Type": "application/x-protobuf",
"Accept-Encoding": "gzip, deflate, br",
},
},
"desktop": {
"client_id": "65b708073fc0480ea92a077233ca87bd",
"client_version": "1.2.52.442.g6c6a4fd0",
"impersonate": "chrome142",
"headers": {
"Accept": "application/x-protobuf",
"Content-Type": "application/x-protobuf",
"User-Agent": "Spotify/126400342 Win32_x86_64/0 (PC laptop)",
},
},
}
# ---- Mobil cihaz havuzları (per-hesap tutarlı seçim) ----
# Her hesapta buradan TEK bir cihaz seçilir; o hesabın clienttoken UA'sı,
# clienttoken protobuf gövdesi ve signup UA'sı hep AYNI cihazı kullanır.
# NOT: client_id / app_version DEĞİŞMEZ (clienttoken geçerliliğine bağlı) —
# sadece cihaz donanımı (model, os, ekran) çeşitlenir. os sürümleri, sabit
# app_version (~2024) ile makul kalması için iOS 17.x-18.x / Android 13-14 aralığında.
# İdeal: bu değerleri gerçek cihazların build.prop/DB değerleriyle doğrula.
IOS_DEVICE_POOL = [
# (model_identifier, ios_version)
("iPhone14,7", "17.6.1"), # iPhone 14
("iPhone14,8", "18.1.1"), # iPhone 14 Plus
("iPhone15,2", "17.5.1"), # iPhone 14 Pro
("iPhone15,3", "18.2.1"), # iPhone 14 Pro Max
("iPhone15,4", "18.3.2"), # iPhone 15
("iPhone15,5", "17.6.1"), # iPhone 15 Plus
("iPhone16,1", "18.5"), # iPhone 15 Pro
("iPhone16,2", "18.6.1"), # iPhone 15 Pro Max
("iPhone17,3", "18.5"), # iPhone 16
("iPhone17,4", "18.6.2"), # iPhone 16 Plus
("iPhone17,1", "18.6"), # iPhone 16 Pro
("iPhone17,2", "18.4.1"), # iPhone 16 Pro Max
("iPhone17,5", "18.3.1"), # iPhone 16e
]
ANDROID_DEVICE_POOL = [
# (model, manufacturer, brand, os_release, sdk, (width_px, height_px, dp_width, density_dpi))
("SM-S911B", "samsung", "samsung", "14", 34, (1080, 2340, 360, 480)), # Galaxy S23
("SM-S918B", "samsung", "samsung", "14", 34, (1440, 3088, 384, 600)), # Galaxy S23 Ultra
("SM-S921B", "samsung", "samsung", "14", 34, (1080, 2340, 360, 480)), # Galaxy S24
("SM-S926B", "samsung", "samsung", "14", 34, (1440, 3120, 384, 600)), # Galaxy S24+
("SM-S928B", "samsung", "samsung", "14", 34, (1440, 3120, 384, 600)), # Galaxy S24 Ultra
("SM-A546B", "samsung", "samsung", "14", 34, (1080, 2340, 360, 480)), # Galaxy A54
("SM-G991B", "samsung", "samsung", "13", 33, (1080, 2340, 360, 480)), # Galaxy S21
("Pixel 7", "Google", "google", "14", 34, (1080, 2400, 411, 420)), # Pixel 7
("Pixel 8", "Google", "google", "14", 34, (1080, 2400, 411, 420)), # Pixel 8
("Pixel 8 Pro", "Google", "google", "14", 34, (1344, 2992, 448, 480)), # Pixel 8 Pro
("2211133G", "Xiaomi", "Xiaomi", "13", 33, (1080, 2400, 393, 440)), # Xiaomi 13
("CPH2449", "OnePlus", "OnePlus", "14", 34, (1440, 3216, 411, 560)), # OnePlus 11
]
# ---- Mobil app sürüm havuzları (per-hesap tutarlı, client_id SABİT) ----
# client_id sadece platformu belirtir (iOS/Android), app_version'a bağlı değildir;
# bu yüzden sürüm çeşitlendirilebilir → "tek app version" cluster sinyali kırılır.
# Güncel + eskiden yaygın sürümler karışık.
# iOS/Android: TEK 4-parça sürüm string'i; UA/header/client_info hep aynı sürüm.
# iOS UA formatı YAPIŞIK: 'Spotify/9.1.60.1970iOS/18.5(iPhone16,1)'. Gerçek sürümler
# (uydurma build → token/istek reddi riski). İstersen buraya daha çok gerçek 4-parça ekle.
IOS_APP_VERSIONS = [
"8.8.54.544", # yakalanmış known-good (4-parça)
"9.1.60.1970", # güncel yakalama (4-parça)
"8.9.82", # 3-parça
"9.0.40", # 3-parça
"9.0.48", # 3-parça
]
ANDROID_APP_VERSIONS = [
"8.7.32.1554", # yakalanmış known-good
"8.9.76.538", # 2024-09
"8.9.98.488", # 2024-12
"9.0.0.487", # 2024-12
"9.0.48.269", # 2025
"9.0.80.1343", # 2025-09
"9.1.6.1124", # 2025-12
]
def _pick_app_version(platform):
"""Per-hesap app sürümü seçer → {"marketing","full"}. Pool kapalı/boş → None."""
if not MOBILE_APP_VERSION_POOL_KULLAN:
return None
if platform == "ios" and IOS_APP_VERSIONS:
v = random.choice(IOS_APP_VERSIONS)
return {"marketing": v, "full": v} # iOS tek 4-parça sürüm (UA/header/body aynı)
if platform == "android" and ANDROID_APP_VERSIONS:
v = random.choice(ANDROID_APP_VERSIONS)
return {"marketing": v, "full": v}
return None
def _pick_mobile_device(platform):
"""Per-hesap tutarlı cihaz seçer. Pool kapalı/boş → None (hardcoded default'a düşülür)."""
if not MOBILE_DEVICE_POOL_KULLAN:
return None
if platform == "ios" and IOS_DEVICE_POOL:
model, osv = random.choice(IOS_DEVICE_POOL)
return {"platform": "ios", "model": model, "os_version": osv}
if platform == "android" and ANDROID_DEVICE_POOL:
model, manuf, brand, rel, sdk, screen = random.choice(ANDROID_DEVICE_POOL)
return {"platform": "android", "model": model, "manufacturer": manuf,
"brand": brand, "release": rel, "sdk": sdk, "screen": screen}
return None
def _mobile_app_ua(platform, device, av=None):
"""Spotify app User-Agent — clienttoken isteği + signup için ORTAK.
device/av=None → eski hardcoded cihaz/sürüm. iOS formatı YAPIŞIK (boşluksuz),
4-parça full sürümle: 'Spotify/9.1.60.1970iOS/16.0.2(iPhone10,3)'."""
appv = av["marketing"] if av else SIGNUP_PROFILES.get(platform, {}).get("app_version")
if platform == "ios":
model = device["model"] if device else "iPhone10,3"
osv = device["os_version"] if device else "16.0.2"
return f"Spotify/{appv}iOS/{osv}({model})"
if platform == "android":
model = device["model"] if device else "SM-N935F"
sdk = device["sdk"] if device else 30
return f"Spotify/{appv} Android/{sdk} ({model})"
return None
ANDROID_ATTESTATION = (
"AADynCg09JL/kts/V/2dLeQgo7SFoedRakEEVpAfTzNsUc+jxiinekfHJ6JMAWLZj2iU+4P5dUc/"
"AggZaJlpUct1fS4iCAqFEKr9pikXzkmZ/A46EZeVYKV0v3qSkWk2Tr/1e5KpXU9fVXPvuoH/Ele"
"FXSBtPNyIoFzs8sxcx5Oztl0C4ghHuVIfUde696Vkb5Cvsuksa6SBbJDWaxe5f7cDNDRixNCX5q"
"0R6gVwXopj0JxVxd60X7HLHdWPG+ydk4vw48EYr3wXI1Moh6VV87608hvp0RRuaMaArOuNSHWai"
"OXK/uO3pJ78+WSx1VC3kDIgdXma1A=="
)
CHALLENGE_HASH_CASH = 3
def _clienttoken_body_android(client_version, client_id, device_id, device=None):
# device=None → eski hardcoded SM-N935F. ANDROID_ATTESTATION her durumda sabit.
if device:
w, h, dp, dpi = device["screen"]
model, manuf, brand = device["model"], device["manufacturer"], device["brand"]
release, sdk = device["release"], device["sdk"]
else:
w, h, dp, dpi = 1080, 1794, 411, 420
model, manuf, brand = "SM-N935F", "samsung", "samsung"
release, sdk = "11", 30
tmpl = [
{"field": 1, "type": "varint", "value": 1},
{"field": 2, "type": "nested", "value": [
{"field": 1, "type": "string", "value": client_version},
{"field": 2, "type": "string", "value": client_id},
{"field": 3, "type": "nested", "value": [
{"field": 1, "type": "nested", "value": [
{"field": 1, "type": "nested", "value": [
{"field": 1, "type": "nested", "value": [
{"field": 1, "type": "varint", "value": w},
{"field": 2, "type": "varint", "value": h},
{"field": 3, "type": "varint", "value": dp},
{"field": 4, "type": "varint", "value": dpi},
{"field": 5, "type": "varint", "value": dpi},
]},
{"field": 2, "type": "string", "value": release},
{"field": 3, "type": "varint", "value": sdk},
{"field": 4, "type": "string", "value": "Android Bluedroid"},
{"field": 5, "type": "string", "value": model},
{"field": 6, "type": "string", "value": manuf},
{"field": 7, "type": "string", "value": brand},
{"field": 8, "type": "varint", "value": 32},
]},
]},
{"field": 2, "type": "string", "value": device_id},
]},
{"field": 5, "type": "string", "value": ANDROID_ATTESTATION},
]},
]
return build_proto(tmpl)
def _clienttoken_body_ios(client_version, client_id, device_id, model="iPhone10,3", os_version="16.0.2"):
tmpl = [
{"field": 1, "type": "varint", "value": 1},
{"field": 2, "type": "nested", "value": [
{"field": 1, "type": "string", "value": client_version},
{"field": 2, "type": "string", "value": client_id},
{"field": 3, "type": "nested", "value": [
{"field": 1, "type": "nested", "value": [
{"field": 2, "type": "nested", "value": [
{"field": 1, "type": "varint", "value": 0},
{"field": 3, "type": "string", "value": model},
{"field": 4, "type": "string", "value": os_version},
]},
]},
{"field": 2, "type": "string", "value": device_id},
]},
]},
]
return build_proto(tmpl)
def _clienttoken_body_windows(client_version, client_id, device_id):
"""NativeWindowsData (desktop/web variant) — librespot'suz, elle build_proto ile.
Çıktı librespot'un SerializeToString'i ile byte-byte aynı (doğrulandı).
Alan eşlemesi: PlatformSpecificData.windows = field 4;
NativeWindowsData something1/3/4/6/7/8/10 = field 1/3/4/6/7/8/10."""
tmpl = [
{"field": 1, "type": "varint", "value": 1}, # REQUEST_CLIENT_DATA_REQUEST
{"field": 2, "type": "nested", "value": [ # client_data
{"field": 1, "type": "string", "value": client_version},
{"field": 2, "type": "string", "value": client_id},
{"field": 3, "type": "nested", "value": [ # connectivity_sdk_data
{"field": 1, "type": "nested", "value": [ # platform_specific_data
{"field": 4, "type": "nested", "value": [ # windows (oneof field 4)
{"field": 1, "type": "varint", "value": 10},
{"field": 3, "type": "varint", "value": 21370},
{"field": 4, "type": "varint", "value": 2},
{"field": 6, "type": "varint", "value": 9},
{"field": 7, "type": "varint", "value": 332},
{"field": 8, "type": "varint", "value": 33404},
{"field": 10, "type": "varint", "value": 1}, # bool True
]},
]},
{"field": 2, "type": "string", "value": device_id},
]},
]},
]
return build_proto(tmpl)
# Hashcash PoW ayrı modülde (_hashcash.py) — process pool worker'ları onu import eder.
# `solve_hash_cash`/`_trailing_zero_bits` ana modülde de erişilebilir kalsın diye re-export.
from _hashcash import solve_hash_cash, _trailing_zero_bits # noqa: E402
# ---- Hashcash process pool ----
# PoW saf-Python SHA1 döngüsü → CPU-bound, GIL'i bırakmaz. 50 thread aynı anda
# length-20 PoW çözünce GIL yüzünden serileşip clienttoken state TTL'ini (~14s)
# aşıyor → cevap POST'u 400 alıyor. Çözüm: PoW'u SÜREÇ havuzunda çöz (gerçek paralel).
_POW_POOL = None
_POW_POOL_LOCK = threading.Lock()
# Tek bir PoW'un süreç havuzunda çözülmesi için tavan süre (saniye). Aşılırsa
# worker thread PER_ACCOUNT_TIMEOUT boyunca asılı kalmasın diye bir kez daha
# (taze submit) denenir; o da aşarsa hata yükseltilir.
POW_RESULT_TIMEOUT = 60
def _get_pow_pool():
global _POW_POOL
if _POW_POOL is None:
with _POW_POOL_LOCK:
if _POW_POOL is None:
# PoW CPU-bound: worker ≈ çekirdek sayısı (thread sayısı DEĞİL).
# 16 ile sınırlı: spawn'da her worker ana modülü yeniden import
# ediyor (PROFILES yükü), aşırı worker maliyetli.
workers = max(2, min(os.cpu_count() or 4, 16))
_POW_POOL = ProcessPoolExecutor(max_workers=workers)
atexit.register(_POW_POOL.shutdown, wait=False)
logger.info(f"hashcash · process pool başlatıldı (workers={workers})")
return _POW_POOL
def _solve_hash_cash_pooled(ctx, prefix, length):
"""PoW'u ayrı süreçte çöz (GIL'i aş).
- Süreç havuzunda çözülür; `.result()` POW_RESULT_TIMEOUT ile sınırlanır.
- Timeout: tek seferlik taze submit denenir (genelde havuz doygunluğudur,
worker boşalınca biter). In-process fallback'e DÜŞMEZ — fallback PoW'u
worker thread'inde GIL tutarak çalıştırıp diğer thread'leri yavaşlatır.
- Gerçek havuz arızası (BrokenProcessPool vb.): son çare in-process çöz.
"""
try:
fut = _get_pow_pool().submit(solve_hash_cash, ctx, prefix, length)
try:
return fut.result(timeout=POW_RESULT_TIMEOUT)
except FuturesTimeoutError:
logger.warning(
f"hashcash · {POW_RESULT_TIMEOUT}s timeout (havuz dolu olabilir) — taze submit"
)
fut.cancel()
fut2 = _get_pow_pool().submit(solve_hash_cash, ctx, prefix, length)
return fut2.result(timeout=POW_RESULT_TIMEOUT)
except Exception as e:
logger.warning(f"hashcash · pool fallback (in-process): {e}")
return solve_hash_cash(ctx, prefix, length)
def _read_clienttoken_response(raw):
tree = parse_proto(raw)
granted = _find(tree, 2)
chall = _find(tree, 3)
if granted and granted["type"] == "nested":
tok = _find(granted["value"], 1)
return ("granted", tok["value"] if tok else None)
if chall and chall["type"] == "nested":
cv = chall["value"]
state = (_find(cv, 1) or {}).get("value", "")
hcs = []
for ch in cv:
if ch["field"] == 2 and ch["type"] == "nested":
params = _find(ch["value"], 4)
if params and params["type"] == "nested":
ln = _find(params["value"], 1)
pf = _find(params["value"], 2)
hcs.append((ln["value"] if ln else 0, pf["value"] if pf else ""))
return ("challenges", state, hcs)
rt = _find(tree, 1)
return ("unknown", rt["value"] if rt else None)
def _clienttoken_answer_body(state, answers):
items = [{"field": 1, "type": "string", "value": state}]
for ctype, suffix_hex in answers:
items.append({"field": 2, "type": "nested", "value": [
{"field": 1, "type": "varint", "value": ctype},
{"field": 4, "type": "nested", "value": [
{"field": 1, "type": "string", "value": suffix_hex},
]},
]})
tmpl = [
{"field": 1, "type": "varint", "value": 2},
{"field": 3, "type": "nested", "value": items},
]
return build_proto(tmpl)
def _mb_proxies(proxy):
return {"http": proxy, "https": proxy} if proxy else None
class ClientTokenHTTPError(Exception):
"""clienttoken uç noktasının non-200 yanıtı — 'beklenen' hata (rate-limit/state
expiry vs.). raise_for_status'un ham HTTPError traceback'i yerine bununla
fırlatılır ki generate_mobile temiz LOG.fail bassın (traceback spam'i olmadan)."""
def __init__(self, status, body=""):
super().__init__(f"clienttoken http {status}")
self.status = status
self.body = body
def _mb_raw_post(url, headers, *, data=None, json=None, impersonate=None, proxy=None):
"""Standalone (session'sız) mobil POST — backend-agnostic (curl_cffi | primp).
clienttoken çağrıları için: her çağrı `impersonate` farklı olabildiğinden
(session ile aynı TLS fingerprint) kalıcı client yerine kısa ömürlü kullanılır.
primp yolunda Client GC'ye bırakılır (impersonate çağrı-başına değiştiği için
reuse edilemez; gövde eagerly buffer'landığından response geçerli kalır)."""
if ACTIVE_BACKEND == "curl_cffi":
return crequests.post(url, headers=headers, data=data, json=json,
impersonate=impersonate, proxies=_mb_proxies(proxy))
# primp — PrimpSession ile birebir: json elle serialize edilip content= gönderilir
client = primp.Client(impersonate=impersonate, proxy=proxy)
body = data
if json is not None:
import json as _j
body = _j.dumps(json).encode()
return client.post(url, headers=headers, content=body)
def _clienttoken_post(prof, body, proxy=None, impersonate=None, ua=None):
# ua verilirse (seçilen cihaza göre) UA header'ı override edilir; body'deki
# model/os ile tutarlı olması için.
hdrs = {**prof["headers"], "User-Agent": ua} if ua else prof["headers"]
r = _mb_raw_post(
"https://clienttoken.spotify.com/v1/clienttoken",
hdrs, data=body,
impersonate=impersonate or prof["impersonate"], proxy=proxy,
)
if r.status_code != 200:
raise ClientTokenHTTPError(r.status_code, (getattr(r, "text", "") or "")[:160])
return r.content
def _clienttoken_web_token(prof, proxy=None, impersonate=None):
payload = {"client_data": {
"client_id": prof["client_id"], "client_version": prof["client_version"],
"js_sdk_data": {"device_brand": "unknown", "device_model": "desktop",
"os": "windows", "os_version": "NT 10.0"}}}
r = _mb_raw_post(
"https://clienttoken.spotify.com/v1/clienttoken",
prof["headers"], json=payload,
impersonate=impersonate or prof["impersonate"], proxy=proxy,
)
if r.status_code != 200:
raise ClientTokenHTTPError(r.status_code, (getattr(r, "text", "") or "")[:160])
data = r.json()
token = (data.get("granted_token") or {}).get("token")
if not token:
raise RuntimeError(f"web clienttoken token alınamadı: {data}")
return token
def get_client_token(variant, device_id, proxy=None, impersonate=None, device=None, app_version=None):
if variant not in GETCLIENT_PROFILES:
raise ValueError(f"bilinmeyen GETCLIENT_VARIANT: {variant!r}")
prof = GETCLIENT_PROFILES[variant]
imp = impersonate or prof["impersonate"] # session ile aynı TLS fingerprint (rotasyon)
# Seçilen cihaz + app sürümüne göre UA + body — clienttoken isteği içinde tutarlı.
ua = _mobile_app_ua(variant, device, app_version) if variant in ("android", "ios") else None
cver = app_version["full"] if app_version else prof["client_version"] # client_id SABİT, sürüm seçili
LOG.info(f"clienttoken · variant={variant} client_id={prof['client_id'][:8]}… v={cver} impersonate={imp}"
+ (f" ua={_short(ua, 40)}" if ua else ""))
if variant == "web":
token = _clienttoken_web_token(prof, proxy, imp)
LOG.info(f"clienttoken · alındı ({len(token)} karakter)")
return token
if variant == "android":
body = _clienttoken_body_android(cver, prof["client_id"], device_id, device)
elif variant == "ios":
model = device["model"] if device else "iPhone10,3"
osv = device["os_version"] if device else "16.0.2"
body = _clienttoken_body_ios(cver, prof["client_id"], device_id, model, osv)
else:
body = _clienttoken_body_windows(prof["client_version"], prof["client_id"], device_id)
MAX_TRIES = 5
for attempt in range(1, MAX_TRIES + 1):
parsed = _read_clienttoken_response(_clienttoken_post(prof, body, proxy, imp, ua=ua))
kind = parsed[0]
if kind == "granted":
token = parsed[1]
LOG.info(f"clienttoken · alındı ({len(token)} karakter)")
return token
if kind == "challenges":
_, state, hcs = parsed
if not hcs:
raise RuntimeError("challenge geldi ama hashcash parametresi yok")
answers = []
for length, prefix_hex in hcs:
LOG.info(f"clienttoken · hashcash çözülüyor (length={length}, deneme {attempt}/{MAX_TRIES})")
suffix = _solve_hash_cash_pooled(b"", bytes.fromhex(prefix_hex), length)
answers.append((CHALLENGE_HASH_CASH, suffix.hex().upper()))
body = _clienttoken_answer_body(state, answers)
continue
raise RuntimeError(f"clienttoken beklenmeyen yanıt: {parsed}")
raise RuntimeError(f"clienttoken: {MAX_TRIES} denemede token alınamadı")
# ---- mobil signup + challenge ----
def _mb_new_session(proxy):
"""Mobil session — aktif backend'e göre curl_cffi (ham) veya primp (wrapper).
Web ile aynı dinamik impersonate/UA havuzunu kullanır. _pick_impersonate
(batch başına rotasyon) + _activate_dynamic_ua sayesinde challenge web-context
çağrıları (visit/solve_v2/invoke/complete) dinamik tarayıcı UA'sı alır. App
çağrıları (clienttoken/signup) kendi app UA'sını header'da set etmeye devam eder."""
if ACTIVE_BACKEND == "curl_cffi":
imp = _pick_impersonate(IMPERSONATE_PROFILES_CURL_CFFI)
_activate_dynamic_ua(imp)
s = crequests.Session(impersonate=imp)
s.impersonate = imp
if proxy:
s.proxies = {"http": proxy, "https": proxy}
return s
# primp — PrimpSession wrapper mobil akışın kullandığı tüm API'yi karşılar:
# .impersonate / .get / .post(headers,data,json) / .cookies(.set/.items) / .close
if not _HAS_PRIMP:
LOG.fail("mobil akış: aktif backend primp ama primp kurulu değil")
return None
return PrimpSession(proxy)
def _mb_send_signup(session, token, payload_json, platform, client_id, device=None, app_version=None):
sp = SIGNUP_PROFILES[platform]
body = build_proto(payload_json)
# Seçilen cihaz + app sürümüne göre UA (clienttoken ile aynı); yoksa profil default'u.
ua = _mobile_app_ua(platform, device, app_version) or sp["headers"]["User-Agent"]
appver_hdr = app_version["full"] if app_version else sp["headers"]["Spotify-App-Version"]
h = {
"Accept": "application/protobuf",
"Client-Token": token,
"Accept-Language": LOCALE_LANG_SHORT, # locale ile tutarlı (gerçek iOS 'tr')
"User-Agent": ua,
"Spotify-App-Version": appver_hdr,
"X-Client-Id": client_id,
"App-Platform": sp["headers"]["App-Platform"],
"Content-Type": "application/x-protobuf",
"Accept-Encoding": "gzip, deflate, br",
}
r = None
for attempt in range(CREATE_MAX_RETRIES):
try:
r = session.post(SPCLIENT_CREATE_URL, headers=h, data=body, timeout=CREATE_TIMEOUT)
break
except Exception as e:
if attempt < CREATE_MAX_RETRIES - 1:
LOG.warn(f"signup · retry {attempt+1}/{CREATE_MAX_RETRIES}: {e}")
time.sleep(RETRY_SLEEP)
else:
LOG.warn(f"signup · exhausted: {e}")
raise
raw = decode_response(r)
parsed = parse_proto(raw)
LOG.info(f"signup · http {r.status_code} ({len(raw)}B)")
if r.status_code != 200:
LOG.warn(f"signup · beklenmeyen status {r.status_code}")
return r.status_code, parsed
def extract_signup_challenge(parsed):
"""Protobuf signup yanıtında challenge varsa session_id/creation_id döndürür."""
ch = next((x for x in parsed if x["field"] == 3 and x["type"] == "nested"), None)
if not ch:
return None
sid_node = next((x for x in ch["value"] if x["field"] == 1), None)
session_id = sid_node["value"] if sid_node else None
if not session_id:
return None
cid_node = next((x for x in parsed if x["field"] == 4), None)
creation_id = cid_node["value"] if cid_node else None
return {"session_id": session_id, "creation_id": creation_id,
"url": f"{CHALLENGE_BASE}/c/{session_id}"}
def _mb_parse_success(parsed):
"""Doğrudan başarı (challenge yok) yanıtından username/login_token (best-effort)."""
node = _find(parsed, 1)
if not node or node.get("type") != "nested":
return None, None
sub = node["value"]
username = (_find(sub, 1) or {}).get("value")
login_token = (_find(sub, 2) or {}).get("value")
return username, login_token
def _mb_resolve_challenge_url(session, session_id):
h = {
"Content-Type": "application/json",
"Accept": "application/json",
"Origin": SPOTIFY_ORIGIN,
"Referer": SPOTIFY_REFERER,
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
}
# Web step2_create ile aynı: geçici ağ hatasında GET_SESSION_MAX_RETRIES kez dene.
for attempt in range(GET_SESSION_MAX_RETRIES):
try:
LOG.info(f"get-session · POST attempt {attempt+1}")
r = session.post(CHALLENGE_GET_SESSION_URL, headers=h,
json={"session_id": session_id}, timeout=CHALLENGE_TIMEOUT)
LOG.info(f"get-session · status={r.status_code}")
if r.status_code != 200:
LOG.warn(f"get-session · http {r.status_code}: {r.text[:160]}")
return None
sd = r.json()
url = (sd.get("in_progress", {}).get("challenge_details", {})
.get("web_challenge_launcher", {}).get("url", ""))
return url or None
except Exception as e:
if attempt < GET_SESSION_MAX_RETRIES - 1:
LOG.warn(f"get-session retry {attempt+1}: {e}")
time.sleep(RETRY_SLEEP)
else:
LOG.warn(f"get-session hatası: {e}")
return None
def _mb_visit_challenge(session, challenge_url):
m = re.search(r'/c/([^/]+)/', challenge_url)
if m:
try:
session.cookies.set("sp_csid", m.group(1))
except Exception:
pass
h = {
"user-agent": current_ua(),
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"accept-language": LOCALE_LANG_HEADER,
"cache-control": "no-cache",
"pragma": "no-cache",
"referer": SPOTIFY_REFERER,
"sec-fetch-site": "same-site",
"sec-fetch-mode": "navigate",
"sec-fetch-dest": "document",
**current_sec_ch_headers(),
}
try:
r = session.get(challenge_url, headers=h, allow_redirects=True, timeout=SUBPROCESS_TIMEOUT)
LOG.info(f"visit challenge · status={getattr(r, 'status_code', '?')}")
parsed = urllib.parse.urlparse(challenge_url)
try:
session.cookies.set("sp_landing",
urllib.parse.quote(f"{CHALLENGE_BASE}{parsed.path}", safe=""))
except Exception:
pass
return r.text
except Exception as e:
LOG.warn(f"visit_challenge failed: {e}")
return None
def _mb_solve_email_challenge(session, full_url, email_ch, profile):
"""emailChallenge (__NEXT_DATA__'dan) → aktif MAIL_SOURCE'tan kodu poll et → invoke → complete.
Spotify bazı hesaplarda reCAPTCHA yerine e-posta PIN doğrulaması istiyor
(`props.pageProps.emailChallenge`). Kod yalnızca gerçek bir kutuya
ulaşabildiğimiz kaynaklarda (MAIL_SOURCE="ucretli" veya "ucretsiz")
okunabilir — "random" modda (var olmayan sahte e-posta) bu challenge
çözülemez, açıkça fail ederiz."""
session_id = email_ch.get("sessionId")
challenge_id = email_ch.get("challengeId")
pin_length = email_ch.get("pinLength") or 6
if not session_id or not challenge_id:
LOG.fail("emailChallenge session/challenge id eksik", detail=str(email_ch))
STATS.record_email_challenge(ok=False)
return None
if MAIL_SOURCE == "ucretli" and not (profile or {}).get("mailSifre"):
LOG.fail("emailChallenge geldi ama ücretli mail order edilmemiş (mailSifre yok) — kod alınamaz")
STATS.record_email_challenge(ok=False)
return None
if MAIL_SOURCE == "ucretsiz" and not (profile or {}).get("ucretsizMailSession"):
LOG.fail("emailChallenge geldi ama ücretsiz mailbox açılamamış — kod alınamaz")
STATS.record_email_challenge(ok=False)
return None
if MAIL_SOURCE == "random":
LOG.fail("emailChallenge geldi ama MAIL_SOURCE=random (gerçek kutu yok) — kod alınamaz")
STATS.record_email_challenge(ok=False)
return None
LOG.info(f"emailChallenge · session={_short(session_id, 12)} id={_short(challenge_id, 12)} "
f"pin={pin_length} mail_source={MAIL_SOURCE}")
t0 = time.time()
def _poll(attempt):
LOG.info(f"emailChallenge · kod bekleniyor {attempt} ({int(time.time()-t0)}s)")
code = wait_for_email_code(profile, pin_length=pin_length, on_poll=_poll)
if not code:
LOG.fail("emailChallenge · kod gelmedi (timeout)")
STATS.record_email_challenge(ok=False)
return None
STATS.record_email_challenge(ok=True)
LOG.info(f"emailChallenge · kod alındı ({time.time()-t0:.1f}s)")
inv = invoke_email_challenge(session, session_id, challenge_id, code, challenge_url=full_url)
if not inv:
LOG.fail("emailChallenge · invoke başarısız")
return None
LOG.info("emailChallenge · invoke ok")
return complete_creation(session, session_id)
def _mb_complete_challenge(session, full_url, ch, sticky_proxy, profile=None):
"""Challenge sayfası → __NEXT_DATA__ siteKey (recaptcha) veya emailChallenge → solve → invoke → complete.
request.py'nin solve_v2/invoke_challenge/complete_creation'ını yeniden kullanır."""
html = _mb_visit_challenge(session, full_url)
if not html:
LOG.fail("challenge sayfası alınamadı")
return None
LOG.info(f"visit challenge · done ({len(list(session.cookies.items()))} cookies)")
nd = _extract_next_data(html)
page_props = nd.get("props", {}).get("pageProps", {})
rc = page_props.get("reCaptchaChallenge", {})
site_key = rc.get("siteKey")
nd_session = rc.get("sessionId")
nd_challenge = rc.get("challengeId")
site_key_type = rc.get("siteKeyType")
if not site_key:
email_ch = page_props.get("emailChallenge")
if email_ch:
return _mb_solve_email_challenge(session, full_url, email_ch, profile)
LOG.fail("siteKey __NEXT_DATA__ içinde bulunamadı")
return None
LOG.info(f"reCaptchaChallenge · siteKey={_short(site_key, 12)} type={site_key_type}")
# session_id adayları (USE_PROTOBUF_SESSION_ID = sıra) + challenge_id (URL path / next_data)
proto_sid = ch.get("session_id")
ordered = [proto_sid, nd_session] if USE_PROTOBUF_SESSION_ID else [nd_session, proto_sid]
seen = set()
session_id_candidates = [s for s in ordered if s and not (s in seen or seen.add(s))]
challenge_id = None
try:
parts = urllib.parse.urlparse(full_url).path.split("/")
if len(parts) >= 4:
challenge_id = parts[3]
except Exception:
pass
if not challenge_id:
challenge_id = nd_challenge
if not session_id_candidates or not challenge_id:
# reason sabit tutulur (UUID'ler detail'e taşındı) ki STATS.fail_reasons
# her hesapta ayrı bir key açıp sınırsız büyümesin — aksi halde özet
# raporda "en sık hata nedenleri" bu satır için anlamsızlaşır.
LOG.fail("session_id/challenge_id eksik",
detail=f"adaylar={session_id_candidates} challenge={challenge_id}")
return None
LOG.info(f"challenge · session={_short(session_id_candidates[0], 12)} "
f"id={_short(challenge_id, 12)} url={_short(full_url, 60)}")
# captcha v2 enterprise çöz — web'le aynı: createTask logu + on_poll periyodik bekleme logu
LOG.info(f"captcha v2 · createTask siteKey={_short(site_key, 12)}")
t0 = time.time()
poll_count = {"n": 0}
def _poll():
poll_count["n"] += 1
LOG.info(f"captcha v2 · polling {poll_count['n']} ({int(time.time()-t0)}s)")
tok, ua_v2, tid, prov = solve_v2(
site_key=site_key, page_url=full_url, proxy=sticky_proxy,
is_invisible=False,
site_key_type=str(site_key_type) if site_key_type is not None else None,
on_poll=_poll,
cookies=_format_session_cookies(session) if CAPTCHA_SEND_COOKIES else None,
)
if not tok:
STATS.record_captcha("v2", ok=False)
LOG.fail("captcha v2 çözülemedi", detail=f"after {time.time()-t0:.0f}s")
return None
STATS.record_captcha("v2", ok=True)
LOG.info(f"captcha v2 · token len={len(tok)} provider={prov} ({time.time()-t0:.1f}s)")
# invoke — adayları AYNI token'la sırayla dene
session_id = None
for cand in session_id_candidates:
inv = invoke_challenge(session, cand, challenge_id, tok, challenge_url=full_url, user_agent=ua_v2)
if inv:
session_id = cand
LOG.info(f"invoke challenge · 200 (session_id={_short(cand, 14)})")
break
if len(session_id_candidates) > 1:
LOG.warn(f"invoke session_id={_short(cand, 14)} başarısız — sıradaki aday")
if not session_id:
_report_bad(prov, tid, "v2")
LOG.fail("invoke challenge başarısız (tüm session_id adayları)")
return None
comp = complete_creation(session, session_id)
if comp is not None:
comp["ua_v2"] = ua_v2 # acct dict'i gerçek captcha çözüm UA'sını kaydedebilsin
return comp
def generate_mobile(proxy=None, sticky_proxy=None, idx=0):
"""Mobil/protobuf hesap açma akışı (SIGNUP_MODE == 'mobile')."""
profile = PROFILES.get()
if USE_PROFILE_DATA:
email = profile["mail"]
password = profile["sifre"]
display_name = profile["display_name"]
dob = profile["birthdate"]
gender = profile["cinsiyet"]
else:
email = rand_email()
password = rand_pass()
display_name = rand_name()
dob = rand_dob()
gender = random.randint(1, 2)
# MAIL_SOURCE'a göre email'i belirler ("ucretli" → anymessage.shop order,
# "ucretsiz" → tempmail.com.tr, "random" → değişmez). Hata/bakiye/network
# sorununda sessizce mevcut email'e fallback yapar, akış kesilmez.
email = mail_kaynagi_uygula(profile, email, tid=idx)
LOG.begin(idx, email)
LOG.info(f"identity · pass={_short(password,4)}*** name={display_name} dob={dob} gender={gender}")
LOG.info(f"mobile · variant={GETCLIENT_VARIANT} platform={SIGNUP_PLATFORM} sync_id={SYNC_SIGNUP_CLIENT_ID}")
LOG.info(f"proxy · http={'on' if proxy else 'off'} captcha={'on' if sticky_proxy else 'off'} | v2={V2_SOLVER}")
t0 = time.time()
s = None
basarili = False # ücretli mail iptali için: return acct'a ulaşılırsa True
try:
# Session'ı önce kur → impersonate aktive olsun (hem challenge web-context hem clienttoken için)
s = _mb_new_session(proxy)
if s is None:
return None
session_impersonate = getattr(s, "impersonate", "") or ""
session_ua = current_ua() or ""
LOG.info(f"session · impersonate={session_impersonate or '?'} ua={_short(session_ua, 40)}")
# Per-hesap tutarlı cihaz + app sürümü (clienttoken variant'ına göre seç);
# aynı ikili clienttoken UA/body + signup UA/body boyunca kullanılır.
device = _pick_mobile_device(GETCLIENT_VARIANT)
app_ver = _pick_app_version(GETCLIENT_VARIANT)
# signup için: platformlar eşitse (önerilen) aynı cihaz+sürüm; farklıysa
# signup platformuna uygun ayrı seçim (her istek kendi içinde tutarlı).
if SIGNUP_PLATFORM == GETCLIENT_VARIANT:
signup_device, signup_app = device, app_ver
else:
signup_device = _pick_mobile_device(SIGNUP_PLATFORM)
signup_app = _pick_app_version(SIGNUP_PLATFORM)
if device:
LOG.info(f"device · {device.get('model')} / {device.get('os_version') or device.get('release')}"
f" app={(app_ver or {}).get('marketing', '-')}")
install_id = _secrets.token_hex(SIGNUP_PROFILES[SIGNUP_PLATFORM]["install_len"])
signup_json = build_signup_json(SIGNUP_PLATFORM, email, password, install_id,
display_name=display_name, dob=dob, gender=gender,
app_version=signup_app)
if SHOW_BODY:
dump_tree(signup_json)
token = get_client_token(GETCLIENT_VARIANT, install_id, proxy=proxy,
impersonate=session_impersonate, device=device, app_version=app_ver)
if SYNC_SIGNUP_CLIENT_ID:
signup_client_id = GETCLIENT_PROFILES[GETCLIENT_VARIANT]["client_id"]
else:
signup_client_id = SIGNUP_PROFILES[SIGNUP_PLATFORM]["x_client_id"]
status_code, parsed = _mb_send_signup(s, token, signup_json, SIGNUP_PLATFORM,
signup_client_id, device=signup_device, app_version=signup_app)
if status_code != 200:
# non-200 → protobuf gövde çöp olabilir; challenge/success araması yanıltıcı
# olur. Web'in step2_create'i gibi burada net hata ver ve kısa devre yap.
LOG.fail(f"signup http {status_code}", detail=json.dumps(parsed)[:200])
return None
ch = extract_signup_challenge(parsed)
via_challenge = False
ua_used = session_ua
if ch:
STATS.record_challenge()
if CHALLENGE_ATLA:
LOG.info("challenge geldi → atlanıyor (CHALLENGE_ATLA)")
return None # ücretli mail açıksa finally iptal/iade eder
LOG.info(f"challenge geldi · session_id={_short(ch['session_id'], 14)} — get-session ile URL çözülüyor")
full_url = _mb_resolve_challenge_url(s, ch["session_id"])
if not full_url:
LOG.fail("challenge url çözülemedi")
return None
comp = _mb_complete_challenge(s, full_url, ch, sticky_proxy, profile=profile)
if not (comp and comp.get("status") == "success"):
LOG.fail("challenge tamamlanamadı", detail=str(comp))
return None
username = comp.get("username")
login_token = comp.get("login_token")
ua_used = comp.get("ua_v2") or session_ua
via_challenge = True
else:
username, login_token = _mb_parse_success(parsed)
if not username or not login_token:
LOG.fail("signup başarısız (ne success ne challenge)", detail=json.dumps(parsed)[:200])
return None
acct = {
"email": email, "password": password, "username": username,
"login_token": login_token, "display_name": display_name,
"birthdate": dob, "gender": gender, "profile": profile,
"cookies": dict(s.cookies.items()),
"via_challenge": via_challenge, "thread_idx": idx,
"duration_s": round(time.time() - t0, 2),
"created_at": datetime.now(timezone.utc).isoformat(),
"impersonation": session_impersonate, "ua": ua_used,
}
fp = _persist(acct, profile)
LOG.info(f"persist · {fp}")
LOG.success(f"saved{' (challenge)' if via_challenge else ''} user={username} duration={acct['duration_s']}s")
basarili = True
return acct
except KeyboardInterrupt:
raise
except ClientTokenHTTPError as e:
# Beklenen HTTP hatası (rate-limit/state expiry) → temiz LOG.fail, traceback yok.
LOG.fail(str(e), detail=e.body)
return None
except Exception as e:
LOG.fail("mobile create exception", detail=str(e))
logger.exception("generate_mobile failed")
return None
finally:
# Session'ı her durumda kapat — web generate()'teki gibi FD/connection leak'ini önler.
if s is not None:
try:
s.close()
except Exception:
pass
# Hesap açılamadıysa (return acct'a ulaşılmadıysa) mail kaynağına göre
# iptal/iade et (yalnızca "ucretli" bir şey yapar — bkz. mail_kaynagi_iptal).
if not basarili:
mail_kaynagi_iptal(profile)
def generate(proxy=None, sticky_proxy=None, idx=0):
if SIGNUP_MODE == "mobile":
return generate_mobile(proxy=proxy, sticky_proxy=sticky_proxy, idx=idx)
# Profile data (opt-in via USE_PROFILE_DATA) — gives realistic names + gender.
profile = PROFILES.get()
if USE_PROFILE_DATA:
email = profile["mail"]
password = profile["sifre"]
display_name = profile["display_name"]
dob = profile["birthdate"]
gender = profile["cinsiyet"]
else:
email = rand_email()
password = rand_pass()
display_name = rand_name()
dob = rand_dob()
# Spotify gender field: 1 = male, 2 = female. Random fallback matches
# `spotifyHesapAcici_request_test.py:ProfileDataManager._generate_random_profile`.
gender = random.randint(1, 2)
# MAIL_SOURCE'a göre email'i belirler ("ucretli" → anymessage.shop order,
# "ucretsiz" → tempmail.com.tr, "random" → değişmez). Hata/bakiye/network
# sorununda sessizce mevcut email'e fallback yapar, akış kesilmez.
email = mail_kaynagi_uygula(profile, email, tid=idx)
LOG.begin(idx, email)
LOG.info(f"identity · pass={_short(password,4)}*** name={display_name} dob={dob} gender={gender}")
LOG.info(f"profile · source={'file' if (USE_PROFILE_DATA and PROFILES._use_file) else 'random'}")
LOG.info(
f"proxy · http={'on' if proxy else 'off'}({ACCOUNT_PROXY_TYPE}) "
f"captcha={'on' if sticky_proxy else 'off'} backend={ACTIVE_BACKEND}"
)
LOG.info(f"solvers · v2={V2_SOLVER} v3={V3_SOLVER}")
t0 = time.time()
s = None
basarili = False # ücretli mail iptali için: return acct'a ulaşılırsa True
try:
# HTTP session: ACCOUNT_PROXY_TYPE'a göre seçilen `proxy`.
# Captcha solver'lar aşağıda `sticky_proxy` ile çağrılmaya devam eder.
s = new_session(proxy)
session_impersonate = getattr(s, 'impersonate', '') or ''
session_ua = current_ua() or ''
LOG.info(f"session · created (impersonate={session_impersonate or '?'})")
LOG.info("consent · seeding cookies")
consent_add(s)
ck_count = len(list(s.cookies.items()))
LOG.info(f"consent · done ({ck_count} cookies)")
LOG.info("init · GET signup page")
init_data = step0_init(s)
if not init_data:
LOG.fail("init failed")
return None
LOG.info(
f"init · build_id={_short(init_data.get('build_id'), 8)} "
f"flow_id={_short(init_data.get('flow_id'), 8)} "
f"api_key={_short(init_data.get('api_key'), 12)}"
)
LOG.info(
f"init · scoring_key={_short(init_data.get('recaptcha_scoring_key'), 12)} "
f"checkbox_key={_short(init_data.get('recaptcha_checkbox_key'), 12)}"
)
flow_id = init_data["flow_id"]
if not flow_id:
LOG.info("init · flow_id missing, fetching via data.json")
flow_id = get_flow_id(s, init_data["build_id"])
if not flow_id:
LOG.fail("no flow id")
return None
LOG.info(f"init · flow_id={_short(flow_id, 8)} (refetched)")
api_key = init_data["api_key"]
creation_point = init_data["creation_point"]
scoring_key = init_data["recaptcha_scoring_key"]
checkbox_key = init_data.get("recaptcha_checkbox_key", scoring_key)
page_inst = init_data["page_inst"]
app_inst = init_data["app_inst"]
inst_id = init_data.get("sp_t") or s.cookies.get("sp_t")
if not inst_id:
inst_id = str(uuid.uuid4())
LOG.info(f"sp_t · generated {_short(inst_id, 8)}")
else:
LOG.info(f"sp_t · reused {_short(inst_id, 8)}")
s.cookies.set("sp_t", inst_id, domain=SPOTIFY_COOKIE_DOMAIN)
page_pw = str(uuid.uuid4())
page_ui = str(uuid.uuid4())
page_terms = str(uuid.uuid4())
LOG.info("events · signup_pre + flow_corr + recaptcha_load")
signup_pre(s, flow_id, inst_id, page_inst, app_inst, page_pw, page_ui, page_terms)
LOG.info("validate · email")
if not step1_validate(s, email, password, inst_id, api_key, creation_point):
LOG.fail("validation failed")
return None
LOG.info("validate · email + password ok")
LOG.info("email-dossier · POST (soft)")
try:
r = s.post(EMAIL_DOSSIER_URL, json={"email": email}, headers=headers())
LOG.info(f"email-dossier · status={r.status_code}")
except Exception as e:
LOG.warn(f"email-dossier soft-fail: {e}")
if init_data["build_id"]:
LOG.info("flow · refreshing flow_id")
fresh = get_flow_id(s, init_data["build_id"]) or flow_id
else:
fresh = flow_id
LOG.info(f"flow · fresh={_short(fresh, 8)}")
LOG.info("events · flow_corr + signup_post")
flow_corr(s, fresh, inst_id)
signup_post(s, flow_id, inst_id, page_inst, app_inst, page_pw, page_ui, page_terms)
LOG.info(f"captcha v3 · solving action={SIGNUP_SUBMIT_ACTION}")
cap_t0 = time.time()
v3_result = solve_v3(
site_key=scoring_key,
page_url=SIGNUP_URL,
action=SIGNUP_SUBMIT_ACTION,
proxy=sticky_proxy,
skip_direct=True,
cookies=_format_session_cookies(s) if CAPTCHA_SEND_COOKIES else None,
)
if v3_result:
tok_create, ua_v3, tid_v3, prov_v3 = v3_result
else:
tok_create, ua_v3, tid_v3, prov_v3 = None, None, None, ""
if not tok_create:
STATS.record_captcha("v3", ok=False)
LOG.fail("captcha failed", detail=f"after {time.time()-cap_t0:.1f}s")
return None
STATS.record_captcha("v3", ok=True)
LOG.info(f"captcha v3 · token len={len(tok_create)} ({time.time()-cap_t0:.1f}s) provider={prov_v3}")
if ua_v3:
LOG.info("captcha v3 · solution UA → using for create call")
LOG.info("events · registration_attempt")
registration_attempt(s, inst_id)
# Stamp callback_uri ONCE here so step2_create + complete_creation share
# the same flow_ctx timestamp (Spotify ties session continuation to this).
callback_timestamp = int(time.time())
callback_uri = SIGNUP_CHALLENGE_CB_TPL.format(
flow_ctx=urllib.parse.quote(f"{fresh}:{callback_timestamp}"),
ubi=urllib.parse.quote(make_ubi(page_terms, app_inst, fresh)),
locale=LOCALE,
)
LOG.info(f"callback · ts={callback_timestamp} uri={_short(callback_uri, 60)}")
for attempt in range(CREATE_MAX_ATTEMPTS):
LOG.info(f"create · POST account/create (attempt {attempt+1}/{CREATE_MAX_ATTEMPTS})")
res = step2_create(s, email, password, display_name, dob, inst_id, tok_create,
fresh, page_terms, app_inst, api_key, creation_point,
gender=gender, callback_uri=callback_uri,
callback_timestamp=callback_timestamp,
user_agent=ua_v3)
if not res:
LOG.fail("create failed")
return None
status = res.get("status", "unknown")
LOG.info(f"create · response status={status}")
if status == "error":
code = res.get("code")
err = res.get("error")
detail = f"http {code}" if code else None
if err:
detail = (detail + ": " if detail else "") + str(err)
# V3 token Spotify tarafında reject edildi (HTTP error response).
# Provider'a reportbad — refund / quality skor için.
_report_bad(prov_v3, tid_v3, "v3")
LOG.fail("request failed", detail=detail)
return None
if status == "success":
username = res.get("username")
login_token = res.get("login_token")
LOG.info(f"create · username={username} login_token={_short(login_token, 16)}")
acct = {
"email": email,
"password": password,
"username": username,
"login_token": login_token,
"display_name": display_name,
"birthdate": dob,
"gender": gender,
"profile": profile,
"cookies": res.get("cookies", {}),
"via_challenge": False,
"thread_idx": idx,
"duration_s": round(time.time() - t0, 2),
"created_at": datetime.now(timezone.utc).isoformat(),
"impersonation": session_impersonate,
"ua": ua_v3 or session_ua,
}
fp = _persist(acct, profile)
LOG.info(f"persist · {fp}")
LOG.success(f"saved user={username} duration={acct['duration_s']}s")
basarili = True
return acct
if status == "challenge":
STATS.record_challenge()
# V3 yüksek skorlu token verseydi challenge'a düşmezdi —
# token zayıf sayılır, provider'a reportbad. En net case.
_report_bad(prov_v3, tid_v3, "v3")
if CHALLENGE_ATLA:
LOG.info("challenge geldi → atlanıyor (CHALLENGE_ATLA)")
return None # ücretli mail açıksa finally iptal/iade eder
ch_session = res.get("session_id")
ch_id = res.get("challenge_id")
ch_url = res.get("challenge_url")
LOG.info(
f"challenge · session={_short(ch_session, 12)} "
f"id={_short(ch_id, 12)} url={_short(ch_url, 60)}"
)
if not ch_url or not ch_session or not ch_id:
LOG.fail("challenge info missing")
return None
LOG.info("visit challenge · curl")
if not visit_challenge(s, ch_url):
LOG.fail("visit challenge failed")
return None
ck_now = len(list(s.cookies.items()))
LOG.info(f"visit challenge · done ({ck_now} cookies)")
LOG.info(f"captcha v2 · createTask siteKey={_short(checkbox_key, 12)}")
ts_cap = time.time()
poll_count = {"n": 0}
def _poll():
poll_count["n"] += 1
LOG.info(f"captcha v2 · polling {poll_count['n']} ({int(time.time()-ts_cap)}s)")
v2_result = solve_v2(
site_key=checkbox_key,
page_url=ch_url,
proxy=sticky_proxy,
is_invisible=False,
site_key_type="1",
on_poll=_poll,
cookies=_format_session_cookies(s) if CAPTCHA_SEND_COOKIES else None,
)
tok_v2_str, ua_v2, tid_v2, prov_v2 = v2_result
if not tok_v2_str:
STATS.record_captcha("v2", ok=False)
LOG.fail("v2 captcha failed", detail=f"after {time.time()-ts_cap:.0f}s")
return None
STATS.record_captcha("v2", ok=True)
LOG.info(f"captcha v2 · token received len={len(tok_v2_str)} ({time.time()-ts_cap:.1f}s) provider={prov_v2}")
if ua_v2:
LOG.info("captcha v2 · solution UA → using for invoke_challenge")
LOG.info("invoke challenge · POST")
inv = invoke_challenge(s, ch_session, ch_id, tok_v2_str, challenge_url=ch_url,
user_agent=ua_v2)
if not inv:
# V2 token Spotify challenge endpoint'i tarafından reject edildi.
# Provider'a reportbad — refund / quality skor için.
_report_bad(prov_v2, tid_v2, "v2")
LOG.fail("invoke challenge failed")
return None
LOG.info("invoke challenge · 200")
LOG.info("complete creation · POST")
# NOTE: orijinal main.py complete_creation çağrısında özel
# referer geçmiyor (default SIGNUP_URL). Kaynak dosyada test
# edilen callback_uri referer Spotify'ın mevcut backend
# davranışıyla uyumsuz görünüyor — orijinal davranışa dönüyoruz.
comp = complete_creation(s, ch_session)
if comp and comp.get("status") == "success":
username = comp.get("username")
login_token = comp.get("login_token")
LOG.info(f"complete · username={username} login_token={_short(login_token, 16)}")
acct = {
"email": email,
"password": password,
"username": username,
"login_token": login_token,
"display_name": display_name,
"birthdate": dob,
"gender": gender,
"profile": profile,
"cookies": comp.get("cookies", {}),
"via_challenge": True,
"thread_idx": idx,
"duration_s": round(time.time() - t0, 2),
"created_at": datetime.now(timezone.utc).isoformat(),
"impersonation": session_impersonate,
"ua": ua_v2 or ua_v3 or session_ua,
}
fp = _persist(acct, profile)
LOG.info(f"persist · {fp}")
LOG.success(f"saved (challenge) user={username} duration={acct['duration_s']}s")
basarili = True
return acct
LOG.fail("complete creation failed")
return None
return None
except KeyboardInterrupt:
raise
except Exception as e:
LOG.fail("connection error", detail=str(e))
logger.exception("unexpected error in generate")
return None
finally:
# Session'ı her durumda kapat — connection pool / FD leak'ini önler.
# Abandoned worker (PER_ACCOUNT_TIMEOUT) C-level libcurl call'da asılı
# kalırsa finally'e ulaşmaz; o nadir patolojik durumda GC'ye bırakırız.
if s is not None:
s.close()
# Hesap açılamadıysa mail kaynağına göre iptal/iade et (yalnızca
# "ucretli" bir şey yapar — bkz. mail_kaynagi_iptal).
if not basarili:
mail_kaynagi_iptal(profile)
def worker(idx):
# Rate-watchdog pause kontrolü — düşük başarı oranı pause'u aktifse bekle.
_wait_if_paused()
if _shutdown_event.is_set():
return False
# Per-worker başlangıç jitter'ı: aynı batch içindeki thread'lerin ilk isteklerini
# birbirinden kaydırıp eşzamanlı-burst imzasını kırar (submit stagger'a ek katman).
if WORKER_START_JITTER:
_shutdown_event.wait(random.uniform(0, WORKER_START_JITTER))
if _shutdown_event.is_set():
return False
# Clear any leftover thread-local context from a previous job on this worker thread
_step_local.tid = None
_step_local.t0 = None
_step_local.email = None
# HTTP session proxy: ACCOUNT_PROXY_TYPE seçimine göre.
# Captcha solver'lar her durumda STICKY_PROXY alır.
if NO_PROXY:
http_proxy = None
sticky = None
else:
http_proxy = STICKY_PROXY if ACCOUNT_PROXY_TYPE == "sticky" else PROXY
sticky = STICKY_PROXY
try:
result = generate(proxy=http_proxy, sticky_proxy=sticky, idx=idx)
if result:
STATS.record(True, duration=result.get("duration_s", 0.0))
else:
STATS.record(False)
return bool(result)
except KeyboardInterrupt:
raise
except Exception as e:
logger.exception(f"worker {idx} crashed: {e}")
STATS.record(False)
return False
def _paced_submit_sleep():
"""Ardışık worker submit'leri arasında INTER_SUBMIT_DELAY + rastgele jitter kadar bekler.
time.sleep yerine _shutdown_event.wait → Ctrl-C/shutdown anında responsive."""
d = INTER_SUBMIT_DELAY + random.uniform(0, INTER_SUBMIT_JITTER)
if d > 0:
_shutdown_event.wait(d)
def _run_pool_continuous(num_threads):
"""Saturate `num_threads` workers indefinitely; FIRST_COMPLETED → resubmit.
PER_ACCOUNT_TIMEOUT'u aşan future'lar abandoned sayılır: STATS.fail kaydı +
yerine yeni worker submit edilir. Böylece tüm pool dış bağımlılık (proxy /
captcha) yüzünden donsa bile self-recovery olur.
"""
ex = ThreadPoolExecutor(max_workers=num_threads, thread_name_prefix="acct")
pending = {} # future -> submit_time
idx = 0
try:
for _ in range(num_threads):
idx += 1
pending[ex.submit(worker, idx)] = time.time()
_paced_submit_sleep()
while True:
# En geç PER_ACCOUNT_TIMEOUT'ta uyan — hiçbir future bitmese bile
# liveness kontrolü yapabilelim. Healthy durumda FIRST_COMPLETED ile
# her hesap bitiminde uyanıyoruz zaten.
done, _ = wait(list(pending.keys()), timeout=PER_ACCOUNT_TIMEOUT,
return_when=FIRST_COMPLETED)
now = time.time()
stuck = [f for f, t0 in pending.items() if f not in done and now - t0 >= PER_ACCOUNT_TIMEOUT]
for f in stuck:
logger.warning(f"worker abandoned (>{PER_ACCOUNT_TIMEOUT}s)")
STATS.record_abandoned()
for f in list(done) + stuck:
del pending[f]
idx += 1
pending[ex.submit(worker, idx)] = time.time()
_paced_submit_sleep()
except KeyboardInterrupt:
raise
finally:
ex.shutdown(wait=False, cancel_futures=True)
def _run_batch(num_threads, batch_size, batch_num):
"""One bounded batch of `batch_size` accounts across `num_threads` workers.
Batch deadline = PER_ACCOUNT_TIMEOUT × ceil(batch_size / num_threads).
Deadline aşıldığında biten future'lar normal, bitmemiş olanlar abandoned
(STATS.record_abandoned() çağrılır, batch ilerler — sonsuza dek beklemeyiz).
"""
logger.info(f"batch #{batch_num} · starting {batch_size} accounts × {num_threads} threads")
ex = ThreadPoolExecutor(max_workers=num_threads, thread_name_prefix=f"acct-b{batch_num}")
futures = []
try:
for i in range(batch_size):
futures.append(ex.submit(worker, (batch_num - 1) * batch_size + i + 1))
_paced_submit_sleep() # burst yerine sıralı-aralıklı submit
# Paralel dalga sayısına göre ölçeklendir (batch_size > num_threads ise
# her thread birden çok hesap işliyor).
waves = (batch_size + num_threads - 1) // max(num_threads, 1)
batch_deadline = PER_ACCOUNT_TIMEOUT * waves
try:
for f in as_completed(futures, timeout=batch_deadline):
try:
f.result()
except Exception as e:
logger.debug(f"worker future error: {e}")
except FuturesTimeoutError:
stuck = [f for f in futures if not f.done()]
logger.warning(
f"batch #{batch_num} · {len(stuck)} worker abandoned "
f"(>{batch_deadline}s batch deadline)"
)
for _ in stuck:
STATS.record_abandoned()
except KeyboardInterrupt:
raise
finally:
ex.shutdown(wait=False, cancel_futures=True)
_sigint_count = 0
def _install_sigint_handler():
"""Ctrl+C anında shutdown_event'i set'le, summary bas, process'i zorla
kapat. Non-daemon ThreadPoolExecutor thread'leri normalde KeyboardInterrupt
sonrası bitene kadar process'i tutuyordu — `os._exit(0)` ile bypass.
İkinci Ctrl+C → hemen sert kapatma (summary bile basılmaz).
"""
def _handler(signum, frame):
global _sigint_count
_sigint_count += 1
if _sigint_count == 1:
logger.warning("Ctrl+C alındı — kapanıyor (tekrar bas: hard kill)")
_shutdown_event.set()
try:
_print_final_summary()
except Exception:
pass
os._exit(0)
else:
os._exit(1)
signal.signal(signal.SIGINT, _handler)
def main():
_install_sigint_handler()
num_threads = DEFAULT_THREADS
output_desc = f"{OUTPUT_DIR}/" if OUTPUT_MODE == "local" else "remote-upload"
logger.info(
f"starting backend={ACTIVE_BACKEND} threads={num_threads} "
f"output={OUTPUT_MODE}({output_desc}) proxy={'on' if not NO_PROXY else 'off'}({ACCOUNT_PROXY_TYPE}) "
f"profile={'file' if USE_PROFILE_DATA else 'random'} mail_source={MAIL_SOURCE} "
f"v2={V2_SOLVER} v3={V3_SOLVER} "
f"infinite={'on' if INFINITE_MODE else 'off'}"
)
# proxy2 API'sinden boot zamanı bir defalık fetch (PROXY2_API_KULLAN=True ise);
# başarısız olursa SystemExit ile program durur.
_init_proxy2()
# Konfigüre edilmiş captcha provider'ların boot zamanı bakiye kontrolü
_log_solver_balances()
reporter = threading.Thread(target=_stats_reporter, name="stats-reporter", daemon=True)
reporter.start()
# Rate-watchdog — her RATE_WATCHDOG_INTERVAL'da rate kontrolü, düşükse pause
watchdog = threading.Thread(target=_rate_watchdog, name="rate-watchdog", daemon=True)
watchdog.start()
logger.info(
f"rate-watchdog · interval={RATE_WATCHDOG_INTERVAL}s threshold={RATE_LOW_THRESHOLD}% "
f"pause={RATE_LOW_PAUSE}s min_samples={RATE_MIN_SAMPLES}"
)
try:
if INFINITE_MODE:
batch_num = 0
while True:
# Rate-watchdog pause aktifse yeni batch submit etme — aksi halde
# 50 worker submit edilir, pause'u beklerken batch_deadline aşılır,
# hepsi abandoned olur ve thread'ler leak eder (Python kill edemez).
_wait_if_paused()
if _shutdown_event.is_set():
break
batch_num += 1
logger.info(f"batch #{batch_num} · checking cookie-control")
if not check_cookie_control(COOKIE_CONTROL_URL):
logger.info(
f"batch #{batch_num} · control returned false, sleeping {INFINITE_CHECK_INTERVAL}s"
)
if _shutdown_event.wait(INFINITE_CHECK_INTERVAL):
break
continue
logger.info(f"batch #{batch_num} · control OK, running batch")
_run_batch(num_threads, INFINITE_BATCH_SIZE, batch_num)
s = STATS.snapshot()
logger.info(
f"batch #{batch_num} complete · ok={s['ok']} fail={s['fail']} "
f"success={s['success_rate']}%"
)
if _shutdown_event.wait(INFINITE_CHECK_INTERVAL):
break
else:
_run_pool_continuous(num_threads)
except KeyboardInterrupt:
# SIGINT handler zaten os._exit'le çıktığı için buraya normalde gelmez;
# gelirse de aynı yolu izle.
logger.warning("interrupt received, shutting down...")
finally:
_shutdown_event.set()
_print_final_summary()
# Worker thread'leri non-daemon → Python interpreter onları beklerdi.
# Force exit ile process anında ölür.
os._exit(0)
def _print_final_summary():
"""Detaylı kapanış raporu — ctrl+c sonrası ekran ve dosyaya basılır."""
s = STATS.snapshot()
sep = "=" * 64
lines = [
"",
sep,
" ÖZET RAPOR — spotifyHesapAcici",
sep,
f" Çalışma süresi : {s['elapsed_s']}s ({s['elapsed_s']/60:.1f} dk)",
f" Toplam deneme : {s['total']}",
f" ✓ Verimli (başarılı) : {s['ok']} ({s['success_rate']}%)",
f" ✗ Verimsiz (başarısız): {s['fail']} ({round(100 - s['success_rate'], 1) if s['total'] else 0}%)",
f" Ortalama süre/hesap : {s['avg_duration_s']}s",
f" Hız : {s['rate_per_min']} hesap/dk",
"",
" Captcha:",
f" v3 (create) : {s['cap_v3_ok']} ok / {s['cap_v3_fail']} fail ({s['cap_v3_rate']}%)",
f" v2 (challenge): {s['cap_v2_ok']} ok / {s['cap_v2_fail']} fail ({s['cap_v2_rate']}%)",
f" E-posta challenge (captcha maliyeti yok): {s['email_challenge_ok']} ok / {s['email_challenge_fail']} fail",
f" Challenge dalına düşen: {s['challenge_count']}",
]
if OUTPUT_MODE == "remote":
lines += [
"",
" Uzak Sunucu Upload:",
f" Başarılı: {s['upload_ok']}",
f" Başarısız: {s['upload_fail']}",
]
if s["fail_reasons"]:
lines += ["", " Hata nedenleri (en sık)"]
sorted_reasons = sorted(s["fail_reasons"].items(), key=lambda kv: kv[1], reverse=True)
for reason, count in sorted_reasons[:10]:
lines.append(f" {count:>4}x {reason}")
lines += [sep, ""]
block = "\n".join(lines)
# stdout (Fore.CYAN), file (plain via logger)
print(Fore.CYAN + block + Style.RESET_ALL, flush=True)
for ln in lines:
if ln.strip():
logger.info(ln)
if __name__ == "__main__":
main()