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