"""TLS tabanlı istekler ve captcha süreci için yardımcı modül.""" from __future__ import annotations import asyncio import copy import inspect import json import logging import os import random import re import secrets import shutil import string import subprocess import threading import time from dataclasses import dataclass from datetime import datetime, timedelta, timezone from email.utils import parsedate_to_datetime from http.cookies import SimpleCookie from pathlib import Path from typing import Any, Callable, Dict, Iterable, List, Literal, Optional, Set, Tuple import requests from anymessage import ( MailCheckResult, get_available_domain_count_with_retry, order_email, reorder_email, wait_for_email, ) from captcha_service import ( CaptchaServiceHTTPError, create_task, get_task, ) from castle_solver import CastleSolverError, request_castle_solves from disposablemail import DisposableMailAPI, DisposableMailAPIError, resolve_mail_service from email_reregistry import EmailReRegistrationManager from email_utils import ( ensure_local_part_has_dot, extract_gmail_dot_positions, is_gmail_domain, is_gmail_domain_alias, normalize_email_key, randomize_gmail_local_part, remove_dots_from_local_part, ) from log_utils import log_debug_event from mail_stats import record_mail_check_result, record_mail_order_result from metrics_tracker import ( record_account_created, record_begin_verification, record_captcha_solved, record_captcha_failure, record_castle_token_solved, record_castle_failure, record_error_398, record_failure, record_signup_denied, record_success, record_mail_success, record_mail_failure, ) from panel_sender import configure_panel_sender, send_account_to_panel from requests import exceptions as requests_exceptions from requests.cookies import RequestsCookieJar, cookiejar_from_dict from requests.utils import dict_from_cookiejar from tid_provider import TransactionIdError, TransactionIdProvider from tls_client_manager import TLSClient, TLSClientConfigError, resolve_emulation_os_from_user_agent from xpff_header import xpff_header_generator CONFIG_PATH = Path(__file__).with_name("config.json") USERNAME_FILE = Path(__file__).with_name("username.txt") ACCOUNT_LOG_FILE = Path(__file__).with_name("accounts.txt") EMAIL_REREG_LOG_FILE = Path( __file__ ).with_name("Email_information_awaiting_re-registration.txt") LOGGER = logging.getLogger(__name__) DEBUG_LOG_PATH = Path(__file__).with_name("log.txt") TLS_PEET_LOG_PATH = Path(__file__).with_name("tls-peet-log.txt") _DEBUG_LOG_LOCK = threading.Lock() _TLS_PEET_LOG_LOCK = threading.Lock() _STEP_SEQ = 0 _STEP_LOCK = threading.Lock() _DISPLAY_NAME_LOCK = threading.Lock() _ACCOUNT_RECORD_BUFFER: Dict[str, Dict[str, str]] = {} _ACCOUNT_RECORD_LOCK = threading.Lock() _EMAIL_REREG_LOG_LOCK = threading.Lock() _EMAIL_REREG_MANAGER_LOCK = threading.Lock() _EMAIL_REREG_MANAGER: Optional[EmailReRegistrationManager] = None _EMAIL_REREG_REUSE_AFTER = timedelta(minutes=100) _EMAIL_REREG_POLL_INTERVAL = 60.0 _REDACTED_PLACEHOLDER = "***" _GLOBAL_SHUTDOWN_EVENT = threading.Event() _GLOBAL_SHUTDOWN_REASON_LOCK = threading.Lock() _GLOBAL_SHUTDOWN_REASON: Optional[str] = None _GLOBAL_RESOURCE_WAIT_EVENT = threading.Event() ANYMESSAGE_STOCK_THRESHOLD = 10 ANYMESSAGE_STOCK_SITE = "x.com" _ANYMESSAGE_RESOURCE_WAIT_INTERVAL = 3.0 _ANYMESSAGE_STOCK_RETRY_ATTEMPTS = 5 _ANYMESSAGE_STOCK_RETRY_DELAY = 1.5 _API_BEARER_TOKEN = ( "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D" "1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" ) _CALLBACK_URL = "https://api.x.com/1.1/onboarding/callback.json" _BEGIN_VERIFICATION_URL = "https://api.x.com/1.1/onboarding/begin_verification.json" _BEGIN_VERIFICATION_SERVICE_DEFAULT = "http://213.239.202.66:6066/begin_verification" _TLS_PEET_TEST_URL = "https://tls.peet.ws/api/all" _PROXY_VERIFICATION_URL = "https://pro.ip-api.com/json/?fields=61439&key=q2SvnxYiNrfhgcW" _CUSTOM_CASTLE_ENDPOINT = "http://37.27.240.144:5000/api/token/createTask" _DISPLAY_NAME_CACHE: Optional[List[str]] = None _PROXY_RANDOM_TOKEN = "random" _PROXY_RANDOM_LENGTH = 10 _PROXY_RANDOM_ALPHABET = string.ascii_letters + string.digits _GMAIL_DOMAIN = "gmail.com" _MAIL_CODE_MIN_LENGTH = 4 _MAIL_CODE_MAX_LENGTH = 8 _MAIL_CODE_REGEX = re.compile(r"(? Optional[str]: digits = re.sub(r"\D", "", text) if _MAIL_CODE_MIN_LENGTH <= len(digits) <= _MAIL_CODE_MAX_LENGTH: return digits return None _MAILAPI_TEMPORARY_CODES = { "WAIT_LINK", "WAIT_MAIL", "WAIT_MESSAGE", "INBOX_EMPTY", "RETRY_LATER", } _MAILAPI_PERMANENT_CODES = { "BAD_TOKEN", "INVALID_ID", "NOT_FOUND", "UNKNOWN_ERROR", } _DEFAULT_MAX_POLL_ATTEMPTS = 300 ANYMESSAGE_POLICY_KEY = "_anymessage_email_source_policy" ANYMESSAGE_POLICY_AUTO = "auto" ANYMESSAGE_POLICY_PURCHASE_ONLY = "purchase_only" ANYMESSAGE_POLICY_REUSE_ONLY = "reuse_only" def request_global_shutdown(reason: str) -> None: """Tüm işçi thread'lerinin kontrollü şekilde sonlanmasını tetikle.""" if isinstance(reason, str): message = reason.strip() else: message = str(reason).strip() if not message: message = "bilinmeyen neden" with _GLOBAL_SHUTDOWN_REASON_LOCK: already_requested = _GLOBAL_SHUTDOWN_EVENT.is_set() if not already_requested: _GLOBAL_SHUTDOWN_REASON = message _GLOBAL_SHUTDOWN_EVENT.set() if not already_requested: LOGGER.error("Global shutdown requested: %s", message) def is_global_shutdown_requested() -> bool: """Global kapatma isteğinin tetiklenip tetiklenmediğini belirt.""" return _GLOBAL_SHUTDOWN_EVENT.is_set() def get_global_shutdown_reason() -> Optional[str]: """Global kapatma isteği mevcutsa nedenini döndür.""" with _GLOBAL_SHUTDOWN_REASON_LOCK: return _GLOBAL_SHUTDOWN_REASON def wait_for_global_shutdown(timeout: float) -> bool: """Belirtilen süre boyunca global kapatma isteğini bekle.""" if timeout <= 0: return _GLOBAL_SHUTDOWN_EVENT.is_set() return _GLOBAL_SHUTDOWN_EVENT.wait(timeout) def is_global_resource_wait_requested() -> bool: """Kaynak bekleme sinyalinin aktif olup olmadığını döndür.""" return _GLOBAL_RESOURCE_WAIT_EVENT.is_set() def request_global_resource_wait(reason: Optional[str] = None) -> None: """Mail kaynağı tükendiğinde küresel bekleme sinyalini tetikle.""" detail = str(reason or "").strip() if detail: LOGGER.warning("Mail source set to wait mode: %s", detail) _GLOBAL_RESOURCE_WAIT_EVENT.set() def clear_global_resource_wait() -> None: """Kaynak bekleme sinyalini temizle.""" _GLOBAL_RESOURCE_WAIT_EVENT.clear() def _generate_proxy_random_value(length: int = _PROXY_RANDOM_LENGTH) -> str: return "".join(secrets.choice(_PROXY_RANDOM_ALPHABET) for _ in range(length)) def _is_castlesolver_balance_error_message(message: str) -> bool: """CastleSolver servisinden gelen hata mesajı bakiyeye işaret ediyor mu kontrol et.""" if not message: return False lowered = message.strip().lower() if not lowered: return False if not any(keyword in lowered for keyword in _CASTLESOLVER_BALANCE_KEYWORDS): return False for token in ("insufficient", "yetersiz", "tükendi", "not enough", "out of", "no"): if token in lowered: return True return False def _normalise_proxy_spec(proxy_spec: Any) -> Optional[str]: if proxy_spec is None: return None if isinstance(proxy_spec, str): value = proxy_spec.strip() else: value = str(proxy_spec).strip() return value or None def _resolve_proxy_placeholders(proxy_spec: str) -> str: if _PROXY_RANDOM_TOKEN not in proxy_spec: return proxy_spec random_part = _generate_proxy_random_value() return proxy_spec.replace(_PROXY_RANDOM_TOKEN, random_part) def _prepare_proxy_spec(proxy_spec: Any) -> Optional[str]: normalised = _normalise_proxy_spec(proxy_spec) if normalised is None: return None return normalised def _build_custom_proxy_payload(proxy_spec: Optional[str]) -> Optional[Dict[str, str]]: spec = _prepare_proxy_spec(proxy_spec) if not spec: return None if "://" in spec: spec = spec.split("://", 1)[1] parts = spec.split(":") if len(parts) < 2: return None address = f"{parts[0].strip()}:{parts[1].strip()}" username = parts[2].strip() if len(parts) >= 3 else "" password = ":".join(parts[3:]).strip() if len(parts) >= 4 else "" payload = {"Address": address} if username: payload["Username"] = username if password: payload["Password"] = password return payload def _random_adjacent_swap(headers: Dict[str, Any]) -> Dict[str, Any]: """Sözlük sırasını komşu elemanlardan biriyle rasgele değiştir.""" items = list(headers.items()) if len(items) <= 1: return dict(items) index = random.randrange(len(items)) if index == 0: swap_with = 1 elif index == len(items) - 1: swap_with = index - 1 else: swap_with = index + random.choice((-1, 1)) items[index], items[swap_with] = items[swap_with], items[index] return dict(items) def _resolve_sec_ch_ua_platform_and_mobile(user_agent: str, default_platform: str = '"Windows"') -> Tuple[str, str]: ua = str(user_agent or '').lower() if 'android' in ua: return '"Android"', '?1' platform = default_platform.strip() if isinstance(default_platform, str) else '' return platform or '"Windows"', '?0' def _extract_email_domain(email_address: str) -> Optional[str]: if not email_address: return None parts = email_address.rsplit("@", 1) if len(parts) != 2: return None domain = parts[1].strip().lower() return domain or None def _is_gmail_address(email_address: str) -> bool: domain = _extract_email_domain(email_address) return is_gmail_domain(domain) def _normalize_gmail_address_case(email_address: str) -> str: if not email_address: return email_address local_part, sep, domain_part = email_address.rpartition("@") if sep != "@": return email_address if not is_gmail_domain(domain_part): return email_address return f"{local_part.lower()}@{domain_part.lower()}" def _randomize_gmail_local_case(email_address: str) -> str: if not email_address: return email_address local_part, sep, domain_part = email_address.rpartition("@") if sep != "@": return email_address domain_lower = domain_part.strip().lower() if not is_gmail_domain(domain_lower): return email_address base_local = local_part.lower() alpha_indexes = [idx for idx, ch in enumerate(base_local) if ch.isalpha()] if not alpha_indexes: return f"{base_local}@{domain_lower}" upper_indexes = { idx for idx in alpha_indexes if random.choice((True, False)) } if not upper_indexes: upper_indexes.add(random.choice(alpha_indexes)) chars = list(base_local) for idx in upper_indexes: chars[idx] = chars[idx].upper() return f"{''.join(chars)}@{domain_lower}" def _parse_duration_value(value: Any) -> Optional[int]: if value is None: return None if isinstance(value, (int, float)): seconds = int(value) elif isinstance(value, str): text = value.strip().lower() if not text: return None match = re.fullmatch(r"(\d+)([smh]?)", text) if not match: return None amount = int(match.group(1)) unit = match.group(2) if unit == "m": seconds = amount * 60 elif unit == "h": seconds = amount * 3600 else: seconds = amount else: return None return max(0, seconds) def load_config(path: Path = CONFIG_PATH) -> Dict[str, Any]: """Verilen yoldaki config dosyasını sözlük olarak döndür.""" if not path.exists(): raise FileNotFoundError(f"Config dosyası bulunamadı: {path}") with path.open("r", encoding="utf-8") as config_file: data = json.load(config_file) if not isinstance(data, dict): raise ValueError("config.json beklenenden farklı veri içeriyor") # Proxy değerindeki "random" yer tutucusu bu aşamada korunur; her thread # kendi kopyasında XGenWorker içinde rastgeleleştirilir. original_proxy = data.get("proxy") prepared_proxy = _prepare_proxy_spec(original_proxy) if prepared_proxy is not None: data["proxy"] = prepared_proxy elif isinstance(original_proxy, str): data["proxy"] = original_proxy.strip() data["show_proxy_in_logs"] = _is_feature_enabled( data.get("show_proxy_in_logs", False) ) data["send_panel"] = _is_feature_enabled(data.get("send_panel")) configure_panel_sender(data["send_panel"]) return data def build_proxy_url(proxy_spec: Optional[str]) -> Optional[str]: """config.json içindeki proxy tanımını standart URL'ye çevir.""" spec = _normalise_proxy_spec(proxy_spec) if not spec: return None if "://" in spec: return spec parts = spec.split(":") if len(parts) < 2: return None host = parts[0].strip() port = parts[1].strip() if not host or not port: return None username = parts[2].strip() if len(parts) >= 3 and parts[2] else "" password = ":".join(parts[3:]).strip() if len(parts) >= 4 else "" auth = "" if username: auth = username if password: auth += f":{password}" auth += "@" return f"http://{auth}{host}:{port}" def _resolve_max_reregistration(config: Dict[str, Any]) -> int: raw_value = config.get("max_re-registration") try: resolved = int(raw_value) except (TypeError, ValueError): return 0 return max(0, resolved) def _get_email_rereg_manager(config: Dict[str, Any]) -> Optional[EmailReRegistrationManager]: max_uses = _resolve_max_reregistration(config) if max_uses <= 0: return None with _EMAIL_REREG_MANAGER_LOCK: global _EMAIL_REREG_MANAGER manager = _EMAIL_REREG_MANAGER if manager is None: manager = EmailReRegistrationManager( EMAIL_REREG_LOG_FILE, file_lock=_EMAIL_REREG_LOG_LOCK, reuse_after=_EMAIL_REREG_REUSE_AFTER, poll_interval=_EMAIL_REREG_POLL_INTERVAL, max_uses=max_uses, ) _EMAIL_REREG_MANAGER = manager else: manager.configure(max_uses=max_uses, reuse_after=_EMAIL_REREG_REUSE_AFTER) return manager def has_available_rereg_email(config: Dict[str, Any]) -> bool: """Email_information_awaiting_re-registration.txt icinde hazir kayit var mi?""" manager = _get_email_rereg_manager(config) if manager is None: return False try: return manager.has_ready_email() except Exception: LOGGER.exception("Error while checking re-registration list.") return False def count_available_rereg_email(config: Dict[str, Any]) -> int: """Hazir re-registration email sayisini dondur.""" manager = _get_email_rereg_manager(config) if manager is None: return 0 try: return manager.count_ready_emails() except Exception: LOGGER.exception("Error while counting re-registration list.") return 0 def build_proxies(proxy_spec: Optional[str]) -> Optional[Dict[str, str]]: """requests/tls-client uyumlu proxy sözlüğü döndür.""" proxy_url = build_proxy_url(proxy_spec) if not proxy_url: return None return {"http": proxy_url, "https": proxy_url} def _is_feature_enabled(value: Any) -> bool: """Bir yapılandırma değerinin 'etkin' olup olmadığını çeşitli formatlara göre belirler.""" if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): normalised = value.strip().lower() return normalised in {"1", "true", "yes", "on"} return False def is_debug_logging_enabled(config: Dict[str, Any]) -> bool: """config.json içindeki debug ayarını bool olarak çöz.""" return _is_feature_enabled(config.get("debug")) def _safe_get(config: Dict[str, Any], key: str, default: Any = "") -> str: value = config.get(key, default) if value is None: return "" return str(value) def _format_for_log(data: Any) -> str: if isinstance(data, dict): try: return json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) except TypeError: return str(data) if isinstance(data, (list, tuple)): try: return json.dumps(list(data), ensure_ascii=False, indent=2) except TypeError: return str(data) return str(data) def _redact_castlebreaker_raw(raw: Any) -> Any: if not isinstance(raw, dict): return raw redacted = dict(raw) data = redacted.get("data") if isinstance(data, dict): data_copy = dict(data) for key in ("tokens", "token", "castle_token"): if key not in data_copy: continue value = data_copy.get(key) if isinstance(value, list): previews = [] for item in value: text = str(item or "") previews.append(f"{text[:8]}...({len(text)})" if text else "") data_copy[key] = previews else: text = str(value or "") data_copy[key] = f"{text[:8]}...({len(text)})" if text else "" redacted["data"] = data_copy return redacted def _summarise_solver_details(details: Optional[Dict[str, Any]]) -> str: if not isinstance(details, dict) or not details: return "" def _shorten(value: str, limit: int = 80) -> str: if len(value) <= limit: return value return value[: limit - 3] + "..." def _format_value(key: str, value: Any) -> Optional[str]: if value is None: return None if isinstance(value, dict): return f"{key}=dict[{len(value)}]" if isinstance(value, (list, tuple, set)): return f"{key}={type(value).__name__}[{len(value)}]" text = str(value).strip() if not text: return None key_lower = key.lower() if "token" in key_lower or "castle" in key_lower: return f"{key}={text[:8]}...({len(text)})" if key_lower in {"useragent", "user_agent", "user-agent"}: return f"{key}={text}" return f"{key}={_shorten(text)}" parts: List[str] = [] for key in sorted(details.keys()): try: item = _format_value(key, details[key]) except Exception: item = None if item: parts.append(item) return "; ".join(parts) _COLOR_OUTPUT_ENABLED: Optional[bool] = None def _is_color_output_enabled() -> bool: """Root logger'daki formatter üzerinden renkli çıktının aktif olup olmadığını belirle.""" global _COLOR_OUTPUT_ENABLED if _COLOR_OUTPUT_ENABLED is not None: return _COLOR_OUTPUT_ENABLED root_logger = logging.getLogger() for handler in root_logger.handlers: formatter = getattr(handler, "formatter", None) colorize = getattr(formatter, "_colorize", None) if isinstance(colorize, bool): _COLOR_OUTPUT_ENABLED = colorize return _COLOR_OUTPUT_ENABLED _COLOR_OUTPUT_ENABLED = False return _COLOR_OUTPUT_ENABLED def _extract_set_cookie_from_headers(headers: Any) -> Optional[Any]: if headers is None: return None try: items = headers.items() except Exception: items = None if items is not None: for key, value in items: if str(key).lower() == "set-cookie": return value getter_all = getattr(headers, "get_all", None) if callable(getter_all): try: values = getter_all("set-cookie") except Exception: values = None if values: if isinstance(values, (list, tuple)) and len(values) == 1: return values[0] return values getter = getattr(headers, "get", None) if callable(getter): for key in ("set-cookie", "Set-Cookie"): try: value = getter(key) except Exception: value = None if value: return value return None def _parse_set_cookie_header(raw_value: Any) -> Dict[str, str]: cookie = SimpleCookie() if raw_value is None: return {} try: if isinstance(raw_value, (list, tuple)): for item in raw_value: if item: cookie.load(str(item)) else: cookie.load(str(raw_value)) except Exception: return {} return {key: morsel.value for key, morsel in cookie.items()} def _extract_cookies_from_cookiejar(cookiejar: Any) -> Optional[Dict[str, Any]]: if cookiejar is None: return None get_dict = getattr(cookiejar, "get_dict", None) if callable(get_dict): try: data = get_dict() except Exception: data = None if data: return data if isinstance(cookiejar, dict): return cookiejar or None try: return {getattr(cookie, "name", index): getattr(cookie, "value", "") for index, cookie in enumerate(cookiejar)} except Exception: return None def _redact_sensitive_fields( payload: Any, fields: Optional[Iterable[str]] ) -> Any: if payload is None or not fields: return payload try: field_set = { str(field).strip().lower() for field in fields if field is not None and str(field).strip() } except Exception: field_set = { str(field).lower() for field in fields if field is not None } if not field_set: return payload items: Optional[Iterable[Any]] if isinstance(payload, dict): items = payload.items() else: getter = getattr(payload, "items", None) if callable(getter): try: items = list(getter()) except Exception: items = None else: items = None if items is None: return payload redacted: Dict[Any, Any] = {} for key, value in items: key_str = str(key) if key_str.lower() in field_set: redacted[key_str] = _REDACTED_PLACEHOLDER else: redacted[key_str] = value return redacted class MailServiceNoEmailsError(RuntimeError): """AnyMessage servisi yeni e-posta sağlayamadığında yükseltilen hata.""" def __init__(self, detail: str) -> None: super().__init__(detail) self.detail = detail class ProxyVerificationError(RuntimeError): """Proxy doğrulama testi başarısız olduğunda yükseltilen hata.""" class XGenWorker: """TLS istemcisi ile captcha görev döngüsünü yöneten işçi.""" def __init__(self, worker_id: int, config: Dict[str, Any]) -> None: self.worker_id = worker_id self.config = dict(config) self.logger = LOGGER.getChild(f"worker-{worker_id}") self._should_abort = False self.debug_enabled = is_debug_logging_enabled(self.config) self._test_tls_peet_enabled = bool(self.config.get("test-tls-peet")) self._manual_email_enabled = ( self.debug_enabled and bool(self.config.get("manuel-email")) ) self._show_proxy_in_logs = _is_feature_enabled( self.config.get("show_proxy_in_logs") ) self._email_source_policy = str( self.config.get(ANYMESSAGE_POLICY_KEY) or ANYMESSAGE_POLICY_AUTO ).strip().lower() if self._email_source_policy not in { ANYMESSAGE_POLICY_AUTO, ANYMESSAGE_POLICY_PURCHASE_ONLY, ANYMESSAGE_POLICY_REUSE_ONLY, }: self._email_source_policy = ANYMESSAGE_POLICY_AUTO self._transaction_provider: Optional[TransactionIdProvider] = None self._last_email_order: Optional[Dict[str, Any]] = None self._last_callback_payload: Optional[_CallbackPayload] = None self._last_js_instrumentation: Optional[Dict[str, Any]] = None self._last_display_name: Optional[str] = None self._last_captcha_result: Optional[Dict[str, Any]] = None self._last_task_response: Optional[Dict[str, Any]] = None self._last_username: Optional[str] = None self._last_password: Optional[str] = None self._last_castle_token: Optional[str] = None self._last_castle_cuid: Optional[str] = None self._last_castle_service: Optional[str] = None self._last_castle_solver_details: Optional[Dict[str, Any]] = None self._last_castle_user_agent: Optional[str] = None self._last_castle_accept_language: Optional[str] = None self._last_castle_task_password_verification: Optional[str] = None self._castlebreaker_header_overrides: Optional[Dict[str, str]] = None self._last_castlebreaker_user_agent: Optional[str] = None self._castle_tls_session: Optional[Any] = None self._castle_tls_user_agent: Optional[str] = None self._last_task_flow_token: Optional[str] = None self._last_csrf_token: Optional[str] = None self._last_auth_token: Optional[str] = None self._last_email_id: Optional[str] = None self._last_email_address: Optional[str] = None self._last_begin_verification_timestamp: Optional[float] = None self._js_instrumentation_attempted = False self._proxy_spec: Optional[str] = None self._proxy_country_code: Optional[str] = None self._manual_email_address: Optional[str] = None self._manual_verification_code: Optional[str] = None self._shutdown_notice_logged = False self._tls_async_loop: Optional[asyncio.AbstractEventLoop] = None proxy_value = self.config.get("proxy") normalised_proxy = _normalise_proxy_spec(proxy_value) if normalised_proxy is not None: if _PROXY_RANDOM_TOKEN in normalised_proxy: resolved_proxy = _resolve_proxy_placeholders(normalised_proxy) if resolved_proxy != normalised_proxy: self.logger.debug( "Proxy 'random' placeholder replaced with a unique value." ) normalised_proxy = resolved_proxy self._proxy_spec = normalised_proxy self.config["proxy"] = normalised_proxy elif isinstance(proxy_value, str): stripped = proxy_value.strip() self.config["proxy"] = stripped if self.debug_enabled: self.logger.info( "Detailed HTTP logs will be written to %s.", DEBUG_LOG_PATH ) if self._manual_email_enabled: self.logger.info( "Manual email mode enabled. Email address and verification code will be read from console." ) self.session: Optional[Any] = None self.poll_interval: float = 1.0 self.max_poll_attempts: int = _DEFAULT_MAX_POLL_ATTEMPTS self.mail_service: str = "" self.mail_api: Optional[DisposableMailAPI] = None self._use_castle_cuid_cookie = _is_feature_enabled( self.config.get("use_castle_cuid_cookie") ) self._initialise_proxy_requirements() if self._should_abort: return self._initialize_worker_components() def _initialize_worker_components(self) -> None: """İşçinin oturum, anket aralığı ve posta servisi gibi ana bileşenlerini başlatır.""" self.session = self._create_session() self.poll_interval = self._resolve_poll_interval() self.max_poll_attempts = self._resolve_max_poll_attempts() self.mail_service = self._determine_mail_service() self.mail_api = self._initialise_mail_api() def _report_terminal_error( self, message: str, *, exc: Optional[BaseException] = None, log_exception: bool = False, emit_console: bool = True, ) -> None: detail = message if exc is not None: detail = f"{message} Detay: {exc}" if log_exception and exc is not None: self.logger.exception(detail) else: self.logger.error(detail) if emit_console: print(f"[Thread {self.worker_id}] {detail}", flush=True) self._should_abort = True def _initialise_proxy_requirements(self) -> None: if not self._proxy_spec: self._report_terminal_error( "Proxy configuration not found; worker thread stopping." ) return try: country_code = self._fetch_proxy_country_code() except ProxyVerificationError as exc: self._report_terminal_error( "Proxy verification test failed; worker thread stopping.", exc=exc, ) return self._proxy_country_code = country_code self.logger.info( "Proxy validation succeeded. Country code: %s", country_code.upper(), ) def _fetch_proxy_country_code(self) -> str: proxies = build_proxies(self._proxy_spec) if not proxies: raise ProxyVerificationError( "Proxy yapılandırmasından geçerli bir proxy URL'si oluşturulamadı." ) self.logger.info("Sending proxy validation request.") try: response = requests.get( _PROXY_VERIFICATION_URL, proxies=proxies, timeout=15, ) response.raise_for_status() except requests_exceptions.ProxyError as exc: raise ProxyVerificationError( "Proxy tünel hatası nedeniyle doğrulama başarısız oldu." ) from exc except requests_exceptions.RequestException as exc: raise ProxyVerificationError( f"Proxy test isteği başarısız oldu: {exc}" ) from exc try: data = response.json() except ValueError as exc: raise ProxyVerificationError( "Proxy test yanıtı geçerli JSON formatında değil." ) from exc status = str(data.get("status") or "").strip().lower() if status != "success": message = str( data.get("message") or data.get("error") or data.get("detail") or data ) raise ProxyVerificationError( f"Proxy test servisi success döndürmedi: {message}" ) country_code = str(data.get("countryCode") or "").strip() if not country_code: raise ProxyVerificationError( "Proxy test yanıtında countryCode alanı bulunamadı." ) query_ip = str(data.get("query") or "").strip() if query_ip: self.logger.info("Proxy IP validated: %s", query_ip) return country_code.lower() def _create_session(self): session = self._build_tls_session( error_context="TLS istemcisi", log_proxy_info=True, ) return session def _build_tls_session( self, *, error_context: str, log_proxy_info: bool, ) -> Optional[Any]: try: overrides: Dict[str, Any] = {} user_agent = _safe_get(self.config, "user_agent") if user_agent: overrides["headers"] = {"User-Agent": user_agent} client = TLSClient( client_overrides=overrides or None, async_mode=True, ) except TLSClientConfigError as exc: self.logger.error("%s could not be loaded: %s", error_context, exc) return None session = client.session if session is None: self.logger.error("Failed to create %s session.", error_context) return None self._configure_tls_session( session, log_proxy_info=log_proxy_info, ) self.logger.debug("%s ready.", error_context) return session def _should_exit_for_shutdown(self, context: str) -> bool: if not is_global_shutdown_requested(): return False if not self._shutdown_notice_logged: reason = get_global_shutdown_reason() or "bilinmeyen neden" self.logger.warning( "Global shutdown requested (%s); ending %s.", reason, context, ) self._shutdown_notice_logged = True return True def _wait_with_shutdown(self, timeout: float, context: str) -> bool: if wait_for_global_shutdown(timeout): self._should_exit_for_shutdown(context) return True return False def _configure_tls_session(self, session: Any, *, log_proxy_info: bool) -> None: log_fn = self.logger.info if log_proxy_info else self.logger.debug proxies = build_proxies(self._proxy_spec) if proxies: try: session.proxies.update(proxies) except AttributeError: session.proxies = dict(proxies) proxy_value = proxies["http"] if not self._show_proxy_in_logs: proxy_value = _mask_proxy_for_log(proxy_value) log_fn("Proxy ayarlandı (host:port): %s", proxy_value) else: log_fn("Proxy kullanılmıyor.") headers: Dict[str, str] = {} user_agent = _safe_get(self.config, "user_agent") if user_agent: headers["user-agent"] = user_agent accept_language = _safe_get(self.config, "accept_language") if accept_language: headers["accept-language"] = accept_language headers.setdefault("accept", "*/*") try: session.headers.update(headers) except AttributeError: session.headers = dict(headers) def _ensure_tls_async_loop(self) -> asyncio.AbstractEventLoop: loop = self._tls_async_loop if loop is not None and not loop.is_closed(): return loop if os.name == "nt" and hasattr(asyncio, "WindowsSelectorEventLoopPolicy"): try: asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) except Exception: pass loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) self._tls_async_loop = loop return loop def _run_tls_coroutine(self, maybe_awaitable: Any) -> Any: if not inspect.isawaitable(maybe_awaitable): return maybe_awaitable loop = self._ensure_tls_async_loop() return loop.run_until_complete(maybe_awaitable) def _close_tls_async_loop(self) -> None: loop = self._tls_async_loop if loop is None: return try: if not loop.is_closed(): try: loop.run_until_complete(loop.shutdown_asyncgens()) except Exception: pass loop.close() finally: self._tls_async_loop = None @staticmethod def _update_session_cookie_values(session: Any, cookies: Dict[str, str]) -> None: if session is None or not cookies: return try: jar = getattr(session, "cookies") except AttributeError: jar = None if jar is None: jar = RequestsCookieJar() try: setattr(session, "cookies", jar) except Exception: return try: jar.update(cookiejar_from_dict(dict(cookies))) return except Exception: pass try: for key, value in dict(cookies).items(): jar.set(str(key), str(value)) except Exception: pass def _copy_session_cookies(self, source: Any, target: Any) -> None: if source is None or target is None: return jar = getattr(source, "cookies", None) if jar is None: return try: cookie_data = dict_from_cookiejar(jar) except Exception: cookie_data = _extract_cookies_from_cookiejar(jar) or {} if cookie_data: self._update_session_cookie_values(target, cookie_data) def _create_tls_session_for_user_agent( self, user_agent: str, *, log_proxy_info: bool, ) -> Optional[Any]: ua_value = str(user_agent or "").strip() overrides: Dict[str, Any] = {} if ua_value: overrides["headers"] = {"User-Agent": ua_value} if "android" in ua_value.lower(): detected_os, _ = resolve_emulation_os_from_user_agent(ua_value) if detected_os is not None: overrides["emulation_os"] = detected_os try: client = TLSClient(client_overrides=overrides or None, async_mode=True) except TLSClientConfigError as exc: self.logger.warning("Castle UA TLS session could not be loaded: %s", exc) return None session = client.session if session is None: self.logger.warning("Failed to create Castle UA TLS session.") return None self._configure_tls_session(session, log_proxy_info=log_proxy_info) if ua_value: try: session.headers.update({"user-agent": ua_value}) except AttributeError: session.headers = {"user-agent": ua_value} return session def _get_castle_tls_session(self, user_agent: str) -> Optional[Any]: ua_value = str(user_agent or "").strip() if not ua_value: return None if ( self._castle_tls_session is not None and self._castle_tls_user_agent == ua_value ): return self._castle_tls_session session = self._create_tls_session_for_user_agent( ua_value, log_proxy_info=False, ) if session is None: return None self._castle_tls_session = session self._castle_tls_user_agent = ua_value return session def _should_use_castle_session( self, request_user_agent: str, base_user_agent: str, ) -> bool: ua_value = str(request_user_agent or "").strip() if not ua_value: return False if "android" in ua_value.lower(): return True if (self._last_castle_service or "").lower() != "castlebreaker.cc": return False base_value = str(base_user_agent or "").strip() return ua_value != base_value def _apply_payload_cookies_to_session(self, payload: "_CallbackPayload") -> None: if payload is None: return try: cookies = dict(payload.cookies) except Exception: cookies = {} self._update_session_cookie_values(self.session, cookies) def _apply_castle_cuid_cookie(self) -> None: if not self._use_castle_cuid_cookie: return cuid = str(self._last_castle_cuid or "").strip() if not cuid: return cookie_keys = ["__cuid"] if (self._last_castle_service or "").lower() == "castlebreaker.cc": cookie_keys.append("cuid") try: if self._last_callback_payload is not None: for key in cookie_keys: self._last_callback_payload.cookies[key] = cuid except Exception: pass cookie_payload = {key: cuid for key in cookie_keys} self._update_session_cookie_values(self.session, cookie_payload) def _consume_castlebreaker_header_overrides(self) -> Dict[str, str]: overrides = self._castlebreaker_header_overrides or {} self._castlebreaker_header_overrides = None return {k: v for k, v in overrides.items() if v} def _resolve_poll_interval(self) -> float: value = self.config.get("captcha_poll_interval", 1) try: interval = float(value) except (TypeError, ValueError): interval = 1.0 return max(0.1, interval) def _resolve_max_poll_attempts(self) -> int: value = self.config.get("captcha_poll_attempts") if value in (None, "", 0): return _DEFAULT_MAX_POLL_ATTEMPTS try: attempts = int(value) except (TypeError, ValueError): return _DEFAULT_MAX_POLL_ATTEMPTS return max(1, attempts) def _determine_mail_service(self) -> str: if self._manual_email_enabled: self.config["mail_service"] = "manual" return "manual" configured = str(self.config.get("mail_service") or "").strip().lower() if configured: return configured try: resolved = resolve_mail_service(self.config) except Exception as exc: self.logger.warning( "Error determining mail service; defaulting to anymessage: %s", exc, ) resolved = "anymessage" self.config["mail_service"] = resolved return resolved def _initialise_mail_api(self) -> Optional[DisposableMailAPI]: if self.mail_service != "disposablemail": return None try: return DisposableMailAPI(self.config) except DisposableMailAPIError as exc: self._report_terminal_error( "Online Disposable Mail client could not be created; worker thread stopping.", exc=exc, ) return None except Exception as exc: # pragma: no cover - beklenmeyen hatalar loglanır self._report_terminal_error( "Unexpected error while creating mail service client; worker thread stopping.", exc=exc, log_exception=True, ) return None def run(self) -> None: if self._should_abort: self.logger.error( "Critical error during startup; worker will not start." ) return if self._should_exit_for_shutdown("işçi başlangıcı"): return self.logger.info("Worker started.") if self.session is None: self.logger.error("No TLS session; ending task.") return try: self._run_captcha_flow() except Exception: # pragma: no cover - beklenmeyen hatalar loglanır self.logger.exception("Worker encountered an unexpected error.") finally: self._close_tls_async_loop() def _run_captcha_flow(self) -> None: if self._should_exit_for_shutdown("captcha akışı"): return # Varsayılan akış: standart code solver API kullan api_key = _safe_get(self.config, "code_solver_api_key") if not api_key: self.logger.error("'code_solver_api_key' not found in config.json.") return user_agent = _safe_get(self.config, "user_agent") # E-posta kaynağı hazır olmadan captcha başlatmayalım try: self._wait_for_email_resources() except Exception as exc: self._report_terminal_error( "Captcha start blocked because email source is not ready.", exc=exc, log_exception=True, ) return self.logger.info("Sending captcha create_task request...") started_at = time.monotonic() try: response = create_task( api_key, proxy_spec=self._proxy_spec, user_agent=user_agent ) except Exception as exc: self._report_terminal_error( "Captcha create_task request failed; worker thread stopping.", exc=exc, log_exception=True, ) return session_id = self._extract_session_id(response) if session_id: self.logger.info("create_task success: session_id=%s", session_id) else: self.logger.error("Could not extract session_id from create_task response.") return self._poll_captcha_task(session_id, api_key, started_at) def _poll_captcha_task( self, session_id: str, api_key: str, started_at: Optional[float] = None ) -> None: attempt = 0 while True: if self._should_exit_for_shutdown("captcha görevi takibi"): return attempt += 1 try: result = get_task(session_id, api_key) except CaptchaServiceHTTPError as exc: error_text = ( f"Captcha service get_task HTTP error ({exc.status_code}): {exc.body}." ) self._report_terminal_error( f"{error_text} Thread is shutting down.", exc=None, ) return except Exception as exc: if self._is_connection_related_error(exc): if not self._handle_get_task_connection_error(exc, attempt): return continue self._report_terminal_error( "Captcha get_task request ended with an unexpected error; worker thread stopping.", exc=exc, log_exception=True, ) return self.logger.debug("get_task response (#%d): %s", attempt, result) log_debug_event( "CaptchaService get_task sonucu", {"worker_id": self.worker_id, "attempt": attempt, "result": result}, ) skip_preview = False preview_text = "" if isinstance(result, dict): task_type = str(result.get("taskType") or "").strip().lower() success = result.get("success") if task_type == "processing" and success is False: skip_preview = True if not skip_preview: try: if isinstance(result, (dict, list)): preview_text = json.dumps( result, ensure_ascii=False, separators=(",", ":") ) else: preview_text = str(result) except Exception: preview_text = str(result) preview_text = preview_text.replace("\n", " ") if _is_color_output_enabled(): prefix = "\033[97m" suffix = "\033[0m" else: prefix = suffix = "" # Ayrı renklerle token ve flow_token özetini yaz. task_token = "" flow_token = "" if isinstance(result, dict): try: task_token = str(result.get("token") or "").strip() except Exception: task_token = "" try: solver_details = result.get("solverDetails") or {} if isinstance(solver_details, dict): flow_token = str( solver_details.get("flow_token") or "" ).strip() except Exception: flow_token = "" if task_token or flow_token: color_enabled = _is_color_output_enabled() label_color = "\033[97m" if color_enabled else "" token_color = "\033[36m" if color_enabled else "" flow_color = "\033[35m" if color_enabled else "" separator_color = "\033[90m" if color_enabled else "" reset_color = "\033[0m" if color_enabled else "" self.logger.info( "%sget_task tokens (#%d):%s %stask%s=%s%s%s %s|%s %sflow%s=%s%s%s", label_color, attempt, reset_color, label_color, reset_color, token_color, task_token or "-", reset_color, separator_color, reset_color, label_color, reset_color, flow_color, flow_token or "-", reset_color, ) elif not skip_preview: self.logger.info( "%sget_task preview (#%d): %s%s", prefix, attempt, preview_text, suffix, ) if self._is_fatal_error(result): self._handle_fatal_captcha_error(result) return if self._is_task_completed(result): self.logger.info("Captcha task completed; sending onboarding request.") duration = None if started_at is not None: duration = max(0.0, time.monotonic() - started_at) record_captcha_solved(duration) self._handle_completed_task(result) break if self.max_poll_attempts and attempt >= self.max_poll_attempts: self.logger.warning( "Reached maximum %d polls; ending task.", self.max_poll_attempts, ) break if self._wait_with_shutdown( self.poll_interval, "captcha görevi bekleme" ): return def _handle_get_task_connection_error(self, exc: Exception, attempt: int) -> bool: """get_task sırasında oluşan bağlantı hatalarını işler ve tekrar deneme gerekip gerekmediğini belirler.""" retry_delay = random.uniform(1.0, 2.0) self.logger.warning( "Captcha get_task request failed due to connection/timeout; retrying in %.1f seconds. error=%s", retry_delay, exc, ) if self.max_poll_attempts and attempt >= self.max_poll_attempts: self.logger.warning( "Reached maximum %d polls after connection errors; ending task.", self.max_poll_attempts, ) return False if self._wait_with_shutdown( retry_delay, "captcha get_task tekrar deneme bekleme" ): return False return True def _handle_fatal_captcha_error(self, result: Any) -> None: """Kaptcha görevinden gelen ölümcül bir hatayı işler ve loglar.""" record_failure() record_captcha_failure() fatal_error = "" message = "" task_type = "" error_code = "" status_code: Optional[Any] = None if isinstance(result, dict): fatal_error = str(result.get("fatal_error") or "").strip() message = str(result.get("message") or "").strip() task_type = str(result.get("taskType") or "").strip() error_code = str(result.get("error") or "").strip() status_code = result.get("status_code") or result.get("statusCode") details: List[str] = [] if fatal_error: details.append(f"fatal_error={fatal_error}") if error_code and error_code != fatal_error: details.append(f"error={error_code}") if task_type: details.append(f"taskType={task_type}") if status_code: details.append(f"status_code={status_code}") if message: details.append(f"message={message}") detail_text = "; ".join(details) if details else _format_for_log(result) if task_type.lower() == "failed": self.logger.error( "Captcha task failed: %s", message or detail_text or "unknown error", ) else: self.logger.error( "Captcha task ended with a fatal error: %s", detail_text, ) self._should_abort = True def _handle_completed_task(self, result: Any) -> None: if self.session is None: self.logger.error("TLS session not available; onboarding request not sent.") return try: payload = self._extract_callback_payload(result) except ValueError as exc: self.logger.error("Could not convert captcha solution into onboarding request: %s", exc) return if isinstance(result, dict): self._last_captcha_result = dict(result) else: self._last_captcha_result = None self._last_callback_payload = payload if self._should_exit_for_shutdown("onboarding akışı"): return self._apply_payload_cookies_to_session(payload) self._last_task_flow_token = None self._last_castle_token = None self._last_castle_cuid = None self._last_castle_service = None self._last_castle_solver_details = None self._last_castle_user_agent = None self._last_castle_accept_language = None self._last_castle_task_password_verification = None self._castlebreaker_header_overrides = None self._last_castlebreaker_user_agent = None self._castle_tls_session = None self._castle_tls_user_agent = None self._last_castlebreaker_user_agent = None self._castle_tls_session = None self._castle_tls_user_agent = None self._js_instrumentation_attempted = False self._last_js_instrumentation = self._extract_js_instrumentation(result) if self._last_js_instrumentation: self.logger.debug("js_instrumentation payload captured.") else: self.logger.debug("js_instrumentation data not found.") if isinstance(self._last_email_order, dict): self._last_email_order.update( { "flow_token": payload.flow_token, "guest_token": payload.guest_token, "sec_ch_ua": payload.sec_ch_ua, "languages": payload.languages, "user_agent": payload.user_agent, "cookies": dict(payload.cookies), } ) if self._last_js_instrumentation: self._last_email_order["js_instrumentation"] = self._last_js_instrumentation self.logger.debug( "Preparing onboarding request: user_agent=%s, languages=%s, guest_id=%s, cookie_keys=%s", payload.user_agent, payload.languages, payload.encoded_guest_id, list(payload.cookies.keys()), ) try: callback_response = self._send_onboarding_callback( payload, step_no=1 ) except Exception: self.logger.exception("Onboarding callback request failed.") return status_code = getattr(callback_response, "status_code", "?") preview = self._summarise_response_text(callback_response) self.logger.info("Onboarding callback response: %s | %s", status_code, preview) self._last_begin_verification_timestamp = time.time() if isinstance(self._last_email_order, dict): self._last_email_order[ "begin_verification_timestamp" ] = self._last_begin_verification_timestamp try: begin_response = self._send_begin_verification(payload, step_no=2) except MailServiceNoEmailsError as exc: self.logger.warning( "AnyMessage service could not supply a new email; worker thread stopping. Detail: %s", exc, ) request_global_resource_wait( "AnyMessage stoğu veya yeniden kayıt listesi boş görünüyor." ) self._report_terminal_error( "AnyMessage service could not provide a new email; worker thread stopping.", exc=exc, ) return except Exception as exc: self._report_terminal_error( "Begin verification request failed due to email service; worker thread stopping.", exc=exc, log_exception=True, ) return # _send_begin_verification already logs and handles the response; nothing further here. def _run_mail_checker(self) -> None: if self._should_exit_for_shutdown("mail_checker"): return order = self._last_email_order if not isinstance(order, dict) or not order: self.logger.warning( "mail_checker.py could not run: email order info missing." ) return service_name, email_address = self._resolve_mail_checker_context(order) result: MailCheckResult preview_text = "" email_id = "" if self._is_manual_email_mode(): if not email_address: self.logger.warning( "Manual email mode enabled but email address not found." ) return self.logger.info( "Requesting manual email verification code (email=%s).", email_address, ) code = self._prompt_manual_verification_code(email_address) if not code: self.logger.warning( "Manual verification code not obtained; task.json request will not be sent." ) return self._manual_verification_code = code result = MailCheckResult(True, 0, "", None, 1, None, code) else: email_id = str(order.get("email_id") or "").strip() if not email_id: self.logger.warning( "mail_checker.py could not run: email order missing email_id (email=%s).", email_address or "?", ) return previous_code = str(order.get("verification_code") or "").strip() or None if previous_code and service_name != "disposablemail": self.logger.debug( "mail_checker.py will ignore previous verification code: %s", previous_code, ) def _log_mail_checker(message: str) -> None: self.logger.info(message) api_key: Optional[str] = None if service_name == "disposablemail": if self.mail_api is None: try: self.mail_api = DisposableMailAPI(self.config) except DisposableMailAPIError as exc: self.logger.warning( "mail_checker.py could not run: DisposableMail client could not be created (%s).", exc, ) return except Exception: self.logger.exception( "mail_checker.py could not run: DisposableMail client creation failed." ) return else: api_key = str(order.get("api_key") or "").strip() if not api_key: api_key = _safe_get(self.config, "AnyMessage").strip() if not api_key: message = ( "mail_checker.py could not run: 'AnyMessage' API key missing." ) request_global_shutdown(message) self._report_terminal_error(message) return self.logger.info( "Starting mail_checker.py (%s): email_id=%s email=%s", service_name, email_id, email_address or "?", ) duplicate_retry_done = False try: if service_name == "disposablemail": result = self._wait_for_disposablemail_email( email_id, log_fn=_log_mail_checker, email_address=email_address, not_before=self._last_begin_verification_timestamp, ) else: result = wait_for_email( api_key or "", email_id, log_fn=_log_mail_checker, reject_code=previous_code, ) if ( result.error == "duplicate_code" and not duplicate_retry_done and api_key ): duplicate_retry_done = True prev_code_text = result.code or previous_code or "" self.logger.info( "duplicate_code detected; triggering reorder and re-wait (attempts=%d).", result.attempts, ) try: reorder_email( api_key, email_id, log_fn=_log_mail_checker, ) except Exception: self.logger.debug( "reorder_email threw an exception; continuing to wait_for_email.", exc_info=True, ) result = wait_for_email( api_key or "", email_id, log_fn=_log_mail_checker, reject_code=prev_code_text, ) except Exception: self.logger.exception( "Unexpected error while running mail_checker.py." ) record_mail_failure() return preview_text = self._build_mail_checker_preview(result) if result.error in { "email_banned", "disposablemail_invalid", "no_activation", "no_activation_limit", "json_decode_error", }: log_message = ( f"Mail service returned a fatal error ('{result.error}'); worker thread stopping." ) self._handle_mail_check_failure( result, log_message, email_address=email_address, email_id=email_id, ) return order_info = self._build_order_info_from_mail_result(order, result) self._last_email_order = order_info if result.ok: record_mail_success() code_display = result.code or "?" self.logger.info( "mail_checker.py completed successfully: status_code=%s attempts=%d code=%s", result.status_code, result.attempts, code_display, ) if preview_text: self.logger.debug("mail_checker.py response: %s", preview_text) if result.data: self.logger.debug( "mail_checker.py data: %s", _format_for_log(result.data), ) try: task_response = self._submit_onboarding_task( result.code or "", step_no=3, ) except Exception: self.logger.exception( "Unexpected error while sending task.json request." ) else: if task_response is not None: status = getattr(task_response, "status_code", "?") preview = self._summarise_response_text(task_response) task_text_raw = getattr(task_response, "text", "") task_text = ( task_text_raw if isinstance(task_text_raw, str) else str(task_text_raw) ) task_text_lower = task_text.lower() if self._response_has_error_code(task_response, "398"): error_message = ( self._extract_error_message_from_response( task_response, "398" ) or preview or "task.json returned error code 398" ) self._report_terminal_error( f"task.json returned error 398 (human verification failed); terminating thread. Detail: {error_message}", emit_console=False, ) record_error_398() record_failure() return has_enter_password = "enterpassword" in task_text_lower log_method = ( self.logger.warning if ( self._is_error_status_code(status) or self._response_contains_error_code(task_response) ) else self.logger.info ) if has_enter_password and log_method == self.logger.info: self.logger.info("Next Step EnterPassword") else: log_method("task.json response: %s | %s", status, preview) order_info["task_json_response"] = { "status_code": status, "preview": preview, } if ( task_text and "account/begin_password_reset" in task_text_lower ): email_for_log = ( str( order_info.get("email") or email_address or self._last_email_address or "" ).strip() or "?" ) email_id_for_log = ( str( order_info.get("email_id") or self._last_email_id or "" ).strip() ) _remove_email_from_retry_logs( email_id_for_log, email_for_log ) self._report_terminal_error( f"task.json response indicates email is already registered (password reset link). email={email_for_log}" ) return order_info["enter_password_response"] = None try: password_response = self._submit_password_strength_check( self._last_callback_payload, order_info, step_no=4, ) except Exception: self.logger.exception( "Unexpected error while sending password_strength request." ) order_info["password_strength_response"] = None else: if password_response is not None: status = getattr(password_response, "status_code", "?") preview = self._summarise_response_text(password_response) self.logger.info( "password_strength response: %s | %s", status, preview, ) order_info["password_strength_response"] = { "status_code": status, "preview": preview, } if self._is_password_strength_successful( password_response ): try: enter_password_response = ( self._submit_enter_password_subtask( order_info, step_no=5, ) ) except Exception: self.logger.exception( "Unexpected error while sending EnterPassword request." ) else: if enter_password_response is not None: status = getattr( enter_password_response, "status_code", "?" ) preview = self._summarise_response_text( enter_password_response ) color_enabled = _is_color_output_enabled() reset_color = "\033[0m" if color_enabled else "" success_color = ( "\033[32m" if color_enabled else "" ) username_color = ( "\033[35m" if color_enabled else "" ) flow_token_value = "" try: payload = enter_password_response.json() except Exception: payload = None if isinstance(payload, dict): flow_token_value = str( payload.get("flow_token") or payload.get("flowToken") or "" ).strip() username_for_log = _normalise_username( order_info.get("username") or self._last_username or "" ) auth_token_for_response = ( self._extract_cookie_from_response( enter_password_response, "auth_token" ) or "" ) success_flow = ( self._status_equals(status, 200) and bool(auth_token_for_response) ) if success_flow: record_account_created() record_success() if username_for_log: self.logger.info( "%sCreate Account successfully%s | Username: %s%s%s", success_color, reset_color, username_color, username_for_log, reset_color, ) else: self.logger.info( "%sCreate Account successfully%s", success_color, reset_color, ) else: self.logger.info( "EnterPassword response: %s | %s", status, preview, ) order_info["enter_password_response"] = { "status_code": status, "preview": preview, } if ( self._status_equals(status, 400) and self._response_has_error_code( enter_password_response, "366" ) ): error_message = ( self._extract_error_message_from_response( enter_password_response, "366", ) or preview ) email_for_log = ( str( order_info.get("email") or email_address or self._last_email_address or "" ).strip() ) if not email_for_log: email_for_log = "?" self.logger.error( "EnterPassword request returned error 366: email=%s detail=%s", email_for_log, error_message, ) cleanup_email_id = str( order_info.get("email_id") or self._last_email_id or "" ).strip() cleanup_email_address = ( str(order_info.get("email") or "") .strip() ) if not cleanup_email_address: cleanup_email_address = str( self._last_email_address or email_address or "" ).strip() if ( cleanup_email_id or cleanup_email_address ): try: _remove_email_from_retry_logs( cleanup_email_id, cleanup_email_address, ) except Exception: self.logger.exception( "Error while clearing email retry records." ) return else: order_info["enter_password_response"] = None else: order_info["password_strength_response"] = None else: order_info["task_json_response"] = None order_info["password_strength_response"] = None order_info["enter_password_response"] = None else: self._handle_mail_checker_failure( result=result, preview_text=preview_text, service_name=service_name, email_id=email_id, email_address=email_address, ) def _resolve_mail_checker_context(self, order: Dict[str, Any]) -> Tuple[str, str]: service_name = str(order.get("service") or self.mail_service or "").strip().lower() if not service_name: service_name = "anymessage" email_address = str(order.get("email") or "").strip() if not email_address and self._manual_email_address: email_address = self._manual_email_address return service_name, email_address @staticmethod def _build_mail_checker_preview(result: MailCheckResult) -> str: preview_text = "" if result.text: preview_text = str(result.text).replace("\n", "\\n") if len(preview_text) > 500: preview_text = preview_text[:497] + "..." return preview_text @staticmethod def _build_order_info_from_mail_result( order: Dict[str, Any], result: MailCheckResult ) -> Dict[str, Any]: order_info = dict(order) order_info["mail_checker_result"] = { "ok": result.ok, "status_code": result.status_code, "attempts": result.attempts, "error": result.error, "code": result.code, } if result.code: order_info["verification_code"] = result.code return order_info def _handle_mail_checker_failure( self, *, result: MailCheckResult, preview_text: str, service_name: str, email_id: str, email_address: str, ) -> None: self.logger.error( "mail_checker.py failed: status_code=%s attempts=%d error=%s", result.status_code, result.attempts, result.error or "?", ) record_mail_failure() if preview_text: self.logger.debug("mail_checker.py response: %s", preview_text) if result.data: self.logger.debug( "mail_checker.py data: %s", _format_for_log(result.data), ) if service_name == "anymessage" and result.error == "no_activation_limit": try: _remove_email_from_retry_logs(email_id, email_address) except Exception: self.logger.exception( "Error clearing retry records due to no activation limit." ) def _resolve_email_service_api_key(self, service_name: str) -> str: service_text = str(service_name or "").strip() if not service_text: return "" lowered = service_text.lower() if lowered == "disposablemail": api_key: str = "" if self.mail_api is not None: api_key_value = getattr(self.mail_api, "api_key", "") if isinstance(api_key_value, str): api_key = api_key_value.strip() if api_key: return api_key return _safe_get(self.config, "disposablemail_apikey").strip() if lowered == "anymessage": return _safe_get(self.config, "AnyMessage").strip() candidate = _safe_get(self.config, service_text).strip() if candidate: return candidate if service_text != lowered: return _safe_get(self.config, lowered).strip() return "" def _record_email_for_reregistration(self) -> None: email_id = (self._last_email_id or "").strip() email_address = (self._last_email_address or "").strip() if not email_id and not email_address: return if isinstance(self._last_email_order, dict) and self._last_email_order.get("reused"): # Yeniden kullanılabilir e-postalar rezervasyon sırasında güncellendi. return if isinstance(self._last_email_order, dict): base_email = str(self._last_email_order.get("email_base") or "").strip() if base_email: email_address = base_email service_name = "" if isinstance(self._last_email_order, dict): service_name = str(self._last_email_order.get("service") or "").strip() if not service_name: service_name = str(self.mail_service or "").strip() api_key = self._resolve_email_service_api_key(service_name) verification_code: Optional[str] = None if isinstance(self._last_email_order, dict): raw_code = self._last_email_order.get("verification_code") if raw_code is not None: verification_code = str(raw_code).strip() or None try: _persist_email_reregistration_info( email_id=email_id, email_address=email_address, service=service_name, api_key=api_key, verification_code=verification_code, max_uses=_resolve_max_reregistration(self.config), ) except Exception: self.logger.exception( "Unexpected error while writing re-registration email info." ) def _append_tls_peet_log(self, message: str) -> None: timestamp = datetime.utcnow().isoformat(timespec="seconds") + "Z" entry = f"[UTC {timestamp}] [WORKER {self.worker_id}] {message}\n" try: with _TLS_PEET_LOG_LOCK: TLS_PEET_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with TLS_PEET_LOG_PATH.open("a", encoding="utf-8") as log_file: log_file.write(entry) except Exception: self.logger.debug("Failed to write TLS Peet log.", exc_info=True) def _run_tls_peet_probe( self, *, context: str, request_headers: Optional[Dict[str, Any]], step_no: Optional[int], ) -> None: if not self._test_tls_peet_enabled: return base_session = self.session if base_session is None: self._append_tls_peet_log( f"{context} [step={step_no if step_no is not None else '--'}] TLS oturumu mevcut değil" ) return try: headers = dict(request_headers or {}) except Exception: headers = {} try: headers = {str(key): value for key, value in headers.items()} except Exception: pass headers["host"] = "tls.peet.ws" headers["origin"] = "https://tls.peet.ws" headers["referer"] = "https://tls.peet.ws/" headers.setdefault("accept", "*/*") headers.pop("content-length", None) headers.pop("Content-Length", None) status_description = "?" preview = "" error_message: Optional[str] = None try: response = self._run_tls_coroutine( base_session.get(_TLS_PEET_TEST_URL, headers=headers) ) status_description = str(getattr(response, "status_code", "?")) preview = self._summarise_response_text(response) self.logger.info( "TLS Peet test completed: context=%s status=%s", context, status_description, ) except Exception as exc: error_message = f"{exc.__class__.__name__}: {exc}" self.logger.warning( "TLS Peet test failed: context=%s error=%s", context, error_message, ) step_repr = f"{step_no:02d}" if step_no is not None else "--" if error_message: log_line = f"{context} [step={step_repr}] error={error_message}" else: log_line = f"{context} [step={step_repr}] status={status_description} summary={preview}" self._append_tls_peet_log(log_line) def _log_http_interaction( self, *, method: str, url: str, request_headers: Dict[str, Any], request_body: Any, response: Any = None, error: Optional[BaseException] = None, force: bool = False, redact_fields: Optional[Iterable[str]] = None, step_no: Optional[int] = None, phase: Literal["SEND", "RECV", "ERROR"] = "RECV", ) -> None: method_upper = method.upper() if ( self._test_tls_peet_enabled and phase == "RECV" and not str(url).startswith(_TLS_PEET_TEST_URL) ): try: self._run_tls_peet_probe( context=f"{method_upper} {url}", request_headers=request_headers, step_no=step_no, ) except Exception: self.logger.debug( "Unexpected error while triggering TLS Peet test.", exc_info=True, ) if not self.debug_enabled: return global _STEP_SEQ if step_no is None: with _STEP_LOCK: _STEP_SEQ += 1 timestamp = datetime.utcnow().isoformat(timespec="seconds") + "Z" monotonic_ns = time.monotonic_ns() step_str = f"{step_no:02d}/06" if step_no is not None else "--/--" lines = [ ( f"[UTC {timestamp}] [MONO {monotonic_ns}] [WORKER {self.worker_id}] " f"[STEP {step_str}] [PHASE {phase}] HTTP {method_upper} {url}" ) ] include_request_details = phase in {"SEND", "ERROR"} if include_request_details and request_headers: lines.append("--> Headers:") lines.append(_format_for_log(dict(request_headers))) if include_request_details: if redact_fields: request_body_for_log = _redact_sensitive_fields( request_body, redact_fields ) else: request_body_for_log = request_body if request_body_for_log is not None: lines.append("--> Body:") lines.append(_format_for_log(request_body_for_log)) if response is not None: status = getattr(response, "status_code", "?") reason = getattr(response, "reason", None) status_line = f"<-- Status: {status}" if reason: status_line += f" {reason}" lines.append(status_line) response_headers = getattr(response, "headers", None) if response_headers: try: header_mapping = dict(response_headers) except Exception: header_mapping = response_headers lines.append("<-- Response Headers:") lines.append(_format_for_log(header_mapping)) set_cookie = _extract_set_cookie_from_headers(response_headers) else: set_cookie = None if not set_cookie: set_cookie = _extract_cookies_from_cookiejar( getattr(response, "cookies", None) ) if set_cookie: lines.append("<-- Set-Cookie:") lines.append(_format_for_log(set_cookie)) response_text = getattr(response, "text", None) if response_text is None: response_content = getattr(response, "content", b"") if isinstance(response_content, bytes): response_text = response_content.decode("utf-8", "replace") else: response_text = str(response_content) if response_text: lines.append("<-- Response Body:") lines.append(str(response_text)) if error is not None: lines.append("<-- Error:") lines.append(f"{error.__class__.__name__}: {error}") lines.append("") entry = "\n".join(lines) try: with _DEBUG_LOG_LOCK: DEBUG_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with DEBUG_LOG_PATH.open("a", encoding="utf-8") as log_file: log_file.write(entry) except Exception: self.logger.debug("Error while writing detailed log.", exc_info=True) @staticmethod def _is_connection_related_error(error: BaseException) -> bool: if isinstance(error, requests_exceptions.RequestException): connection_types: Tuple[type, ...] = ( requests_exceptions.ConnectionError, requests_exceptions.Timeout, requests_exceptions.ProxyError, requests_exceptions.SSLError, ) return isinstance(error, connection_types) name = error.__class__.__name__ return name.endswith("ConnectionError") or name.endswith("Timeout") def _handle_request_exception( self, *, method: str, url: str, error: BaseException, step_no: Optional[int] = None, ) -> None: if not self._is_connection_related_error(error): return method_upper = method.upper() message = ( f"HTTP {method_upper} {url} request failed due to connection error; worker thread stopping." ) if step_no is not None: message += f" (step {step_no})" self._report_terminal_error(message, exc=error, log_exception=True) def _get_transaction_provider(self) -> TransactionIdProvider: if self.session is None: raise RuntimeError("TLS oturumu hazır değil") user_agent = _safe_get(self.config, "user_agent").strip() accept_language = _safe_get(self.config, "accept_language").strip() if ( self._transaction_provider is None or not self._transaction_provider.matches(user_agent, accept_language) ): self._transaction_provider = TransactionIdProvider( XGenRequestContext(self.session), user_agent=user_agent, accept_language=accept_language, ) return self._transaction_provider def _generate_transaction_id_or_raise(self, method: str, endpoint: str) -> str: try: return self._get_transaction_provider().generate(method, endpoint) except TransactionIdError as exc: self._transaction_provider = None raise RuntimeError(f"Transaction ID could not be generated: {exc}") from exc except Exception as exc: self._transaction_provider = None raise RuntimeError( f"Transaction ID generation failed unexpectedly: {exc}" ) from exc def _generate_transaction_id_or_log( self, method: str, endpoint: str, *, context: str ) -> Optional[str]: try: return self._get_transaction_provider().generate(method, endpoint) except TransactionIdError as exc: self._transaction_provider = None self.logger.error( "Could not generate Transaction ID for %s: %s", context, exc ) return None except Exception as exc: self._transaction_provider = None self.logger.error( "Transaction ID generation for %s failed unexpectedly: %s", context, exc, ) return None def _send_with_http_logging( self, *, method: str, url: str, request_headers: Dict[str, Any], request_body: Any, send_fn: Callable[[], Any], step_no: Optional[int] = None, redact_fields: Optional[Iterable[str]] = None, ) -> Any: self._log_http_interaction( method=method, url=url, request_headers=request_headers, request_body=request_body, force=True, redact_fields=redact_fields, step_no=step_no, phase="SEND", ) try: response = send_fn() except Exception as exc: self._log_http_interaction( method=method, url=url, request_headers=request_headers, request_body=request_body, error=exc, force=True, redact_fields=redact_fields, step_no=step_no, phase="ERROR", ) self._handle_request_exception( method=method, url=url, error=exc, step_no=step_no, ) raise self._log_http_interaction( method=method, url=url, request_headers=request_headers, request_body=request_body, response=response, force=True, redact_fields=redact_fields, step_no=step_no, phase="RECV", ) return response def _extract_callback_payload(self, result: Any) -> _CallbackPayload: if not isinstance(result, dict): raise ValueError("Captcha sonucu sözlük formatında değil") flow_token_raw = _search_value(result, "flow_token", "flowtoken") flow_token = str(flow_token_raw or "").strip() if not flow_token: raise ValueError("flow_token değeri bulunamadı") sec_ch_ua_raw = _search_value(result, "sec-ch-ua") sec_ch_ua = _normalise_sec_ch_ua(sec_ch_ua_raw) if not sec_ch_ua: raise ValueError("sec-ch-ua değeri bulunamadı") languages_raw = _search_value(result, "languages", "accept-language", "accept_language") languages = str(languages_raw or "").strip() if not languages: languages = _safe_get(self.config, "accept_language").strip() if not languages: languages = "en-US,en" user_agent_raw = _search_value(result, "useragent", "user_agent", "user-agent") user_agent = str(user_agent_raw or "").strip() if not user_agent: user_agent = _safe_get(self.config, "user_agent").strip() if not user_agent: raise ValueError("user_agent değeri bulunamadı") cookie_raw = _search_value(result, "cookie", "cookies") if cookie_raw is None: raise ValueError("cookie verisi bulunamadı") cookies = _normalise_cookie_map(cookie_raw) guest_token_raw = _search_value(result, "gt", "guest_token", "x-guest-token") if guest_token_raw is None: guest_token_raw = _search_value(cookies, "gt", "guest_token") guest_token = str(guest_token_raw or "").strip() if not guest_token: raise ValueError("guest token (gt) değeri bulunamadı") guest_id_raw = _search_value(result, "guest_id") if guest_id_raw is None: guest_id_raw = _search_value(cookies, "guest_id") guest_id = str(guest_id_raw or "").strip() if not guest_id: raise ValueError("guest_id değeri bulunamadı") return _CallbackPayload( flow_token=flow_token, sec_ch_ua=sec_ch_ua, languages=languages, user_agent=user_agent, cookies=cookies, guest_token=guest_token, guest_id=guest_id, ) def _build_default_headers( self, payload: _CallbackPayload, transaction_id: str, xpff_token: str, *, effective_user_agent: Optional[str] = None, accept_language: Optional[str] = None, extra_headers: Optional[Dict[str, str]] = None, ) -> Dict[str, str]: user_agent = effective_user_agent or payload.user_agent language = accept_language or payload.languages default_platform = self.config.get("sec_ch_ua_platform", '"Windows"') sec_ch_ua_platform, sec_ch_ua_mobile = _resolve_sec_ch_ua_platform_and_mobile( user_agent, default_platform ) headers = { "authorization": _API_BEARER_TOKEN, "content-type": "application/json", "user-agent": user_agent, "x-twitter-client-language": "en", "x-twitter-active-user": "yes", "x-client-transaction-id": transaction_id, "x-guest-token": payload.guest_token, # "x-xp-forwarded-for": xpff_token, "origin": "https://x.com", "referer": "https://x.com/", "sec-ch-ua": payload.sec_ch_ua, "sec-ch-ua-platform": sec_ch_ua_platform, "sec-ch-ua-mobile": sec_ch_ua_mobile, "accept": "*/*", "accept-language": language, "accept-encoding": "gzip, deflate, br, zstd", "sec-fetch-site": "same-site", "sec-fetch-mode": "cors", "sec-fetch-dest": "empty", "priority": "u=1, i", } if extra_headers: headers.update(extra_headers) return headers def _post_metric_tracking_callback( self, payload: _CallbackPayload, flow_token: str, subtask_id: str, *, timestamp_ms: Optional[int] = None, extra_headers: Optional[Dict[str, str]] = None, step_no: Optional[int] = None, ): base_session = self.session if base_session is None: raise RuntimeError("TLS oturumu mevcut değil") request_session = base_session resolved_step_no = step_no if resolved_step_no is None: if subtask_id == "EmailVerification": resolved_step_no = 1 elif subtask_id == "OpenAccount": resolved_step_no = 6 transaction_id = self._generate_transaction_id_or_raise( "POST", "/1.1/onboarding/callback.json" ) xpff_token, xpff_timestamp = xpff_header_generator.generate( payload.user_agent, payload.encoded_guest_id ) if timestamp_ms is None: timestamp_ms = xpff_timestamp headers = self._build_default_headers( payload, transaction_id, xpff_token, extra_headers=extra_headers, ) body = { "identifier": "generic-metric-tracking-cb", "params": json.dumps( { "data": [ { "dataset": "signup_events", "signup_event_data": { "subtask_id": subtask_id, "flow_token": flow_token, "flow_name": "signup", "flow_start_location": "splash_screen", "signup_flow_name": "signup", }, } ] } ), "product": "metric-tracking", "timestampMs": timestamp_ms, } url = _CALLBACK_URL method = "POST" request_headers = dict(headers) request_body = body response = self._send_with_http_logging( method=method, url=url, request_headers=request_headers, request_body=request_body, step_no=resolved_step_no, send_fn=lambda: self._run_tls_coroutine( request_session.post( url=url, headers=headers, json=body, ) ), ) return response def _send_onboarding_callback( self, payload: _CallbackPayload, *, step_no: Optional[int] = None ): return self._post_metric_tracking_callback( payload=payload, flow_token=payload.flow_token, subtask_id="EmailVerification", step_no=step_no, ) def _send_begin_verification( self, payload: _CallbackPayload, *, step_no: Optional[int] = None ): base_session = self.session if base_session is None: raise RuntimeError("TLS oturumu mevcut değil") # E-posta/flow bilgisi hazirlanir email_address = self._obtain_email_address() if _is_gmail_address(email_address): email_address = _normalize_gmail_address_case(email_address) dot_history: List[List[int]] = [] if _is_gmail_address(email_address): existing_history = None if isinstance(self._last_email_order, dict): existing_history = self._last_email_order.get("used_dot_positions") data = self._last_email_order.get("data") if isinstance(data, dict) and data.get("used_dot_positions") is not None: existing_history = data.get("used_dot_positions") dot_history = _merge_used_dot_positions( existing_history, extract_gmail_dot_positions(email_address) ) def _update_retry_email_address() -> None: nonlocal email_address, dot_history if not _is_gmail_address(email_address): return updated_email, updated_history = randomize_gmail_local_part( email_address, used_dot_positions=dot_history, advance_history=True, ) updated_email = ensure_local_part_has_dot(updated_email) updated_email = _normalize_gmail_address_case(updated_email) email_address = updated_email dot_history = updated_history self._last_email_address = updated_email if isinstance(self._last_email_order, dict): self._last_email_order["email"] = updated_email if dot_history: self._last_email_order["used_dot_positions"] = dot_history elif "used_dot_positions" in self._last_email_order: self._last_email_order.pop("used_dot_positions", None) display_name = self._choose_display_name() self._last_display_name = display_name if isinstance(self._last_email_order, dict): self._last_email_order["display_name"] = display_name email_for_request = _randomize_gmail_local_case(email_address) if isinstance(self._last_email_order, dict): self._last_email_order["email_base"] = email_address self._last_email_order["email"] = email_for_request self._last_email_address = email_for_request self.logger.debug( "Begin verification body prepared: display_name=%s, email=%s", display_name, email_for_request, ) # Begin Verification via external service (single attempt, no retries) service_url = str( self.config.get("begin_verification_service_url") or _BEGIN_VERIFICATION_SERVICE_DEFAULT ).strip() proxy_url = build_proxy_url(self._proxy_spec) if not proxy_url: raise RuntimeError( "Proxy yapılandırmasından geçerli bir proxy URL'si oluşturulamadı." ) payload_body = { "proxy": proxy_url, "display_name": display_name, "email_address": email_for_request, "flow_token": payload.flow_token, "user_agent": payload.user_agent, } self._last_begin_verification_timestamp = time.time() if isinstance(self._last_email_order, dict): self._last_email_order["begin_verification_timestamp"] = ( self._last_begin_verification_timestamp ) self._last_email_order["email"] = email_for_request self._last_email_order["display_name"] = display_name record_begin_verification() service_headers = {"Content-Type": "application/json"} api_key_header = ( str(self.config.get("begin_verification_api_key") or "").strip() or str(self.config.get("auth-api-key") or "").strip() ) if api_key_header: service_headers["X-API-Key"] = api_key_header try: response = requests.post( service_url, json=payload_body, headers=service_headers, timeout=30, proxies={"http": None, "https": None}, # avoid stripping headers via upstream proxies ) except requests_exceptions.RequestException as exc: self.logger.error( "Begin verification service request failed: %s", exc ) return None try: resp_json = response.json() except Exception: resp_json = {} status_text = str(resp_json.get("status") or "").lower() preview = (response.text or "")[:200] self.logger.info( "Begin verification service response: http=%s status=%s preview=%s", getattr(response, "status_code", "?"), status_text or "-", preview, ) if status_text == "success": self.logger.info("Begin verification service success; triggering mail_checker.py.") self._run_mail_checker() return None self.logger.error("Begin verification service returned failure.") return None def _handle_mail_check_failure( self, result: MailCheckResult, log_message: str, *, email_address: Optional[str] = None, email_id: Optional[str] = None, ): """Posta kontrolündeki ölümcül bir hatayı işler, loglar ve iş parçacığını durdurur.""" if result.error in ("email_banned", "disposablemail_invalid"): email_for_log = email_address or self._last_email_address or "" email_id_for_log = email_id or self._last_email_id or "" _remove_email_from_retry_logs(email_id_for_log, email_for_log) self._report_terminal_error(log_message) record_mail_failure() preview_text = "" if result.text: preview_text = str(result.text).replace("\n", "\\n") if len(preview_text) > 500: preview_text = preview_text[:497] + "..." if preview_text: self.logger.debug("mail_checker.py response: %s", preview_text) if result.data: self.logger.debug( "mail_checker.py data: %s", _format_for_log(result.data), ) def _submit_onboarding_task( self, verification_code: str, step_no: Optional[int] = None ) -> Optional[Any]: code = str(verification_code or "").strip() if not code: self.logger.error( "task.json request not sent: verification code not found." ) return None session = self.session if session is None: self.logger.error( "task.json request not sent: TLS session not available." ) return None payload = self._last_callback_payload if payload is None: self.logger.warning( "task.json request not sent: callback payload not saved." ) return None order: Dict[str, Any] = ( self._last_email_order if isinstance(self._last_email_order, dict) else {} ) email_address = str(order.get("email") or "").strip() if not email_address: self.logger.warning( "task.json request not sent: email address not found." ) return None display_name = ( self._last_display_name or str(order.get("display_name") or "").strip() ) display_name = display_name.strip() if isinstance(display_name, str) else "" if not display_name: self.logger.warning( "task.json request not sent: display name not found." ) return None completion_deeplink = self._resolve_completion_deeplink(order) if not completion_deeplink: self.logger.warning( "task.json request not sent: completion_deeplink value missing." ) return None js_payload: Optional[Dict[str, Any]] = None if isinstance(order, dict): js_payload = self._extract_js_instrumentation(order) if not js_payload and self._last_js_instrumentation: js_payload = self._last_js_instrumentation if js_payload: js_payload = self._normalise_js_instrumentation(js_payload) if not js_payload: ensured_js = self._ensure_js_instrumentation(payload.user_agent) js_payload = self._normalise_js_instrumentation(ensured_js) if not js_payload: self.logger.error( "task.json request not sent: js_instrumentation data missing." ) return None birthday = self._resolve_birthdate() personalization = self._resolve_personalization_settings() transaction_id = self._generate_transaction_id_or_log( "POST", "/1.1/onboarding/task.json", context="task.json" ) if transaction_id is None: return None xpff_token, _ = xpff_header_generator.generate( payload.user_agent, payload.encoded_guest_id ) headers = self._build_default_headers( payload, transaction_id, xpff_token, ) headers = _random_adjacent_swap(headers) body = { "flow_token": payload.flow_token, "subtask_inputs": [ { "sign_up": { "birthday": birthday, "email": email_address, "js_instrumentation": js_payload, "link": "email_next_link", "name": display_name, "personalization_settings": personalization, }, "subtask_id": "Signup", },{ "subtask_id":"SignupSettingsListEmail", "settings_list":{ "setting_responses":[ { "key":"allow_emails_about_activity", "response_data":{ "boolean_data":{ "result":False } } }, { "key":"find_by_email", "response_data":{ "boolean_data":{ "result":False } } }, { "key":"personalize_ads", "response_data":{ "boolean_data":{ "result":False } } } ], "link":"next_link", #"castle_token": castle_token } }, { "subtask_id": "ArkoseEmail", "web_modal": { "completion_deeplink": completion_deeplink, "link": "signup_with_email_next_link", }, }, { "email_verification": { "code": code, "email": email_address, "link": "next_link", }, "subtask_id": "EmailVerification", }, ], } if isinstance(order, dict): order["display_name"] = display_name order["js_instrumentation"] = js_payload order["flow_token"] = payload.flow_token self._last_js_instrumentation = js_payload url = "https://api.x.com/1.1/onboarding/task.json" method = "POST" request_headers = dict(headers) request_body = body self.logger.info( "Sending task.json request: email=%s, display_name=%s", email_address, display_name, ) response = self._send_with_http_logging( method=method, url=url, request_headers=request_headers, request_body=request_body, redact_fields={"password"}, step_no=step_no, send_fn=lambda: self._run_tls_coroutine( session.post( url=url, headers=headers, json=body, ) ), ) self._handle_task_json_response(response, payload, order) return response def _submit_password_strength_check( self, payload: Optional[_CallbackPayload], order: Optional[Dict[str, Any]], step_no: Optional[int] = None, ) -> Optional[Any]: session = self.session if session is None: self.logger.error( "password_strength request not sent: TLS session not available." ) return None callback_payload = payload or self._last_callback_payload if callback_payload is None: self.logger.debug( "password_strength request skipped: callback payload not saved." ) return None username = self._last_username or "" if not username and isinstance(order, dict): username = _normalise_username(order.get("username")) if not username and isinstance(self._last_task_response, dict): username = _normalise_username( _search_value(self._last_task_response, "username") ) username = username.strip() if not username: self.logger.debug( "password_strength request skipped: username not found." ) return None transaction_id = self._generate_transaction_id_or_log( "POST", "/1.1/account/password_strength.json", context="password_strength" ) if transaction_id is None: return None xpff_token, _ = xpff_header_generator.generate( callback_payload.user_agent, callback_payload.encoded_guest_id ) password = self._generate_random_password() self._last_password = password self._last_username = username email_source: Optional[Dict[str, Any]] = None if isinstance(order, dict): order["username"] = username order["password"] = password email_source = order elif isinstance(self._last_email_order, dict): email_source = self._last_email_order email_id = "" email_address = "" if isinstance(email_source, dict): email_id = str(email_source.get("email_id") or "").strip() email_address = str(email_source.get("email") or "").strip() self._last_email_id = email_id self._last_email_address = email_address _persist_account_record(username, password, email_id, email_address) headers = self._build_default_headers( callback_payload, transaction_id, xpff_token, ) headers["content-type"] = "application/x-www-form-urlencoded" headers = _random_adjacent_swap(headers) from urllib.parse import urlencode data = urlencode( { "password": password, "username": username, } ) url = "https://api.x.com/1.1/account/password_strength.json" method = "POST" request_headers = dict(headers) request_body = data self.logger.info( "Sending password_strength request: username=%s", username, ) response = self._send_with_http_logging( method=method, url=url, request_headers=request_headers, request_body=request_body, redact_fields={"password"}, step_no=step_no, send_fn=lambda: self._run_tls_coroutine( session.post( url=url, headers=headers, data=data, ) ), ) return response def _submit_enter_password_subtask( self, order: Optional[Dict[str, Any]], step_no: Optional[int] = None ) -> Optional[Any]: session = self.session if session is None: self.logger.error( "EnterPassword request not sent: TLS session not available." ) return None payload = self._last_callback_payload if payload is None: self.logger.debug( "EnterPassword request skipped: callback payload not saved." ) return None flow_token = (self._last_task_flow_token or "").strip() if not flow_token: self.logger.debug( "EnterPassword request skipped: task flow_token not found." ) return None password = (self._last_password or "").strip() if not password: self.logger.debug( "EnterPassword request skipped: password not stored." ) return None # Castle token artık EnterPassword aşamasında alınıyor. email_for_solver = "" if isinstance(order, dict): email_for_solver = str(order.get("email") or "").strip() if not email_for_solver: email_for_solver = str(self._last_email_address or "").strip() display_name = ( self._last_display_name or str(order.get("display_name") if isinstance(order, dict) else "").strip() ) castle_token = self._obtain_castle_token( payload.user_agent, email_for_solver, display_name, ) self._last_castle_token = castle_token if isinstance(order, dict): order["castle_token"] = castle_token self._apply_castle_cuid_cookie() header_overrides = self._consume_castlebreaker_header_overrides() effective_user_agent = header_overrides.get("user-agent", payload.user_agent) accept_language_value = header_overrides.get("accept-language", payload.languages) if self._last_castle_service == "code_solver_castle": solver_user_agent = str(self._last_castle_user_agent or "").strip() if not solver_user_agent: self.logger.error( "EnterPassword request aborted: code_solver_castle user_agent missing." ) return None effective_user_agent = solver_user_agent solver_accept_language = str(self._last_castle_accept_language or "").strip() if solver_accept_language: accept_language_value = solver_accept_language if self._last_castle_service == "castlebreaker.cc": castle_user_agent = str(self._last_castlebreaker_user_agent or "").strip() if castle_user_agent: effective_user_agent = castle_user_agent transaction_id = self._generate_transaction_id_or_log( "POST", "/1.1/onboarding/task.json", context="EnterPassword" ) if transaction_id is None: return None xpff_token, _ = xpff_header_generator.generate( effective_user_agent, payload.encoded_guest_id ) headers = self._build_default_headers( payload, transaction_id, xpff_token, effective_user_agent=effective_user_agent, accept_language=accept_language_value, ) #headers = _random_adjacent_swap(headers) if header_overrides: for key in ("sec-ch-ua", "sec-ch-ua-platform", "user-agent", "accept-language"): value = header_overrides.get(key) if value: headers[key] = value body = { "flow_token": flow_token, "subtask_inputs": [ { "enter_password": { "castle_token": castle_token, "link": "next_link", "password": password, }, "subtask_id": "EnterPassword", } ], } task_password_token = str( self._last_castle_task_password_verification or "" ).strip() if task_password_token: body["subtask_inputs"][0]["enter_password"][ "task_password_verification_castle" ] = task_password_token if isinstance(order, dict): order["enter_password"] = {"flow_token": flow_token} url = "https://api.x.com/1.1/onboarding/task.json" method = "POST" request_headers = dict(headers) request_body = body request_session = session using_castle_session = False if self._should_use_castle_session(effective_user_agent, payload.user_agent): castle_session = self._get_castle_tls_session(effective_user_agent) if castle_session is not None: self._copy_session_cookies(session, castle_session) request_session = castle_session using_castle_session = True self.logger.info("Sending EnterPassword request.") response = self._send_with_http_logging( method=method, url=url, request_headers=request_headers, request_body=request_body, redact_fields={"password", "castle_token"}, step_no=step_no, send_fn=lambda: self._run_tls_coroutine( request_session.post( url=url, headers=headers, json=body, ) ), ) if using_castle_session: self._copy_session_cookies(request_session, session) self._handle_task_json_response(response, payload, order) status_code = getattr(response, "status_code", None) if self._status_equals(status_code, 200): auth_token = self._extract_cookie_from_response(response, "auth_token") csrf_token = self._extract_cookie_from_response(response, "ct0") if auth_token and csrf_token: self._last_auth_token = auth_token self._last_csrf_token = csrf_token self._record_email_for_reregistration() self._send_enter_password_metric_callback( payload, response, auth_token, csrf_token ) else: self.logger.debug( "EnterPassword metric-tracking callback skipped: auth_token or ct0 cookie missing." ) return response def _send_enter_password_metric_callback( self, payload: _CallbackPayload, response: Any, auth_token: str, csrf_token: str, ) -> None: flow_token = (self._last_task_flow_token or "").strip() if not flow_token: self.logger.debug( "EnterPassword metric-tracking callback skipped: flow_token missing." ) return csrf_token_text = str(csrf_token or "").strip() if not csrf_token_text: self.logger.debug( "EnterPassword metric-tracking callback skipped: ct0 value empty." ) return try: payload.cookies["ct0"] = csrf_token_text except Exception: self.logger.debug( "Could not update ct0 cookie for EnterPassword metric-tracking callback.", exc_info=True, ) else: self._apply_payload_cookies_to_session(payload) extra_headers: Optional[Dict[str, str]] = {"x-csrf-token": csrf_token_text} timestamp_ms = int(time.time() * 1000) try: self.logger.info( "Sending EnterPassword metric-tracking callback request." ) callback_response = self._post_metric_tracking_callback( payload=payload, flow_token=flow_token, subtask_id="OpenAccount", timestamp_ms=timestamp_ms, extra_headers=extra_headers, step_no=6, ) except Exception: self.logger.exception( "EnterPassword metric-tracking callback failed." ) return status_code = getattr(callback_response, "status_code", "?") preview = self._summarise_response_text(callback_response) new_csrf_token = ( self._extract_cookie_from_response(callback_response, "ct0") or csrf_token_text ) if new_csrf_token: self._last_csrf_token = new_csrf_token try: payload.cookies["ct0"] = new_csrf_token except Exception: self.logger.debug( "ct0 cookie could not be updated after EnterPassword metric-tracking callback.", exc_info=True, ) else: self._apply_payload_cookies_to_session(payload) self._last_auth_token = auth_token username = (self._last_username or "").strip() password = (self._last_password or "").strip() email_id = (self._last_email_id or "").strip() email_address = (self._last_email_address or "").strip() if username and password: _persist_account_record( username, password, email_id, email_address, auth_token=auth_token, csrf_token=new_csrf_token, ) self.logger.info( "EnterPassword metric-tracking callback response: %s | %s", status_code, preview, ) def _is_password_strength_successful(self, response: Any) -> bool: status = getattr(response, "status_code", None) if not self._status_equals(status, 200): return False payload: Optional[Any] try: payload = response.json() except Exception: payload = None if isinstance(payload, dict): pass_value = payload.get("pass") if isinstance(pass_value, bool): return pass_value if isinstance(pass_value, str): if pass_value.strip().lower() == "true": return True text = getattr(response, "text", "") if isinstance(text, str): lowered = text.lower() if "\"pass\":true" in lowered or "\"pass\": true" in lowered: return True return False def _generate_random_password(self) -> str: length = secrets.choice(range(10, 15)) alphabet = string.ascii_letters + string.digits return "".join(secrets.choice(alphabet) for _ in range(length)) def _choose_display_name(self) -> str: names = _load_display_names() if names: display_name = random.choice(names) if len(names) > 1 else names[0] display_name = _swap_random_adjacent(display_name) display_name = _inject_display_name_separators(display_name) self.logger.debug("Display name selected: %s", display_name) return display_name fallback = _safe_get(self.config, "display_name").strip() if fallback: self.logger.debug("Display name will use configured value: %s", fallback) return fallback raise RuntimeError("display_name değeri bulunamadı") def _resolve_completion_deeplink( self, order: Optional[Dict[str, Any]] ) -> str: token = self._extract_completion_access_token(order) if token: return ( "twitter://onboarding/web_modal/next_link?access_token=" f"{token}" ) return "" def _extract_completion_access_token( self, order: Optional[Dict[str, Any]] ) -> Optional[str]: ignore_values: Set[str] = set() def _normalise(value: Any) -> None: if isinstance(value, str): text = value.strip() if text: ignore_values.add(text) payload = self._last_callback_payload if payload: _normalise(payload.flow_token) if isinstance(order, dict): _normalise(order.get("flow_token")) _normalise(order.get("flowToken")) _normalise(order.get("castle_token")) _normalise(order.get("castleToken")) if isinstance(self._last_captcha_result, dict): _normalise( _search_value( self._last_captcha_result, "flow_token", "flowtoken" ) ) sources: List[Any] = [] if isinstance(order, dict): sources.append(order) nested_order = order.get("data") if isinstance(nested_order, (dict, list, tuple)): sources.append(nested_order) if isinstance(self._last_captcha_result, dict): sources.append(self._last_captcha_result) for key in ("data", "result", "task"): nested = self._last_captcha_result.get(key) if isinstance(nested, (dict, list, tuple)): sources.append(nested) def _coerce(candidate: Any) -> Optional[str]: if isinstance(candidate, str): text = candidate.strip() if text and text not in ignore_values: return text return None def _search_token(source: Any) -> Optional[str]: stack: List[Any] = [source] while stack: current = stack.pop() if isinstance(current, dict): for key in ( "completion_deeplink_token", "deeplink_access_token", "deeplinkToken", "deeplink_token", "access_token", "accessToken", "access-token", ): token_candidate = _coerce(current.get(key)) if token_candidate: return token_candidate token_candidate = _coerce(current.get("token")) if token_candidate: return token_candidate for value in current.values(): if isinstance(value, (dict, list, tuple)): stack.append(value) elif isinstance(current, (list, tuple)): stack.extend(current) return None for source in sources: token_value = _search_token(source) if token_value: return token_value return None def _resolve_birthdate(self) -> Dict[str, int]: default = {"day": 20, "month": 12, "year": 1986} raw = self.config.get("birthdate") if not isinstance(raw, dict): alt = self.config.get("birthday") raw = alt if isinstance(alt, dict) else None if isinstance(raw, dict): try: day = int(raw.get("day", default["day"])) month = int(raw.get("month", default["month"])) year = int(raw.get("year", default["year"])) except (TypeError, ValueError): return default return {"day": day, "month": month, "year": year} return default def _resolve_personalization_settings(self) -> Dict[str, bool]: settings = { "allow_ads_personalization": True, "allow_cookie_use": True, "allow_device_personalization": True, "allow_partnerships": True, } overrides = self.config.get("personalization_settings") if isinstance(overrides, dict): for key, value in overrides.items(): settings[key] = _is_feature_enabled(value) return settings def _ensure_js_instrumentation(self, user_agent: str) -> Optional[Dict[str, Any]]: if self._last_js_instrumentation: return self._last_js_instrumentation if self._js_instrumentation_attempted: return None self._js_instrumentation_attempted = True generated = self._generate_js_instrumentation(user_agent) if generated: self._last_js_instrumentation = generated self.logger.debug("Generated with JS instrumentation script.") return self._last_js_instrumentation def _generate_js_instrumentation(self, user_agent: str) -> Optional[Dict[str, Any]]: script_path = Path(__file__).with_name("gen_jsui.mjs") if not script_path.exists(): self.logger.debug( "JS instrumentation script not found: %s", script_path, ) return None node_path = shutil.which("node") if not node_path: self.logger.debug( "Node.js not found; skipping js_instrumentation generation." ) return None try: completed = subprocess.run( [node_path, str(script_path), user_agent], capture_output=True, text=True, check=True, timeout=90, ) except subprocess.TimeoutExpired: self.logger.warning("JS instrumentation script timed out.") return None except subprocess.CalledProcessError as exc: stderr = exc.stderr.strip() if isinstance(exc.stderr, str) else "" if stderr: self.logger.error( "JS instrumentation script failed: %s | stderr: %s", exc, stderr[:200], ) else: self.logger.error( "JS instrumentation script failed: %s", exc, ) return None except OSError as exc: self.logger.warning( "JS instrumentation script could not be executed: %s", exc, ) return None stdout = (completed.stdout or "").strip() if not stdout: self.logger.warning("JS instrumentation script returned empty output.") return None lines = [line.strip() for line in stdout.splitlines() if line.strip()] if not lines: self.logger.warning( "JS instrumentation output could not be parsed (empty lines)." ) return None last_line = lines[-1] try: data = json.loads(last_line) except json.JSONDecodeError as exc: self.logger.warning( "JS instrumentation output not in JSON format: %s", exc, ) return None if not isinstance(data, dict): self.logger.warning( "JS instrumentation output not a dictionary." ) return None stderr_text = (completed.stderr or "").strip() if stderr_text: self.logger.debug( "JS instrumentation script stderr: %s", stderr_text[:200], ) normalised = self._normalise_js_instrumentation(data) if not normalised: self.logger.warning( "JS instrumentation output not in expected format." ) return None return normalised def _obtain_castle_token( self, user_agent: str, email_address: str, display_name: str, ) -> str: self._last_castle_cuid = None self._last_castle_service = None self._last_castle_solver_details = None self._last_castle_user_agent = None self._last_castle_accept_language = None self._last_castle_task_password_verification = None self._castlebreaker_header_overrides = None raw_value = self.config.get("castle_token") if isinstance(raw_value, str): trimmed = raw_value.strip() lowered = trimmed.lower() if trimmed and lowered not in {"true", "false"}: self.logger.debug("Using Castle token from config.") return trimmed elif raw_value not in (None, ""): text = str(raw_value).strip() if text and text.lower() not in {"true", "false"}: self.logger.debug("Using Castle token from config.") return text # code_solver_castle öne alındı; diğer Castle servisleri sırayla denenir. service_candidates = ( ("code_solver_castle", self.config.get("code_solver_castle")), ("castlesolver.cc", self.config.get("castlesolver.cc")), ("castlebreaker.cc", self.config.get("castlebreaker.cc")), ) errors: List[str] = [] for service_name, service_config in service_candidates: if not isinstance(service_config, dict): continue service_config = dict(service_config) if service_name == "code_solver_castle": test_flag = _is_feature_enabled(self.config.get("code_solver_test_server")) service_config.setdefault("test_server", test_flag) if not _is_feature_enabled(service_config.get("enabled")): continue api_key = str(service_config.get("api_key") or "").strip() if not api_key: errors.append(f"{service_name}: missing api_key") continue count_value = service_config.get("count") try: count = int(count_value or 1) except (TypeError, ValueError): count = 1 timeout_value = service_config.get("timeout") try: timeout = int(timeout_value or 30) except (TypeError, ValueError): timeout = 30 cuid_value = str(service_config.get("cuid") or "").strip() if service_name == "code_solver_castle": solver_user_agent = _safe_get(self.config, "user_agent").strip() if not solver_user_agent: solver_user_agent = user_agent try: result = request_castle_solves( api_key, count=count, user_agent=solver_user_agent, email=email_address, display_name=display_name, cuid=cuid_value, timeout=timeout, service=service_name, proxy_spec=self._proxy_spec, service_config=service_config, proxy_country=self._proxy_country_code, ) except CastleSolverError as exc: record_castle_failure() error_text = str(exc) errors.append(f"{service_name}: {error_text}") self.logger.warning( "Castle token request failed (%s): %s", service_name, error_text, ) continue except Exception as exc: # pragma: no cover - d?? ba??ml?l?k hatalar? record_castle_failure() error_text = str(exc) errors.append(f"{service_name}: {error_text}") self.logger.warning( "Castle token request ended with unexpected error (%s): %s", service_name, error_text, exc_info=self.debug_enabled, ) continue else: try: result = request_castle_solves( api_key, count=count, user_agent=user_agent, email=email_address, display_name=display_name, cuid=cuid_value, timeout=timeout, service=service_name, proxy_spec=self._proxy_spec, service_config=service_config, proxy_country=self._proxy_country_code, ) except CastleSolverError as exc: record_castle_failure() error_text = str(exc) if _is_castlesolver_balance_error_message(error_text): shutdown_reason = ( "CastleSolver bakiyesi t?kenmi? g?r?n?yor, uygulama g?venli ?ekilde " "sonland?r?lacak." ) self.logger.error( "%s. Service response: %s", shutdown_reason, error_text, ) request_global_shutdown( f"CastleSolver bakiyesi t?kenmi? olabilir: {error_text}" ) message = f"{service_name}: {error_text}" errors.append(message) self.logger.warning( "Castle token request failed (%s): %s", service_name, exc ) continue except Exception as exc: # pragma: no cover - d?? ba??ml?l?k hatalar? record_castle_failure() message = f"{service_name}: {exc}" errors.append(message) self.logger.warning( "Castle token request ended with unexpected error (%s): %s", service_name, exc, exc_info=self.debug_enabled, ) continue if result.tokens: token = result.tokens[0] self.logger.debug( "Castle token obtained from %s service (cuid=%s, balance=%s).", service_name, result.cuid or "?", result.balance, ) self._last_castle_cuid = (result.cuid or cuid_value or "").strip() self._last_castle_service = service_name if service_name == "code_solver_castle": solver_user_agent = str(result.user_agent or "").strip() if not solver_user_agent: record_castle_failure() raise RuntimeError( "code_solver_castle user_agent missing; castle token rejected." ) self._last_castle_user_agent = solver_user_agent solver_accept_language = str(result.accept_lang or "").strip() self._last_castle_accept_language = ( solver_accept_language if solver_accept_language else None ) task_password_token = str( result.task_password_verification_castle or "" ).strip() self._last_castle_task_password_verification = ( task_password_token if task_password_token else None ) raw_details = None if isinstance(result.raw, dict): raw_details = result.raw.get("solverDetails") or result.raw.get("solver_details") if isinstance(raw_details, dict): self._last_castle_solver_details = dict(raw_details) else: self._last_castle_solver_details = None self._castlebreaker_header_overrides = None if service_name == "castlebreaker.cc": overrides: Dict[str, str] = {} accept_lang_value = str(result.accept_lang or "").strip() sec_ch_ua_value = str(result.sec_ch_ua or "").strip() sec_ch_ua_platform_value = str(result.sec_ch_ua_platform or "").strip() user_agent_value = str(result.user_agent or "").strip() if accept_lang_value: overrides["accept-language"] = accept_lang_value if sec_ch_ua_value: overrides["sec-ch-ua"] = sec_ch_ua_value if sec_ch_ua_platform_value: overrides["sec-ch-ua-platform"] = sec_ch_ua_platform_value if user_agent_value: overrides["user-agent"] = user_agent_value self._last_castlebreaker_user_agent = user_agent_value if overrides: self._castlebreaker_header_overrides = overrides extras: List[str] = [] if result.cuid: extras.append(f"cuid={result.cuid}") if result.balance is not None: balance_text = f"{result.balance:.6f}".rstrip("0").rstrip(".") extras.append(f"balance={balance_text}") if accept_lang_value: extras.append(f"lang={accept_lang_value}") if sec_ch_ua_platform_value: extras.append(f"platform={sec_ch_ua_platform_value}") if sec_ch_ua_value: extras.append(f"sec_ch_ua={sec_ch_ua_value}") if user_agent_value: extras.append(f"ua={user_agent_value}") if extras: self.logger.info( "Castlebreaker extras: %s", "; ".join(extras), ) if self.debug_enabled and isinstance(result.raw, dict): self.logger.debug( "Castlebreaker raw (redacted): %s", _format_for_log(_redact_castlebreaker_raw(result.raw)), ) if result.accept_lang: self.config["accept_language"] = result.accept_lang if result.sec_ch_ua: self.config["sec_ch_ua"] = result.sec_ch_ua if result.sec_ch_ua_platform: self.config["sec_ch_ua_platform"] = result.sec_ch_ua_platform if result.user_agent and service_name != "code_solver_castle": self.config["user_agent"] = result.user_agent record_castle_token_solved() return token errors.append(f"{service_name}: response without tokens") detail = f" ({'; '.join(errors)})" if errors else "" raise RuntimeError(f"Castle token elde edilemedi{detail}") def _is_manual_email_mode(self) -> bool: return self._manual_email_enabled def _prompt_manual_email_address(self) -> str: if not self._is_manual_email_mode(): return "" if self._manual_email_address: return self._manual_email_address prompt = "[Manuel E-posta] Kullanmak istediğiniz e-posta adresini girin" while True: try: user_input = input(f"{prompt}: ").strip() except EOFError as exc: raise RuntimeError( "Manuel e-posta modu etkin ancak e-posta adresi için konsol girdisi alınamadı." ) from exc if user_input: break self.logger.warning( "Manual email mode enabled but empty value entered; please try again." ) self._manual_email_address = user_input order_data = { "email": user_input, "email_id": "", "service": "manual", "manual": True, } self._last_email_order = order_data self._last_email_id = "" self._last_email_address = user_input self.logger.info("Manual email address will be used: %s", user_input) return user_input def _prompt_manual_verification_code(self, email_address: str) -> str: if not self._is_manual_email_mode(): return "" prompt = ( f"[Manuel E-posta] {email_address} adresi için doğrulama kodunu girin" ) while True: try: user_input = input(f"{prompt}: ").strip() except EOFError as exc: raise RuntimeError( "Manuel e-posta modu etkin ancak doğrulama kodu için konsol girdisi alınamadı." ) from exc if not user_input: self.logger.warning( "Manual verification code left empty; please try again." ) continue normalised = _normalise_mail_code(user_input) or user_input self._manual_verification_code = normalised return normalised def _obtain_email_address(self) -> str: if self._is_manual_email_mode(): return self._prompt_manual_email_address() prioritize_anymessage_stock = self._should_prioritize_anymessage_stock() if prioritize_anymessage_stock and self._has_sufficient_anymessage_stock(): return self._obtain_email_address_from_anymessage() while True: policy = self._email_source_policy force_purchase = policy == ANYMESSAGE_POLICY_PURCHASE_ONLY force_rereg = policy == ANYMESSAGE_POLICY_REUSE_ONLY reused_email = self._try_reuse_email_address( enable_reregistry=not force_purchase ) if reused_email: return reused_email if force_rereg: if self._has_sufficient_anymessage_stock(): self.logger.info( "Re-registration lists are empty; AnyMessage stock is sufficient, switching to purchase mode." ) self._email_source_policy = ANYMESSAGE_POLICY_PURCHASE_ONLY return self._obtain_email_address_from_anymessage() # Yeniden kayıt listesi zorunluysa ve e-posta bulunamadıysa, # bu işçinin başka yapabileceği bir şey yok. Hata fırlatarak sonlanmalı. raise MailServiceNoEmailsError( "Yeniden kayıt listesi zorunlu ancak kullanılabilir e-posta bulunamadı." ) if self.mail_service == "disposablemail": return self._obtain_email_address_from_disposablemail() return self._obtain_email_address_from_anymessage() def _wait_for_email_resources(self) -> None: if self.mail_service and self.mail_service != "anymessage": self._email_source_policy = ANYMESSAGE_POLICY_AUTO clear_global_resource_wait() return if not _is_feature_enabled(self.config.get("anymessage")): self._email_source_policy = ANYMESSAGE_POLICY_AUTO clear_global_resource_wait() return interval = _ANYMESSAGE_RESOURCE_WAIT_INTERVAL prefer_purchase = _is_feature_enabled( self.config.get("anymessage_prioritize_purchase", True) ) self.logger.debug( "Email source wait loop running every %.1f seconds.", interval ) while True: reuse_sources = [] if has_available_rereg_email(self.config): reuse_sources.append("Email_information_awaiting_re-registration.txt") has_reuse_email = bool(reuse_sources) has_stock = self._has_sufficient_anymessage_stock() if prefer_purchase: if has_stock: if self._email_source_policy == ANYMESSAGE_POLICY_REUSE_ONLY: self.logger.info( "AnyMessage Gmail stock is sufficient; switching to purchase mode." ) self._email_source_policy = ANYMESSAGE_POLICY_PURCHASE_ONLY return if has_reuse_email: if self._email_source_policy != ANYMESSAGE_POLICY_REUSE_ONLY: self.logger.warning( "AnyMessage Gmail stock is low; using re-registration lists (%s).", " ve ".join(reuse_sources), ) self._email_source_policy = ANYMESSAGE_POLICY_REUSE_ONLY return else: if has_reuse_email: if self._email_source_policy != ANYMESSAGE_POLICY_REUSE_ONLY: self.logger.info( "Re-registration lists (%s) prioritized; continuing from lists.", " ve ".join(reuse_sources), ) self._email_source_policy = ANYMESSAGE_POLICY_REUSE_ONLY return if has_stock: if self._email_source_policy == ANYMESSAGE_POLICY_REUSE_ONLY: self.logger.info( "Re-registration lists prioritized but empty; stock available, switching to purchase mode." ) self._email_source_policy = ANYMESSAGE_POLICY_PURCHASE_ONLY return if not _GLOBAL_RESOURCE_WAIT_EVENT.is_set(): self.logger.error( "No available email source found. Putting entire system into wait mode." ) _GLOBAL_RESOURCE_WAIT_EVENT.set() if self._wait_with_shutdown(interval, "global kaynak bekleme"): raise RuntimeError("Global kapatma isteği nedeniyle kaynak bekleme durduruldu.") def _should_prioritize_anymessage_stock(self) -> bool: if self.mail_service and self.mail_service != "anymessage": return False if not _is_feature_enabled(self.config.get("anymessage_prioritize_purchase", True)): return False if not _is_feature_enabled(self.config.get("anymessage")): return False domain = str(self.config.get("anymessage_domain") or "").strip().lower() return is_gmail_domain_alias(domain) def _has_sufficient_anymessage_stock(self) -> bool: if not _is_feature_enabled(self.config.get("anymessage")): return False if self.mail_service and self.mail_service != "anymessage": return False api_key = str(self.config.get("AnyMessage") or "").strip() if not api_key: return False domain = str(self.config.get("anymessage_domain") or "").strip() or _GMAIL_DOMAIN stock_count = get_available_domain_count_with_retry( api_key, domain=domain, site=ANYMESSAGE_STOCK_SITE, attempts=_ANYMESSAGE_STOCK_RETRY_ATTEMPTS, retry_delay=_ANYMESSAGE_STOCK_RETRY_DELAY, logger=self.logger, ) if stock_count is None: return False if stock_count >= ANYMESSAGE_STOCK_THRESHOLD: self.logger.info( "AnyMessage Gmail stock appears sufficient (count=%d).", stock_count, ) return True self.logger.debug( "AnyMessage Gmail stock appears insufficient (count=%s).", stock_count, ) return False def _try_reuse_email_address( self, *, enable_reregistry: bool = True ) -> Optional[str]: if not enable_reregistry: manager = None else: manager = _get_email_rereg_manager(self.config) if manager is None: return None try: record = manager.acquire_ready_email() except Exception: self.logger.exception( "Could not read email re-registration list." ) return None if not isinstance(record, dict): return None reused = self._apply_reused_email_record( record, source="reuse", log_template="Reusable email selected: id=%s email=%s", ) return reused def _apply_reused_email_record( self, record: Dict[str, Any], *, source: str, log_template: str, ) -> Optional[str]: email_address = str(record.get("email_address") or "").strip() if not email_address: return None should_randomize = source == "reuse" reserved_variant = str(record.get("reserved_email_variant") or "").strip() used_dot_positions = record.get("used_dot_positions") if should_randomize: if reserved_variant: randomized_email = reserved_variant else: randomized_email, _ = randomize_gmail_local_part( email_address, used_dot_positions=used_dot_positions, advance_history=False, ) else: randomized_email = email_address prepared_email = ensure_local_part_has_dot(randomized_email) email_id = str(record.get("email_id") or "").strip() service = str(record.get("service") or "").strip() or source service_lower = service.lower() order_data = { "email": prepared_email, "email_id": email_id, "service": service, "data": dict(record), "reused": True, "retry_source": source, } api_key = str(record.get("api_key") or "").strip() if service_lower == "anymessage": if not api_key: api_key = _safe_get(self.config, "AnyMessage").strip() if not api_key: message = ( "AnyMessage API key missing for reused email entry; shutting down." ) request_global_shutdown(message) raise RuntimeError(message) if api_key: order_data["api_key"] = api_key verification_code = str(record.get("verification_code") or "").strip() if verification_code: order_data["verification_code"] = verification_code self._last_email_order = order_data self._last_email_id = email_id self._last_email_address = prepared_email self.logger.info(log_template, email_id or "?", prepared_email) return prepared_email def _obtain_email_address_from_anymessage(self) -> str: api_key = _safe_get(self.config, "AnyMessage").strip() if not api_key: raise RuntimeError("AnyMessage değeri bulunamadı") try: result = order_email(api_key) except Exception as exc: # pragma: no cover - dış bağımlılık hataları raise RuntimeError(f"Mail servisi isteği başarısız: {exc}") from exc if not result.ok or not isinstance(result.data, dict): detail_parts: List[str] = [] if result.error: error_text = str(result.error).strip() if error_text: detail_parts.append(error_text) if result.text: text_detail = str(result.text).strip() if text_detail: if len(text_detail) > 500: text_detail = text_detail[:497] + "..." if text_detail not in detail_parts: detail_parts.append(text_detail) detail = " | ".join(detail_parts) if detail_parts else "bilinmeyen hata" self.logger.error("Mail service failed: %s", detail) detail_lower = detail.lower() if "no emails" in detail_lower: raise MailServiceNoEmailsError(f"Mail servisi başarısız: {detail}") raise RuntimeError(f"Mail servisi başarısız: {detail}") data = dict(result.data) raw_email_address = str( data.get("email") or data.get("address") or "" ).strip() email_address = ensure_local_part_has_dot(raw_email_address) if not email_address: raise RuntimeError("Mail servisi geçerli e-posta döndürmedi") email_id = str( data.get("id") or data.get("email_id") or data.get("order_id") or "" ).strip() self._last_email_order = { "email": email_address, "email_id": email_id, "data": data, "service": "anymessage", } self._last_email_id = email_id self._last_email_address = email_address if email_id: self.logger.debug( "Mail order completed: id=%s, email=%s", email_id, email_address, ) else: self.logger.debug("Mail order completed: email=%s", email_address) return email_address def _obtain_email_address_from_disposablemail(self) -> str: if self.mail_api is None: raise RuntimeError("Mail servisi yapılandırılamadı") try: response = self.mail_api.order_mail() except DisposableMailAPIError as exc: record_mail_order_result(False) raise RuntimeError(f"Mail servisi isteği başarısız: {exc}") from exc except Exception as exc: # pragma: no cover - dış bağımlılık hataları record_mail_order_result(False) raise RuntimeError(f"Mail servisi isteği başarısız: {exc}") from exc if not isinstance(response, dict): record_mail_order_result(False) raise RuntimeError("Mail servisi beklenmeyen yanıt döndürdü") code_value = response.get("code") if code_value != 200: detail = self._summarise_mail_error(response) self.logger.error("Mail service failed: %s", detail) record_mail_order_result(False) raise RuntimeError(f"Mail servisi başarısız: {detail}") data = response.get("data") if not isinstance(data, dict): record_mail_order_result(False) raise RuntimeError("Mail servisi geçerli veri döndürmedi") order_info = data.get("order") if not isinstance(order_info, dict): record_mail_order_result(False) raise RuntimeError("Mail servisi sipariş bilgisi döndürmedi") email_address = str(order_info.get("email") or "").strip() if not email_address: record_mail_order_result(False) raise RuntimeError("Mail servisi geçerli e-posta döndürmedi") order_id = str(order_info.get("orderId") or "").strip() if not order_id: record_mail_order_result(False) raise RuntimeError("Mail servisi geçerli orderId döndürmedi") self._last_email_order = { "email": email_address, "email_id": order_id, "order_id": order_id, "data": data, "service": "disposablemail", } self._last_email_id = order_id self._last_email_address = email_address links_value = data.get("links") if links_value: self._last_email_order["links"] = links_value self.logger.debug( "Mail order completed (disposablemail): id=%s, email=%s", order_id, email_address, ) record_mail_order_result(True) return email_address @staticmethod def _summarise_mail_error(payload: Dict[str, Any]) -> str: detail_parts: List[str] = [] temporary_codes = _MAILAPI_TEMPORARY_CODES for key in ( "message", "error", "detail", "description", "status_message", "statusMessage", "value", "msg", "Msg", ): value = payload.get(key) if isinstance(value, (str, int, float)): text = str(value).strip() if text: upper_text = text.upper() if upper_text in temporary_codes: return f"Geçici durum: {upper_text}" detail_parts.append(text) if not detail_parts: try: detail_parts.append(json.dumps(payload, ensure_ascii=False)) except (TypeError, ValueError): detail_parts.append(str(payload)) return " | ".join(detail_parts) @staticmethod def _find_first_scalar_value(payload: Any, keys: Iterable[str]) -> Optional[str]: key_set = {str(key).lower() for key in keys} stack: List[Any] = [payload] while stack: current = stack.pop() if isinstance(current, dict): for key, value in current.items(): lowered = str(key).lower() if lowered in key_set and isinstance(value, (str, int, float)): text = str(value).strip() if text: return text stack.extend(current.values()) elif isinstance(current, list): stack.extend(current) return None def _extract_email_address_from_payload(self, payload: Dict[str, Any]) -> str: result = self._find_first_scalar_value( payload, ( "email", "mail", "address", "email_address", "emailAddress", "mail_address", ), ) return result or "" def _extract_email_id_from_payload(self, payload: Dict[str, Any]) -> str: result = self._find_first_scalar_value( payload, ( "id", "email_id", "order_id", "emailId", "emailID", "mail_id", "mailId", ), ) return result or "" @staticmethod def _serialise_mail_payload(payload: Any) -> str: if isinstance(payload, (dict, list)): try: return json.dumps(payload, ensure_ascii=False) except (TypeError, ValueError): return str(payload) return str(payload) @staticmethod def _extract_code_from_mail_payload(payload: Any) -> Optional[str]: if isinstance(payload, str): match = _MAIL_CODE_REGEX.search(payload) if match: normalised = _normalise_mail_code(match.group(0)) if normalised: return normalised fallback = _normalise_mail_code(payload) if fallback: return fallback return None if isinstance(payload, dict): for value in payload.values(): code = XGenWorker._extract_code_from_mail_payload(value) if code: return code return None if isinstance(payload, list): for item in payload: code = XGenWorker._extract_code_from_mail_payload(item) if code: return code return None def _wait_for_disposablemail_email( self, email_id: str, *, log_fn: Optional[Callable[[str], None]] = None, email_address: Optional[str] = None, not_before: Optional[float] = None, ) -> MailCheckResult: def _finalise(result: MailCheckResult) -> MailCheckResult: record_mail_check_result(result.ok) return result if self.mail_api is None: return _finalise( MailCheckResult( False, 0, "mail_api_not_configured", None, 0, "mail_api_not_configured", ) ) if self._should_exit_for_shutdown("disposablemail bekleme"): return _finalise( MailCheckResult( False, 0, "shutdown_requested", None, 0, "shutdown_requested", ) ) timeout_value = self.config.get("mail_timeout_sec") try: timeout_sec = int(timeout_value or 180) except (TypeError, ValueError): timeout_sec = 180 interval_value = self.config.get("mail_interval_sec") try: interval_sec = float(interval_value or 2) except (TypeError, ValueError): interval_sec = 2.0 interval_sec = max(0.5, interval_sec) deadline = time.time() + max(1, timeout_sec) attempts = 0 last_error: Optional[str] = None last_payload: Optional[Dict[str, Any]] = None last_text = "" stale_email_detected = False if isinstance(not_before, (int, float)): not_before_ts: Optional[float] = float(not_before) else: not_before_ts = None def _parse_receive_timestamp(value: Any) -> Optional[float]: if value is None: return None if isinstance(value, (int, float)): ts = float(value) if ts > 1_000_000_000_000: ts /= 1000.0 return ts if isinstance(value, str): stripped = value.strip() if not stripped: return None try: ts = float(stripped) except ValueError: candidate = stripped if candidate.endswith("Z"): candidate = candidate[:-1] + "+00:00" dt_obj: Optional[datetime] try: dt_obj = datetime.fromisoformat(candidate) except ValueError: try: dt_obj = parsedate_to_datetime(stripped) except (TypeError, ValueError): return None if dt_obj is None: return None if dt_obj.tzinfo is None: dt_obj = dt_obj.replace(tzinfo=timezone.utc) else: dt_obj = dt_obj.astimezone(timezone.utc) return dt_obj.timestamp() else: ts = float(ts) if ts > 1_000_000_000_000: ts /= 1000.0 return ts return None def _find_receive_timestamp(payload_obj: Any) -> Optional[float]: if isinstance(payload_obj, dict): for key, value in payload_obj.items(): if key == "receiveTimestamp": parsed = _parse_receive_timestamp(value) if parsed is not None: return parsed nested = _find_receive_timestamp(value) if nested is not None: return nested elif isinstance(payload_obj, list): for item in payload_obj: nested = _find_receive_timestamp(item) if nested is not None: return nested return None while time.time() < deadline: if self._should_exit_for_shutdown("disposablemail bekleme"): return _finalise( MailCheckResult( False, 0, "shutdown_requested", None, attempts, "shutdown_requested", ) ) attempts += 1 try: payload = self.mail_api.check_mail(email_id) except DisposableMailAPIError as exc: last_error = str(exc) lowered_error = last_error.lower() if ( "email is no longer valid" in lowered_error and "cannot be used" in lowered_error ): status_code = 0 match = re.search(r"\((\d{3})\)", last_error) if match: try: status_code = int(match.group(1)) except ValueError: status_code = 0 if log_fn: log_fn( "[DisposableMail] invalid email detected; terminating mail check." ) return _finalise( MailCheckResult( False, status_code, last_error, None, attempts, "disposablemail_invalid", ) ) if log_fn: log_fn(f"[DisposableMail] attempt={attempts} error={last_error}") if self._wait_with_shutdown( interval_sec, "disposablemail bekleme" ): return _finalise( MailCheckResult( False, 0, "shutdown_requested", None, attempts, "shutdown_requested", ) ) continue except Exception as exc: # pragma: no cover - dış bağımlılık hataları last_error = str(exc) message = f"[DisposableMail] attempt={attempts} exception={last_error}" if log_fn: log_fn(message) if self._wait_with_shutdown( interval_sec, "disposablemail bekleme" ): return _finalise( MailCheckResult( False, 0, "shutdown_requested", None, attempts, "shutdown_requested", ) ) continue if not isinstance(payload, dict): last_error = "invalid_response" message = ( f"[DisposableMail] attempt={attempts} invalid response type=" f"{type(payload).__name__}" ) if log_fn: log_fn(message) if self._wait_with_shutdown( interval_sec, "disposablemail bekleme" ): return _finalise( MailCheckResult( False, 0, "shutdown_requested", None, attempts, "shutdown_requested", ) ) continue last_payload = payload last_text = self._serialise_mail_payload(payload) code_value = payload.get("code") if code_value not in (None, 200): message_text = str(payload.get("msg") or "api_error").strip() error_code = f"api_{code_value}" if code_value else "api_error" last_error = error_code if log_fn: log_fn( f"[DisposableMail] attempt={attempts} error: {message_text or error_code}" ) return _finalise( MailCheckResult( False, int(code_value) if isinstance(code_value, (int, float)) else 0, last_text, payload, attempts, error_code, ) ) data = payload.get("data") code = self._extract_code_from_mail_payload(data) if not_before_ts is not None: receive_ts = _find_receive_timestamp(payload) if receive_ts is not None and receive_ts < not_before_ts: stale_email_detected = True last_error = "stale_email" if log_fn: log_fn( f"[DisposableMail] attempt={attempts} stale verification email detected; waiting for a newer message." ) if self._wait_with_shutdown( interval_sec, "disposablemail bekleme" ): return _finalise( MailCheckResult( False, 0, "shutdown_requested", None, attempts, "shutdown_requested", ) ) continue if code: success_msg = f"[DisposableMail] attempt={attempts} status=OK code={code}" if log_fn: log_fn(success_msg) return _finalise( MailCheckResult(True, 200, last_text, payload, attempts, None, code) ) if log_fn: log_fn(f"[DisposableMail] attempt={attempts} code not found") last_error = "code_not_found" if self._wait_with_shutdown(interval_sec, "disposablemail bekleme"): return _finalise( MailCheckResult( False, 0, "shutdown_requested", None, attempts, "shutdown_requested", ) ) timeout_message = f"timeout after {attempts} attempts" if last_payload is not None: last_text = self._serialise_mail_payload(last_payload) elif not last_text: last_text = timeout_message final_error = last_error if stale_email_detected and final_error in (None, "code_not_found", "stale_email"): final_error = "stale_email_timeout" if log_fn: log_fn( "[DisposableMail] no verification email newer than begin_verification was received before timeout; terminating mail check." ) return _finalise( MailCheckResult(False, 0, last_text, last_payload, attempts, final_error) ) @staticmethod def _normalise_js_instrumentation( payload: Any, ) -> Optional[Dict[str, Any]]: if not isinstance(payload, dict): return None result: Dict[str, Any] = dict(payload) base_payload = dict(result) response = result.get("response") if isinstance(response, str): trimmed = response.strip() if trimmed: result["response"] = trimmed return result result.pop("response", None) base_payload = dict(result) elif response is not None: try: serialised = json.dumps( response, separators=(",", ":"), ensure_ascii=False ) except (TypeError, ValueError): result.pop("response", None) base_payload = dict(result) else: trimmed = serialised.strip() if trimmed: result["response"] = trimmed return result result.pop("response", None) base_payload = dict(result) else: base_payload = dict(result) try: serialised_payload = json.dumps( base_payload, separators=(",", ":"), ensure_ascii=False ) except (TypeError, ValueError): return None trimmed_payload = serialised_payload.strip() if not trimmed_payload: return None result["response"] = trimmed_payload return result def _handle_task_json_response( self, response: Any, payload: Optional[_CallbackPayload], order: Optional[Dict[str, Any]], ) -> None: if payload is None or response is None: return try: response_payload = response.json() except Exception: # pragma: no cover - ağ yanıtı beklenmedik biçimdeyse self.logger.debug( "task.json response could not be parsed as JSON.", exc_info=True, ) return if not isinstance(response_payload, dict): return new_flow_token_raw = _search_value(response_payload, "flow_token", "flowtoken") new_flow_token = str(new_flow_token_raw or "").strip() if new_flow_token: self._last_task_flow_token = new_flow_token if isinstance(order, dict): order["latest_flow_token"] = new_flow_token processed_payload = _replace_flow_token_in_response( response_payload, payload.flow_token ) self._last_task_response = processed_payload if isinstance(order, dict): order["task_json_payload"] = processed_payload username_value = _search_value(processed_payload, "username") username_text = _normalise_username(username_value) if username_text: self._last_username = username_text if isinstance(order, dict): order["username"] = username_text _persist_username(username_text) @classmethod def _coerce_js_instrumentation( cls, value: Any ) -> Optional[Dict[str, Any]]: if isinstance(value, dict): return cls._normalise_js_instrumentation(value) if isinstance(value, (list, tuple)): for item in value: candidate = cls._coerce_js_instrumentation(item) if candidate: return candidate return None if isinstance(value, str): text = value.strip() if not text: return None try: parsed = json.loads(text) except json.JSONDecodeError: return cls._normalise_js_instrumentation({"response": text}) else: if isinstance(parsed, dict): return cls._normalise_js_instrumentation(parsed) return cls._normalise_js_instrumentation({"response": text}) return None @staticmethod def _extract_js_instrumentation(source: Any) -> Optional[Dict[str, Any]]: if source is None: return None value = _search_value( source, "js_instrumentation", "jsinstrumentation", "js_instrumentation_response", ) if value is None: return None if isinstance(value, dict): return XGenWorker._normalise_js_instrumentation(value) if isinstance(value, (list, tuple)): return XGenWorker._coerce_js_instrumentation(value) if isinstance(value, str): return XGenWorker._coerce_js_instrumentation(value) return None @staticmethod def _status_equals(status: Any, expected: int) -> bool: try: if isinstance(status, str): status = status.strip() return int(status) == int(expected) except (TypeError, ValueError): return False @staticmethod def _is_signup_denied_response(response: Any) -> bool: if response is None: return False payload: Any = None try: payload = response.json() except Exception: payload = None def _message_contains_signup_denied(message: Any) -> bool: if not isinstance(message, str): return False return "signup denied" in message.strip().lower() if isinstance(payload, dict): errors = payload.get("errors") if isinstance(errors, list): for item in errors: if not isinstance(item, dict): continue code_value = item.get("code") code_text = str(code_value).strip() if code_value is not None else "" message_text = item.get("message") if code_text == "498" and _message_contains_signup_denied(message_text): return True if _message_contains_signup_denied(message_text): return True detail_value = payload.get("detail") if _message_contains_signup_denied(detail_value): return True text_value = getattr(response, "text", "") if isinstance(text_value, bytes): try: text_value = text_value.decode("utf-8", "ignore") except Exception: text_value = str(text_value) if isinstance(text_value, str) and _message_contains_signup_denied(text_value): return True return False @staticmethod def _is_error_status_code(status: Any) -> bool: try: return int(str(status).strip()) >= 400 except (TypeError, ValueError): return False @staticmethod def _response_contains_error_code(response: Any) -> bool: if response is None: return False try: payload = response.json() except Exception: return False if isinstance(payload, dict): errors = payload.get("errors") if isinstance(errors, list): for item in errors: if isinstance(item, dict) and item.get("code") not in (None, ""): return True return False @staticmethod def _response_has_error_code(response: Any, code: str) -> bool: if response is None: return False target_code = str(code).strip() if not target_code: return False try: payload = response.json() except Exception: return False if isinstance(payload, dict): errors = payload.get("errors") if isinstance(errors, list): for item in errors: if not isinstance(item, dict): continue value = item.get("code") if value is None: continue if str(value).strip() == target_code: return True return False @staticmethod def _extract_error_message_from_response( response: Any, code: Optional[str] = None ) -> Optional[str]: if response is None: return None target_code = None if code is not None: target_code = str(code).strip() if not target_code: target_code = None try: payload = response.json() except Exception: payload = None if isinstance(payload, dict): errors = payload.get("errors") if isinstance(errors, list): for item in errors: if not isinstance(item, dict): continue if target_code is not None: value = item.get("code") if str(value).strip() != target_code: continue message = item.get("message") if isinstance(message, str): text = message.strip() if text: return text detail = payload.get("detail") if target_code is None and isinstance(detail, str): text = detail.strip() if text: return text text_value = getattr(response, "text", None) if isinstance(text_value, bytes): try: text_value = text_value.decode("utf-8", "ignore") except Exception: text_value = str(text_value) if isinstance(text_value, str): text = text_value.strip() if text: return text return None @staticmethod def _summarise_response_text(response: Any) -> str: text = getattr(response, "text", "") if not isinstance(text, str): text = str(text) text = text.replace("\n", "\\n") return text @staticmethod def _extract_cookie_from_response( response: Any, cookie_name: str ) -> Optional[str]: if response is None: return None cookie_key = str(cookie_name or "").strip() if not cookie_key: return None try: cookies = getattr(response, "cookies", None) if cookies is not None: value = cookies.get(cookie_key) if value is not None: text = str(value).strip() if text: return text except Exception: pass headers = getattr(response, "headers", None) if headers: raw_set_cookie = headers.get("set-cookie") or headers.get("Set-Cookie") if raw_set_cookie: cookie = SimpleCookie() try: cookie.load(raw_set_cookie) except Exception: pass else: morsel = cookie.get(cookie_key) if morsel: text = str(morsel.value).strip() if text: return text return None @staticmethod def _extract_csrf_token_from_response(response: Any) -> Optional[str]: return XGenWorker._extract_cookie_from_response(response, "ct0") @staticmethod def _extract_session_id(response: Any) -> Optional[str]: if isinstance(response, str): return response.strip() or None if not isinstance(response, dict): return None keys_to_check: Iterable[str] = ( "sessionId", "sessionID", "session_id", "SessionId", "taskId", "taskID", "task_id", ) for key in keys_to_check: value = response.get(key) if value: return str(value) for nested_key in ("data", "result", "task"): nested = response.get(nested_key) if isinstance(nested, dict): nested_id = XGenWorker._extract_session_id(nested) if nested_id: return nested_id return None @staticmethod def _is_fatal_error(result: Any) -> bool: if not isinstance(result, dict): return False fatal = str(result.get("fatal_error") or "").strip() if fatal: return True success = result.get("success") task_type = str(result.get("taskType") or "").strip().lower() status = str(result.get("status") or "").strip().lower() pending_states = { "processing", "pending", "in_progress", "inprogress", "queued", "waiting", "working", } fatal_states = { "failed", "fail", "error", "fatal", "timeout", "timed_out", "expired", "stopped", } if task_type in pending_states or status in pending_states: return False if task_type in fatal_states or status in fatal_states: return True if isinstance(success, bool) and success is False: return True return False @staticmethod def _is_task_completed(result: Any) -> bool: if not isinstance(result, dict): return False success = result.get("success") task_type = str(result.get("taskType") or "").lower() status = str(result.get("status") or "").lower() if task_type in {"solved", "completed", "done"}: return True if status in {"solved", "completed", "done", "finished", "success"}: return True if success is True and (result.get("token") or result.get("answer")): return True return False def run_worker(worker_id: int, config: Dict[str, Any]) -> None: """İlgili config ile işçiyi çalıştır.""" worker = XGenWorker(worker_id, config) worker.run() def create_worker_thread(config: Dict[str, Any], worker_id: int) -> threading.Thread: """Yeni bir işçi thread'i oluşturup başlat.""" thread = threading.Thread( target=run_worker, name=f"Thread{worker_id}", args=(worker_id, dict(config)), ) thread.start() return thread def start_worker_threads(config: Dict[str, Any], thread_count: int) -> List[threading.Thread]: """Verilen sayıda işçi thread'i başlat ve liste olarak döndür.""" threads: List[threading.Thread] = [] for index in range(thread_count): worker_id = index + 1 thread = create_worker_thread(config, worker_id) threads.append(thread) return threads def _mask_proxy_for_log(proxy_url: str) -> str: if not proxy_url: return proxy_url cleaned = proxy_url if "@" in proxy_url: host_part = proxy_url.rsplit("@", 1)[-1] cleaned = f"***:***@{host_part}" return cleaned __all__ = [ "ANYMESSAGE_POLICY_AUTO", "ANYMESSAGE_POLICY_KEY", "ANYMESSAGE_POLICY_PURCHASE_ONLY", "ANYMESSAGE_POLICY_REUSE_ONLY", "ANYMESSAGE_STOCK_THRESHOLD", "ANYMESSAGE_STOCK_SITE", "CONFIG_PATH", "LOGGER", "XGenRequestContext", "XGenWorker", "build_proxy_url", "build_proxies", "is_debug_logging_enabled", "load_config", "get_global_shutdown_reason", "is_global_shutdown_requested", "run_worker", "request_global_shutdown", "request_global_resource_wait", "start_worker_threads", "create_worker_thread", "wait_for_global_shutdown", "has_available_rereg_email", "count_available_rereg_email", "clear_global_resource_wait", "is_global_resource_wait_requested", ] def _build_cookie_header(cookies: Dict[str, str]) -> str: return "; ".join(f"{key}={value}" for key, value in cookies.items()) def _load_display_names(path: Path = USERNAME_FILE) -> List[str]: global _DISPLAY_NAME_CACHE with _DISPLAY_NAME_LOCK: if _DISPLAY_NAME_CACHE is not None: return list(_DISPLAY_NAME_CACHE) try: content = path.read_text(encoding="utf-8") except FileNotFoundError: _DISPLAY_NAME_CACHE = [] return [] names = [line.strip() for line in content.splitlines() if line.strip()] _DISPLAY_NAME_CACHE = names return list(names) def _search_value(payload: Any, *candidates: str) -> Optional[Any]: if not candidates: return None target_keys = {candidate.lower() for candidate in candidates if candidate} if not target_keys: return None stack: List[Any] = [payload] while stack: current = stack.pop() if isinstance(current, dict): for key, value in current.items(): key_lower = str(key).lower() if key_lower in target_keys: return value if isinstance(value, (dict, list, tuple)): stack.append(value) elif isinstance(current, (list, tuple)): stack.extend(current) return None def _replace_flow_token_in_response( payload: Dict[str, Any], new_token: str ) -> Dict[str, Any]: if not isinstance(payload, dict): return payload if not isinstance(new_token, str) or not new_token.strip(): return copy.deepcopy(payload) original_value = _search_value(payload, "flow_token", "flowtoken") old_token = str(original_value).strip() if isinstance(original_value, str) else None cloned = copy.deepcopy(payload) _apply_flow_token_replacement(cloned, new_token.strip(), old_token) return cloned def _apply_flow_token_replacement( value: Any, replacement: str, original: Optional[str] ) -> Any: token_keys = {"flow_token", "flowtoken"} if isinstance(value, dict): for key in list(value.keys()): key_lower = str(key).lower() if key_lower in token_keys: value[key] = replacement continue value[key] = _apply_flow_token_replacement( value[key], replacement, original ) return value if isinstance(value, list): for index, item in enumerate(value): value[index] = _apply_flow_token_replacement(item, replacement, original) return value if isinstance(value, tuple): return tuple( _apply_flow_token_replacement(item, replacement, original) for item in value ) if isinstance(value, str) and original: return value.replace(original, replacement) return value def _normalise_username(value: Any) -> str: if isinstance(value, str): text = value.strip() if text: return text return "" def _swap_random_adjacent(text: str) -> str: """Swap a random character with its neighbor to slightly randomize display names.""" if len(text) < 2: return text if len(text) == 2: return text[::-1] index = secrets.randbelow(len(text)) if index == 0: swap_index = 1 elif index == len(text) - 1: swap_index = index - 1 else: swap_index = index - 1 if secrets.choice((True, False)) else index + 1 chars = list(text) chars[index], chars[swap_index] = chars[swap_index], chars[index] return "".join(chars) def _insert_separator(text: str, separator: str) -> str: if len(text) < 2: return text positions = [ idx for idx in range(1, len(text)) if text[idx - 1].isalnum() and text[idx].isalnum() ] if not positions: positions = list(range(1, len(text))) if not positions: return text insert_at = random.choice(positions) return f"{text[:insert_at]}{separator}{text[insert_at:]}" def _inject_display_name_separators(text: str) -> str: cleaned = text.strip() if len(cleaned) < 2: return cleaned has_space = " " in cleaned has_word_sep = any(sep in cleaned for sep in (".", "_", "-")) if has_space and has_word_sep: return cleaned result = cleaned if not has_word_sep: result = _insert_separator(result, random.choice((".", "_", "-"))) if not has_space: result = _insert_separator(result, " ") return result def _persist_username(username: str) -> None: """username.txt dosyasının üzerine yazmamak için artık no-op.""" if isinstance(username, str): _ = username.strip() def _persist_account_record( username: str, password: str, email_id: str, email_address: str, auth_token: Optional[str] = None, csrf_token: Optional[str] = None, ) -> None: username_text = username.strip() password_text = password.strip() if not username_text or not password_text: return email_id_text = email_id.strip() email_address_text = email_address.strip() auth_token_text = auth_token.strip() if isinstance(auth_token, str) else "" csrf_token_text = csrf_token.strip() if isinstance(csrf_token, str) else "" key = f"{username_text}:{password_text}" with _ACCOUNT_RECORD_LOCK: record = _ACCOUNT_RECORD_BUFFER.get( key, { "username": username_text, "password": password_text, "email_id": email_id_text, "email_address": email_address_text, }, ) record["username"] = username_text record["password"] = password_text if email_id_text or "email_id" not in record: record["email_id"] = email_id_text if email_address_text or "email_address" not in record: record["email_address"] = email_address_text if auth_token_text: record["auth_token"] = auth_token_text if csrf_token_text: record["csrf_token"] = csrf_token_text _ACCOUNT_RECORD_BUFFER[key] = record if not record.get("auth_token") or not record.get("csrf_token"): return base_line = ( f"{record['username']}:{record['password']}:{record.get('email_id', '')}" f"|{record.get('email_address', '')}" ) token_segments = [ value for value in ( record.get("auth_token", ""), record.get("csrf_token", ""), ) if value ] if token_segments: base_line += ":" + ":".join(token_segments) line = base_line + "\n" should_send_panel = bool(record.get("auth_token") and record.get("csrf_token")) if should_send_panel and record.get("_panel_sent"): should_send_panel = False panel_payload = dict(record) if should_send_panel else None persisted = False try: ACCOUNT_LOG_FILE.parent.mkdir(parents=True, exist_ok=True) existing_lines: List[str] = [] if ACCOUNT_LOG_FILE.exists(): with ACCOUNT_LOG_FILE.open("r", encoding="utf-8") as log_file: existing_lines = log_file.readlines() prefix = f"{record['username']}:{record['password']}:" updated_lines: List[str] = [] replaced = False for existing in existing_lines: if existing.startswith(prefix): updated_lines.append(line) replaced = True else: updated_lines.append(existing) if not replaced: updated_lines.append(line) with ACCOUNT_LOG_FILE.open("w", encoding="utf-8") as log_file: log_file.writelines(updated_lines) persisted = True except Exception: # pragma: no cover - dosya izin hataları log için LOGGER.getChild("account").debug( "Hesap bilgileri dosyaya kaydedilemedi.", exc_info=True, ) finally: if persisted: _ACCOUNT_RECORD_BUFFER.pop(key, None) if panel_payload: record["_panel_sent"] = True send_account_to_panel(panel_payload) def _atomic_write_lines(file_path: Path, lines: Iterable[str]) -> None: temp_path = file_path.with_suffix(file_path.suffix + ".tmp") content = "\n".join(lines) + "\n" try: with temp_path.open("w", encoding="utf-8") as handle: handle.write(content) temp_path.replace(file_path) except OSError: try: if temp_path.exists(): temp_path.unlink() except OSError: pass raise def _persist_email_entry( file_path: Path, lock: threading.Lock, *, email_id: str, email_address: str, service: str, api_key: str, verification_code: Optional[str] = None, used_dot_positions: Optional[Iterable[Iterable[int]]] = None, max_uses: int = 0, initial_count: int = 1, increment_on_update: bool = True, persist_usage: bool = True, ) -> None: email_id_text = str(email_id or "").strip() email_address_text = str(email_address or "").strip() original_email_address = email_address_text initial_dot_positions = extract_gmail_dot_positions(original_email_address) if not email_id_text and not email_address_text: return if email_address_text and not _is_gmail_address(email_address_text): return email_address_storage = remove_dots_from_local_part(email_address_text) email_address_text = email_address_storage or email_address_text service_text = str(service or "").strip().lower() api_key_text = str(api_key or "").strip() verification_code_text: Optional[str] = None if verification_code is not None: verification_code_text = str(verification_code).strip() or None timestamp = int(time.time()) limit = max(0, int(max_uses)) if persist_usage else 0 base_count = 0 if persist_usage: try: base_count = int(initial_count) except (TypeError, ValueError): base_count = 0 base_count = max(0, base_count) with lock: entries: List[Dict[str, Any]] = [] if file_path.exists(): try: with file_path.open("r", encoding="utf-8") as handle: for line in handle: stripped = line.strip() if not stripped: continue try: payload = json.loads(stripped) except json.JSONDecodeError: continue if isinstance(payload, dict): record = dict(payload) current_address = str( record.get("email_address") or "" ).strip() normalized_address = remove_dots_from_local_part( current_address ) if normalized_address: record["email_address"] = normalized_address if not persist_usage: record.pop("count", None) record.pop("last_used", None) entries.append(record) except OSError: pass updated_entries: List[Dict[str, Any]] = [] found = False for record in entries: existing_id = str(record.get("email_id") or "").strip() existing_address = str(record.get("email_address") or "").strip() existing_count = 0 if persist_usage: try: existing_count = int(record.get("count", 0)) except (TypeError, ValueError): existing_count = 0 if persist_usage and limit and existing_count >= limit: continue is_target = False if email_id_text and existing_id == email_id_text: is_target = True elif not email_id_text and email_address_text and existing_address == email_address_text: is_target = True if is_target and not found: found = True if email_id_text: record["email_id"] = email_id_text if email_address_text: record["email_address"] = email_address_text if service_text: record["service"] = service_text if api_key_text: record["api_key"] = api_key_text if verification_code_text: record["verification_code"] = verification_code_text merged_positions = _merge_used_dot_positions( record.get("used_dot_positions"), initial_dot_positions ) if used_dot_positions: merged_positions = _merge_used_dot_positions( merged_positions, used_dot_positions ) if merged_positions: record["used_dot_positions"] = merged_positions elif "used_dot_positions" in record: record.pop("used_dot_positions", None) if persist_usage: updated_count = max(0, existing_count) if increment_on_update: updated_count += 1 record["count"] = updated_count record["last_used"] = timestamp if limit and updated_count >= limit: continue updated_entries.append(record) if not found: new_record = { "email_id": email_id_text, "email_address": email_address_text, "service": service_text, "api_key": api_key_text, } if verification_code_text: new_record["verification_code"] = verification_code_text if persist_usage: new_record["count"] = base_count new_record["last_used"] = timestamp initial_history = _merge_used_dot_positions([], initial_dot_positions) if used_dot_positions: initial_history = _merge_used_dot_positions( initial_history, used_dot_positions ) if initial_history: new_record["used_dot_positions"] = initial_history if not (persist_usage and limit and new_record.get("count", 0) >= limit): updated_entries.append(new_record) if not updated_entries: if file_path.exists(): try: file_path.unlink() except OSError: pass return try: file_path.parent.mkdir(parents=True, exist_ok=True) except OSError: pass lines = [json.dumps(record, ensure_ascii=False) for record in updated_entries] try: _atomic_write_lines(file_path, lines) except OSError: pass def _persist_email_reregistration_info( email_id: str, email_address: str, service: str, api_key: str, verification_code: Optional[str] = None, *, max_uses: int = 0, ) -> None: _persist_email_entry( EMAIL_REREG_LOG_FILE, _EMAIL_REREG_LOG_LOCK, email_id=email_id, email_address=email_address, service=service, api_key=api_key, verification_code=verification_code, max_uses=max_uses, initial_count=1, increment_on_update=True, ) def _remove_email_entry_from_file( file_path: Path, lock: threading.Lock, *, email_id: str, email_address: str, ) -> None: email_id_text = str(email_id or "").strip() email_address_text = str(email_address or "").strip() normalized_target_address = remove_dots_from_local_part(email_address_text) if normalized_target_address: email_address_text = normalized_target_address if not email_id_text and not email_address_text: return if not file_path.exists(): return with lock: try: content = file_path.read_text(encoding="utf-8") except OSError: return lines = content.splitlines() if not lines: return updated_lines: List[str] = [] removed = False for line in lines: stripped = line.strip() if not stripped: continue try: record = json.loads(stripped) except json.JSONDecodeError: updated_lines.append(line) continue existing_id = str(record.get("email_id") or "").strip() existing_address = str(record.get("email_address") or "").strip() normalized_existing = remove_dots_from_local_part(existing_address) if normalized_existing: existing_address = normalized_existing matches_id = email_id_text and existing_id == email_id_text matches_address = email_address_text and existing_address == email_address_text if matches_id or matches_address: removed = True continue updated_lines.append(line) if not removed: return if not updated_lines: try: file_path.unlink() except FileNotFoundError: pass except OSError: pass return try: _atomic_write_lines(file_path, updated_lines) except OSError: pass def _remove_email_from_retry_logs(email_id: str, email_address: str) -> None: _remove_email_entry_from_file( EMAIL_REREG_LOG_FILE, _EMAIL_REREG_LOG_LOCK, email_id=email_id, email_address=email_address, ) def _normalise_cookie_map(raw_cookie: Any) -> Dict[str, str]: cookie_data: Any = raw_cookie if isinstance(raw_cookie, str): text = raw_cookie.strip() if not text: raise ValueError("Captcha yaniti gecersiz cookie degeri dondurdu") decode_error: Optional[Exception] = None try: cookie_data = json.loads(text) except json.JSONDecodeError as exc: # pragma: no cover - loglama amacli decode_error = exc fallback_text = None try: # Bazi solver'lar JSON'u kacisli string olarak donebiliyor; once stringi coz. fallback_text = json.loads(f'"{text}"') except Exception: fallback_text = None if isinstance(fallback_text, str): try: cookie_data = json.loads(fallback_text) decode_error = None except Exception: decode_error = exc if decode_error is not None: raise ValueError("Captcha cookie JSON verisi cozumlenemedi") from decode_error if isinstance(cookie_data, list): cookie_dict: Dict[str, str] = {} for item in cookie_data: if isinstance(item, dict): for key, value in item.items(): if value is not None: cookie_dict[str(key)] = str(value) elif isinstance(item, str) and "=" in item: name, _, value = item.partition("=") cookie_dict[name] = value cookie_data = cookie_dict if not isinstance(cookie_data, dict): raise ValueError("Captcha yaniti desteklenmeyen cookie formati dondurdu") normalised: Dict[str, str] = {} for key, value in cookie_data.items(): if value is None: continue normalised[str(key)] = str(value) if not normalised: raise ValueError("Captcha yanitindaki cookie bilgisi bos") return normalised def _normalise_sec_ch_ua(raw_value: Any) -> str: def _format_from_list(items: Iterable[Any]) -> str: parts: List[str] = [] for item in items: if isinstance(item, dict): brand = item.get("brand") version = item.get("version") if not brand or version is None: continue brand_text = str(brand) version_text = str(version) if not brand_text: continue escaped_brand = brand_text.replace("\"", r"\"") parts.append(f'"{escaped_brand}";v="{version_text}"') elif isinstance(item, str): text = item.strip() if text: parts.append(text) return ", ".join(parts) if raw_value is None: return "" if isinstance(raw_value, str): text = raw_value.strip() if not text: return "" def _try_format(candidate: str) -> str: if not candidate.startswith("[") or not candidate.endswith("]"): return "" try: parsed = json.loads(candidate) except json.JSONDecodeError: return "" formatted = _format_from_list(parsed if isinstance(parsed, list) else [parsed]) return formatted formatted_text = _try_format(text) if formatted_text: return formatted_text if "\\\"" in text: try: unescaped = json.loads(f'"{text}"') except json.JSONDecodeError: unescaped = "" if unescaped: formatted_text = _try_format(unescaped) if formatted_text: return formatted_text text = str(unescaped).strip() cleaned_text = text.replace("\\\"", "\"").replace("\\\\", "\\") return cleaned_text if isinstance(raw_value, list): formatted = _format_from_list(raw_value) if formatted: return formatted if isinstance(raw_value, dict): formatted = _format_from_list([raw_value]) if formatted: return formatted return str(raw_value).strip() def _merge_used_dot_positions( existing: Any, additional: Any ) -> List[List[int]]: merged: List[List[int]] = [] seen: Set[Tuple[int, ...]] = set() def _add_pattern(pattern: Iterable[Any]) -> None: cleaned: Set[int] = set() for value in pattern: try: numeric = int(value) except (TypeError, ValueError): continue if numeric >= 0: cleaned.add(numeric) if not cleaned: return key = tuple(sorted(cleaned)) if key in seen: return seen.add(key) merged.append(list(key)) def _add_patterns(value: Any) -> None: if value is None: return if isinstance(value, list): if value and all( isinstance(item, Iterable) and not isinstance(item, (str, bytes)) for item in value ): for item in value: _add_pattern(item) return _add_pattern(value) if isinstance(existing, list): for item in existing: if isinstance(item, Iterable) and not isinstance(item, (str, bytes)): _add_pattern(item) _add_patterns(additional) return merged class _NodeHTTPResponse: def __init__( self, status_code: Any, reason: Optional[str], headers: Optional[Dict[str, Any]], text: Any, ) -> None: try: self.status_code = int(status_code) except (TypeError, ValueError): self.status_code = 0 self.reason = reason or None self.headers = dict(headers or {}) self.text = text if isinstance(text, str) else str(text or "") self.cookies: Optional[RequestsCookieJar] = None self._json_cache: Optional[Any] = None self._json_error: Optional[Exception] = None def json(self) -> Any: if self._json_error is not None: raise self._json_error if self._json_cache is not None: return self._json_cache try: self._json_cache = json.loads(self.text) except Exception as exc: self._json_error = exc raise return self._json_cache @dataclass class _CallbackPayload: flow_token: str sec_ch_ua: str languages: str user_agent: str cookies: Dict[str, str] guest_token: str guest_id: str @property def encoded_guest_id(self) -> str: guest_id = self.guest_id.strip() if guest_id.lower().startswith("v1:"): prefix, _, remainder = guest_id.partition(":") if remainder: return f"{prefix}%3A{remainder}" return guest_id @property def cookie_header(self) -> str: return _build_cookie_header(self.cookies) class XGenRequestContext: def __init__(self, session: Any): self.session = session