time_tool.py

"""Timezone helper for the Reachy Mini Time Tool Space: offline compute, keyless geo-IP for local time."""

import json
import logging
from datetime import datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from functools import lru_cache
from ipaddress import ip_address
from urllib.error import URLError, HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen

import gradio as gr


logger = logging.getLogger(__name__)

IPWHOIS_URL = "https://ipwho.is"
TIMEOUT_S = 3.0
USER_AGENT = "reachy-mini-time-tool/2.0 (+https://huggingface.co/spaces/pollen-robotics/reachy-mini-time-tool)"
CLIENT_IP_HEADERS = (
    "cf-connecting-ip",
    "x-real-ip",
    "x-forwarded-for",
    "x-client-ip",
    "fastly-client-ip",
)
UTC = ZoneInfo("UTC")


def get_time(
    timezone: str = "",
    compare_timezone: str = "",
    request: gr.Request | None = None,
) -> dict[str, object]:
    """Return the current time for an IANA timezone (empty resolves the caller's local zone), with an optional difference."""
    if not isinstance(timezone, str) or not isinstance(compare_timezone, str):
        return {"error": "timezone and compare_timezone must be strings"}

    base_name = timezone.strip()
    if base_name:
        base_zone = _load_timezone(base_name)
        if base_zone is None:
            return {"error": f"unknown timezone '{base_name}'; pass an IANA name like 'Europe/Paris'"}
    else:
        base_name, base_zone = _local_timezone(request)

    logger.info("get_time timezone=%s compare=%s", base_name, compare_timezone.strip())
    now_utc = datetime.now(UTC)
    base_now = now_utc.astimezone(base_zone)
    result = _format_time(base_now, base_name)

    compare_name = compare_timezone.strip()
    if compare_name:
        compare_zone = _load_timezone(compare_name)
        if compare_zone is None:
            return {"error": f"unknown timezone '{compare_name}'; pass an IANA name like 'Asia/Tokyo'"}
        compare_now = now_utc.astimezone(compare_zone)
        difference_minutes = _offset_minutes(compare_now) - _offset_minutes(base_now)
        result["compare"] = _format_time(compare_now, compare_name)
        result["time_difference_minutes"] = difference_minutes
        result["time_difference_hours"] = difference_minutes / 60
        result["time_difference_summary"] = _difference_summary(compare_name, base_name, difference_minutes)

    return result


def _local_timezone(request: gr.Request | None) -> tuple[str, ZoneInfo]:
    """Resolve the caller's local timezone by keyless geo-IP, falling back to UTC when it cannot be determined."""
    caller_ip = _caller_ip(request)
    if caller_ip is None:
        logger.warning("caller public IP was not provided; local time falls back to UTC")
        return "UTC", UTC

    name = _timezone_for_ip(caller_ip)
    zone = _load_timezone(name) if name else None
    if zone is None:
        return "UTC", UTC
    return name, zone


@lru_cache(maxsize=1024)
def _timezone_for_ip(caller_ip: str) -> str | None:
    """Look up an IANA timezone for a public IP via keyless ipwho.is, cached per IP to stay under the shared quota."""
    url = f"{IPWHOIS_URL}/{quote(caller_ip, safe='.:')}?fields=success,timezone"
    try:
        request = Request(url, headers={"Accept": "application/json", "User-Agent": USER_AGENT})
        with urlopen(request, timeout=TIMEOUT_S) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except (HTTPError, URLError, TimeoutError, ValueError) as exc:
        logger.warning("geo-IP timezone lookup failed for %s: %s", caller_ip, exc)
        return None

    if not payload.get("success", True):
        logger.warning("geo-IP lookup unsuccessful for %s: %s", caller_ip, payload.get("message"))
        return None
    timezone = payload.get("timezone")
    if not isinstance(timezone, dict):
        return None
    return str(timezone.get("id", "")).strip() or None


def _caller_ip(request: gr.Request | None) -> str | None:
    if request is None:
        return None

    headers = getattr(request, "headers", {}) or {}
    for header_name in CLIENT_IP_HEADERS:
        header_value = headers.get(header_name)
        if isinstance(header_value, str):
            public_ip = _first_public_ip(header_value)
            if public_ip is not None:
                return public_ip

    host = getattr(getattr(request, "client", None), "host", None)
    return _first_public_ip(host) if isinstance(host, str) else None


def _first_public_ip(value: str) -> str | None:
    for raw_candidate in value.split(","):
        candidate = raw_candidate.strip()
        if candidate.count(":") == 1:
            candidate = candidate.rsplit(":", maxsplit=1)[0]
        try:
            parsed_ip = ip_address(candidate)
        except ValueError:
            continue
        if not (
            parsed_ip.is_private
            or parsed_ip.is_loopback
            or parsed_ip.is_link_local
            or parsed_ip.is_multicast
            or parsed_ip.is_unspecified
        ):
            return str(parsed_ip)
    return None


def _load_timezone(name: str) -> ZoneInfo | None:
    try:
        return ZoneInfo(name)
    except (ZoneInfoNotFoundError, ValueError):
        return None


def _format_time(now: datetime, name: str) -> dict[str, object]:
    return {
        "iso": now.isoformat(timespec="seconds"),
        "date": now.strftime("%Y-%m-%d"),
        "time": now.strftime("%H:%M"),
        "weekday": now.strftime("%A"),
        "timezone": name,
        "utc_offset_minutes": _offset_minutes(now),
        "summary": now.strftime("%A, %d %B %Y, %H:%M"),
    }


def _offset_minutes(now: datetime) -> int:
    offset = now.utcoffset()
    return int(offset.total_seconds() // 60) if offset else 0


def _difference_summary(compared_name: str, base_name: str, difference_minutes: int) -> str:
    if difference_minutes == 0:
        return f"{compared_name} has the same current UTC offset as {base_name}."

    direction = "ahead of" if difference_minutes > 0 else "behind"
    hours, remaining_minutes = divmod(abs(difference_minutes), 60)
    duration = f"{hours} hour{'s' if hours != 1 else ''}"
    if remaining_minutes:
        duration = f"{duration} {remaining_minutes} minute{'s' if remaining_minutes != 1 else ''}"
    return f"{compared_name} is {duration} {direction} {base_name}."