diff --git a/worker/src/scraper.py b/worker/src/scraper.py index b1095aa..6a289e3 100644 --- a/worker/src/scraper.py +++ b/worker/src/scraper.py @@ -11,6 +11,9 @@ logger = logging.getLogger(__name__) _client: httpx.AsyncClient | None = None _proxy_ip_logged = False +_system_public_ip: str | None = None +_proxy_public_ip: str | None = None +_last_used_public_ip: str | None = None def _get_proxy_url() -> str | None: @@ -56,7 +59,7 @@ async def _fetch_public_ip(client: httpx.AsyncClient) -> str | None: async def _log_proxy_ip_comparison(proxy_url: str) -> None: - global _proxy_ip_logged + global _proxy_ip_logged, _system_public_ip, _proxy_public_ip, _last_used_public_ip if _proxy_ip_logged: return @@ -69,6 +72,9 @@ async def _log_proxy_ip_comparison(proxy_url: str) -> None: try: direct_ip = await _fetch_public_ip(direct_client) proxy_ip = await _fetch_public_ip(proxy_client) + _system_public_ip = direct_ip + _proxy_public_ip = proxy_ip + _last_used_public_ip = proxy_ip or direct_ip logger.info( "HTTPS proxy enabled: %s | public_ip_without_proxy=%s | public_ip_with_proxy=%s", _redact_proxy_url(proxy_url), @@ -82,7 +88,7 @@ async def _log_proxy_ip_comparison(proxy_url: str) -> None: async def get_client() -> httpx.AsyncClient: """Return a shared AsyncClient with keepalive connection pool.""" - global _client + global _client, _system_public_ip, _last_used_public_ip if _client is None or _client.is_closed: max_conns = int(os.getenv("HTTP_MAX_CONNECTIONS", "10")) @@ -91,6 +97,13 @@ async def get_client() -> httpx.AsyncClient: if proxy_url: await _log_proxy_ip_comparison(proxy_url) + elif _system_public_ip is None: + direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False) + try: + _system_public_ip = await _fetch_public_ip(direct_client) + _last_used_public_ip = _system_public_ip + finally: + await direct_client.aclose() _client = httpx.AsyncClient( timeout=float(os.getenv("HTTP_TIMEOUT_S", "30.0")), @@ -112,6 +125,49 @@ async def get_client() -> httpx.AsyncClient: return _client +async def refresh_network_status() -> dict[str, str | bool | None]: + """Ensure network status has best-effort values even before first scrape.""" + global _system_public_ip, _proxy_public_ip, _last_used_public_ip + + proxy_url = _get_proxy_url() + + if _system_public_ip is None: + direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False) + try: + _system_public_ip = await _fetch_public_ip(direct_client) + finally: + await direct_client.aclose() + + if proxy_url and _proxy_public_ip is None: + proxy_client = httpx.AsyncClient(timeout=10.0, trust_env=False, proxy=proxy_url) + try: + _proxy_public_ip = await _fetch_public_ip(proxy_client) + finally: + await proxy_client.aclose() + + if proxy_url: + _last_used_public_ip = _proxy_public_ip or _system_public_ip + else: + _last_used_public_ip = _system_public_ip + + return get_network_status() + + +def get_network_status() -> dict[str, str | bool | None]: + """Return best-effort network status for UI display.""" + proxy_enabled = bool(_get_proxy_url()) + last_used = _last_used_public_ip + if last_used is None: + last_used = _proxy_public_ip if proxy_enabled else _system_public_ip + + return { + "proxy_enabled": proxy_enabled, + "system_public_ip": _system_public_ip, + "proxy_public_ip": _proxy_public_ip, + "last_used_public_ip": last_used, + } + + async def close_client() -> None: """Close the shared AsyncClient. Call during shutdown.""" global _client diff --git a/worker/src/templates/dashboard.html b/worker/src/templates/dashboard.html index 05fb9fb..23362bd 100644 --- a/worker/src/templates/dashboard.html +++ b/worker/src/templates/dashboard.html @@ -56,6 +56,32 @@ +
+
Network Egress
+
+
+
System IP
+
+ {{ data.system_country_flag }} + {{ data.system_public_ip or 'Unknown' }} + {% if data.system_country_code %} + ({{ data.system_country_code }}) + {% endif %} +
+
+
+
Last Used IP
+
+ {{ data.last_used_country_flag }} + {{ data.last_used_public_ip or 'Unknown' }} + {% if data.last_used_country_code %} + ({{ data.last_used_country_code }}) + {% endif %} +
+
+
+
+
Delivery Queue
diff --git a/worker/src/web.py b/worker/src/web.py index b454c03..947c16a 100644 --- a/worker/src/web.py +++ b/worker/src/web.py @@ -1,17 +1,22 @@ import os import logging +import time from contextlib import asynccontextmanager +import httpx from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.templating import Jinja2Templates import asyncpg from db import get_pool +from scraper import get_network_status, refresh_network_status logger = logging.getLogger(__name__) templates = Jinja2Templates(directory="templates") +_country_cache: dict[str, tuple[float, str]] = {} +_COUNTRY_CACHE_TTL_S = 3600 async def query(sql: str, *args) -> list: @@ -78,6 +83,50 @@ def _proxy_enabled() -> bool: return bool((os.getenv("HTTPS_PROXY") or os.getenv("https_proxy") or "").strip()) +def _flag_from_country_code(country_code: str | None) -> str: + if not country_code or len(country_code) != 2: + return "" + code = country_code.upper() + if not code.isalpha(): + return "" + return chr(127397 + ord(code[0])) + chr(127397 + ord(code[1])) + + +async def _country_code_for_ip(ip_addr: str | None) -> str | None: + if not ip_addr: + return None + + now = time.time() + cached = _country_cache.get(ip_addr) + if cached and now - cached[0] < _COUNTRY_CACHE_TTL_S: + return cached[1] + + try: + async with httpx.AsyncClient(timeout=5.0, trust_env=False) as client: + resp = await client.get(f"https://ipapi.co/{ip_addr}/country/") + resp.raise_for_status() + code = resp.text.strip().upper() + if len(code) == 2 and code.isalpha(): + _country_cache[ip_addr] = (now, code) + return code + except Exception: + logger.debug("Could not resolve country for IP %s", ip_addr, exc_info=True) + + try: + async with httpx.AsyncClient(timeout=5.0, trust_env=False) as client: + resp = await client.get(f"https://ipwho.is/{ip_addr}") + resp.raise_for_status() + payload = resp.json() + code = str(payload.get("country_code", "")).upper() + if len(code) == 2 and code.isalpha(): + _country_cache[ip_addr] = (now, code) + return code + except Exception: + logger.debug("Fallback country lookup failed for IP %s", ip_addr, exc_info=True) + + return None + + @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Web UI starting up") @@ -110,6 +159,16 @@ async def dashboard(request: Request): {"request": request, "error": "Database unavailable", "data": None}, ) + try: + network = await refresh_network_status() + except Exception: + logger.debug("Could not refresh network status", exc_info=True) + network = get_network_status() + system_ip = network.get("system_public_ip") + last_used_ip = network.get("last_used_public_ip") + system_country = await _country_code_for_ip(system_ip if isinstance(system_ip, str) else None) + last_used_country = await _country_code_for_ip(last_used_ip if isinstance(last_used_ip, str) else None) + data = { "total_keywords": total_keywords or 0, "active_keywords": active_keywords or 0, @@ -120,6 +179,12 @@ async def dashboard(request: Request): "queue_dead": queue_dead or 0, "last_scheduler": last_scheduler["scraped_at"] if last_scheduler else None, "proxy_enabled": _proxy_enabled(), + "system_public_ip": system_ip, + "last_used_public_ip": last_used_ip, + "system_country_code": system_country, + "last_used_country_code": last_used_country, + "system_country_flag": _flag_from_country_code(system_country), + "last_used_country_flag": _flag_from_country_code(last_used_country), } return templates.TemplateResponse("dashboard.html", {"request": request, "error": None, "data": data})