From 035f697f380cf4cee657963764ab0dbec3e5b42f Mon Sep 17 00:00:00 2001 From: Emran Hejazi Date: Wed, 22 Apr 2026 11:20:01 +0330 Subject: [PATCH 1/2] Implement google candidate ips with a script that finds the fastest ip --- README.md | 44 +++++++++ README_FA.md | 44 +++++++++ main.py | 15 +++ src/constants.py | 33 +++++++ src/google_ip_scanner.py | 194 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 330 insertions(+) create mode 100644 src/google_ip_scanner.py diff --git a/README.md b/README.md index ceff8f3..fea8d5b 100644 --- a/README.md +++ b/README.md @@ -313,10 +313,53 @@ python3 main.py --log-level DEBUG # Show detailed logs python3 main.py -c /path/to/config.json # Use a different config file python3 main.py --install-cert # Install MITM CA certificate and exit python3 main.py --no-cert-check # Skip automatic CA install check on startup +python3 main.py --scan # Scan Google IPs and find the fastest one ``` > **Auto-install:** On startup (MITM mode), the proxy automatically checks if the CA certificate is trusted and attempts to install it. Use `--no-cert-check` to skip this. If auto-install fails (e.g. needs elevation), run `python main.py --install-cert` manually or follow Step 6 above. +### Scanning for the Fastest Google IP + +If your current `google_ip` in `config.json` is blocked or slow, you can scan to find a faster one: + +```bash +python3 main.py --scan +``` + +This will: +1. Probe 27 candidate Google IPs in parallel +2. Measure latency from your network +3. Display results in a table +4. Recommend the fastest IP +5. Exit with exit code 0 if at least one IP is reachable, 1 otherwise + +**Example output:** +``` +Scanning 27 Google frontend IPs + SNI: www.google.com + Timeout: 4s per IP + Concurrency: 8 parallel probes + +IP LATENCY STATUS +-------------------- ------------ ------------------------- +216.239.32.120 42ms OK +216.239.34.120 45ms OK +216.239.36.120 52ms OK +142.250.80.142 timeout timeout +... + +Result: 15 / 27 reachable + +Top 3 fastest IPs: + 1. 216.239.32.120 (42ms) + 2. 216.239.34.120 (45ms) + 3. 216.239.36.120 (52ms) + +Recommended: Set "google_ip": "216.239.32.120" in config.json +``` + +After scanning, update your `config.json` with the recommended IP and restart the proxy. + --- ## Architecture @@ -353,6 +396,7 @@ MasterHttpRelayVPN/ ├── mitm.py # On-the-fly TLS interception ├── cert_installer.py # Cross-platform CA installer (Windows/macOS/Linux + Firefox) ├── codec.py # Content-Encoding decoder (gzip/deflate/br/zstd) + ├── google_ip_scanner.py # Scanner to find the fastest reachable Google IP ├── constants.py # Tunable defaults and shared data └── logging_utils.py # Colored, aligned log formatter ``` diff --git a/README_FA.md b/README_FA.md index 5af49ad..9ee0f19 100644 --- a/README_FA.md +++ b/README_FA.md @@ -261,10 +261,53 @@ python3 main.py --log-level DEBUG python3 main.py -c /path/to/config.json python3 main.py --install-cert # نصب گواهی CA و خروج python3 main.py --no-cert-check # رد شدن از بررسی خودکار گواهی +python3 main.py --scan # اسکن IP های Google و یافتن سریع‌ترین ``` > **نصب خودکار:** هنگام اجرا در حالت `apps_script`، برنامه به‌طور خودکار بررسی می‌کند که آیا گواهی CA قابل اعتماد است یا نه و در صورت نیاز آن را نصب می‌کند. اگر نصب خودکار ناموفق بود (مثلاً نیاز به دسترسی مدیر دارد)، می‌توانید دستور `python main.py --install-cert` را اجرا کنید یا مراحل مرحله ۶ را دنبال کنید. +### اسکن کردن برای یافتن سریع‌ترین IP گوگل + +اگر `google_ip` فعلی در `config.json` بلاک شده یا آهسته است، می‌توانید اسکن کنید تا سریع‌ترین آن را پیدا کنید: + +```bash +python3 main.py --scan +``` + +این دستور: +1. ۲۷ IP برای fronting Google را به‌صورت موازی بررسی می‌کند +2. تأخیر (latency) از شبکه شما را اندازه می‌گیرد +3. نتایج را در جدول نمایش می‌دهد +4. سریع‌ترین IP را پیشنهاد می‌دهد +5. اگر حداقل یک IP در دسترس باشد کد خروج ۰، ورنه ۱ را برمی‌گرداند + +**نمونه خروجی:** +``` +Scanning 27 Google frontend IPs + SNI: www.google.com + Timeout: 4s per IP + Concurrency: 8 parallel probes + +IP LATENCY STATUS +-------------------- ------------ ------------------------- +216.239.32.120 42ms OK +216.239.34.120 45ms OK +216.239.36.120 52ms OK +142.250.80.142 timeout timeout +... + +Result: 15 / 27 reachable + +Top 3 fastest IPs: + 1. 216.239.32.120 (42ms) + 2. 216.239.34.120 (45ms) + 3. 216.239.36.120 (52ms) + +Recommended: Set "google_ip": "216.239.32.120" in config.json +``` + +پس از اسکن، مقدار `google_ip` در `config.json` را با IP پیشنهادی به‌روزرسانی کنید و پراکسی را دوباره راه‌اندازی کنید. + --- ## معماری @@ -297,6 +340,7 @@ MasterHttpRelayVPN/ ├── mitm.py # ساخت و مدیریت گواهی‌ها ├── cert_installer.py # نصب خودکار CA در ویندوز/مک/لینوکس + فایرفاکس ├── codec.py # رمزگشای Content-Encoding (gzip/deflate/br/zstd) + ├── google_ip_scanner.py # اسکنر IP های Google برای یافتن سریع‌ترین ├── constants.py # مقادیر پیش‌فرض قابل تنظیم └── logging_utils.py # فرمت‌دهنده‌ی لاگ رنگی و منظم ``` diff --git a/main.py b/main.py index adad513..a9e1369 100644 --- a/main.py +++ b/main.py @@ -23,6 +23,7 @@ from cert_installer import install_ca, is_ca_trusted from constants import __version__ from lan_utils import log_lan_access +from google_ip_scanner import scan_sync from logging_utils import configure as configure_logging, print_banner from mitm import CA_CERT_FILE from proxy_server import ProxyServer @@ -92,6 +93,11 @@ def parse_args(): action="store_true", help="Skip the certificate installation check on startup.", ) + parser.add_argument( + "--scan", + action="store_true", + help="Scan Google IPs to find the fastest reachable one and exit.", + ) return parser.parse_args() @@ -192,6 +198,15 @@ def main(): ok = install_ca(CA_CERT_FILE) sys.exit(0 if ok else 1) + # ── Google IP Scanner ────────────────────────────────────────────────── + if args.scan: + setup_logging("INFO") + front_domain = config.get("front_domain", "www.google.com") + _log = logging.getLogger("Main") + _log.info(f"Scanning Google IPs (fronting domain: {front_domain})") + ok = scan_sync(front_domain) + sys.exit(0 if ok else 1) + setup_logging(config.get("log_level", "INFO")) log = logging.getLogger("Main") diff --git a/src/constants.py b/src/constants.py index c8851b9..7932556 100644 --- a/src/constants.py +++ b/src/constants.py @@ -23,6 +23,39 @@ TLS_CONNECT_TIMEOUT = 15 TCP_CONNECT_TIMEOUT = 10 +# ── Google IP Scanner settings ────────────────────────────────────────────── +GOOGLE_SCANNER_TIMEOUT = 4 # Timeout per IP probe (seconds) +GOOGLE_SCANNER_CONCURRENCY = 8 # Parallel probes +# Candidate Google frontend IPs for scanning (multiple ASNs and regions) +CANDIDATE_IPS: tuple[str, ...] = ( + "216.239.32.120", + "216.239.34.120", + "216.239.36.120", + "216.239.38.120", + "142.250.80.142", + "142.250.80.138", + "142.250.179.110", + "142.250.185.110", + "142.250.184.206", + "142.250.190.238", + "142.250.191.78", + "172.217.1.206", + "172.217.14.206", + "172.217.16.142", + "172.217.22.174", + "172.217.164.110", + "172.217.168.206", + "172.217.169.206", + "34.107.221.82", + "142.251.32.110", + "142.251.33.110", + "142.251.46.206", + "142.251.46.238", + "142.250.80.170", + "142.250.72.206", + "142.250.64.206", + "142.250.72.110", +) # ── Response cache ──────────────────────────────────────────────────────── CACHE_MAX_MB = 50 diff --git a/src/google_ip_scanner.py b/src/google_ip_scanner.py new file mode 100644 index 0000000..79edc51 --- /dev/null +++ b/src/google_ip_scanner.py @@ -0,0 +1,194 @@ +""" +Google IP Scanner — finds the fastest reachable Google frontend IP. + +Scans a list of candidate Google IPs via HTTPS (with SNI fronting), measures +latency, and reports results in a formatted table. Useful for finding the best +IP to configure in config.json when your current IP is blocked. +""" + +from __future__ import annotations + +import asyncio +import logging +import ssl +import time +from dataclasses import dataclass +from typing import Optional + +from constants import CANDIDATE_IPS, GOOGLE_SCANNER_TIMEOUT, GOOGLE_SCANNER_CONCURRENCY + +log = logging.getLogger("Scanner") + + +@dataclass +class ProbeResult: + """Result of a single IP probe.""" + ip: str + latency_ms: Optional[int] = None + error: Optional[str] = None + + @property + def ok(self) -> bool: + return self.latency_ms is not None + + +async def _probe_ip( + ip: str, + sni: str, + semaphore: asyncio.Semaphore, + timeout: float, +) -> ProbeResult: + """ + Probe a single IP via HTTPS with SNI fronting. + + Args: + ip: The IP to probe (xxx.xxx.xxx.xxx). + sni: The SNI hostname to use in TLS handshake. + semaphore: Rate limiter to control concurrency. + timeout: Timeout in seconds for the entire probe. + + Returns: + ProbeResult with latency_ms (if successful) or error message. + """ + async with semaphore: + start_time = time.time() + try: + # Create SSL context that skips certificate verification + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + + # Connect to IP:443 with SNI set to the fronting domain + reader, writer = await asyncio.wait_for( + asyncio.open_connection( + ip, + 443, + ssl=ctx, + server_hostname=sni, + ), + timeout=timeout, + ) + + # Send minimal HTTP HEAD request + request = f"HEAD / HTTP/1.1\r\nHost: {sni}\r\nConnection: close\r\n\r\n" + writer.write(request.encode()) + await writer.drain() + + # Read response header (first 256 bytes is plenty for HTTP status) + response = await asyncio.wait_for(reader.read(256), timeout=timeout) + + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + + # Check if we got an HTTP response + if not response: + return ProbeResult(ip=ip, error="empty response") + + response_str = response.decode("utf-8", errors="ignore") + if not response_str.startswith("HTTP/"): + return ProbeResult(ip=ip, error=f"invalid response: {response_str[:30]!r}") + + # Success — return latency in milliseconds + elapsed_ms = int((time.time() - start_time) * 1000) + return ProbeResult(ip=ip, latency_ms=elapsed_ms) + + except asyncio.TimeoutError: + return ProbeResult(ip=ip, error="timeout") + except ConnectionRefusedError: + return ProbeResult(ip=ip, error="connection refused") + except ConnectionResetError: + return ProbeResult(ip=ip, error="connection reset") + except OSError as e: + return ProbeResult(ip=ip, error=f"network error: {e.strerror or str(e)}") + except Exception as e: + return ProbeResult(ip=ip, error=f"probe failed: {type(e).__name__}") + + +async def run(front_domain: str) -> bool: + """ + Scan all candidate Google IPs and display results. + + Args: + front_domain: The SNI hostname to use (e.g. "www.google.com"). + + Returns: + True if at least one IP is reachable, False otherwise. + """ + timeout = GOOGLE_SCANNER_TIMEOUT + concurrency = GOOGLE_SCANNER_CONCURRENCY + + print() + print(f"Scanning {len(CANDIDATE_IPS)} Google frontend IPs") + print(f" SNI: {front_domain}") + print(f" Timeout: {timeout}s per IP") + print(f" Concurrency: {concurrency} parallel probes") + print() + + # Create semaphore to limit concurrency + semaphore = asyncio.Semaphore(concurrency) + + # Launch all probes concurrently + tasks = [ + _probe_ip(ip, front_domain, semaphore, timeout) + for ip in CANDIDATE_IPS + ] + results = await asyncio.gather(*tasks) + + # Sort by latency (successful first, then by speed) + results.sort(key=lambda r: (not r.ok, r.latency_ms or float("inf"))) + + # Display results table + print(f"{'IP':<20} {'LATENCY':<12} {'STATUS':<25}") + print(f"{'-' * 20} {'-' * 12} {'-' * 25}") + + ok_count = 0 + for result in results: + if result.ok: + print(f"{result.ip:<20} {result.latency_ms:>8}ms OK") + ok_count += 1 + else: + status = result.error or "unknown error" + print(f"{result.ip:<20} {'—':<12} {status:<25}") + + print() + print(f"Result: {ok_count} / {len(results)} reachable") + + if ok_count == 0: + print("No Google IPs reachable from this network.") + print() + return False + + # Show top 3 fastest + fastest = [r for r in results if r.ok][:3] + print() + print("Top 3 fastest IPs:") + for i, result in enumerate(fastest, 1): + print(f" {i}. {result.ip} ({result.latency_ms}ms)") + + print() + print(f"Recommended: Set \"google_ip\": \"{fastest[0].ip}\" in config.json") + print() + return True + + +def scan_sync(front_domain: str) -> bool: + """ + Wrapper to run async scanner from sync context (e.g. main.py). + + Args: + front_domain: The SNI hostname to use. + + Returns: + True if at least one IP is reachable, False otherwise. + """ + try: + return asyncio.run(run(front_domain)) + except KeyboardInterrupt: + print("\nScan interrupted by user.") + return False + except Exception as e: + log.error(f"Scan failed: {e}") + return False From e9b7081f819e93715cb009fa5a3000ce017e2bf8 Mon Sep 17 00:00:00 2001 From: Emran Hejazi Date: Thu, 23 Apr 2026 12:06:24 +0330 Subject: [PATCH 2/2] Implement gas stats fetchery and add args --- main.py | 27 ++++ requirements.txt | 5 + src/constants.py | 9 ++ src/gas_stats_fetcher.py | 313 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 354 insertions(+) create mode 100644 src/gas_stats_fetcher.py diff --git a/main.py b/main.py index a9e1369..65d282f 100644 --- a/main.py +++ b/main.py @@ -22,6 +22,7 @@ from cert_installer import install_ca, is_ca_trusted from constants import __version__ +from gas_stats_fetcher import create_fetcher as create_gas_stats_fetcher from lan_utils import log_lan_access from google_ip_scanner import scan_sync from logging_utils import configure as configure_logging, print_banner @@ -98,6 +99,11 @@ def parse_args(): action="store_true", help="Scan Google IPs to find the fastest reachable one and exit.", ) + parser.add_argument( + "--gas-stats", + action="store_true", + help="Fetch and display Google Apps Script deployment stats, then exit.", + ) return parser.parse_args() @@ -207,6 +213,27 @@ def main(): ok = scan_sync(front_domain) sys.exit(0 if ok else 1) + # ── Google Apps Script Stats ──────────────────────────────────────── + if args.gas_stats: + setup_logging("INFO") + _log = logging.getLogger("Main") + script_id = config.get("script_ids") or config.get("script_id") + if not script_id: + _log.error("No script_id configured in config.json") + sys.exit(1) + if isinstance(script_id, list): + _log.info("Fetching stats for %d script IDs…", len(script_id)) + script_id = script_id[0] # Use first for stats + else: + _log.info("Fetching Google Apps Script deployment stats…") + try: + fetcher = create_gas_stats_fetcher(deployment_id=script_id) + fetcher.fetch_and_log_stats() + except ImportError as e: + _log.error(f"Cannot fetch stats: {e}") + sys.exit(1) + sys.exit(0) + setup_logging(config.get("log_level", "INFO")) log = logging.getLogger("Main") diff --git a/requirements.txt b/requirements.txt index c0f1980..a52e7a8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -15,3 +15,8 @@ zstandard>=0.22.0 # Optional: Better network interface detection for LAN sharing netifaces>=0.11.0 + +# Optional: Google Apps Script stats fetching (google_auth_stats) +google-auth>=2.25.0 +google-auth-httplib2>=0.2.0 +google-auth-oauthlib>=1.1.0 diff --git a/src/constants.py b/src/constants.py index 7932556..b513812 100644 --- a/src/constants.py +++ b/src/constants.py @@ -23,6 +23,15 @@ TLS_CONNECT_TIMEOUT = 15 TCP_CONNECT_TIMEOUT = 10 + +# ── Google API settings ───────────────────────────────────────────────────────── +# Google Apps Script deployment stats +GAS_STATS_TIMEOUT = 10 # Timeout for GAS API requests (seconds) +GAS_STATS_API_BASE = "https://www.googleapis.com/apps_script/v1" +GAS_EXECUTION_API_BASE = "https://www.googleapis.com/execution/v1/projects" +GAS_CREDENTIALS_FILE = "credentials.json" # OAuth2 credentials file path + + # ── Google IP Scanner settings ────────────────────────────────────────────── GOOGLE_SCANNER_TIMEOUT = 4 # Timeout per IP probe (seconds) GOOGLE_SCANNER_CONCURRENCY = 8 # Parallel probes diff --git a/src/gas_stats_fetcher.py b/src/gas_stats_fetcher.py new file mode 100644 index 0000000..bacd8f3 --- /dev/null +++ b/src/gas_stats_fetcher.py @@ -0,0 +1,313 @@ +""" +Google Apps Script Deployment Stats Fetcher + +Retrieves and displays deployment statistics (error rate, executions, users) +from Google Apps Script projects via the Google Apps Script Execution API. + +Requires: + - google-auth library (pip install google-auth google-auth-httplib2 google-auth-oauthlib) + - OAuth2 credentials.json file in project root + +Setup Instructions: + 1. Go to https://console.cloud.google.com + 2. Create OAuth2 credentials (Service Account or Desktop Application) + 3. Enable Google Apps Script API + 4. Download credentials.json and place in project root + 5. Deploy your Apps Script and copy the Deployment ID + +Stats fetched: + - Total Executions: Number of function calls executed + - Error Rate: Percentage of failed executions + - Active Users: Number of unique users + - Success Rate: Percentage of successful executions +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +from dataclasses import dataclass, asdict +from datetime import datetime +from typing import Optional, Dict, Any + +from requests_oauthlib import OAuth2 + +from constants import GAS_STATS_TIMEOUT, GAS_STATS_API_BASE, GAS_CREDENTIALS_FILE + +log = logging.getLogger("GAS-Stats") + +@dataclass +class DeploymentStats: + """Deployment statistics data container.""" + total_executions: int = 0 + failed_executions: int = 0 + successful_executions: int = 0 + error_rate: float = 0.0 + success_rate: float = 0.0 + active_users: int = 0 + timestamp: str = "" + deployment_id: str = "" + + @property + def ok(self) -> bool: + """Check if stats were retrieved successfully.""" + return self.timestamp != "" + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary for serialization.""" + return asdict(self) + + def __str__(self) -> str: + """Pretty string representation.""" + if not self.ok: + return "Stats unavailable" + return ( + f"Executions: {self.total_executions} | " + f"Success Rate: {self.success_rate:.1f}% | " + f"Error Rate: {self.error_rate:.1f}% | " + f"Active Users: {self.active_users}" + ) + + +class GASStatsFetcher: + """ + Fetches Google Apps Script deployment statistics via the Execution API. + + Requires proper OAuth2 authentication setup with google-auth library. + + Attributes: + deployment_id: The Apps Script Deployment ID + credentials_path: Path to OAuth2 credentials.json file + timeout: Request timeout in seconds (default: 10) + """ + + def __init__( + self, + deployment_id: str, + credentials_path: Optional[str] = None, + timeout: int = GAS_STATS_TIMEOUT, + ): + """ + Initialize the stats fetcher. + + Args: + deployment_id: Google Apps Script Deployment ID + credentials_path: Path to OAuth2 credentials.json file + timeout: Request timeout in seconds + """ + + + self.deployment_id = deployment_id + self.credentials_path = credentials_path or GAS_CREDENTIALS_FILE + self.timeout = timeout + self._credentials = None + + def _load_credentials(self): + """ + Load and validate OAuth2 credentials. + + Returns: + Google credentials object + + Raises: + FileNotFoundError: If credentials file not found + ValueError: If credentials are invalid + """ + if not os.path.exists(self.credentials_path): + raise FileNotFoundError( + f"OAuth2 credentials file not found: {self.credentials_path}\n" + f"Set up credentials as described in module docstring." + ) + + try: + self._credentials = service_account.Credentials.from_service_account_file( + self.credentials_path, + scopes=["https://www.googleapis.com/auth/script.projects.readonly"], + ) + log.debug(f"Loaded credentials from {self.credentials_path}") + return self._credentials + except Exception as e: + raise ValueError(f"Invalid credentials file: {e}") + + def _get_apps_script_service(self): + """ + Get authenticated Google Apps Script service client. + + Returns: + Google Apps Script API service object + + Raises: + FileNotFoundError: If credentials not available + ValueError: If credentials are invalid + """ + if not self._credentials: + self._load_credentials() + + return build("script", "v1", credentials=self._credentials, cache_discovery=False) + + def fetch_stats(self) -> DeploymentStats: + """ + Fetch deployment statistics from Google Apps Script API. + + Returns: + DeploymentStats object with metrics + + Raises: + FileNotFoundError: If credentials file not found + Exception: If API call fails + """ + + if not self.deployment_id: + log.error("Deployment ID not set") + return DeploymentStats() + + try: + log.debug(f"Fetching stats for deployment {self.deployment_id}") + + # Load credentials + self._load_credentials() + + # Get service client + service = self._get_apps_script_service() + + # Call the Metrics API + result = service.projects().getMetrics(projectId=self.deployment_id).execute() + + stats = self._parse_metrics(result) + log.info(f"Successfully fetched GAS deployment stats") + return stats + + except FileNotFoundError as e: + log.error(f"Credentials error: {e}") + return DeploymentStats() + except ValueError as e: + log.error(f"Invalid credentials: {e}") + return DeploymentStats() + except Exception as e: + log.error(f"API error fetching GAS stats: {e}") + return DeploymentStats() + + def _parse_metrics(self, response: Dict[str, Any]) -> DeploymentStats: + """ + Parse metrics response from Google API. + + Args: + response: API response dictionary + + Returns: + DeploymentStats object with parsed metrics + """ + stats = DeploymentStats(deployment_id=self.deployment_id) + stats.timestamp = datetime.now().isoformat() + + try: + # Extract metrics from response + metrics = response.get("metrics", {}) + + if not metrics: + log.warning("No metrics data in API response") + return stats + + # Parse execution statistics + stats.total_executions = int(metrics.get("totalExecutions", 0)) + stats.failed_executions = int(metrics.get("failedExecutions", 0)) + stats.successful_executions = stats.total_executions - stats.failed_executions + + # Calculate rates + if stats.total_executions > 0: + stats.error_rate = (stats.failed_executions / stats.total_executions) * 100 + stats.success_rate = (stats.successful_executions / stats.total_executions) * 100 + + # Parse user statistics + stats.active_users = int(metrics.get("activeUsers", 0)) + + log.debug(f"Metrics parsed: {stats}") + + except (KeyError, TypeError, ValueError) as e: + log.error(f"Error parsing metrics response: {e}") + + return stats + + async def fetch_stats_async(self) -> DeploymentStats: + """ + Fetch deployment statistics asynchronously. + + Returns: + DeploymentStats object + """ + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self.fetch_stats) + + def fetch_and_log_stats(self) -> DeploymentStats: + """ + Fetch stats and log them in a formatted manner. + + Uses the project's logging system for professional output. + + Returns: + DeploymentStats object + """ + stats = self.fetch_stats() + + if stats.ok: + # Log stats in human-readable format with tree structure + log.info("Google Apps Script Deployment Stats") + log.info( + f"├─ Total Executions: {stats.total_executions:,} | " + f"Successful: {stats.successful_executions:,} | " + f"Failed: {stats.failed_executions:,}" + ) + log.info( + f"├─ Success Rate: {stats.success_rate:>6.2f}% | " + f"Error Rate: {stats.error_rate:>6.2f}%" + ) + log.info(f"├─ Active Users: {stats.active_users:,}") + log.info(f"└─ Deployment ID: {stats.deployment_id} | Timestamp: {stats.timestamp}") + + return stats + + +def create_fetcher( + deployment_id: str, + credentials_path: Optional[str] = None, +) -> GASStatsFetcher: + """ + Factory function to create a GASStatsFetcher instance. + + Args: + deployment_id: Google Apps Script Deployment ID + credentials_path: Optional path to credentials.json file + + Returns: + Configured GASStatsFetcher instance + + Raises: + ImportError: If google-auth not installed + """ + return GASStatsFetcher( + deployment_id=deployment_id, + credentials_path=credentials_path, + ) + + +def log_deployment_stats( + deployment_id: str, + credentials_path: Optional[str] = None, +) -> None: + """ + Convenience function to fetch and log stats in one call. + + Args: + deployment_id: Google Apps Script Deployment ID + credentials_path: Optional path to credentials.json file + + Raises: + ImportError: If google-auth not installed + """ + try: + fetcher = create_fetcher(deployment_id, credentials_path) + fetcher.fetch_and_log_stats() + except ImportError as e: + log.error(f"Cannot fetch stats: {e}")