import copy import json import os import queue import random import re import threading import time from dataclasses import dataclass from typing import Any, Dict, List, Optional, Set, Tuple import requests # ========================= # Config / Debug # ========================= BASE_DIR = os.path.abspath(os.path.dirname(__file__)) LOG_DIR = os.path.join(BASE_DIR, "logs") CONFIG_PATH = os.path.join(BASE_DIR, "config.json") DEFAULT_CONFIG = { "debug": True, "proxy": {"enabled": False, "url": ""}, "max_threads": 1, "auth_token_checker": False, } def _coerce_positive_int(value: Any, default: int) -> int: try: number = int(value) except (TypeError, ValueError): return default return number if number > 0 else default def _load_config() -> Dict[str, Any]: config: Dict[str, Any] = copy.deepcopy(DEFAULT_CONFIG) try: with open(CONFIG_PATH, "r", encoding="utf-8") as handle: loaded = json.load(handle) if isinstance(loaded, dict): for key in DEFAULT_CONFIG: if key not in loaded: continue if key == "proxy" and isinstance(config.get(key), dict): proxy_loaded = loaded.get(key) if isinstance(proxy_loaded, dict): for proxy_key in config[key]: if proxy_key in proxy_loaded: config[key][proxy_key] = proxy_loaded[proxy_key] continue config[key] = loaded[key] except FileNotFoundError: pass except Exception as exc: print(f"[config] failed to read config.json: {exc}") return config CONFIG = _load_config() DEBUG = bool(CONFIG.get("debug", True)) MAX_THREADS = _coerce_positive_int(CONFIG.get("max_threads"), DEFAULT_CONFIG["max_threads"]) AUTH_TOKEN_CHECKER = bool(CONFIG.get("auth_token_checker", DEFAULT_CONFIG["auth_token_checker"])) SENSITIVE_HDR_KEYS = {"authorization", "cookie", "x-csrf-token", "x-guest-token"} # ========================= # Models # ========================= @dataclass class SettingsModel: proxy_enabled: bool = False proxy_url: Optional[str] = None @dataclass class UserNormal: username: str password: str mail_raw: str mail: str mail_pass: str two_factor_code: str ct0: str auth_token: str @dataclass class UserToken: username: str ct0: str auth_token: str # ========================= # Global state # ========================= settings = SettingsModel() proxy_cfg_raw = CONFIG.get("proxy") proxy_cfg = proxy_cfg_raw if isinstance(proxy_cfg_raw, dict) else {} settings.proxy_enabled = bool(proxy_cfg.get("enabled", False)) proxy_value = proxy_cfg.get("url") settings.proxy_url = ( proxy_value.strip() if isinstance(proxy_value, str) and proxy_value.strip() else None ) semaphore = threading.Semaphore(MAX_THREADS) token_lock = threading.Lock() counter_lock = threading.Lock() results_lock = threading.Lock() valid_file_lock = threading.Lock() locked_file_lock = threading.Lock() suspended_file_lock = threading.Lock() base_dir = BASE_DIR pool_path = os.path.join(base_dir, "havuz.txt") user_path = os.path.join(base_dir, "users.txt") valid_user_path = os.path.join(base_dir, "valid.txt") locked_user_path = os.path.join(base_dir, "lock.txt") suspended_user_path = os.path.join(base_dir, "suspend.txt") user_token_list: List[UserToken] = [] user_normal_list: List[UserNormal] = [] user_raw_map: Dict[str, str] = {} guest_token: Optional[str] = None guest_cookies: str = "" success_count = 0 suspended_count = 0 access_count = 0 scan_results: List[Dict[str, str]] = [] # ========================= # Helpers # ========================= COLOR_RESET = "\033[0m" COLOR_GREEN = "\033[32m" COLOR_RED = "\033[31m" COLOR_YELLOW = "\033[33m" STATUS_COLOR_MAP = { "ok": COLOR_GREEN, "locked": COLOR_YELLOW, "suspended": COLOR_RED, } def color_text(text: str, color: str) -> str: return f"{color}{text}{COLOR_RESET}" def colorize_by_status(text: str, status_key: str) -> str: color = STATUS_COLOR_MAP.get(status_key) if not color: return text return color_text(text, color) def print_message(message: str) -> None: print(message) def build_proxy_url() -> Optional[str]: if not settings.proxy_enabled or not settings.proxy_url: return None if settings.proxy_url.startswith("http://") or settings.proxy_url.startswith("https://"): return settings.proxy_url parts = settings.proxy_url.split(":") if len(parts) < 2: return None host, port = parts[0], parts[1] credentials = "" if len(parts) >= 4: user, password = parts[2], parts[3] credentials = f"{user}:{password}@" return f"http://{credentials}{host}:{port}" def build_proxy_dict() -> Optional[Dict[str, str]]: proxy_url = build_proxy_url() if proxy_url: return {"http": proxy_url, "https": proxy_url} return None def update_console_title(title: str) -> None: try: print(f"\x1b]0;{title}\x07", end="", flush=True) except Exception: pass def _ensure_log_dir(): try: os.makedirs(LOG_DIR, exist_ok=True) except Exception: pass def _redact_headers(h: Dict[str, str]) -> Dict[str, str]: if not h: return {} redacted = {} for k, v in h.items(): if k.lower() in SENSITIVE_HDR_KEYS: redacted[k] = "***redacted***" else: redacted[k] = v return redacted def dump_http_debug(resp: requests.Response, label: str, username: str) -> None: """ Persist the response to disk and print a short summary. Sensitive headers are redacted. """ if not DEBUG: return try: _ensure_log_dir() ts = time.strftime("%Y%m%d-%H%M%S") safe_user = re.sub(r"[^a-zA-Z0-9_.-]+", "_", username or "unknown") # Short summary on console print_message(f"[DEBUG] {label} | {resp.status_code} {resp.reason} | {resp.url}") # File names meta_path = os.path.join(LOG_DIR, f"{ts}_{safe_user}_{label}_meta.json") body_path = os.path.join(LOG_DIR, f"{ts}_{safe_user}_{label}_body.txt") # Meta (headers, url, method, vs.) try: meta = { "label": label, "status_code": resp.status_code, "reason": resp.reason, "url": resp.url, "method": getattr(resp.request, "method", None), "request_headers": _redact_headers(getattr(resp.request, "headers", {})), "response_headers": dict(resp.headers or {}), } with open(meta_path, "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2) except Exception as e: print_message(f"[DEBUG] failed to write meta: {e}") # Body (ham metin) try: text = resp.text with open(body_path, "w", encoding="utf-8") as f: f.write(text if text is not None else "") print_message( f"[DEBUG] Saved: {os.path.basename(meta_path)}, {os.path.basename(body_path)}" ) except Exception as e: print_message(f"[DEBUG] failed to write body: {e}") except Exception as e: print_message(f"[DEBUG] dump error: {e}") # ========================= # Data Loading # ========================= def record_scan_result(username: str, status: str, detail: str = "") -> None: entry = { "username": username, "status": status, "detail": detail, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), } with results_lock: scan_results.append(entry) def load_pool() -> None: user_token_list.clear() if not os.path.exists(pool_path): print_message("havuz.txt not found.") return with open(pool_path, "r", encoding="utf-8") as handle: for raw_line in handle: line = raw_line.strip() if not line: continue parts = line.split(":") if len(parts) > 6: user_token_list.append( UserToken(username=parts[0], ct0=parts[5], auth_token=parts[6]) ) def _normalize_token_pair(ct0_raw: str, auth_raw: str) -> Tuple[str, str]: """Auto-corrects ct0/auth_token order based on length.""" ct0_clean = (ct0_raw or "").strip() auth_clean = (auth_raw or "").strip() # Based on user feedback, ct0 is always longer. # If ct0 is shorter, assume the fields are swapped and fix the order. if ct0_clean and auth_clean and len(ct0_clean) < len(auth_clean): ct0_clean, auth_clean = auth_clean, ct0_clean return ct0_clean, auth_clean def load_user_data() -> None: """ Parses users.txt while accepting multiple formats: 7 parts (full format): username:password:mail:mail_pass:two_factor_code:ct0:auth_token 6 parts (no 2FA): username:password:mail:mail_pass:ct0:auth_token -> two_factor_code = "" 5 parts (phone|email + ct0 + auth_token): username:password:(phone|email):ct0:auth_token -> mail_raw = field as-is, mail = email (if after pipe), mail_pass = "", two_factor_code = "" """ user_normal_list.clear() user_raw_map.clear() if not os.path.exists(user_path): print_message("users.txt not found.") return def pick_email(s: str) -> str: # "68167545006|andersonfield...@gmail.com" -> "andersonfield...@gmail.com" if "|" in s: right = s.split("|", 1)[1].strip() return right or s return s with open(user_path, "r", encoding="utf-8") as handle: for raw_line in handle: line = raw_line.strip() if not line: continue parts = [seg for seg in re.split(r"[:;]", line) if seg != ""] try: if len(parts) == 7: username, password, mail_raw, mail_pass, twofa, ct0, auth = parts mail = mail_raw elif len(parts) == 6: username, password, mail_raw, mail_pass, auth, ct0 = parts mail = mail_raw twofa = "" elif len(parts) == 5: username, password, mail_mixed, auth, ct0 = parts mail_raw = mail_mixed mail = pick_email(mail_mixed) mail_pass = "" twofa = "" else: print_message(f"[users.txt] Skipped malformed line: {line}") continue ct0_clean, auth_clean = _normalize_token_pair(ct0, auth) if not username or not ct0_clean or not auth_clean: print_message(f"[users.txt] Missing required field(s): {line}") continue username_clean = username.strip() user_raw_map[username_clean] = line user_normal_list.append( UserNormal( username=username_clean, password=(password or "").strip(), mail_raw=(mail_raw or "").strip(), mail=(mail or "").strip(), mail_pass=(mail_pass or "").strip(), two_factor_code=(twofa or "").strip(), ct0=ct0_clean, auth_token=auth_clean, ) ) except Exception: print_message(f"[users.txt] Parse error: {line}") continue # ========================= # Utilities # ========================= def get_random_user_token() -> Optional[UserToken]: if not user_token_list: return None return random.choice(user_token_list) def get_token(max_retries: int = 3) -> None: """Guest token flow (may be required for 429 responses).""" global guest_token, guest_cookies url = "https://api.x.com/1.1/guest/activate.json" proxies = build_proxy_dict() headers = { "host": "api.x.com", "sec-ch-ua": '"Google Chrome";v="142", "Chromium";v="142", "Not.A/Brand";v="24"', "x-twitter-client-language": "en", "sec-ch-ua-mobile": "?0", "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36", "x-client-transaction-id": "pNF52uL9wJ1XPYZt5OadUHbzeHmgoLm7ZoT8fwew5v80MBVug7uBTwAE4fBgtxJKUyB7v6Yz2L3F4OkeGoHr4ydcc8nYpw", "x-twitter-active-user": "yes", "sec-ch-ua-platform": '"Windows"', "accept": "*/*", "origin": "https://x.com", "sec-fetch-site": "same-site", "sec-fetch-mode": "cors", "sec-fetch-dest": "empty", "referer": "https://x.com/", "accept-language": "en-US,en;q=0.9", "priority": "u=1, i", } for attempt in range(1, max_retries + 1): try: response = requests.post( url, headers=headers, data="", timeout=10, proxies=proxies ) if response.status_code == 429: print_message("[+] Waiting for rate limit window...") time.sleep(300) continue response.raise_for_status() payload = response.json() guest_token = payload.get("guest_token") cookies = response.cookies guest_cookies = "; ".join(f"{c.name}={c.value}" for c in cookies) return except Exception as exc: if attempt >= max_retries: raise RuntimeError("Failed to obtain token after maximum retries.") from exc time.sleep(1) def append_valid_user(user: UserNormal) -> None: os.makedirs(os.path.dirname(valid_user_path), exist_ok=True) with valid_file_lock: existing: Set[str] = set() if os.path.exists(valid_user_path): with open(valid_user_path, "r", encoding="utf-8") as handle: existing = {line.split(":", 1)[0] for line in handle if ":" in line} if user.username in existing: print_message(f"[!] {user.username} already recorded.") return mail_value = user.mail_raw.strip() if hasattr(user, "mail_raw") else user.mail mail_pass = (user.mail_pass or "").strip() twofa = (user.two_factor_code or "").strip() password = (user.password or "").strip() ct0 = (user.ct0 or "").strip() auth = (user.auth_token or "").strip() if not mail_pass and not twofa: fields = [user.username, password, mail_value, ct0, auth] elif not twofa: fields = [user.username, password, mail_value, mail_pass, ct0, auth] else: fields = [user.username, password, mail_value, mail_pass, twofa, ct0, auth] line = ":".join(fields) with open(valid_user_path, "a", encoding="utf-8") as handle: handle.write(f"{line}\n") def append_locked_user(username: str) -> None: try: entry = user_raw_map.get(username, username) with locked_file_lock: with open(locked_user_path, "a", encoding="utf-8") as handle: handle.write(f"{entry}\n") except Exception as exc: if DEBUG: print_message(f"[DEBUG] failed to write lock.txt: {exc}") def append_suspended_user(username: str) -> None: try: entry = user_raw_map.get(username, username) with suspended_file_lock: with open(suspended_user_path, "a", encoding="utf-8") as handle: handle.write(f"{entry}: Account Suspended!\n") except Exception as exc: if DEBUG: print_message(f"[DEBUG] failed to write suspend.txt: {exc}") # ========================= # Core logic (TXT ONLY) # ========================= def build_common_headers(random_token: UserToken, username: str, transaction_id: str) -> Dict[str, str]: cookie_parts = [] if guest_cookies: cookie_parts.append(guest_cookies) cookie_parts.append(f"auth_token={random_token.auth_token}") cookie_parts.append(f"ct0={random_token.ct0}") cookies = "; ".join(cookie_parts) return { "host": "x.com", "sec-ch-ua-platform": '"Windows"', "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA", "x-csrf-token": random_token.ct0, "sec-ch-ua": '"Not)A;Brand";v="8", "Chromium";v="138"', "x-twitter-client-language": "en", "sec-ch-ua-mobile": "?0", "x-twitter-active-user": "yes", "x-client-transaction-id": transaction_id, "x-twitter-auth-type": "OAuth2Session", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", "content-type": "application/json", "accept": "*/*", "sec-fetch-site": "same-origin", "sec-fetch-mode": "cors", "sec-fetch-dest": "empty", "referer": f"https://x.com/{username}", "accept-language": "en-US,en;q=0.9", "accept-encoding": "gzip, deflate, br", "priority": "u=1, i", "x-xp-forwarded-for": "a50c0429593aac9f16831ccc8d4b47c5dccc4489073b9175332228e50fffa880da98c34e2455d4153b452446268497f83a298b4ec788073ce89d50f9c690baa2eb6f23f4ac9dadcaf38ab466dcca127408662ac64744e0b5691a62b6b7c7b9d4d06ef5532f6a3454f496543f8c9085a039d9563875364a659594dc0d897b86e9036f85663a927ce8dc110ad19fc2b6ed4f65f4f946855e9dc0606d5ac9e3536dea687a8613d0b98c5b6cd133c3c34648f97afc10b881457f27ab20495f84f9c7fafac53c0bdf299e78d3289a4138fcdb9b96450dd57700e8b9ef4e728f105be1445dbc5c94c6a93bc10123ef79a24334e19ee66af07d159ebd5735", "cookie": cookies, } def handle_status(username: str, result: Dict) -> None: global success_count, suspended_count, access_count reason = result.get("reason") legacy = result.get("legacy", {}) profile_type = legacy.get("profile_interstitial_type") if reason == "Suspended": print(color_text(f"[*] {username}: Account Suspended!", COLOR_RED)) with counter_lock: suspended_count += 1 append_suspended_user(username) record_scan_result(username, "suspended") return if profile_type == "sensitive_media" or profile_type == "": print(color_text(f"[+] {username}: Account OK!", COLOR_GREEN)) with counter_lock: success_count += 1 record_scan_result(username, "ok") return if profile_type in {"fake_account", "offensive_profile_content"}: print(color_text(f"[-] {username}: Account Locked!", COLOR_YELLOW)) with counter_lock: access_count += 1 append_locked_user(username) record_scan_result(username, "locked") return print_message(f"[?] {username}: Unable to determine status.") record_scan_result(username, "unknown_status") def checker_normal(user: UserNormal) -> None: if user is None: return if AUTH_TOKEN_CHECKER: tokens_to_try = [ UserToken(username=user.username, ct0=user.ct0, auth_token=user.auth_token) ] else: if not user_token_list: print_message("Pool data unavailable.") record_scan_result(user.username, "pool_unavailable", "havuz.txt data not found") return tokens_to_try = list(user_token_list) random.shuffle(tokens_to_try) for random_token in tokens_to_try: with semaphore: proxies = build_proxy_dict() variables = {"screen_name": user.username} features = { "responsive_web_grok_bio_auto_translation_is_enabled": False, "hidden_profile_subscriptions_enabled": True, "payments_enabled": False, "profile_label_improvements_pcf_label_in_post_enabled": True, "rweb_tipjar_consumption_enabled": True, "verified_phone_label_enabled": False, "subscriptions_verification_info_is_identity_verified_enabled": True, "subscriptions_verification_info_verified_since_enabled": True, "highlights_tweets_tab_ui_enabled": True, "responsive_web_twitter_article_notes_tab_enabled": True, "subscriptions_feature_can_gift_premium": True, "creator_subscriptions_tweet_preview_api_enabled": True, "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False, "responsive_web_graphql_timeline_navigation_enabled": True, } field_toggles = {"withAuxiliaryUserLabels": True} transaction_id = "OyACZxI/PzNoMGj+G35OeDsbtCI0ywIu15A4V8Oq7Z/Eb7T1gfEC8Z+poTLVeCs60+TnFz9vJFsJgWQkN0iEqCScMPD3OA" params = { "variables": json.dumps(variables, separators=(",", ":")), "features": json.dumps(features, separators=(",", ":")), "fieldToggles": json.dumps(field_toggles, separators=(",", ":")), } headers = build_common_headers(random_token, user.username, transaction_id) resp: Optional[requests.Response] = None try: resp = requests.get( "https://x.com/i/api/graphql/U15Q5V7hgjzCEg6WpSWhqg/UserByScreenName", params=params, headers=headers, timeout=6, proxies=proxies, ) status = resp.status_code if status == 429: print_message("[+] Rate limit bypass in progress!") dump_http_debug(resp, "rate_limit_429", user.username) record_scan_result(user.username, "rate_limited", "Refreshing guest token") with token_lock: get_token() return if status == 401: print_message("[!] Pool token unauthorized, will try next.") dump_http_debug(resp, "unauthorized_401", user.username) record_scan_result(user.username, "token_unauthorized") continue # Raise an exception for 4xx/5xx responses resp.raise_for_status() payload = resp.json() if payload.get("errors"): print_message("[!] GraphQL returned errors, skipping token.") dump_http_debug(resp, "graphql_errors", user.username) record_scan_result(user.username, "graphql_error") continue data = payload.get("data", {}) result = ((data.get("user") or {}).get("result")) or {} if result: handle_status(user.username, result) if result.get("reason") != "Suspended": profile_type = (result.get("legacy") or {}).get("profile_interstitial_type", "") if profile_type in {"sensitive_media", ""}: append_valid_user(user) return except requests.HTTPError as http_err: print_message(f"Error: {http_err}") if resp is not None: dump_http_debug(resp, "http_error", user.username) record_scan_result(user.username, "http_error", str(http_err)) except Exception as exc: print_message(f"Error: {exc}") # No response to dump; general exception. record_scan_result(user.username, "unexpected_error", str(exc)) finally: title = f"[SUCCESS:{success_count}] | [ACCESS:{access_count}] | [SUSPENDED:{suspended_count}] | [USERSCOUNT:{len(user_normal_list)}]" update_console_title(title) if AUTH_TOKEN_CHECKER: print_message( "[!] Account auth_token is unauthorized or invalid. Please update users.txt." ) record_scan_result(user.username, "auth_token_invalid") else: print_message("[!] All pool tokens were unauthorized. Please refresh havuz.txt.") record_scan_result(user.username, "all_tokens_unauthorized") # ========================= # Process loop (TXT ONLY) # ========================= def process_start_txt_only() -> None: """Iterates over every entry in users.txt once.""" if not user_normal_list: print_message("users.txt is empty or not found.") return total = len(user_normal_list) print_message(f"[INFO] Total accounts: {total}") users = list(user_normal_list) if MAX_THREADS <= 1: for idx, user in enumerate(users, start=1): print_message(f"[SCAN] {idx}/{total} -> {user.username}") checker_normal(user) time.sleep(0.5) # small delay return print_message(f"[INFO] Using {MAX_THREADS} threads for scanning.") job_queue: "queue.Queue[Optional[Tuple[int, UserNormal]]]" = queue.Queue() stop_event = threading.Event() def worker() -> None: while not stop_event.is_set(): try: item = job_queue.get(timeout=0.5) except queue.Empty: continue if item is None: job_queue.task_done() break idx, account = item try: print_message(f"[SCAN] {idx}/{total} -> {account.username}") checker_normal(account) except Exception as exc: print_message(f"[ERROR] Worker raised an exception: {exc}") finally: job_queue.task_done() threads: List[threading.Thread] = [] for index in range(MAX_THREADS): thread = threading.Thread(target=worker, name=f"x-checker-worker-{index + 1}", daemon=True) threads.append(thread) thread.start() try: for idx, account in enumerate(users, start=1): job_queue.put((idx, account)) time.sleep(0.05) finally: for _ in threads: job_queue.put(None) job_queue.join() stop_event.set() for thread in threads: thread.join() # ========================= # Final summary # ========================= def log_scan_summary(start_time: Optional[float]) -> None: end_time = time.time() with counter_lock: ok_count = success_count locked_count = access_count suspended_total = suspended_count with results_lock: results_copy = list(scan_results) total_duration = None if start_time is not None: total_duration = max(0.0, end_time - start_time) status_breakdown: Dict[str, int] = {} for entry in results_copy: status_breakdown[entry["status"]] = status_breakdown.get(entry["status"], 0) + 1 total_entries = len(user_normal_list) def format_percent(count: int, total: int) -> str: if total <= 0: return "0.00%" return f"{(count / total) * 100:.2f}%" summary_lines = ["=" * 60, "X Checker Scan Summary", "=" * 60] summary_lines.append( f"Start: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time or end_time))}" ) summary_lines.append(f"End: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))}") if total_duration is not None: summary_lines.append(f"Total Duration: {total_duration:.2f} seconds") summary_lines.append(f"users.txt total entries: {total_entries}") summary_lines.append("-- Counters --") summary_lines.append( colorize_by_status( f" Success: {ok_count} ({format_percent(ok_count, total_entries)})", "ok" ) ) summary_lines.append( colorize_by_status( f" Locked: {locked_count} ({format_percent(locked_count, total_entries)})", "locked", ) ) summary_lines.append( colorize_by_status( f" Suspended: {suspended_total} ({format_percent(suspended_total, total_entries)})", "suspended", ) ) if status_breakdown and DEBUG: summary_lines.append("-- Status Breakdown --") for status, count in sorted(status_breakdown.items()): summary_lines.append(colorize_by_status(f" {status}: {count}", status)) if results_copy and DEBUG: summary_lines.append("-- Details --") for entry in results_copy: detail = entry["detail"].strip() status_text = colorize_by_status(entry["status"], entry["status"]) if detail: summary_lines.append( f" {entry['timestamp']} | {entry['username']} | {status_text} | {detail}" ) else: summary_lines.append( f" {entry['timestamp']} | {entry['username']} | {status_text}" ) summary_text = "\n".join(summary_lines) print_message(summary_text) try: _ensure_log_dir() timestamp = time.strftime("%Y%m%d-%H%M%S", time.localtime(end_time)) log_path = os.path.join(LOG_DIR, f"scan_summary_{timestamp}.log") with open(log_path, "w", encoding="utf-8") as handle: handle.write(summary_text + "\n") print_message(f"[LOG] Scan summary saved: {log_path}") except Exception as exc: print_message(f"[LOG] Failed to write summary: {exc}") # ========================= # Entry point # ========================= def main() -> None: if DEBUG: print_message("[CONFIG] Debug mode enabled.") else: print_message("[CONFIG] User-focused mode enabled.") if settings.proxy_enabled and settings.proxy_url: print_message("[CONFIG] Proxy enabled.") else: print_message("[CONFIG] Proxy disabled.") print_message(f"[CONFIG] Thread workers: {MAX_THREADS}") print_message( "[CONFIG] Auth token checker: {}".format("Enabled" if AUTH_TOKEN_CHECKER else "Disabled") ) load_user_data() if AUTH_TOKEN_CHECKER: print_message("[CONFIG] Using per-account auth tokens from users.txt.") else: load_pool() scan_started_at = time.time() try: process_start_txt_only() except KeyboardInterrupt: print_message("Process cancelled.") finally: log_scan_summary(scan_started_at) if __name__ == "__main__": main()