thermograph/backend/notifications/discord_interactions.py

"""Discord slash commands over the HTTP Interactions endpoint.

Discord POSTs each interaction (a slash command, a button, etc.) to a URL we
register in the developer portal, so commands are just an ordinary FastAPI route
(app.py wires POST {BASE}/discord/interactions) with no gateway connection and no
persistent bot process. Every request is signed; we verify the Ed25519 signature
over the raw body before trusting anything (Discord actively probes this with bad
signatures and disables the endpoint if we don't reject them).

Commands so far:
  /grade <city>  -> today's grade for a curated city, from the warm cache.

Register the command definitions with scripts/register_discord_commands.py.
"""
from __future__ import annotations

import datetime
import difflib
import json
import os
import re
import unicodedata

from nacl.exceptions import BadSignatureError
from nacl.signing import VerifyKey

from api import homepage
from data import cities
from notifications import discord

# App's Ed25519 public key (Developer Portal -> General Information). Unset =>
# every request fails verification, which is the safe default for an unconfigured
# deployment.
PUBLIC_KEY = os.environ.get("THERMOGRAPH_DISCORD_PUBLIC_KEY", "").strip()

# Interaction types (Discord).
_PING = 1
_APPLICATION_COMMAND = 2
# Response types.
_PONG = 1
_CHANNEL_MESSAGE = 4
# Message flags: 64 = ephemeral (only the caller sees it).
_EPHEMERAL = 64


def verify(signature: str, timestamp: str, body: bytes, public_key: str | None = None) -> bool:
    """True iff `signature` is a valid Ed25519 signature of timestamp+body under the
    app public key. Any malformed input is a verification failure, never an error."""
    key = public_key if public_key is not None else PUBLIC_KEY
    if not key or not signature or not timestamp:
        return False
    try:
        VerifyKey(bytes.fromhex(key)).verify(timestamp.encode() + body, bytes.fromhex(signature))
        return True
    except (BadSignatureError, ValueError):
        return False


def _fold(s: str) -> str:
    """Casefold and strip accents, so `Zurich` matches `Zürich` and `Sao Paulo`
    matches `São Paulo`. People type city names without the diacritics far more
    often than with them."""
    s = unicodedata.normalize("NFKD", s or "")
    return "".join(ch for ch in s if not unicodedata.combining(ch)).casefold().strip()


def _qualifier_matches(city: dict, tail: str) -> bool:
    """Does the bit after the comma in "Vilnius, Lithuania" describe this city?
    Accepts the country, its ISO code, or the admin1 region, since all three are
    things people write there ("Paris, FR", "Springfield, Illinois")."""
    if not tail:
        return True
    return tail in {
        _fold(city.get("country", "")),
        _fold(city.get("country_code", "")),
        _fold(city.get("admin1", "")),
    }


# Below this length, a name is too collision-prone to look for inside a
# sentence — real entries like "Of" or "Ube" would match ordinary English.
_MIN_FREE_TEXT_NAME = 4


def _resolve_city(query: str) -> dict | None:
    """Best curated-city match for whatever someone actually typed.

    The old version matched the query against `name` alone, exact-or-prefix.
    That failed the two most natural things to write. "Vilnius, Lithuania" never
    matched, because the comma form was compared verbatim against a bare name —
    the country was in the record all along and simply never consulted. And a
    single transposed letter ("Vilinus") fell straight through to "I don't track
    a city called …", which reads as *we've never heard of Vilnius* rather than
    *you typed it wrong*.

    So, in order of how confident each step is:
      1. exact name, honouring a "City, Country/Region" qualifier;
      2. the same, ignoring an unrecognised qualifier — better to grade Paris
         and say so in the reply than to refuse over "Paris, Wherever";
      3. prefix ("san fran");
      4. a close-enough name, for typos;
      5. a city or country named somewhere inside a sentence, so
         "tell me about weather in lithuania" lands on Vilnius.

    All of it prefers the largest city, since cities.json is population-sorted
    and the biggest is the likeliest thing meant by an ambiguous name.
    """
    q = _fold(query)
    if not q:
        return None

    # "Vilnius, Lithuania" -> head "vilnius", tail "lithuania". Only the first
    # comma splits: "Washington, DC, USA" keeps "dc, usa" as the qualifier, and
    # the ISO/admin1 check below still matches it on "dc".
    head, _, tail = q.partition(",")
    head, tail = head.strip(), tail.strip()

    all_cities = cities.all_cities()       # already sorted by population, desc

    named = None                           # name matched, qualifier did not
    prefix = None
    for c in all_cities:
        name = _fold(c["name"])
        if name == head:
            if _qualifier_matches(c, tail):
                return c
            named = named or c
        elif prefix is None and head and name.startswith(head):
            prefix = c
    if named:
        return named
    if prefix:
        return prefix

    # Typos. 0.82 is tight enough that "vilinus" reaches "vilnius" while
    # genuinely different names stay apart; get_close_matches ranks by ratio,
    # and ties resolve to the more populous city because we scan in order.
    names = [_fold(c["name"]) for c in all_cities]
    close = difflib.get_close_matches(head, names, n=1, cutoff=0.82)
    if close:
        return all_cities[names.index(close[0])]

    # A sentence, or a country on its own. Look for the longest city name
    # mentioned in it, then fall back to the largest city of a country named in
    # it — "weather in lithuania" is a perfectly reasonable thing to ask.
    words = set(re.findall(r"[a-z0-9]+", q))
    best, best_len = None, 0
    for c in all_cities:
        name = _fold(c["name"])
        if len(name) < _MIN_FREE_TEXT_NAME or len(name) <= best_len:
            continue
        # Whole-word containment, so "Nice" does not fire on "nicely".
        if set(re.findall(r"[a-z0-9]+", name)) <= words:
            best, best_len = c, len(name)
    if best:
        return best
    for c in all_cities:
        country = _fold(c.get("country", ""))
        if len(country) >= _MIN_FREE_TEXT_NAME and set(re.findall(r"[a-z0-9]+", country)) <= words:
            return c                       # population-sorted: the largest
    return None


def _grade_message(query: str) -> dict:
    """The message data for /grade <city>: an embed with today's grade line, or a
    plain ephemeral note when the city is unknown or not yet warmed."""
    city = _resolve_city(query)
    if city is None:
        # Don't read a whole sentence back as if it were a city name — quoting
        # "Tell me about weather in lithuania" as a place we don't track is what
        # makes this look broken rather than merely unmatched.
        looks_like_a_name = len(query) <= 40 and len(query.split()) <= 4
        detail = f"I don't track a city called '{query}' yet." if looks_like_a_name \
            else "I couldn't find a city in that."
        return {"content": f"{detail} Try a city name — \"Vilnius\" or "
                           "\"Vilnius, Lithuania\" both work.", "flags": _EPHEMERAL}
    card = homepage._grade_city(city, datetime.date.today())
    if card is None:
        return {"content": f"No recent reading is cached for {city['name']} yet. "
                           "It should appear once the data warms up.", "flags": _EPHEMERAL}
    color = discord._COLD if card.get("tail") == "cold" else discord._HOT
    return {"embeds": [{
        "title": f"{card.get('display') or city['name']} — today",
        "description": discord._line(card),
        "url": f"https://thermograph.org/climate/{city['slug']}",
        "color": color,
        "footer": {"text": "thermograph.org · graded against ~45 years of local history"},
    }]}


def _command_response(data: dict) -> dict:
    """Dispatch an application-command interaction to its handler."""
    name = (data.get("name") or "").lower()
    opts = {o.get("name"): o.get("value") for o in (data.get("options") or [])}
    if name == "grade":
        return {"type": _CHANNEL_MESSAGE, "data": _grade_message(opts.get("city", ""))}
    # Unknown command: acknowledge rather than error, so a stale registration can't
    # look broken to the user.
    return {"type": _CHANNEL_MESSAGE,
            "data": {"content": "Unknown command.", "flags": _EPHEMERAL}}


def handle(body: bytes, signature: str, timestamp: str) -> tuple[int, dict]:
    """Verify and route one interaction. Returns (http_status, json_body).

    401 on a bad/missing signature (what Discord's security probe expects); a PING
    is answered with a PONG; an application command is dispatched. The caller reads
    the RAW body for verification and passes it here unparsed."""
    if not verify(signature, timestamp, body):
        return 401, {"error": "bad signature"}
    try:
        payload = json.loads(body)
    except ValueError:
        return 400, {"error": "bad json"}
    itype = payload.get("type")
    if itype == _PING:
        return 200, {"type": _PONG}
    if itype == _APPLICATION_COMMAND:
        return 200, _command_response(payload.get("data") or {})
    # Components/modals/etc. aren't used yet; a PONG-shaped 200 is harmless.
    return 200, {"type": _CHANNEL_MESSAGE,
                 "data": {"content": "Unsupported interaction.", "flags": _EPHEMERAL}}