Jinemi/thermograph
4
0
Fork 0
Code Issues 2 Pull requests 10 Projects Releases Packages Wiki Activity Actions 1
thermograph/content.py

497 lines
22 KiB
Python
Raw Blame History

"""Server-rendered, crawlable content pages (climate hub / per-city / month /
records / glossary / about) plus robots.txt and sitemap.xml.
Ported from backend/web/content.py (repo-split Stage 3): every route handler
here fetches its data from the backend's SSR content API (api_client.py)
instead of importing backend code and computing in-process. Template shape is
unchanged -- only the data source changed, from a function call to an HTTP
call. page_title/page_description/canonical_path/breadcrumb/jsonld are all
computed by the backend now (backend/api/content_payloads.py), not here.
"""
import datetime
import hashlib
import json
import logging
import os
import httpx
from fastapi import HTTPException, Request, Response
from fastapi.responses import PlainTextResponse
from jinja2 import Environment, FileSystemLoader, select_autoescape
from markupsafe import Markup
import api_client
import content_loader
import format as fmt
import paths
_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""
# Backend's browser-facing origin+base (e.g. "http://192.168.1.5:8137/thermograph"),
# for templates referencing backend-owned things -- the interactive SPA pages
# (_page()-served, never by this service), /digest, and every static JS/CSS/
# favicon/manifest file (this service mounts no static files of its own; see
# app.py). Empty by default: today's exact same-origin behavior, where those
# references stay relative (ASSET_BASE == BASE). Repo-split Stage 5.
ASSET_BASE = os.environ.get("THERMOGRAPH_API_BASE_PUBLIC", "").rstrip("/") or BASE
_env = Environment(
loader=FileSystemLoader(paths.TEMPLATES_DIR),
autoescape=select_autoescape(["html", "xml", "j2"]),
trim_blocks=True,
lstrip_blocks=True,
# Templates only change on a redeploy (a fresh process), not while this
# one is running -- the implicit default (True) stats every template file
# on every single render just to notice a change that can never happen.
auto_reload=False,
)
_env.globals["temp_class"] = fmt.temp_class
_env.globals["temp"] = fmt.temp
_env.globals["temp_bare"] = fmt.temp_bare
_env.filters["ordinal"] = fmt.pct_ordinal
_log = logging.getLogger(__name__)
# --- helpers -------------------------------------------------------------
def _origin(request: Request) -> str:
# x-forwarded-host takes precedence over host: when reached via backend's
# internal proxy fallback (no Caddy in front -- see _proxy_to_frontend in
# backend/web/app.py), Host is the internal hop's own address, not the
# browser-facing one.
proto = request.headers.get("x-forwarded-proto") or request.url.scheme
host = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc
return f"{proto}://{host}"
def _respond_html(request: Request, template: str, **ctx) -> Response:
o = _origin(request)
# og:image (an Open Graph tag, which the spec requires be absolute) is the
# one place a *static asset* reference needs an absolute URL always, not
# just when THERMOGRAPH_API_BASE_PUBLIC happens to be set -- ASSET_BASE is
# already absolute in that case, otherwise it's the same relative BASE
# base_url itself is built from, so prefixing with this request's own
# origin recovers exactly today's behavior in the default topology.
asset_base_url = ASSET_BASE if ASSET_BASE.startswith(("http://", "https://")) else f"{o}{ASSET_BASE}"
ctx.setdefault("unit_default", fmt.current_unit())
html = _env.get_template(template).render(
base=BASE, asset_base=ASSET_BASE, asset_base_url=asset_base_url,
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})
def _breadcrumb_tuples(breadcrumb: list[dict]) -> list[tuple]:
"""The templates do `{% for label, href in breadcrumb %}` (tuple
unpacking) -- the backend returns [{"name","href"}, ...] (JSON-safe), so
convert once here rather than touching every template."""
return [(c["name"], c["href"]) for c in breadcrumb]
def _raise_api_error(e: httpx.HTTPStatusError | httpx.TransportError):
"""The backend's 404 (unknown city/month) and 503 (warming) map straight
through -- same status codes _resolve_city raised in-process before. A
TransportError (ConnectError, TimeoutException, ...) never got a response
at all -- a backend blip or restart -- so it becomes a 503 too, rather
than escaping this handler and surfacing as a raw framework 500."""
if isinstance(e, httpx.HTTPStatusError):
raise HTTPException(status_code=e.response.status_code, detail=e.response.text) from e
raise HTTPException(status_code=503, detail="backend unavailable") from e
# --- 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"
f"Disallow: {BASE}/alerts\n"
f"Sitemap: {base_url}/sitemap.xml\n"
)
return PlainTextResponse(body)
# A stable per-process "content last built" date -- crawlers should see
# <lastmod> change on a real rebuild, not churn on every request.
_BOOT_DATE = datetime.date.today().isoformat()
def sitemap_xml(request: Request) -> Response:
base_url = f"{_origin(request)}{BASE}"
try:
entries = api_client.sitemap()
except (httpx.HTTPStatusError, httpx.TransportError) as e:
_raise_api_error(e)
parts = ['<?xml version="1.0" encoding="UTF-8"?>',
'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">']
for e in entries:
parts.append(
f"<url><loc>{base_url}{e['path']}</loc><lastmod>{_BOOT_DATE}</lastmod>"
f"<changefreq>{e['changefreq']}</changefreq><priority>{e['priority']}</priority></url>"
)
parts.append("</urlset>")
# 2.35 MB and crawled repeatedly (~0.5-1s per hit) -- a short cache saves
# every crawler hit within the window a full api_client.sitemap() call
# plus the string build above, on top of the response body itself.
return Response("\n".join(parts), media_type="application/xml",
headers={"Cache-Control": "public, max-age=300"})
def head_verify_html() -> Markup:
"""Search-engine ownership-verification <meta> tags, from env (empty when
unset). No backend dependency -- same env vars, read directly here."""
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 _fmt_months(months: list[dict]) -> list[dict]:
out = []
for m in months:
out.append({
**m,
"high": fmt.temp(m["high_f"]), "low": fmt.temp(m["low_f"]),
"precip": fmt.precip(m["precip_f"]),
"bar": fmt.range_bar(m["range_lo_f"], m["range_hi_f"]),
})
return out
def city_page(request: Request, slug: str) -> Response:
try:
j = api_client.city(slug, origin=_origin(request))
except (httpx.HTTPStatusError, httpx.TransportError) as e:
_raise_api_error(e)
with fmt.unit_scope(j["default_unit"]):
months = _fmt_months(j["months"])
warmest = next((m for m in months if m["slug"] == j["warmest_month_slug"]), None)
coldest = next((m for m in months if m["slug"] == j["coldest_month_slug"]), None)
wettest = next((m for m in months if m["slug"] == j["wettest_month_slug"]), None)
records = j["all_time_records"]
today = None
if j["today_vs_normal"]:
today = {
"date": j["today_vs_normal"]["date"],
"cards": [{**c, "value": fmt.fmt(c["metric"], c["value_f"])}
for c in j["today_vs_normal"]["cards"]],
}
ctx = {
"section": "climate",
"city": j["city"], "display": j["display"], "name": j["city"]["name"],
"year_range": j["year_range"], "n_years": j["n_years"],
"months": months, "warmest": warmest, "coldest": coldest, "wettest": wettest,
"records": records, "today": today,
"tool_hash": f"{j['city']['lat']:.5f},{j['city']['lon']:.5f}",
"flavor": j["flavor"], "event": j["event"],
"compare_url": f"{BASE}/compare#loc={j['city']['lat']:.4f},{j['city']['lon']:.4f}",
"breadcrumb": _breadcrumb_tuples(j["breadcrumb"]),
"canonical_path": j["canonical_path"],
"page_title": j["page_title"], "page_description": j["page_description"],
"jsonld_str": json.dumps(j["jsonld"], ensure_ascii=False, separators=(",", ":")),
}
return _respond_html(request, "city.html.j2", **ctx)
# --- month & records pages --------------------------------------------------
_METRIC_KEY_BY_LABEL = {
"High": "tmax", "Low": "tmin", "Feels-like": "feels",
"Humidity": "humid", "Wind": "wind", "Gust": "gust", "Precip": "precip",
}
def month_page(request: Request, slug: str, month: str) -> Response:
try:
j = api_client.city_month(slug, month)
city = api_client.city(slug)["city"]
except (httpx.HTTPStatusError, httpx.TransportError) as e:
_raise_api_error(e)
with fmt.unit_scope(fmt.unit_for_country(city.get("country_code"))):
stats = []
if j["avg_high_f"] is not None:
stats.append(("Average high", fmt.temp(j["avg_high_f"])))
lo, hi = j["typical_high_range_f"]
stats.append(("Typical high range", fmt.temp(lo) + " to " + fmt.temp(hi)))
if j["avg_low_f"] is not None:
stats.append(("Average low", fmt.temp(j["avg_low_f"])))
lo, hi = j["typical_low_range_f"]
stats.append(("Typical low range", fmt.temp(lo) + " to " + fmt.temp(hi)))
if j["avg_precip_f"] is not None:
stats.append(("Average daily precipitation", fmt.precip(j["avg_precip_f"])))
records = []
for r in j["records"]:
metric = _METRIC_KEY_BY_LABEL[r["label"]]
records.append({
"label": r["label"], "high_tag": r["high_tag"], "low_tag": r["low_tag"],
"high": fmt.fmt(metric, r["high_f"]), "high_date": r["high_date"],
"low": fmt.fmt(metric, r["low_f"]), "low_date": r["low_date"],
})
ctx = {
"section": "climate", "city": city, "display": j.get("display", city["name"]), "name": city["name"],
"month_name": j["month_name"], "month_slug": j["month_slug"],
"year_range": j["year_range"], "n_years": j["n_years"],
"avg_high": fmt.temp(j["avg_high_f"]), "avg_low": fmt.temp(j["avg_low_f"]),
"avg_high_cls": fmt.temp_class(j["avg_high_f"]), "avg_low_cls": fmt.temp_class(j["avg_low_f"]),
"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": j["prev"], "next": j["next"],
"breadcrumb": _breadcrumb_tuples(j["breadcrumb"]),
"canonical_path": j["canonical_path"],
"page_title": j["page_title"], "page_description": j["page_description"],
}
return _respond_html(request, "month.html.j2", **ctx)
def _fmt_extreme(e: dict | None) -> dict | None:
if not e:
return None
return {"txt": fmt.temp(e["value_f"]), "cls": fmt.temp_class(e["value_f"]), "date": e["date"]}
def _fmt_period_extremes(side: dict) -> dict:
"""{'warm': {'value_f','date'}|None, 'cold': ...} -> the {'cls','txt','date'}
shape records.html.j2's `extremes()` macro reads."""
return {k: _fmt_extreme(v) for k, v in side.items()}
def records_page(request: Request, slug: str) -> Response:
try:
j = api_client.city_records(slug, origin=_origin(request))
city = api_client.city(slug)["city"]
except (httpx.HTTPStatusError, httpx.TransportError) as e:
_raise_api_error(e)
with fmt.unit_scope(fmt.unit_for_country(city.get("country_code"))):
rows = []
for r in j["rows"]:
metric = _METRIC_KEY_BY_LABEL[r["label"]]
if r["dry_streak_days"] is not None:
low, low_date = f"{r['dry_streak_days']}-day dry spell", r["low_date"] or "—"
else:
low, low_date = fmt.fmt(metric, r["low_f"]), r["low_date"]
is_temp = metric in ("tmax", "tmin", "feels")
rows.append({
"label": r["label"],
"high": fmt.fmt(metric, r["high_f"]), "high_date": r["high_date"],
"high_f": r["high_f"] if is_temp else None,
"low": low, "low_date": low_date,
"low_f": r["low_f"] if is_temp else None,
})
monthly = [{"name": m["name"], "slug": m["slug"],
"high": _fmt_period_extremes(m["high"]), "low": _fmt_period_extremes(m["low"])}
for m in j["monthly"]]
seasonal = [{"name": s["name"], "span": s["span"],
"high": _fmt_period_extremes(s["high"]), "low": _fmt_period_extremes(s["low"])}
for s in j["seasonal"]]
ctx = {
"section": "climate", "city": city, "display": j.get("display", city["name"]), "name": city["name"],
"year_range": j["year_range"], "n_years": j["n_years"],
"rows": rows, "monthly": monthly, "seasonal": seasonal,
"hemisphere": j["hemisphere"], "all_time": j["all_time_records"],
"canonical_path": j["canonical_path"],
"breadcrumb": _breadcrumb_tuples(j["breadcrumb"]),
"page_title": j["page_title"], "page_description": j["page_description"],
"jsonld_str": json.dumps(j["jsonld"], ensure_ascii=False, separators=(",", ":")),
}
return _respond_html(request, "records.html.j2", **ctx)
# --- hub / glossary / about -------------------------------------------------
def hub_page(request: Request) -> Response:
try:
h = api_client.hub()
except (httpx.HTTPStatusError, httpx.TransportError) as e:
_raise_api_error(e)
groups = {c["country"]: c["cities"] for c in h["countries"]}
ctx = {
"section": "climate",
"groups": groups,
"n_cities": h["n_cities"],
"n_countries": h["n_countries"],
"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)
GLOSSARY: dict[str, dict] = content_loader.load_glossary()
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 ----------------------------------------------------------------
_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."),
}
def home_page(request: Request) -> Response:
try:
h = api_client.home()
except (httpx.HTTPStatusError, httpx.TransportError) as e:
_raise_api_error(e)
ctx = {
"section": "home",
"brand_tag": "p",
"main_class": "",
"canonical_path": "/",
"unusual": h["unusual"],
"stale": h["stale"],
"ranked": h["ranked"],
"cities": h["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 key file ---------------------------------------------------
# IndexNow verification works by serving a file at /<key>.txt -- so the
# key isn't just response *content*, it's baked into the route *path*
# itself, which FastAPI needs at registration time. That used to mean an
# eager, unretried `api_client.indexnow_key()` call right here in
# register() -- called at import time, before the app has served a single
# request -- so if the backend was unreachable (down, mid-restart, a
# renamed/removed route on a newer/older version, a network blip) this
# frontend process would never finish booting at all. That's exactly the
# coupling this task exists to remove: frontend and backend now deploy
# asynchronously, so "backend happens to be briefly unreachable" must be a
# normal, survivable condition at frontend boot, not a crash.
#
# Fix: try the eager fetch once (this preserves today's exact route/
# behavior -- a single static route at /<key>.txt -- for the overwhelming
# common case where the backend IS reachable at boot). If it fails, don't
# let the exception propagate out of register() / crash boot -- log a
# warning and fall back to a catch-all route that lazily (re)fetches the
# key on each request via api_client.indexnow_key(), which already caches
# successful lookups for THERMOGRAPH_SSR_CACHE_TTL seconds (api_client.py's
# TTL cache), so once the backend comes back up the correct key starts
# being served automatically -- no frontend restart required, and no
# request ever gets a permanently wrong/empty key.
try:
_inkey = api_client.indexnow_key()
except Exception:
_log.warning(
"indexnow_key() fetch failed at boot (backend unreachable?) -- "
"continuing boot without it; falling back to lazy per-request "
"lookup at /<key>.txt so the frontend doesn't depend on backend "
"liveness to start up",
exc_info=True,
)
_inkey = None
if _inkey is not None:
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,
)
else:
def _indexnow_key_file_lazy(token: str) -> Response:
try:
key = api_client.indexnow_key()
except Exception as e:
raise HTTPException(status_code=503, detail="backend unavailable") from e
if token != key:
raise HTTPException(status_code=404)
return PlainTextResponse(key + "\n")
app.add_api_route(
f"{BASE}/{{token}}.txt", _indexnow_key_file_lazy,
methods=["GET", "HEAD"], include_in_schema=False,
)
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)
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)
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)
Reference in a new issue View git blame Copy permalink