Function run

src/core/google_ip_scanner.py:110–174

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.

(front_domain: str)

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:")