Jinemi/thermograph
4
0
Fork 0
Code Issues 2 Pull requests 3 Projects Releases Packages Wiki Activity Actions 2

Repo-split Stage 4: cut prod traffic to backend + frontend, dual-service (#14)

Browse source
This commit is contained in:
emi 2026-07-21 20:01:30 +00:00
parent 4d86bbce83
commit f21aa2c2e0
12 changed files with 163 additions and 2007 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
api/content_payloads.py
View file

@ -70,6 +70,15 @@ def _clamp_desc(text: str, limit: int = 155) -> str:
return text[:limit].rsplit(" ", 1)[0].rstrip(",;—-") + "…"
def _month_year(date_s) -> str:
"""'1995-07-13' -> 'Jul 1995'. Meta descriptions have ~155 characters to
spend, so a record's date gives up its day."""
try:
return datetime.date.fromisoformat(str(date_s)).strftime("%b %Y")
except (ValueError, TypeError):
return str(date_s)
def _temp_text(f, unit: str) -> str:
if f is None:
return "—"
@ -375,8 +384,8 @@ def records_payload(request_origin: str, base: str, city: dict, history) -> dict
"page_title": _titled(f"{title} weather records: hottest & coldest days since {year_range[0]}"),
"page_description": _clamp_desc(
f"{title}'s hottest day hit {_temp_text(hi_rec, unit)}"
f"{f' ({hi_date})' if hi_date else ''}; its coldest fell to "
f"{_temp_text(lo_rec, unit)}{f' ({lo_date})' if lo_date else ''}. "
f"{f' ({_month_year(hi_date)})' if hi_date else ''}; its coldest fell to "
f"{_temp_text(lo_rec, unit)}{f' ({_month_year(lo_date)})' if lo_date else ''}. "
f"Every day graded against {n_years} years of local history."),
"jsonld": jsonld,
}

13
deploy/entrypoint.sh
View file

@ -7,6 +7,17 @@
# ready to accept the migration.
set -euo pipefail
# THERMOGRAPH_SERVICE_ROLE (repo-split Stage 4): which app this container
# actually runs -- backend (default, today's exact behavior: migrations, then
# backend/app.py) or frontend (frontend_ssr/app.py -- stateless, no DB, no
# migrations at all). Not to be confused with THERMOGRAPH_ROLE (web|worker),
# which only applies within the backend role.
if [ "${THERMOGRAPH_SERVICE_ROLE:-backend}" = "frontend" ]; then
cd /app/frontend_ssr
echo "==> Starting uvicorn (frontend) on 0.0.0.0:${PORT:-8137} with ${WORKERS:-4} worker(s)"
exec uvicorn app:app --host 0.0.0.0 --port "${PORT:-8137}" --workers "${WORKERS:-4}"
fi
cd /app/backend
# --- Swarm secrets shim ------------------------------------------------------
@ -77,5 +88,5 @@ if [ "${RUN_MIGRATIONS:-1}" != "0" ]; then
run_migrations
fi
echo "==> Starting uvicorn on 0.0.0.0:${PORT:-8137} with ${WORKERS:-4} worker(s)"
echo "==> Starting uvicorn (backend) on 0.0.0.0:${PORT:-8137} with ${WORKERS:-4} worker(s)"
exec uvicorn app:app --host 0.0.0.0 --port "${PORT:-8137}" --workers "${WORKERS:-4}"

4
tests/conftest.py
View file

@ -28,6 +28,10 @@ _TMP = tempfile.mkdtemp(prefix="thermograph-tests-")
# start the background notifier thread during tests. Set before db is imported.
os.environ.setdefault("THERMOGRAPH_ACCOUNTS_DB", os.path.join(_TMP, "accounts.sqlite"))
os.environ.setdefault("THERMOGRAPH_ENABLE_NOTIFIER", "0")
# web/app.py (repo-split Stage 4) fails loud at import if this is unset -- tests
# never actually hit a frontend-owned path through the live proxy (that's
# frontend_ssr/tests' job), so an unreachable placeholder is fine here.
os.environ.setdefault("THERMOGRAPH_FRONTEND_BASE_INTERNAL", "http://127.0.0.1:1")
# Pin the IndexNow key so it isn't generated + written into the repo's data/ dir.
os.environ.setdefault("THERMOGRAPH_INDEXNOW_KEY", "test0000indexnow0000key000000000")

9
tests/notifications/test_digest.py
View file

@ -112,10 +112,5 @@ def test_signup_records_event(client):
assert sum(sum(days.values()) for days in after.values()) == before_n + 1
def test_form_is_hidden_for_now(client):
"""The footer digest signup is temporarily hidden (commented out in the
base template). The /digest endpoint stays live; only the UI is pulled.
Restore the form and flip this back to asserting presence."""
for path in ["/", "/about", "/climate", "/privacy", "/glossary"]:
html = client.get(f"{B}{path}").text
assert "data-digest" not in html, path
# test_form_is_hidden_for_now moved to frontend_ssr/tests/test_content.py
# (repo-split Stage 4) -- every page it checked is frontend-owned now.

30
tests/test_indexnow.py
View file

@ -57,3 +57,33 @@ def test_default_base_url_falls_back_to_env(monkeypatch, tmp_path):
monkeypatch.setattr(indexnow, "submit_all", lambda base, **kw: calls.append(base) or _ok())
indexnow.submit_if_changed()
assert calls == ["https://from-env.example"]
# --- submit_all / url_signature / state file, direct (ported from
# tests/web/test_content.py, repo-split Stage 4) ----------------------------
def test_indexnow_submit_all_builds_payload(monkeypatch):
calls = []
class _Resp:
status_code = 200
text = "ok"
monkeypatch.setattr(indexnow.httpx, "post",
lambda url, json=None, **kw: (calls.append((url, json)), _Resp())[1])
res = indexnow.submit_all("https://thermograph.org")
assert res["submitted"] == res["total"] > 100
first = calls[0][1]
assert first["host"] == "thermograph.org"
assert first["keyLocation"] == f"https://thermograph.org/{indexnow.key()}.txt"
assert first["urlList"][0] == "https://thermograph.org/"
assert all(len(c[1]["urlList"]) <= 10000 for c in calls) # batched under the IndexNow cap
def test_indexnow_if_changed_state(monkeypatch, tmp_path):
monkeypatch.setattr(indexnow, "_STATE_PATH", str(tmp_path / "state.txt"))
sig = indexnow.url_signature()
assert len(sig) == 40 and sig == indexnow.url_signature() # stable sha1 of the URL set
assert indexnow._read_state() == "" # first deploy → would submit
indexnow._write_state(sig)
assert indexnow._read_state() == sig # unchanged → deploy would skip

6
tests/web/test_api.py
View file

@ -186,7 +186,11 @@ def test_suggest_falls_back_to_upstream_geocoder(client, monkeypatch):
def test_pages_serve_with_origin_filled_in(client):
r = client.get("/thermograph/")
"""/ moved to frontend_ssr (repo-split Stage 4) -- it renders via real Jinja
template vars now, not a __ORIGIN__ string substitution, so there's nothing
left to check about it here. /calendar is still a _page()-served static
shell with the substitution, unaffected by the split."""
r = client.get("/thermograph/calendar")
assert r.status_code == 200
assert "text/html" in r.headers["content-type"]
assert "__ORIGIN__" not in r.text

660
tests/web/test_content.py
View file

@ -1,660 +0,0 @@
"""Tests for the crawlable SEO content pages: the city set, robots/sitemap, and
that the server-rendered pages contain the stats in the HTML (with the weather
layer faked, like test_api.py)."""
import math
import re
import pytest
from fastapi.testclient import TestClient
from web import app as appmod
from data import cities
from data import climate
from data import grading
# BASE defaults to /thermograph in tests (THERMOGRAPH_BASE unset).
B = "/thermograph"
SLUG = "london-england-gb" # present in cities.json
@pytest.fixture
def client(monkeypatch, history, recent):
monkeypatch.setattr(climate, "load_cached_history", lambda cell: history.clone())
monkeypatch.setattr(climate, "get_recent_forecast", lambda cell: recent.clone())
monkeypatch.setattr(climate, "get_history", lambda cell: (history.clone(), {"cached": True}))
return TestClient(appmod.app)
def test_city_set_slugs_unique_and_lookup():
slugs = cities.all_slugs()
assert len(slugs) == len(set(slugs)) >= 100
assert cities.get(SLUG) is not None
assert cities.get("does-not-exist") is None
def test_robots_txt(client):
r = client.get(f"{B}/robots.txt")
assert r.status_code == 200
assert "Sitemap:" in r.text and "/sitemap.xml" in r.text
assert "Disallow: /thermograph/api/" in r.text
def test_sitemap_lists_city_urls(client):
r = client.get(f"{B}/sitemap.xml")
assert r.status_code == 200
assert "<urlset" in r.text
assert f"/climate/{SLUG}</loc>" in r.text
assert f"/climate/{SLUG}/july</loc>" in r.text
assert f"/climate/{SLUG}/records</loc>" in r.text
def test_city_page_renders_stats_in_html(client):
r = client.get(f"{B}/climate/{SLUG}")
assert r.status_code == 200
b = r.text
assert "London" in b and "climate" in b.lower()
assert 'rel="canonical"' in b and f"/climate/{SLUG}" in b
assert '"@type":"Dataset"' in b
assert "average temperatures by month" in b.lower()
# a real number rendered in the normals/today text (London is GB, so °C)
assert re.search(r"-?\d+°C</span>", b)
def _jsonld_breadcrumbs(html: str) -> list[dict]:
import json, re
m = re.search(r'<script type="application/ld\+json">(.*?)</script>', html, re.S)
assert m, "no JSON-LD block on page"
graph = json.loads(m.group(1))["@graph"]
return [n for n in graph if n["@type"] == "BreadcrumbList"]
@pytest.mark.parametrize("path", [f"/climate/{SLUG}", f"/climate/{SLUG}/records"])
def test_breadcrumb_jsonld_items_valid_for_google(client, path):
# Google: every ListItem except the last must carry an ``item`` URL, so
# unlinked intermediate crumbs (the country) must not appear in JSON-LD.
(bc,) = _jsonld_breadcrumbs(client.get(B + path).text)
els = bc["itemListElement"]
assert [e["position"] for e in els] == list(range(1, len(els) + 1))
for e in els[:-1]:
assert e["item"].startswith("http"), f"crumb {e['name']!r} missing item"
def test_city_404(client):
assert client.get(f"{B}/climate/nope-not-a-city").status_code == 404
def test_curated_events_have_valid_slugs():
from data import city_events
slugs = set(cities.all_slugs())
assert len(city_events.EVENTS) >= 20
assert all(s in slugs for s in city_events.EVENTS), \
[s for s in city_events.EVENTS if s not in slugs]
def test_curated_event_renders(client):
b = client.get(f"{B}/climate/new-orleans-louisiana-us").text
assert "Notable weather in New Orleans" in b
assert "Hurricane Katrina" in b
def test_uncurated_city_has_no_event_section(client):
b = client.get(f"{B}/climate/shanghai-cn").text
assert "city-event" not in b # falls back to the flavor blurb
def test_city_travel_cta_prefills_compare(client):
b = client.get(f"{B}/climate/{SLUG}").text
assert "Thinking of visiting" in b
assert "/compare#loc=" in b # the city is pre-loaded on the compare page
def test_city_blurb_renders_with_attribution(client, monkeypatch):
monkeypatch.setattr(cities, "flavor",
lambda slug: {"extract": "Testville is a lovely place to test.",
"url": "https://en.wikipedia.org/wiki/Testville", "title": "Testville"})
b = client.get(f"{B}/climate/{SLUG}").text
assert "Testville is a lovely place to test." in b
assert "via Wikipedia" in b # CC BY-SA attribution link
def test_month_and_records(client):
assert client.get(f"{B}/climate/{SLUG}/july").status_code == 200
assert client.get(f"{B}/climate/{SLUG}/records").status_code == 200
assert client.get(f"{B}/climate/{SLUG}/notamonth").status_code == 404
def test_month_records_cover_every_metric_high_and_low(client):
# The month page's records show a record high AND low for each metric present, not
# just warmest/coldest temperature — one card per metric with both extremes. (The
# test fixture carries tmax/tmin/precip; real cities add feels/humidity/wind/gust.)
b = client.get(f"{B}/climate/{SLUG}/july").text
assert "July records" in b
for label in ("High", "Low", "Precip"):
assert f'class="rc-metric">{label}<' in b
# Every card carries a labelled high and low row (temp metrics use Highest/Lowest).
n = b.count('rc-row rc-high')
assert n >= 3 and b.count('rc-row rc-low') == n
assert "Highest" in b and "Lowest" in b
# Precip's extremes are the wettest/driest whole month by total accumulation.
assert "Wettest" in b and "Driest" in b
def test_records_page_has_monthly_and_seasonal(client):
b = client.get(f"{B}/climate/{SLUG}/records").text
# Monthly records: all 12 months, each linking to its month page.
assert "Records by month" in b
for month in ("January", "July", "December"):
assert month in b
assert f"/climate/{SLUG}/january" in b
# Seasonal records: labelled for London's Northern Hemisphere, with month spans.
assert "Records by season" in b
assert "Northern Hemisphere" in b
assert 'Winter <span class="season-span">(Dec–Feb)</span>' in b
assert 'Summer <span class="season-span">(Jun–Aug)</span>' in b
# All-time table still present, plus structured data and real values.
assert "All-time records" in b
assert '"@type":"Dataset"' in b
assert re.search(r"-?\d+°C</span>", b)
# The Dataset structured data still advertises the month/season coverage.
assert "monthly" in b.lower() and "seasonal" in b.lower()
def test_records_seasons_flip_in_southern_hemisphere(client):
# São Paulo is south of the equator, so Dec–Feb is summer, not winter.
b = client.get(f"{B}/climate/sao-paulo-br/records").text
assert "Southern Hemisphere" in b
assert 'Summer <span class="season-span">(Dec–Feb)</span>' in b
assert 'Winter <span class="season-span">(Jun–Aug)</span>' in b
def test_climate_pages_are_colour_coded(client):
# Temperatures wear the diverging cold→hot palette (a heat map, not a plain grid),
# and the city page carries the monthly temperature-range strip.
city = client.get(f"{B}/climate/{SLUG}").text
assert "t-cell t-" in city # tinted temperature cells in the normals table
assert "range-fill" in city and "--c1:var(--" in city # the monthly range strip's gradient bars
recs = client.get(f"{B}/climate/{SLUG}/records").text
assert "t-inline t-" in recs # tinted record chips
month = client.get(f"{B}/climate/{SLUG}/july").text
assert "t-inline t-" in month # tinted hero numbers + records
def test_records_show_both_extremes_per_metric(client):
# Each metric column shows BOTH its record warmest (▲) and coldest (▼) — so the
# daytime high carries its record-low high, and the overnight low its record-high low.
b = client.get(f"{B}/climate/{SLUG}/records").text
assert "Daytime high" in b and "Overnight low" in b
assert "▲" in b and "▼" in b
# 12 months × 2 metric columns × 2 extremes = 48 chips just in the monthly table.
assert b.count('class="rec-ext"') >= 48
def test_hub_glossary_about(client):
assert client.get(f"{B}/climate").status_code == 200
assert client.get(f"{B}/glossary").status_code == 200
assert client.get(f"{B}/glossary/percentile").status_code == 200
assert client.get(f"{B}/glossary/not-a-term").status_code == 404
assert client.get(f"{B}/about").status_code == 200
def test_static_page_titles_come_from_content_yaml(client):
"""These four pages' SEO title/description are loaded from content/pages.yaml
(content_loader.load_pages) rather than hardcoded — assert the rendered
<title> still carries the expected text, so the extraction is provably
render-unchanged, not just "the file parses". Jinja autoescapes the
{{ page_title }} interpolation, so compare against the same markupsafe
escape the template applies (glossary_index's "&amp;" round-trips through
it as "&amp;amp;" both before and after this extraction — a pre-existing
render quirk this test intentionally preserves, not fixes)."""
from markupsafe import escape
from web import content
for path, key in ((f"{B}/climate", "hub"), (f"{B}/glossary", "glossary_index"),
(f"{B}/about", "about"), (f"{B}/privacy", "privacy")):
html = client.get(path).text
assert f"<title>{escape(content.PAGES[key]['title'])}</title>" in html, path
def test_glossary_terms_come_from_content_yaml(client):
from web import content
html = client.get(f"{B}/glossary").text
for entry in content.GLOSSARY.values():
assert entry["term"] in html
term_html = client.get(f"{B}/glossary/percentile").text
assert content.GLOSSARY["percentile"]["short"] in term_html
def test_footer_has_no_leaked_jinja_comment(client):
# A comment whose prose contained a literal "#}" closed early and leaked
# "wrapper. #}" into the footer. Guard every base-template page against a
# stray Jinja comment delimiter reaching the rendered HTML.
import re
for path in (f"{B}/about", f"{B}/climate/{SLUG}", f"{B}/climate/{SLUG}/records"):
b = client.get(path).text
footer = re.search(r"<footer.*?</footer>", b, re.S)
assert footer, f"no footer on {path}"
assert "#}" not in footer.group(0) and "{#" not in footer.group(0), path
def test_title_name_is_short_and_unique():
# Titles lead with the bare city name so the payload ("… in July") survives
# SERP truncation — display_name()'s region+country is what pushed it off the
# end. The exception is a name that recurs: those must still be told apart,
# and every rendered title must stay unique across the whole city set.
all_cities = cities.all_cities()
titles = [cities.title_name(c) for c in all_cities]
assert len(titles) == len(set(titles)), "two cities share a title"
# The overwhelming majority pay no disambiguation cost at all.
assert sum(1 for t in titles if "," not in t) > 0.9 * len(titles)
by_slug = {c["slug"]: cities.title_name(c) for c in all_cities}
assert by_slug["seattle-washington-us"] == "Seattle"
# Same name, different countries -> country code is enough.
assert by_slug["london-england-gb"] == "London, GB"
assert by_slug["london-ontario-ca"] == "London, CA"
# Same name AND same country -> the country code would collide, so fall back
# to admin1. Without this, both Columbuses would render "Columbus, US".
assert by_slug["columbus-ohio-us"] == "Columbus, Ohio"
assert by_slug["columbus-georgia-us"] == "Columbus, Georgia"
def test_page_titles_lead_with_city_and_payload(client):
for path, must_start, payload in (
(f"{B}/climate/{SLUG}", "London, GB climate", "how unusual it is now"),
(f"{B}/climate/{SLUG}/july", "London, GB in July", "normal weather"),
(f"{B}/climate/{SLUG}/records", "London, GB weather records", "hottest & coldest"),
):
b = client.get(path).text
title = re.search(r"<title>(.*?)</title>", b, re.S).group(1)
title = title.replace("&amp;", "&")
assert title.startswith(must_start), title
assert payload in title, title
# The region+country the old template spent its width on is gone.
assert "England, United Kingdom" not in title
# Whatever else happens, the city and the page's subject are inside the
# first ~40 chars, so a SERP cut can no longer remove the payload.
assert len(must_start) <= 40
# og:title mirrors <title> via base.html.j2's self.title().
og = re.search(r'<meta property="og:title" content="(.*?)"', b, re.S).group(1)
assert og == re.search(r"<title>(.*?)</title>", b, re.S).group(1)
def test_meta_descriptions_lead_with_real_numbers(client):
for path in (f"{B}/climate/{SLUG}", f"{B}/climate/{SLUG}/july",
f"{B}/climate/{SLUG}/records"):
b = client.get(path).text
desc = re.search(r'<meta name="description" content="(.*?)"', b, re.S).group(1)
assert len(desc) <= 155, f"{len(desc)}: {desc}"
# A real temperature, not a generic "averages, records and rainfall" —
# in the city's own unit, since a meta description is never converted
# client-side. London is GB, so these read °C.
assert re.search(r"-?\d+°C", desc), desc
assert desc.endswith("years of local history."), desc
# The records description names the dates the extremes were set.
rec = client.get(f"{B}/climate/{SLUG}/records").text
desc = re.search(r'<meta name="description" content="(.*?)"', rec, re.S).group(1)
assert re.search(r"\([A-Z][a-z]{2} \d{4}\)", desc), desc
_TEMP_SPAN = re.compile(r'<span class="temp" data-temp-f="(-?[\d.]+)"([^>]*)>(-?\d+)°([CF]?)</span>')
def _temp_spans(html):
"""(fahrenheit_attr, shown_number, shown_letter) for every rendered temp span.
Reading the spans rather than the raw document matters: base.html.j2 mentions
"°F/°C" in an HTML comment, so a bare `"°F" in html` assertion passes on a page
whose every temperature is Celsius.
"""
return [(float(f), int(n), letter) for f, _attrs, n, letter in _TEMP_SPAN.findall(html)]
def test_climate_pages_carry_convertible_temps(client):
# Every temperature is a client-convertible span; the old inline "(NN°C)" dual
# form is gone (the toggle now supplies the other unit).
for path in (f"{B}/climate/{SLUG}", f"{B}/climate/{SLUG}/july", f"{B}/climate/{SLUG}/records"):
b = client.get(path).text
spans = _temp_spans(b)
assert spans, f"no temperature spans on {path}"
assert re.search(r"\d+°F \(-?\d+°C\)", b) is None # no leftover dual-unit strings
def test_units_default_to_the_citys_country(client):
# The search snippet shows whatever the server rendered, and impressions skew
# to °C countries — so a UK city page must say °C in the HTML itself, with no
# JS and no stored preference. data-temp-f stays Fahrenheit either way: it is
# what climate.js converts from.
gb = client.get(f"{B}/climate/{SLUG}").text # London, GB
us = client.get(f"{B}/climate/seattle-washington-us").text
gb_spans, us_spans = _temp_spans(gb), _temp_spans(us)
assert gb_spans and us_spans
assert {letter for _f, _n, letter in gb_spans if letter} == {"C"}
assert {letter for _f, _n, letter in us_spans if letter} == {"F"}
# The attribute is still Fahrenheit, and really does differ from the Celsius
# text — this is what breaks if someone "helpfully" converts the attribute too.
assert any(math.floor(f + 0.5) != n for f, n, _l in gb_spans), "data-temp-f was converted"
# On a °F page the attribute and the text agree by definition. Rounded half-up,
# not with Python's round(): the page renders the number JS would.
assert all(math.floor(f + 0.5) == n for f, n, _l in us_spans)
assert 'data-unit-default="C"' in gb
assert 'data-unit-default="F"' in us
_PRECIP_SPAN = re.compile(r'<span class="precip" data-precip-in="([\d.]+)">([\d.]+) (in|mm)</span>')
def test_precip_and_wind_follow_the_citys_unit(client):
# One toggle drives every measure: a °C page shows mm and km/h, not °C mixed
# with inches and mph. The data-* attribute stays imperial in both, so
# climate.js converts from the same source of truth it does for temperature.
gb = client.get(f"{B}/climate/{SLUG}/records").text # London, GB
us = client.get(f"{B}/climate/seattle-washington-us/records").text
gb_p, us_p = _PRECIP_SPAN.findall(gb), _PRECIP_SPAN.findall(us)
assert gb_p and us_p
assert {u for _raw, _shown, u in gb_p} == {"mm"}
assert {u for _raw, _shown, u in us_p} == {"in"}
# The conversion is real, not just a relabel.
for raw, shown, _u in gb_p:
assert abs(float(shown) - float(raw) * 25.4) < 0.51, (raw, shown)
# On an imperial page the attribute and the text agree.
for raw, shown, _u in us_p:
assert abs(float(shown) - float(raw)) < 0.005, (raw, shown)
def test_wind_and_humidity_render_per_unit_system():
# The synthetic history carries only tmax/tmin/precip, so wind and humidity
# never reach a rendered page in tests — exercise the renderers directly
# rather than assert on a page that structurally cannot show them.
from web import content
with content._unit_scope("F"):
assert content._wind_text(10) == "10 mph"
assert '<span class="wind" data-wind-mph="10.0">10 mph</span>' == content._wind(10)
assert content._fmt("humid", 7.75) == "7.8 g/m³"
with content._unit_scope("C"):
# 10 mph -> 16.09 km/h, rounded half-up like JS.
assert content._wind_text(10) == "16 km/h"
# data-wind-mph stays imperial: it is what climate.js converts from.
assert 'data-wind-mph="10.0"' in content._wind(10)
# Absolute humidity has no imperial equivalent anyone would recognise, so
# it reads the same in both systems.
assert content._fmt("humid", 7.75) == "7.8 g/m³"
def test_precip_precision_differs_by_unit():
# Rainfall is reported in whole millimetres; a tenth of a mm is below what
# the source resolves. Inches keep two decimals.
from web import content
with content._unit_scope("F"):
assert content._precip_text(0.04) == "0.04 in"
with content._unit_scope("C"):
assert content._precip_text(0.04) == "1 mm"
assert content._precip_text(1.81) == "46 mm"
def test_measure_conversion_constants_match_the_frontend():
# Same drift risk as the country list: two hand-maintained copies.
import os
from web import content
units_js = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
os.pardir, "frontend", "units.js")
with open(units_js, encoding="utf-8") as f:
src = f.read()
mm = float(re.search(r"MM_PER_IN\s*=\s*([\d.]+)", src).group(1))
kmh = float(re.search(r"KMH_PER_MPH\s*=\s*([\d.]+)", src).group(1))
assert mm == content.MM_PER_IN
assert kmh == content.KMH_PER_MPH
def test_unit_default_absent_without_a_city(client):
# No city -> no attribute -> units.js keeps today's locale behaviour. Pinning
# these to "F" would silently regress the interactive pages.
for path in (f"{B}/", f"{B}/climate", f"{B}/about", f"{B}/glossary"):
assert "data-unit-default" not in client.get(path).text, path
def test_unit_does_not_leak_between_requests(client):
# The page handlers are sync `def`, so Starlette runs them on a recycled
# threadpool thread: a unit left set would bleed into the next request served
# by that thread. Interleave and re-check rather than trusting the reset.
assert 'data-unit-default="C"' in client.get(f"{B}/climate/{SLUG}").text
about = client.get(f"{B}/about").text
assert "data-unit-default" not in about
# about.html.j2 renders an illustrative temperature; it must stay °F.
assert {l for _f, _n, l in _temp_spans(about) if l} == {"F"}
us = client.get(f"{B}/climate/seattle-washington-us").text
assert {l for _f, _n, l in _temp_spans(us) if l} == {"F"}
gb_again = client.get(f"{B}/climate/{SLUG}").text
assert {l for _f, _n, l in _temp_spans(gb_again) if l} == {"C"}
def test_f_country_list_matches_the_frontend():
# Two hand-maintained copies of the same 14 codes; nothing else stops them
# drifting apart, and drift means the server and the client disagree about
# what a first-time visitor should see.
import os
from web import content
units_js = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
os.pardir, "frontend", "units.js")
with open(units_js, encoding="utf-8") as f:
src = f.read()
literal = re.search(r"F_REGIONS = new Set\(\[(.*?)\]\)", src, re.S).group(1)
assert set(re.findall(r'"([A-Z]{2})"', literal)) == set(content.F_COUNTRIES)
def test_celsius_rounding_matches_js():
# Python's round() goes to even, JS's Math.round goes up. Where they disagree
# the server prints one number and climate.js repaints another — a flicker for
# the visitor whose unit already matched.
from web import content
assert content._round_half_up(0.5) == 1 # round(0.5) would give 0
assert content._round_half_up(1.5) == 2
assert content._round_half_up(2.5) == 3 # round(2.5) would give 2
assert content._round_half_up(-0.5) == 0 # Math.round(-0.5) === 0
# 31.1°F -> -0.5°C exactly: the case that actually shows up on a cold page.
assert content._c(31.1) == 0
def test_city_page_etag_is_stable(client):
# The unit comes from the URL's city, never from a request header, so the
# response stays a pure function of the URL — no Vary, and the ETag still works.
r1 = client.get(f"{B}/climate/{SLUG}")
r2 = client.get(f"{B}/climate/{SLUG}")
assert r1.headers["etag"] == r2.headers["etag"]
r3 = client.get(f"{B}/climate/{SLUG}", headers={"If-None-Match": r1.headers["etag"]})
assert r3.status_code == 304
def test_seo_pages_load_unit_and_account_scripts(client):
# Header parity with the interactive pages: the °F/°C toggle, account/bell, and
# the temperature converter are all loaded.
b = client.get(f"{B}/climate/{SLUG}").text
for src in (f"{B}/units.js", f"{B}/account.js", f"{B}/climate.js"):
assert f'src="{src}"' in b
def test_indexnow_key_file_served(client):
import indexnow
k = indexnow.key()
r = client.get(f"{B}/{k}.txt")
assert r.status_code == 200
assert r.text.strip() == k
assert r.headers["content-type"].startswith("text/plain")
def test_sitemap_lastmod_is_stable(client):
import datetime
import re
body = client.get(f"{B}/sitemap.xml").text
mods = set(re.findall(r"<lastmod>([^<]+)</lastmod>", body))
assert len(mods) == 1 # one build date, not a per-request churn
datetime.date.fromisoformat(next(iter(mods))) # a valid ISO date
def test_indexnow_submit_all_builds_payload(monkeypatch):
import indexnow
calls = []
class _Resp:
status_code = 200
text = "ok"
monkeypatch.setattr(indexnow.httpx, "post",
lambda url, json=None, **kw: (calls.append((url, json)), _Resp())[1])
res = indexnow.submit_all("https://thermograph.org")
assert res["submitted"] == res["total"] > 100
first = calls[0][1]
assert first["host"] == "thermograph.org"
assert first["keyLocation"] == f"https://thermograph.org/{indexnow.key()}.txt"
assert first["urlList"][0] == "https://thermograph.org/"
assert all(len(c[1]["urlList"]) <= 10000 for c in calls) # batched under the IndexNow cap
def test_indexnow_if_changed_state(monkeypatch, tmp_path):
import indexnow
monkeypatch.setattr(indexnow, "_STATE_PATH", str(tmp_path / "state.txt"))
sig = indexnow.url_signature()
assert len(sig) == 40 and sig == indexnow.url_signature() # stable sha1 of the URL set
assert indexnow._read_state() == "" # first deploy → would submit
indexnow._write_state(sig)
assert indexnow._read_state() == sig # unchanged → deploy would skip
def test_search_verification_meta(client, monkeypatch):
monkeypatch.setenv("THERMOGRAPH_GOOGLE_VERIFY", "gtok123")
monkeypatch.setenv("THERMOGRAPH_BING_VERIFY", "btok456")
seo = client.get(f"{B}/climate/{SLUG}").text # Jinja SEO page
home = client.get(f"{B}/").text # static homepage via _page()
for b in (seo, home):
assert '<meta name="google-site-verification" content="gtok123">' in b
assert '<meta name="msvalidate.01" content="btok456">' in b
# The brand lockup is the site-wide "go home" affordance. The static tool pages
# and the Jinja pages build their headers separately, so this is easy to add to
# one and forget on the other — assert every page carries it.
BRAND_PAGES = ["/", "/calendar", "/day", "/compare", "/legend", "/alerts",
"/about", "/privacy", "/climate", "/glossary"]
@pytest.mark.parametrize("path", BRAND_PAGES)
def test_brand_lockup_links_home(client, path):
html = client.get(f"{B}{path}").text
brand = html.split('<div class="brand">', 1)[1].split("</header>", 1)[0]
head = brand.split("</h1>")[0] if "<h1" in brand else brand.split("</p>")[0]
# The link must wrap the whole lockup, so clicking the mark works too, not
# just the word.
assert "<a href=" in head, f"{path}: brand is not a link"
href = head.split('<a href="', 1)[1].split('"', 1)[0]
assert href in ("./", f"{B}/"), f"{path}: brand links to {href!r}, not home"
assert '<span class="logo">' in head.split("<a href=", 1)[1], \
f"{path}: the mark sits outside the home link"
# Percentiles are shown on the Day page, the calendar tooltip, the chart, the
# city pages and the homepage strip. They all have to say the same thing about
# the same reading, so the rule lives in grading.pct_ordinal() and every surface
# defers to it (the frontend mirrors it as pctOrd in shared.js).
def test_pct_ordinal_rule():
assert grading.pct_ordinal(66.4) == "66th"
assert grading.pct_ordinal(1.2) == "1st"
assert grading.pct_ordinal(22) == "22nd"
assert grading.pct_ordinal(3) == "3rd"
assert grading.pct_ordinal(11) == "11th"
# Floored into 1..99: an empirical percentile is a rank against the sample,
# so "100th percentile" reads as a measurement error.
assert grading.pct_ordinal(99.6) == "99th"
assert grading.pct_ordinal(100) == "99th"
assert grading.pct_ordinal(0) == "1st"
assert grading.pct_ordinal(None) == "—"
def test_city_page_percentiles_are_real_ordinals(client):
"""The city page used to render `{{ c.percentile }}th pct` against a float,
producing "97.1th pct" and "1th pct" on every one of the ~1000 city pages."""
html = client.get(f"{B}/climate/{SLUG}").text
assert "th pct" in html or "st pct" in html or "nd pct" in html or "rd pct" in html
# No float ordinals, no impossible ones, no "1th"/"2th"/"3th".
for bad in re.findall(r"\d+\.\d+(?:st|nd|rd|th)|100th|\b0th|\b\d*[123]th", html):
raise AssertionError(f"malformed percentile ordinal: {bad!r}")
# --- content API parity (repo-split Stage 2) ----------------------------------
# The new /api/v2/content/* endpoints (backend/api/content_routes.py) recompute
# the same numbers this file's SSR pages render, independently (content.py's
# Markup-wrapped spans vs. the API's raw floats). These assert the two never
# drift apart while both exist side by side -- see backend/api/content_payloads.py.
def _data_temp_fs(html: str) -> set[float]:
return {round(float(v), 1) for v in re.findall(r'data-temp-f="(-?[\d.]+)"', html)}
def test_content_api_city_matches_rendered_html(client):
html = client.get(f"{B}/climate/{SLUG}").text
rendered = _data_temp_fs(html)
j = client.get(f"{B}/api/v2/content/city/{SLUG}").json()
assert j["canonical_path"] == f"/climate/{SLUG}"
assert j["city"]["slug"] == SLUG
# Every month's high/low the API reports must appear somewhere in the
# page's client-convertible spans -- same underlying grading.climatology()
# call, so a real drift here means one side silently diverged.
for m in j["months"]:
if m["high_f"] is not None:
assert round(m["high_f"], 1) in rendered, f"{m['slug']} high_f not on page"
if m["low_f"] is not None:
assert round(m["low_f"], 1) in rendered, f"{m['slug']} low_f not on page"
# All-time tmax/tmin extremes likewise.
rec = j["all_time_records"]
assert round(rec["tmax"]["max"], 1) in rendered
assert round(rec["tmin"]["min"], 1) in rendered
def test_content_api_month_matches_rendered_html(client):
html = client.get(f"{B}/climate/{SLUG}/july").text
rendered = _data_temp_fs(html)
j = client.get(f"{B}/api/v2/content/city/{SLUG}/month/july").json()
assert j["month_slug"] == "july"
if j["avg_high_f"] is not None:
assert round(j["avg_high_f"], 1) in rendered
if j["avg_low_f"] is not None:
assert round(j["avg_low_f"], 1) in rendered
def test_content_api_records_matches_rendered_html(client):
html = client.get(f"{B}/climate/{SLUG}/records").text
rendered = _data_temp_fs(html)
j = client.get(f"{B}/api/v2/content/city/{SLUG}/records").json()
for row in j["rows"]:
if row["label"] == "Precip":
continue # precip isn't a data-temp-f span
assert round(row["high_f"], 1) in rendered
if row["low_f"] is not None:
assert round(row["low_f"], 1) in rendered
def test_content_api_unknown_city_404s(client):
assert client.get(f"{B}/api/v2/content/city/not-a-real-city").status_code == 404
def test_content_api_unknown_month_404s(client):
assert client.get(f"{B}/api/v2/content/city/{SLUG}/month/notamonth").status_code == 404
def test_content_api_home_city_slugs_match_content_py():
"""Two hand-maintained copies (content.py's HOME_CITY_SLUGS and
content_payloads.py's) -- pin them together like the other parity tests
in this file."""
from web import content
from api import content_payloads
assert content.HOME_CITY_SLUGS == content_payloads.HOME_CITY_SLUGS

139
tests/web/test_content_loader.py
View file

@ -1,139 +0,0 @@
"""Tests for content_loader.py: schema validation (fail loud on a malformed
content/ file) and that the real glossary.yaml/pages.yaml load correctly."""
import pytest
from web import content_loader
def _write(tmp_path, name, text):
path = tmp_path / name
path.write_text(text, encoding="utf-8")
return str(path)
# --- load_glossary: schema validation --------------------------------------
def test_load_glossary_rejects_non_mapping_top_level(tmp_path):
path = _write(tmp_path, "g.yaml", "- just\n- a\n- list\n")
with pytest.raises(ValueError, match="top-level mapping"):
content_loader.load_glossary(path)
def test_load_glossary_rejects_missing_terms_key(tmp_path):
path = _write(tmp_path, "g.yaml", "not_terms: []\n")
with pytest.raises(ValueError, match="non-empty list"):
content_loader.load_glossary(path)
def test_load_glossary_rejects_empty_terms(tmp_path):
path = _write(tmp_path, "g.yaml", "terms: []\n")
with pytest.raises(ValueError, match="non-empty list"):
content_loader.load_glossary(path)
def test_load_glossary_rejects_missing_required_field(tmp_path):
path = _write(tmp_path, "g.yaml", """
terms:
- slug: foo
term: Foo
short: A short definition.
""")
with pytest.raises(ValueError, match="foo.*body"):
content_loader.load_glossary(path)
def test_load_glossary_rejects_blank_required_field(tmp_path):
path = _write(tmp_path, "g.yaml", """
terms:
- slug: foo
term: Foo
short: ""
body: Full body.
""")
with pytest.raises(ValueError, match="foo.*short"):
content_loader.load_glossary(path)
def test_load_glossary_rejects_duplicate_slug(tmp_path):
path = _write(tmp_path, "g.yaml", """
terms:
- slug: foo
term: Foo
short: s1
body: b1
- slug: foo
term: Foo Again
short: s2
body: b2
""")
with pytest.raises(ValueError, match="duplicate slug"):
content_loader.load_glossary(path)
def test_load_glossary_accepts_well_formed_content(tmp_path):
path = _write(tmp_path, "g.yaml", """
terms:
- slug: foo
term: Foo
short: A short definition.
body: A full body with <b>markup</b>.
- slug: bar
term: Bar
short: Another short definition.
body: Another full body.
""")
glossary = content_loader.load_glossary(path)
assert list(glossary) == ["foo", "bar"] # file order preserved
assert glossary["foo"] == {
"term": "Foo", "short": "A short definition.", "body": "A full body with <b>markup</b>.",
}
# --- load_pages: schema validation ------------------------------------------
def test_load_pages_rejects_missing_pages_key(tmp_path):
path = _write(tmp_path, "p.yaml", "not_pages: {}\n")
with pytest.raises(ValueError, match="non-empty mapping"):
content_loader.load_pages(path)
def test_load_pages_rejects_missing_required_field(tmp_path):
path = _write(tmp_path, "p.yaml", """
pages:
about:
title: About
""")
with pytest.raises(ValueError, match="about.*description"):
content_loader.load_pages(path)
def test_load_pages_accepts_well_formed_content(tmp_path):
path = _write(tmp_path, "p.yaml", """
pages:
about:
title: About Us
description: A description.
""")
pages = content_loader.load_pages(path)
assert pages == {"about": {"title": "About Us", "description": "A description."}}
# --- the real content/ files ------------------------------------------------
def test_real_glossary_loads_and_matches_the_rendered_site():
"""The module-level GLOSSARY content.py loads at import IS content_loader's
output — this just re-asserts the real file is well-formed and non-trivial,
since a broken content/glossary.yaml would already have failed every other
test at collection time (content.py loads it at import)."""
glossary = content_loader.load_glossary()
assert len(glossary) >= 8
assert "percentile" in glossary
for slug, entry in glossary.items():
assert entry["term"] and entry["short"] and entry["body"], slug
def test_real_pages_loads_the_four_static_pages():
pages = content_loader.load_pages()
assert set(pages) == {"about", "privacy", "hub", "glossary_index"}
for key, entry in pages.items():
assert entry["title"] and entry["description"], key

182
tests/web/test_homepage.py
View file

@ -1,5 +1,7 @@
"""Tests for the homepage: the cache-only "unusual right now" precompute, the
server-rendered surfaces, and the digest signup.
"""Tests for the homepage's cache-only "unusual right now" precompute
(api/homepage.py) -- the rendered-page assertions (repo-split Stage 4) moved
to frontend_ssr/tests/test_homepage.py, since content.py's Jinja rendering
lives there now, not in this process.
The precompute must never touch the network — a sweep over ~1000 cities that
fetched would burst the archive quota — so the tests here assert that the
@ -7,48 +9,17 @@ fetching climate helpers are not called at all.
"""
import datetime
import json
import pathlib
import time
import pytest
from fastapi.testclient import TestClient
from api import homepage
from web import app as appmod
from data import cities
from data import climate
from data import store
from web import content
B = "/thermograph"
@pytest.fixture
def client():
return TestClient(appmod.app)
@pytest.fixture
def no_feed(monkeypatch):
"""The homepage with no precomputed feed at all."""
monkeypatch.setattr(homepage, "load", lambda: None)
return None
def _pick(name, pct, cls, grade, tail, slug=None):
return {
"slug": slug or (name.lower().replace(" ", "-") + "-xx"),
"name": name, "display": f"{name}, Somewhere",
"lat": 1.0, "lon": 2.0, "cell_id": "1_2",
"date": datetime.date.today().isoformat(),
"metric": "tmax", "metric_label": "High temp", "window_label": "mid-July",
"value": 99.0, "normal": 88.0, "percentile": pct,
"pct_label": homepage._pct_label(pct),
"grade": grade, "grade_label": homepage._grade_label(grade, tail), "cls": cls,
"departure": abs(pct - 50), "tail": tail,
}
# --- the precompute ----------------------------------------------------------
def test_build_never_fetches_upstream(monkeypatch):
"""The sweep uses the cache-only readers, never the fetching ones."""
@ -210,146 +181,15 @@ def test_staleness(monkeypatch):
assert homepage.is_stale({"date": "1999-01-01", "generated_at": time.time()})
# --- the rendered page -------------------------------------------------------
def test_homepage_is_complete_without_js(client, no_feed):
"""A crawler (or a reader with JS off) gets the whole page as real HTML."""
r = client.get(f"{B}/")
assert r.status_code == 200
html = r.text
assert "How unusual is your weather?" in html
assert "How it works" in html
assert "Climate context for 1,000 cities" in html
assert "Free. No ads. No tracking." in html
# The interactive tool's DOM contract survives the move to Jinja.
for dom_id in ('id="find-btn"', 'id="date-input"', 'id="panel"',
'id="placeholder"', 'id="results"'):
assert dom_id in html
# Exactly one h1: the hero headline. The brand degrades to a <p>.
assert html.count("<h1") == 1
assert 'p class="site-name"' in html
def test_hero_locate_has_a_status_slot(client, no_feed):
"""The locate button needs somewhere to report a declined prompt or a
timeout. Without it the control fails silently, which is how it shipped
broken: navigator.geolocation exists on plain http:// but always errors."""
html = client.get(f"{B}/").text
assert 'id="hero-locate"' in html
assert 'id="hero-locate-msg"' in html
assert 'aria-live="polite"' in html
def test_homepage_brand_keeps_the_lockup_styling(client, no_feed):
"""The homepage renders the brand as <p class="site-name"> so the hero owns
the page's only h1. The lockup styling is written against `.brand h1`, so
the class MUST be matched there too — otherwise the brand silently falls
back to block layout, which baseline-aligns the mark instead of centring it
and drops the monospace wordmark. Nothing else catches this: the page still
renders, and every other test still passes, while the header looks wrong.
"""
html = client.get(f"{B}/").text
assert 'class="site-name"' in html
assert "<h1 class=\"site-name\"" not in html # the hero owns the h1
css = (pathlib.Path(__file__).parents[3] / "frontend" / "style.css").read_text()
lockup = css.split(".brand h1", 1)[1].split("}", 1)[0]
assert ".brand .site-name" in lockup, (
"the .brand h1 lockup rule must also match .brand .site-name"
)
def test_city_chips_all_resolve(client, no_feed):
"""Every homepage chip must be a real, routable city — a stale slug would
render an empty chip list and leak a 404 link."""
resolved = content._home_cities()
assert len(resolved) == len(content.HOME_CITY_SLUGS)
html = client.get(f"{B}/").text
for city in resolved:
assert f"/climate/{city['slug']}" in html
assert cities.get(city["slug"]) is not None
def test_records_strip_shows_both_tails(client, monkeypatch):
"""Two-sided honesty: a qualifying cold-tail city appears alongside the heat."""
feed = {
"generated_at": time.time(), "date": datetime.date.today().isoformat(),
"considered": 800,
"picks": {"extreme": _pick("Phoenix", 99.4, "rec-hot", "Near Record", "warm")},
"ranked": [_pick("Phoenix", 99.4, "rec-hot", "Near Record", "warm"),
_pick("Ushuaia", 2.1, "very-cold", "Very Low", "cold")],
}
monkeypatch.setattr(homepage, "load", lambda: feed)
html = client.get(f"{B}/").text
assert "Unusual right now" in html
assert "Phoenix" in html and "Ushuaia" in html
# Band colour comes from the shared tokens, and the band word is always
# present too — colour is never the only signal.
assert "var(--rec-hot)" in html and "var(--very-cold)" in html
assert "Near Record" in html and "Very Low" in html
def test_strip_cards_carry_value_metric_and_normal(client, monkeypatch):
feed = {
"generated_at": time.time(), "date": datetime.date.today().isoformat(),
"considered": 800, "picks": {},
"ranked": [_pick("Tehran", 99.6, "rec-hot", "Near Record", "warm"),
_pick("Ushuaia", 0.4, "rec-cold", "Near Record", "cold")],
}
monkeypatch.setattr(homepage, "load", lambda: feed)
html = client.get(f"{B}/").text
# Both tails must be tellable apart in WORDS, not just by colour.
assert "Near Record hot" in html and "Near Record cold" in html
assert "normal" in html and "High temp" in html
assert "100th" not in html and "0th" not in html
# Temperatures render as convertible spans, and the converter is loaded, or
# the strip would stay in °F while the °F/°C toggle says °C.
assert 'class="temp" data-temp-f' in html
assert "climate.js" in html
def test_hero_uses_most_unusual_default(client, monkeypatch):
feed = {
"generated_at": time.time(), "date": datetime.date.today().isoformat(),
"considered": 800,
"picks": {"extreme": _pick("Phoenix", 96.0, "very-hot", "Very High", "warm")},
"ranked": [],
}
monkeypatch.setattr(homepage, "load", lambda: feed)
html = client.get(f"{B}/").text
assert "Today in Phoenix, Somewhere" in html
assert "96th percentile" in html
assert "the most unusual weather we're tracking" in html
def test_homepage_renders_without_feed(client, no_feed):
"""A cold checkout with no precomputed feed still serves a complete page."""
html = client.get(f"{B}/").text
assert "How unusual is your weather?" in html
assert "Unusual right now" not in html # strip is omitted, not broken
assert 'id="hero-grade"' in html # the frame is still there
def test_homepage_etag_revalidates(client, no_feed):
r = client.get(f"{B}/")
etag = r.headers["etag"]
again = client.get(f"{B}/", headers={"If-None-Match": etag})
assert again.status_code == 304
# --- static frontend, unaffected by the SSR content split --------------------
@pytest.fixture
def client():
from fastapi.testclient import TestClient
from web import app as appmod
return TestClient(appmod.app)
def test_old_static_index_is_gone(client):
"""index.html must not linger behind the static mount as indexable duplicate
content now that / is server-rendered."""
content now that / is server-rendered (by frontend_ssr)."""
assert client.get(f"{B}/index.html").status_code == 404
def test_privacy_page(client):
r = client.get(f"{B}/privacy")
assert r.status_code == 200
assert "no analytics library" in r.text.lower() or "no analytics" in r.text.lower()
assert "Use my location" in r.text
def test_privacy_is_linked_but_not_in_sitemap(client):
assert f"{B}/privacy" in client.get(f"{B}/").text
# It's a policy page, not a crawl target we want ranked.
assert "/privacy" not in client.get(f"{B}/sitemap.xml").text

101
web/app.py
View file

@ -11,6 +11,9 @@ import queue
import threading
import time
import html as html_escape
import httpx
from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import RedirectResponse
@ -20,7 +23,6 @@ from accounts import api_accounts
from api import content_routes
from core import audit
from data import climate
from web import content
from notifications import digest
from notifications import discord_interactions as discord_interactions_mod
from notifications import discord_link
@ -51,6 +53,15 @@ FRONTEND_DIR = paths.FRONTEND_DIR
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""
# The SSR content service (repo-split Stage 4): backend no longer renders the
# climate hub/city/month/records/glossary/home pages itself (frontend_ssr does,
# over the content API above), and this process now needs a way back TO it for
# every environment with no reverse proxy in front to do the split (LAN dev,
# bare-metal run.sh) -- see _proxy_to_frontend below. Required, fails loud: every
# environment is dual-service from this stage on, no single-service fallback.
FRONTEND_BASE = os.environ["THERMOGRAPH_FRONTEND_BASE_INTERNAL"]
_frontend_client = httpx.AsyncClient(base_url=FRONTEND_BASE, timeout=10.0)
# Which duties this process performs: "web" (serve requests, no notifier),
# "worker" (own the notifier, still serves requests today — see _lifespan), or
# "all" (both — today's single-process default, so an unset ROLE is unchanged
@ -167,6 +178,7 @@ async def _lifespan(app):
yield
notify.stop()
scheduler.stop()
await _frontend_client.aclose()
app = FastAPI(title="Thermograph", version="0.2.0", lifespan=_lifespan)
@ -717,13 +729,67 @@ v2.add_api_route("/event", api_event, methods=["POST"], include_in_schema=False)
app.include_router(v1, prefix=f"{BASE}/api") # legacy unversioned == v1 (backward compatible)
app.include_router(v1, prefix=f"{BASE}/api/v1")
app.include_router(v2, prefix=f"{BASE}/api/v2")
# SSR content API (repo-split Stage 2) -- purely additive, not yet consumed by
# anything in this repo; content.py still renders its own SSR pages in-process.
# The eventual frontend SSR service will call these instead. See
# SSR content API (repo-split Stage 2), now consumed by the frontend_ssr
# service (Stage 3) instead of anything in this process. See
# backend/api/content_routes.py.
app.include_router(content_routes.router, prefix=f"{BASE}/api/v2")
# --- proxy to the frontend SSR service (repo-split Stage 4) ------------------
_PROXY_EXCLUDED_REQUEST_HEADERS = {"host", "content-length"}
_PROXY_EXCLUDED_RESPONSE_HEADERS = {"content-encoding", "content-length", "transfer-encoding", "connection"}
async def _proxy_to_frontend(request: Request) -> Response:
"""Forward a GET/HEAD request verbatim to the frontend SSR service and relay
its response back untouched (status, body, headers -- including ETag, so
frontend's own If-None-Match/304 logic keeps working through this hop).
Every frontend-owned path (content.register()'s old routes) is read-only,
so this never needs to forward a request body.
The primary same-origin mechanism is host Caddy in prod/beta -- it
path-splits directly to backend vs frontend, so this proxy is never
actually exercised there. This is the fallback for every environment with
no Caddy in front (LAN dev, bare-metal run.sh) -- see the repo-split
plan's Stage 4 section."""
headers = {k: v for k, v in request.headers.items()
if k.lower() not in _PROXY_EXCLUDED_REQUEST_HEADERS}
upstream = await _frontend_client.request(
request.method, request.url.path, params=request.url.query, headers=headers)
resp_headers = {k: v for k, v in upstream.headers.items()
if k.lower() not in _PROXY_EXCLUDED_RESPONSE_HEADERS}
return Response(content=upstream.content, status_code=upstream.status_code, headers=resp_headers)
# The exact path list content.register() used to own -- see backend/web/app.py's
# git history (repo-split Stage 3) for the Jinja implementation, now in
# frontend_ssr/content.py. /climate/{slug}/records must be registered before
# the {month} param route so the literal path wins, same ordering content.py
# itself used.
_FRONTEND_ROUTES = [
f"{BASE}/",
f"{BASE}/about",
f"{BASE}/privacy",
f"{BASE}/glossary",
f"{BASE}/glossary/{{term}}",
f"{BASE}/climate",
f"{BASE}/climate/{{slug}}/records",
f"{BASE}/climate/{{slug}}/{{month}}",
f"{BASE}/climate/{{slug}}",
f"{BASE}/robots.txt",
f"{BASE}/sitemap.xml",
]
for _frontend_path in _FRONTEND_ROUTES:
app.add_api_route(_frontend_path, _proxy_to_frontend, methods=["GET", "HEAD"], include_in_schema=False)
# The IndexNow ownership-verification key file, at a dynamic path fixed at
# startup -- mirrors how content.register() used to register it.
import indexnow # noqa: E402 - lazy, same as content.register() did
app.add_api_route(f"{BASE}/{indexnow.key()}.txt", _proxy_to_frontend,
methods=["GET", "HEAD"], include_in_schema=False)
# --- Discord slash commands (HTTP interactions) ----------------------------
async def discord_interactions(request: Request) -> Response:
"""Discord posts each slash-command interaction here. Verification runs over the
@ -767,6 +833,21 @@ app.include_router(discord_link.router, prefix=f"{BASE}/api/v2")
# --- static frontend (served under BASE) -----------------------------------
def head_verify_html() -> str:
"""Search-engine ownership-verification <meta> tags, from env (empty when
unset). Injected into every _page()-served page's <head> — Google verifies
the homepage, which frontend_ssr's own copy of this same logic handles for
the SSR content pages."""
metas = []
google = os.environ.get("THERMOGRAPH_GOOGLE_VERIFY", "").strip()
bing = os.environ.get("THERMOGRAPH_BING_VERIFY", "").strip()
if google:
metas.append(f'<meta name="google-site-verification" content="{html_escape.escape(google, quote=True)}">')
if bing:
metas.append(f'<meta name="msvalidate.01" content="{html_escape.escape(bing, quote=True)}">')
return "\n ".join(metas)
def _page(name):
"""Route handler for one HTML page. Serves the file with its __ORIGIN__
placeholders (the link-preview/Open Graph tags) filled in as the request's
@ -784,7 +865,7 @@ def _page(name):
html = html.replace("__ORIGIN__", f"{proto}://{host}{BASE}")
# Search-engine verification <meta> tags (same source as the SEO pages), so
# the homepage Google/Bing verify against carries them too.
verify = content.head_verify_html()
verify = head_verify_html()
if verify:
html = html.replace("<head>", f"<head>\n {verify}", 1)
etag = f'W/"{hashlib.sha1(html.encode()).hexdigest()[:20]}"'
@ -802,9 +883,8 @@ def _page(name):
# Pages also answer HEAD — link-preview crawlers probe with it before fetching.
if BASE:
app.add_api_route(BASE, lambda: RedirectResponse(url=f"{BASE}/"), methods=["GET", "HEAD"], include_in_schema=False)
# NOTE: "/" is not here — the homepage is server-rendered by content.register()
# below, so its hero, records strip and city links reach crawlers and no-JS
# readers as real HTML.
# NOTE: "/" is not here — the homepage (and the other crawlable SSR pages) are
# proxied to the frontend_ssr service, registered above (_FRONTEND_ROUTES).
app.add_api_route(f"{BASE}/calendar", _page("calendar.html"), methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/day", _page("day.html"), methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/score", _page("score.html"), methods=["GET", "HEAD"], include_in_schema=False)
@ -817,11 +897,6 @@ app.add_api_route(f"{BASE}/alerts", _page("subscriptions.html"), methods=["GET",
# already subscribed.
app.add_api_route(f"{BASE}/digest", digest.signup, methods=["POST"], include_in_schema=False)
# Crawlable, server-rendered SEO pages (the homepage, climate hub / per-city /
# month / records / glossary / about / privacy) + robots.txt + sitemap.xml.
# Registered before the static mount so these explicit routes win.
content.register(app)
# Everything else under BASE (app.js, style.css, nav.js, …) is a static asset.
# Registered last so the explicit page routes above win. Mounts must start with
# "/", so at the root (BASE == "") mount at "/" — the earlier routes still win

944
web/content.py
View file

@ -1,944 +0,0 @@
"""Server-rendered, crawlable content pages (climate hub / per-city / month /
records / glossary / about) plus robots.txt and sitemap.xml.
These are the SEO surface: real URLs with the climate stats in the HTML, rendered
with Jinja2 from the same builders the API uses, linking into the interactive tool.
Registered on the app (via register()) BEFORE the StaticFiles mount so they win.
"""
import contextlib
import contextvars
import datetime
import functools
import hashlib
import json
import math
import os
import polars as pl
from fastapi import HTTPException, Request, Response
from fastapi.responses import PlainTextResponse
from jinja2 import Environment, FileSystemLoader, select_autoescape
from markupsafe import Markup
from api import homepage
from api import sitemap
from api.payloads import OBS_COLS
from data import cities
from data import city_events
from data import climate
from data import grading
from data import grid
from web import content_loader
import paths
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""
TEMPLATES_DIR = paths.TEMPLATES_DIR
_env = Environment(
loader=FileSystemLoader(TEMPLATES_DIR),
autoescape=select_autoescape(["html", "xml", "j2"]),
trim_blocks=True,
lstrip_blocks=True,
)
MONTHS = ["january", "february", "march", "april", "may", "june",
"july", "august", "september", "october", "november", "december"]
MONTHS_TITLE = [m.capitalize() for m in MONTHS]
MONTH_INDEX = {m: i + 1 for i, m in enumerate(MONTHS)}
# Meteorological seasons as (northern-hemisphere label, southern-hemisphere label,
# month numbers, span text). The same three months are one season everywhere — only
# the name flips across the equator (Dec–Feb is winter in the north, summer in the
# south), so a city's latitude picks the label.
SEASONS = [
("Winter", "Summer", [12, 1, 2], "Dec–Feb"),
("Spring", "Autumn", [3, 4, 5], "Mar–May"),
("Summer", "Winter", [6, 7, 8], "Jun–Aug"),
("Autumn", "Spring", [9, 10, 11], "Sep–Nov"),
]
# Display labels for the graded metrics (order = how they appear on the page).
METRIC_LABELS = [
("tmax", "High"), ("tmin", "Low"), ("feels", "Feels-like"),
("humid", "Humidity"), ("wind", "Wind"), ("gust", "Gust"), ("precip", "Precip"),
]
_TEMP_METRICS = {"tmax", "tmin", "feels"}
# Countries that actually use Fahrenheit. Mirrors F_REGIONS in frontend/units.js —
# the client applies it to the visitor's locale, we apply it to the city's country.
# A test asserts the two lists stay identical.
F_COUNTRIES = frozenset({"US", "PR", "GU", "VI", "AS", "MP", "UM",
"BS", "BZ", "KY", "PW", "FM", "MH", "LR"})
# One flag drives every measure: a °C page also gets mm and km/h. Mirrors the same
# decision in frontend/units.js.
MM_PER_IN = 25.4
KMH_PER_MPH = 1.609344
# The unit the current page renders temperatures in. None = °F, and also means
# "this page has no city", which is what keeps the interactive pages on the
# client-side locale default instead of being pinned server-side.
#
# A ContextVar rather than a parameter because _temp() is reached from both sides:
# templates call it as a Jinja global, and the context builders call it directly
# (the records tables are pre-rendered into strings in Python). Threading a `unit`
# argument would mean touching ~20 call sites across three levels of nesting, where
# forgetting one yields a silently wrong unit rather than an error.
_UNIT: contextvars.ContextVar[str | None] = contextvars.ContextVar("display_unit", default=None)
@contextlib.contextmanager
def _unit_scope(unit: str | None):
"""Render temperatures in `unit` for the duration of the block.
The reset is not optional: the page handlers are sync `def`, so Starlette runs
them on a recycled threadpool thread, and a value left set would leak into
whatever request that thread serves next — including pages with no city.
"""
token = _UNIT.set(unit)
try:
yield
finally:
_UNIT.reset(token)
def unit_for_country(code: str | None) -> str:
"""The temperature unit a reader in this country expects."""
return "F" if (code or "").upper() in F_COUNTRIES else "C"
def _round_half_up(x: float) -> int:
"""Round like JS's Math.round, not like Python's round().
Python rounds halves to even, JS rounds them up. Where they disagree the
server would render one number and climate.js would repaint a different one —
a visible flicker for exactly the visitor whose unit already matched.
"""
return math.floor(x + 0.5)
def _c(f: float) -> int:
return _round_half_up((f - 32) * 5 / 9)
def _shown(f: float) -> int:
"""The number to print, in the active unit."""
return _c(f) if _UNIT.get() == "C" else _round_half_up(f)
def _letter() -> str:
return "C" if _UNIT.get() == "C" else "F"
def _temp(f):
"""A Fahrenheit temperature as a client-convertible span, e.g.
'<span class="temp" data-temp-f="72.3">72°F</span>'. The text is rendered in
the page's unit (the city's country convention — see _unit_scope), so crawlers
and no-JS visitors get the local convention rather than a conversion they have
to do themselves; climate.js repaints on toggle. data-temp-f stays Fahrenheit
either way — it is the conversion source of truth. '—' for a missing value."""
if f is None:
return "—"
return Markup('<span class="temp" data-temp-f="{:.1f}">{}°{}</span>').format(
f, _shown(f), _letter())
def _temp_bare(f):
"""Like _temp() but with no unit letter ('72°'), for the range strip where the
axis already implies the unit. Still carries data-temp-f so it converts."""
if f is None:
return "—"
return Markup('<span class="temp" data-temp-f="{:.1f}" data-bare>{}°</span>').format(
f, _shown(f))
def _temp_text(f) -> str:
"""'72°F' as plain text. The span _temp() returns can't go in a
<meta content="…">, and a meta description is the one place a temperature is
never converted client-side — it is what the search snippet quotes, which is
why it has to be in the city's own unit at render time."""
if f is None:
return "—"
return f"{_shown(f)}°{_letter()}"
def _precip(v):
"""Precipitation as a client-convertible span, the same contract as _temp():
text in the page's unit, data-precip-in always inches."""
if v is None:
return "—"
return Markup('<span class="precip" data-precip-in="{:.3f}">{}</span>').format(
v, _precip_text(v))
def _precip_text(v) -> str:
"""Whole millimetres — that is how rainfall is reported, and a tenth of a
millimetre is below what the source resolves. Inches keep two decimals, since
a whole inch of rain is a lot to round to."""
if v is None:
return "—"
return (f"{_round_half_up(v * MM_PER_IN)} mm" if _UNIT.get() == "C"
else f"{v:.2f} in")
def _wind(v):
"""Wind/gust as a client-convertible span; data-wind-mph is always mph."""
if v is None:
return "—"
return Markup('<span class="wind" data-wind-mph="{:.1f}">{}</span>').format(
v, _wind_text(v))
def _wind_text(v) -> str:
if v is None:
return "—"
return (f"{_round_half_up(v * KMH_PER_MPH)} km/h" if _UNIT.get() == "C"
else f"{_round_half_up(v)} mph")
def _month_year(date_s: str) -> str:
"""'1995-07-13' -> 'Jul 1995'. Meta descriptions have ~155 characters to
spend, so a record's date gives up its day."""
try:
return datetime.date.fromisoformat(str(date_s)).strftime("%b %Y")
except (ValueError, TypeError):
return str(date_s)
def _clamp_desc(text: str, limit: int = 155) -> str:
"""Meta descriptions are cut at ~155 characters in the SERP. Trim on a word
boundary so we choose where the sentence ends rather than Google doing it."""
text = " ".join(text.split())
if len(text) <= limit:
return text
return text[:limit].rsplit(" ", 1)[0].rstrip(",;—-") + "…"
def _titled(text: str, suffix: str = " · Thermograph", limit: int = 60) -> str:
"""Brand the title only when it costs nothing — the payload ("… in July",
"hottest & coldest days") has to survive truncation first."""
return text + suffix if len(text) + len(suffix) <= limit else text
def _fmt(metric: str, v) -> str:
if v is None:
return "—"
if metric in _TEMP_METRICS:
return _temp(v)
if metric == "precip":
return _precip(v)
if metric == "humid":
# Absolute humidity is g/m³ in both systems — there is no imperial unit for
# it anyone would recognise — so it is never converted.
return f"{v:.1f} g/m³"
return _wind(v) # wind, gust
# Absolute-temperature colour tiers (°F upper bounds) mapped to the site's diverging
# cold→hot palette — the same 9 tiers the interactive grader uses. Colouring records
# and normals by these turns the tables into a heat map in the site's own visual
# language, rather than a plain grid.
_TEMP_TIERS = [
(20, "rec-cold"), (32, "very-cold"), (45, "cold"), (58, "cool"),
(70, "normal"), (80, "warm"), (90, "hot"), (100, "very-hot"),
]
# °F axis for the monthly temperature-range strip on the city page.
_AXIS_LO, _AXIS_HI = -10.0, 115.0
def temp_class(f) -> str:
"""Diverging-palette tier name for an absolute Fahrenheit temperature (or 'none')."""
if f is None:
return "none"
for upper, cls in _TEMP_TIERS:
if f < upper:
return cls
return "rec-hot"
def _range_bar(low_f, high_f) -> dict | None:
"""Geometry for one month's low→high bar on the shared _AXIS_LO.._AXIS_HI axis:
left offset + width as percentages, and the tier colour at each end (for a
gradient fill). None when a value is missing."""
if low_f is None or high_f is None:
return None
lo = max(_AXIS_LO, min(_AXIS_HI, low_f))
hi = max(_AXIS_LO, min(_AXIS_HI, high_f))
span = _AXIS_HI - _AXIS_LO
return {
"left": round((lo - _AXIS_LO) / span * 100, 1),
"width": round(max(2.0, (hi - lo) / span * 100), 1),
"c1": temp_class(low_f), "c2": temp_class(high_f),
}
def _ordinal(n) -> str:
"""Percentile -> display ordinal. Delegates to grading so every surface
(Day page, calendar, chart, city pages, homepage strip) agrees."""
return grading.pct_ordinal(n)
_env.globals["temp_class"] = temp_class
_env.globals["temp"] = _temp
_env.globals["temp_bare"] = _temp_bare
_env.filters["ordinal"] = _ordinal
def _month_doy(month_idx: int) -> int:
return datetime.date(2001, month_idx, 15).timetuple().tm_yday
# --- helpers -----------------------------------------------------------------
def origin(request: Request) -> str:
proto = request.headers.get("x-forwarded-proto") or request.url.scheme
host = request.headers.get("host") or request.url.netloc
return f"{proto}://{host}"
def _breadcrumb_jsonld(o: str, breadcrumb: list[tuple[str, str | None]]) -> dict:
"""BreadcrumbList structured data. Google requires ``item`` on every
ListItem except the last, so unlinked intermediate crumbs (e.g. the
country, which has no page of its own) are omitted here even though the
visible breadcrumb shows them."""
crumbs = [c for c in breadcrumb[:-1] if c[1]] + breadcrumb[-1:]
return {"@type": "BreadcrumbList",
"itemListElement": [
{"@type": "ListItem", "position": i + 1, "name": nm,
**({"item": f"{o}{href}"} if href else {})}
for i, (nm, href) in enumerate(crumbs)]}
def _respond_html(request: Request, template: str, **ctx) -> Response:
o = origin(request)
# Taken from the same ContextVar the numbers were rendered through, so the
# attribute units.js reads can't drift from what the page actually says.
# None on any page without a city -> base.html.j2 omits it entirely.
ctx.setdefault("unit_default", _UNIT.get())
html = _env.get_template(template).render(base=BASE, origin=o, base_url=f"{o}{BASE}", **ctx)
etag = f'W/"{hashlib.sha1(html.encode()).hexdigest()[:20]}"'
inm = request.headers.get("if-none-match")
if inm and etag in {t.strip() for t in inm.split(",")}:
return Response(status_code=304, headers={"ETag": etag})
return Response(html, media_type="text/html", headers={"ETag": etag})
# --- robots.txt & sitemap.xml -------------------------------------------------
def robots_txt(request: Request) -> Response:
base_url = f"{origin(request)}{BASE}"
body = (
"User-agent: *\n"
"Allow: /\n"
f"Disallow: {BASE}/api/\n" # JSON endpoints, nothing to index
f"Disallow: {BASE}/alerts\n" # per-user, requires login
f"Sitemap: {base_url}/sitemap.xml\n"
)
return PlainTextResponse(body)
@functools.lru_cache(maxsize=1)
def _content_lastmod() -> str:
"""A stable 'content last built' date for <lastmod> — the newest mtime of the
city list and this module (both change on a content/code rebuild, and the
process restarts on deploy). Honest and stable, unlike a per-request today()
that churns every fetch and trains crawlers to ignore lastmod entirely."""
srcs = [paths.CITIES_JSON, __file__]
mtimes = [os.path.getmtime(p) for p in srcs if os.path.exists(p)]
day = datetime.date.fromtimestamp(max(mtimes)) if mtimes else datetime.date.today()
return day.isoformat()
def sitemap_xml(request: Request) -> Response:
base_url = f"{origin(request)}{BASE}"
lastmod = _content_lastmod()
parts = ['<?xml version="1.0" encoding="UTF-8"?>',
'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">']
for path, cf, pr in sitemap.sitemap_entries():
parts.append(
f"<url><loc>{base_url}{path}</loc><lastmod>{lastmod}</lastmod>"
f"<changefreq>{cf}</changefreq><priority>{pr}</priority></url>"
)
parts.append("</urlset>")
return Response("\n".join(parts), media_type="application/xml")
def head_verify_html() -> Markup:
"""Search-engine ownership-verification <meta> tags, from env (empty when
unset). Injected into every page's <head> — Google verifies the homepage."""
metas = []
google = os.environ.get("THERMOGRAPH_GOOGLE_VERIFY", "").strip()
bing = os.environ.get("THERMOGRAPH_BING_VERIFY", "").strip()
if google:
metas.append(Markup('<meta name="google-site-verification" content="{}">').format(google))
if bing:
metas.append(Markup('<meta name="msvalidate.01" content="{}">').format(bing))
return Markup("\n ").join(metas)
_env.globals["head_verify"] = head_verify_html
# --- per-city climate pages ---------------------------------------------------
def _history_for(cell: dict):
"""Cached archive for a cell, fetching once if missing (self-heals like the
notifier). Returns a polars frame or None."""
hist = climate.load_cached_history(cell)
if hist is not None and not hist.is_empty():
return hist
try:
hist, _ = climate.get_history(cell)
except Exception: # noqa: BLE001 - upstream unavailable: caller renders a 503
return None
return hist if (hist is not None and not hist.is_empty()) else None
def _monthly_normals(history) -> list[dict]:
"""One row per month: average high/low and average precip, from the ±7-day
climatology around each month's 15th."""
rows = []
for i, name in enumerate(MONTHS_TITLE, start=1):
clim = grading.climatology(history, _month_doy(i))
tmax, tmin, precip = clim.get("tmax"), clim.get("tmin"), clim.get("precip")
high_f = tmax["mean"] if tmax else None
low_f = tmin["mean"] if tmin else None
# The range strip spans the typical spread: 10th-percentile daily low to
# 90th-percentile daily high (the band most days fall within).
rng_lo = tmin["p10"] if tmin else None
rng_hi = tmax["p90"] if tmax else None
rows.append({
"name": name,
"slug": MONTHS[i - 1],
"high": _temp(tmax["mean"]) if tmax else "—",
"high_f": high_f,
"low": _temp(tmin["mean"]) if tmin else "—",
"low_f": low_f,
"range_lo_f": rng_lo,
"range_hi_f": rng_hi,
"precip": _precip(precip["mean"]) if precip else "—",
"precip_v": precip["mean"] if precip else None,
"bar": _range_bar(rng_lo, rng_hi),
})
return rows
def _extreme(metric_rec, key: str) -> dict | None:
"""One extreme of a metric — key 'max' (warmest) or 'min' (coldest) — as the
two-unit value, its heat-map tier, and the date it occurred."""
if not metric_rec:
return None
v = metric_rec[key]
return {"txt": _temp(v), "f": v, "cls": temp_class(v), "date": metric_rec[f"{key}_date"]}
def _period_records(history, months: list[int]) -> dict:
"""For a set of calendar months, the record *warmest and coldest* of BOTH the
daytime high (tmax) and the overnight low (tmin) — four extremes with dates.
So each metric shows both ends: the daytime high's hottest day and the coldest a
day ever stayed (its record-low high), and the overnight low's mildest night and
its record low. Reuses grading.all_time_records on the month-filtered archive."""
if len(months) == 1:
sub = history.filter(pl.col("date").dt.month() == months[0])
else:
sub = history.filter(pl.col("date").dt.month().is_in(months))
rec = grading.all_time_records(sub) if not sub.is_empty() else {}
tmax, tmin = rec.get("tmax"), rec.get("tmin")
return {
"high": {"warm": _extreme(tmax, "max"), "cold": _extreme(tmax, "min")},
"low": {"warm": _extreme(tmin, "max"), "cold": _extreme(tmin, "min")},
}
def _monthly_records(history) -> list[dict]:
"""Record high and low for each of the 12 months (each row links to its month page)."""
return [
{"name": name, "slug": MONTHS[i - 1], **_period_records(history, [i])}
for i, name in enumerate(MONTHS_TITLE, start=1)
]
def _seasonal_records(history, lat: float) -> list[dict]:
"""Record high and low for each meteorological season, labelled for the city's
hemisphere (Dec–Feb reads as winter north of the equator, summer south of it)."""
south = lat < 0
return [
{"name": (south_lbl if south else north_lbl), "span": span,
**_period_records(history, months)}
for north_lbl, south_lbl, months, span in SEASONS
]
def _today_vs_normal(history, cell) -> dict | None:
"""Grade the latest recorded day against its climatology, for the hero block."""
try:
recent = climate.get_recent_forecast(cell)
except Exception: # noqa: BLE001
return None
if recent is None or recent.is_empty() or "date" not in recent.columns:
return None
today = datetime.date.today()
observed = recent.filter(pl.col("date") <= today).sort("date")
if observed.is_empty():
return None
row = observed.row(observed.height - 1, named=True)
obs = {k: row[k] for k in OBS_COLS if k in row}
graded = grading.grade_day(history, row["date"], obs)
cards = []
for key, label in METRIC_LABELS:
g = graded.get(key)
if not g:
continue
cards.append({
"label": label, "metric": key,
"value": _fmt(key, g.get("value")),
"percentile": g.get("percentile"),
"grade": g.get("grade"), "cls": g.get("class"),
})
date = row["date"]
return {
"date": date.isoformat() if hasattr(date, "isoformat") else str(date),
"cards": cards,
}
def _city_context(request, city, cell, history) -> dict:
name = city["name"]
display = cities.display_name(city)
title = cities.title_name(city)
years = history["date"].dt.year()
year_range = [int(years.min()), int(years.max())]
months = _monthly_normals(history)
warmest = max((m for m in months if m["high_f"] is not None), key=lambda m: m["high_f"], default=None)
coldest = min((m for m in months if m["low_f"] is not None), key=lambda m: m["low_f"], default=None)
wettest = max((m for m in months if m["precip_v"] is not None), key=lambda m: m["precip_v"], default=None)
records = grading.all_time_records(history)
tool_hash = f"{city['lat']:.5f},{city['lon']:.5f}"
# Unique editorial blurb (Wikipedia, CC BY-SA) so the page isn't just templated
# stats, and a travel/comfort CTA that pre-fills this city on the compare page.
flavor = cities.flavor(city["slug"])
event = city_events.get(city["slug"]) # hand-curated; None → page falls back to the blurb
compare_url = f"{BASE}/compare#loc={city['lat']:.4f},{city['lon']:.4f}"
breadcrumb = [("Home", f"{BASE}/"), ("Climate", f"{BASE}/climate")]
if city.get("country"):
breadcrumb.append((city["country"], None))
breadcrumb.append((name, None))
o = origin(request)
page_url = f"{o}{BASE}/climate/{city['slug']}"
jsonld = {
"@context": "https://schema.org",
"@graph": [
{"@type": "Dataset",
"name": f"{display} climate normals and records",
"description": f"Average temperatures, precipitation and record highs and lows for "
f"{display}, from ~{year_range[1] - year_range[0]} years of daily climate history.",
"url": page_url,
"temporalCoverage": f"{year_range[0]}/{year_range[1]}",
"spatialCoverage": {"@type": "Place", "name": display,
"geo": {"@type": "GeoCoordinates",
"latitude": city["lat"], "longitude": city["lon"]}},
"creator": {"@type": "Organization", "name": "Thermograph"},
"isBasedOn": "https://open-meteo.com/ (ERA5 reanalysis)"},
_breadcrumb_jsonld(o, breadcrumb),
],
}
return {
"section": "climate",
"city": city, "display": display, "name": name,
"year_range": year_range, "n_years": year_range[1] - year_range[0],
"months": months, "warmest": warmest, "coldest": coldest, "wettest": wettest,
"records": records, "today": _today_vs_normal(history, cell),
"tool_hash": tool_hash, "flavor": flavor, "event": event, "compare_url": compare_url,
"breadcrumb": breadcrumb,
"canonical_path": f"/climate/{city['slug']}",
"page_title": _titled(f"{title} climate: daily normals, records & how unusual it is now"),
"page_description": _clamp_desc(
f"{title} averages highs of {_temp_text(warmest['high_f']) if warmest else '—'} in "
f"{warmest['name'] if warmest else 'summer'} and lows of "
f"{_temp_text(coldest['low_f']) if coldest else '—'} in "
f"{coldest['name'] if coldest else 'winter'}. "
f"Every day graded against {year_range[1] - year_range[0]} years of local history."),
"jsonld_str": json.dumps(jsonld, ensure_ascii=False, separators=(",", ":")),
}
def _resolve_city(slug: str):
"""(city, cell, history) for a slug, or raise 404 (unknown) / 503 (warming)."""
city = cities.get(slug)
if city is None:
raise HTTPException(status_code=404, detail="Unknown city.")
cell = grid.snap(city["lat"], city["lon"])
history = _history_for(cell)
if history is None:
raise HTTPException(status_code=503, detail="Climate data is warming up; please retry shortly.")
return city, cell, history
def city_page(request: Request, slug: str) -> Response:
city, cell, history = _resolve_city(slug)
# The scope has to cover the context builder too, not just the render: it is
# evaluated as an argument, so it runs first — and it is where most of the
# page's temperatures are formatted.
with _unit_scope(unit_for_country(city.get("country_code"))):
return _respond_html(request, "city.html.j2",
**_city_context(request, city, cell, history))
# --- month & records pages ----------------------------------------------------
def _month_context(request, city, history, month_idx: int) -> dict:
display = cities.display_name(city)
title = cities.title_name(city)
name = city["name"]
month_name = MONTHS_TITLE[month_idx - 1]
month_slug = MONTHS[month_idx - 1]
years = history["date"].dt.year()
year_range = [int(years.min()), int(years.max())]
clim = grading.climatology(history, _month_doy(month_idx))
tmax, tmin, precip = clim.get("tmax"), clim.get("tmin"), clim.get("precip")
mdf = history.filter(pl.col("date").dt.month() == month_idx)
mrec = grading.all_time_records(mdf) if not mdf.is_empty() else {}
stats = []
if tmax:
stats.append(("Average high", _temp(tmax["mean"])))
stats.append(("Typical high range", _temp(tmax['p10']) + " to " + _temp(tmax['p90'])))
if tmin:
stats.append(("Average low", _temp(tmin["mean"])))
stats.append(("Typical low range", _temp(tmin['p10']) + " to " + _temp(tmin['p90'])))
if precip:
stats.append(("Average daily precipitation", _precip(precip["mean"])))
# Record high and low for every metric in this calendar month (mirrors the
# all-time records cards, but scoped to the month). Precip is special-cased:
# a per-day "record low" is just zero, and the dry-streak helper would wrongly
# bridge year boundaries on month-filtered rows, so the wettest/driest sides
# show this month's largest and smallest total accumulation, dated to the year.
records = []
for key, label in METRIC_LABELS:
r = mrec.get(key)
if not r:
continue
if key == "precip":
# Only whole months count — a partial current month would otherwise win
# "driest" on a fraction of its rainfall.
totals = (mdf.filter(pl.col("precip").is_not_null())
.group_by(pl.col("date").dt.year().alias("yr"))
.agg(pl.col("precip").sum().alias("total"),
pl.len().alias("days"))
.filter(pl.col("days") >= 26)
.sort("total"))
if totals.is_empty():
continue
lo, hi = totals.row(0, named=True), totals.row(totals.height - 1, named=True)
records.append({
"label": label,
"high": _precip(hi["total"]), "high_date": str(hi["yr"]), "high_tag": "Wettest",
"low": _precip(lo["total"]), "low_date": str(lo["yr"]), "low_tag": "Driest",
})
else:
records.append({
"label": label,
"high": _fmt(key, r["max"]), "high_date": r["max_date"], "high_tag": "Highest",
"low": _fmt(key, r["min"]), "low_date": r["min_date"], "low_tag": "Lowest",
})
prev_i = 12 if month_idx == 1 else month_idx - 1
next_i = 1 if month_idx == 12 else month_idx + 1
breadcrumb = [("Home", f"{BASE}/"), ("Climate", f"{BASE}/climate"),
(name, f"{BASE}/climate/{city['slug']}"), (month_name, None)]
return {
"section": "climate", "city": city, "display": display, "name": name,
"month_name": month_name, "month_slug": month_slug,
"year_range": year_range, "n_years": year_range[1] - year_range[0],
"avg_high": _temp(tmax["mean"]) if tmax else "—",
"avg_low": _temp(tmin["mean"]) if tmin else "—",
"avg_high_cls": temp_class(tmax["mean"]) if tmax else "none",
"avg_low_cls": temp_class(tmin["mean"]) if tmin else "none",
"stats": stats, "records": records,
"tool_hash": f"{city['lat']:.5f},{city['lon']:.5f}",
"compare_url": f"{BASE}/compare#loc={city['lat']:.4f},{city['lon']:.4f}",
"prev": {"name": MONTHS_TITLE[prev_i - 1], "slug": MONTHS[prev_i - 1]},
"next": {"name": MONTHS_TITLE[next_i - 1], "slug": MONTHS[next_i - 1]},
"breadcrumb": breadcrumb,
"canonical_path": f"/climate/{city['slug']}/{month_slug}",
"page_title": _titled(f"{title} in {month_name}: normal weather & records"),
"page_description": _clamp_desc(
f"{title} averages {_temp_text(tmax['mean']) if tmax else '—'} highs and "
f"{_temp_text(tmin['mean']) if tmin else '—'} lows in {month_name}. "
f"Every day graded against {year_range[1] - year_range[0]} years of local history."),
}
def month_page(request: Request, slug: str, month: str) -> Response:
if month not in MONTH_INDEX:
raise HTTPException(status_code=404, detail="Unknown month.")
city, cell, history = _resolve_city(slug)
with _unit_scope(unit_for_country(city.get("country_code"))):
return _respond_html(request, "month.html.j2",
**_month_context(request, city, history, MONTH_INDEX[month]))
def _records_context(request, city, history) -> dict:
display = cities.display_name(city)
title = cities.title_name(city)
name = city["name"]
years = history["date"].dt.year()
year_range = [int(years.min()), int(years.max())]
n_years = year_range[1] - year_range[0]
rec = grading.all_time_records(history)
# The two numbers the meta description leads with — the same all-time extremes
# the page's own lede sentence quotes.
hi_rec = (rec.get("tmax") or {}).get("max")
hi_date = (rec.get("tmax") or {}).get("max_date")
lo_rec = (rec.get("tmin") or {}).get("min")
lo_date = (rec.get("tmin") or {}).get("min_date")
rows = []
for key, label in METRIC_LABELS:
r = rec.get(key)
if not r:
continue
if key == "precip":
# "Record low" rain is meaningless (it's just 0), so the low side shows
# the longest dry streak and the date it began instead.
days, dry_start = grading.longest_dry_streak(history)
low = f"{days}-day dry spell" if days else "—"
low_date = dry_start or "—"
else:
low, low_date = _fmt(key, r["min"]), r["min_date"]
is_temp = key in _TEMP_METRICS
rows.append({
"label": label,
"high": _fmt(key, r["max"]), "high_date": r["max_date"],
"high_f": r["max"] if is_temp else None,
"low": low, "low_date": low_date,
"low_f": r["min"] if is_temp else None,
})
monthly = _monthly_records(history)
seasonal = _seasonal_records(history, city["lat"])
hemisphere = "Southern" if city["lat"] < 0 else "Northern"
breadcrumb = [("Home", f"{BASE}/"), ("Climate", f"{BASE}/climate"),
(name, f"{BASE}/climate/{city['slug']}"), ("Records", None)]
o = origin(request)
page_url = f"{o}{BASE}/climate/{city['slug']}/records"
jsonld = {
"@context": "https://schema.org",
"@graph": [
{"@type": "Dataset",
"name": f"{display} monthly and seasonal weather records",
"description": f"Record high and low temperatures for {display} by month and by "
f"meteorological season, with the dates they occurred, from ~{n_years} "
f"years of daily climate history.",
"url": page_url,
"temporalCoverage": f"{year_range[0]}/{year_range[1]}",
"spatialCoverage": {"@type": "Place", "name": display,
"geo": {"@type": "GeoCoordinates",
"latitude": city["lat"], "longitude": city["lon"]}},
"creator": {"@type": "Organization", "name": "Thermograph"},
"isBasedOn": "https://open-meteo.com/ (ERA5 reanalysis)"},
_breadcrumb_jsonld(o, breadcrumb),
],
}
return {
"section": "climate", "city": city, "display": display, "name": name,
"year_range": year_range, "n_years": n_years,
"rows": rows, "monthly": monthly, "seasonal": seasonal,
"hemisphere": hemisphere, "all_time": rec,
"canonical_path": f"/climate/{city['slug']}/records",
"breadcrumb": breadcrumb,
"page_title": _titled(
f"{title} weather records: hottest & coldest days since {year_range[0]}"),
"page_description": _clamp_desc(
f"{title}'s hottest day hit {_temp_text(hi_rec)}"
f"{f' ({_month_year(hi_date)})' if hi_date else ''}; its coldest fell to "
f"{_temp_text(lo_rec)}{f' ({_month_year(lo_date)})' if lo_date else ''}. "
f"Every day graded against {n_years} years of local history."),
"jsonld_str": json.dumps(jsonld, ensure_ascii=False, separators=(",", ":")),
}
def records_page(request: Request, slug: str) -> Response:
city, cell, history = _resolve_city(slug)
with _unit_scope(unit_for_country(city.get("country_code"))):
return _respond_html(request, "records.html.j2",
**_records_context(request, city, history))
# --- hub / glossary / about ---------------------------------------------------
def _breadcrumb(*items):
return list(items)
def hub_page(request: Request) -> Response:
groups = cities.by_country()
ctx = {
"section": "climate",
"groups": groups,
"n_cities": sum(len(v) for v in groups.values()),
"n_countries": len(groups),
"canonical_path": "/climate",
"breadcrumb": [("Home", f"{BASE}/"), ("Climate", None)],
"page_title": PAGES["hub"]["title"],
"page_description": PAGES["hub"]["description"],
}
return _respond_html(request, "hub.html.j2", **ctx)
# Weather-terms glossary, loaded from content/glossary.yaml (see content_loader.py).
GLOSSARY: dict[str, dict] = content_loader.load_glossary()
# Static-page SEO title/description, loaded from content/pages.yaml. The
# per-city/per-month pages build theirs dynamically from city data instead —
# see their own page_title/page_description assignments above.
PAGES: dict[str, dict] = content_loader.load_pages()
def _glossary_body(entry: dict) -> str:
return entry["body"].replace("{base}", BASE)
def glossary_index(request: Request) -> Response:
ctx = {
"terms": [{"slug": s, **e} for s, e in GLOSSARY.items()],
"canonical_path": "/glossary",
"breadcrumb": [("Home", f"{BASE}/"), ("Glossary", None)],
"page_title": PAGES["glossary_index"]["title"],
"page_description": PAGES["glossary_index"]["description"],
}
return _respond_html(request, "glossary.html.j2", **ctx)
def glossary_term(request: Request, term: str) -> Response:
entry = GLOSSARY.get(term)
if entry is None:
raise HTTPException(status_code=404, detail="Unknown term.")
ctx = {
"term": entry["term"], "body": _glossary_body(entry),
"canonical_path": f"/glossary/{term}",
"breadcrumb": [("Home", f"{BASE}/"), ("Glossary", f"{BASE}/glossary"), (entry["term"], None)],
"page_title": f"{entry['term']}: what it means | Thermograph",
"page_description": entry["short"],
"others": [{"slug": s, "term": e["term"]} for s, e in GLOSSARY.items() if s != term],
}
return _respond_html(request, "glossary_term.html.j2", **ctx)
def about_page(request: Request) -> Response:
ctx = {
"canonical_path": "/about",
"breadcrumb": [("Home", f"{BASE}/"), ("About", None)],
"page_title": PAGES["about"]["title"],
"page_description": PAGES["about"]["description"],
}
return _respond_html(request, "about.html.j2", **ctx)
def privacy_page(request: Request) -> Response:
ctx = {
"canonical_path": "/privacy",
"breadcrumb": [("Home", f"{BASE}/"), ("Privacy", None)],
"page_title": PAGES["privacy"]["title"],
"page_description": PAGES["privacy"]["description"],
}
return _respond_html(request, "privacy.html.j2", **ctx)
# --- homepage -----------------------------------------------------------------
# The homepage is the Weekly tool plus the distribution surfaces around it. It is
# server-rendered like the SEO pages (rather than a static file with placeholder
# substitution) so a cold visitor with no JS still gets the headline, a real
# graded example, the records strip and the city links.
_HOME_JSONLD = {
"@context": "https://schema.org",
"@type": "WebApplication",
"name": "Thermograph",
"applicationCategory": "WeatherApplication",
"operatingSystem": "Web, iOS, Android",
"isAccessibleForFree": True,
"offers": {"@type": "Offer", "price": "0", "priceCurrency": "USD"},
"description": ("How unusual is your weather? Any day, anywhere on Earth, graded "
"against 45 years of that place's own history."),
}
# 12 city chips, chosen for geographic spread rather than raw population, so the
# strip reads as "anywhere on Earth" and seeds crawl paths across the hub.
HOME_CITY_SLUGS = (
"new-york-city-new-york-us", "london-england-gb", "tokyo-jp",
"sydney-new-south-wales-au", "sao-paulo-br", "lagos-ng",
"mumbai-maharashtra-in", "mexico-city-mx", "cairo-eg",
"toronto-ontario-ca", "berlin-state-of-berlin-de", "seattle-washington-us",
)
def _home_cities() -> list[dict]:
"""The chip set, skipping any slug not in the routable city list so a
regenerated cities.json can never 404 a homepage link."""
out = []
for slug in HOME_CITY_SLUGS:
city = cities.get(slug)
if city:
out.append({"slug": slug, "name": city["name"]})
return out
def home_page(request: Request) -> Response:
feed = homepage.load()
stale = bool(feed) and homepage.is_stale(feed)
unusual = None
ranked: list = []
if feed:
ranked = feed.get("ranked") or []
pick = (feed.get("picks") or {}).get("extreme")
if pick:
unusual = dict(pick, is_default=True)
ctx = {
"section": "home",
# The hero headline owns the page's sole h1, so the brand degrades to a <p>.
"brand_tag": "p",
# The tool needs the wide app column, not the 880px reading column.
"main_class": "",
"canonical_path": "/",
"unusual": unusual,
"stale": stale,
"ranked": ranked,
"cities": _home_cities(),
"jsonld_str": Markup(json.dumps({**_HOME_JSONLD, "url": f"{origin(request)}{BASE}/"})),
}
return _respond_html(request, "home.html.j2", **ctx)
# --- registration ------------------------------------------------------------
def register(app) -> None:
"""Attach all content routes. Call from app.py BEFORE the StaticFiles mount."""
app.add_api_route(f"{BASE}/robots.txt", robots_txt, methods=["GET"], include_in_schema=False)
app.add_api_route(f"{BASE}/sitemap.xml", sitemap_xml, methods=["GET"], include_in_schema=False)
# IndexNow ownership key, served as a text file at the site root (/{key}.txt)
# so Bing/DuckDuckGo/Yandex can verify our submissions. The key is fixed at
# startup, so registering its literal path here is safe.
import indexnow
_inkey = indexnow.key()
def _indexnow_key_file() -> Response:
return PlainTextResponse(_inkey + "\n")
app.add_api_route(
f"{BASE}/{_inkey}.txt", _indexnow_key_file,
methods=["GET", "HEAD"], include_in_schema=False,
)
# The homepage. Registered here (not as a static file in app.py) so it is
# server-rendered from the same Jinja environment as the SEO pages.
app.add_api_route(f"{BASE}/", home_page, methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/about", about_page, methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/privacy", privacy_page, methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/glossary", glossary_index, methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/glossary/{{term}}", glossary_term, methods=["GET", "HEAD"], include_in_schema=False)
# Hub before /climate/{slug} so the literal path wins.
app.add_api_route(f"{BASE}/climate", hub_page, methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/climate/{{slug}}", city_page, methods=["GET", "HEAD"], include_in_schema=False)
# Records before the {month} param so the literal path wins.
app.add_api_route(f"{BASE}/climate/{{slug}}/records", records_page, methods=["GET", "HEAD"], include_in_schema=False)
app.add_api_route(f"{BASE}/climate/{{slug}}/{{month}}", month_page, methods=["GET", "HEAD"], include_in_schema=False)

69
web/content_loader.py
View file

@ -1,69 +0,0 @@
"""Loads structured SSR copy (glossary, static-page SEO meta) from content/ —
see docs/README.md. Validated at load time so a malformed content file fails
loudly at import rather than silently rendering a blank/broken page.
Separate from cities_flavor.json (backend/cities_flavor.json, machine-generated
from Wikipedia by gen_flavor.py) and the Jinja templates themselves
(backend/templates/*.html.j2, which still own page structure/markup) — this
covers only the two content classes that are pure static data with no embedded
template logic: the glossary, and each static page's SEO title/description.
"""
import os
import yaml
import paths
_GLOSSARY_PATH = os.path.join(paths.CONTENT_DIR, "glossary.yaml")
_PAGES_PATH = os.path.join(paths.CONTENT_DIR, "pages.yaml")
_REQUIRED_GLOSSARY_FIELDS = ("term", "short", "body")
_REQUIRED_PAGE_FIELDS = ("title", "description")
def _load_yaml(path: str) -> dict:
with open(path, encoding="utf-8") as f:
data = yaml.safe_load(f)
if not isinstance(data, dict):
raise ValueError(f"{path}: expected a top-level mapping, got {type(data).__name__}")
return data
def load_glossary(path: str = _GLOSSARY_PATH) -> "dict[str, dict]":
"""slug -> {term, short, body}, in file order. Raises ValueError if any
entry is missing a required field, has the wrong shape, or repeats a slug —
a bad content edit fails loudly (at content.py's module-level import,
normally) rather than rendering a blank glossary card."""
data = _load_yaml(path)
terms = data.get("terms")
if not isinstance(terms, list) or not terms:
raise ValueError(f"{path}: 'terms' must be a non-empty list")
glossary: "dict[str, dict]" = {}
for i, entry in enumerate(terms):
if not isinstance(entry, dict) or "slug" not in entry:
raise ValueError(f"{path}: terms[{i}] must be a mapping with a 'slug' key")
slug = entry["slug"]
missing = [f for f in _REQUIRED_GLOSSARY_FIELDS if not entry.get(f)]
if missing:
raise ValueError(f"{path}: term '{slug}' is missing {missing}")
if slug in glossary:
raise ValueError(f"{path}: duplicate slug '{slug}'")
glossary[slug] = {f: entry[f] for f in _REQUIRED_GLOSSARY_FIELDS}
return glossary
def load_pages(path: str = _PAGES_PATH) -> "dict[str, dict]":
"""page key -> {title, description}. Same fail-loud contract as load_glossary."""
data = _load_yaml(path)
pages = data.get("pages")
if not isinstance(pages, dict) or not pages:
raise ValueError(f"{path}: 'pages' must be a non-empty mapping")
result: "dict[str, dict]" = {}
for key, entry in pages.items():
if not isinstance(entry, dict):
raise ValueError(f"{path}: page '{key}' must be a mapping")
missing = [f for f in _REQUIRED_PAGE_FIELDS if not entry.get(f)]
if missing:
raise ValueError(f"{path}: page '{key}' is missing {missing}")
result[key] = {f: entry[f] for f in _REQUIRED_PAGE_FIELDS}
return result