qtalker---/backend/app/haunts.py

"""Haunted geography — real places near the seeker, from real sources.

Two free, keyless, well-licensed APIs rather than scraping:

  - Wikipedia geosearch + extracts: encyclopedic coverage of a place, which
    doubles as our notability gate. If a site has no Wikipedia article, it
    does not appear. That single rule does most of the ethical and quality
    work here — it keeps the map to documented history rather than rumour,
    and it makes every entry independently checkable by the seeker.
  - OpenStreetMap Overpass: cemeteries, ruins, memorials, battlefields,
    former prisons and asylums — the physical furniture of a haunted map,
    contributed and verified by people on the ground.

Both are consulted for the same point and merged. Neither is scraped: both
publish documented APIs with clear reuse terms (CC BY-SA / ODbL), which is
also why every haunt carries its source and a link back.

WHAT IS DELIBERATELY EXCLUDED, and why
--------------------------------------
Recent crimes at residential addresses. People live in those houses now and
get harassed by visitors; the victims' families are usually still alive.
This is a well-documented harm of true-crime tourism, not a hypothetical.
So:

  - `HISTORICAL_CUTOFF_YEAR` gates crime-flavoured entries to events far
    enough back that no one is being pointed at a living family's door.
  - Anything that resolves to a dwelling is reported at street/area
    precision (`RESIDENTIAL_PRECISION_M`), never a door number.
  - The Wikipedia-article requirement means only events with genuine
    encyclopedic coverage qualify in the first place.

Historic sites — battlefields, plague pits, executions, gaols, asylums,
famous centuries-old cases — are unaffected and fully included. The point
is a map of documented history, not a map of somebody's address.

PRIVACY
-------
The seeker's exact coordinate is never sent upstream. Queries are snapped
to `QUERY_GRID_DEG` (~1km) before leaving this process, which is far finer
than the radius we search and coarse enough that the upstream services
never learn where anybody actually is. It also makes the cache useful,
since everyone in a neighbourhood shares a cache key.
"""

import asyncio
import math
import time

import httpx

WIKI_GEOSEARCH_URL = "https://en.wikipedia.org/w/api.php"
OVERPASS_URL = "https://overpass-api.de/api/interpreter"

# Nothing user-facing waits long on a third party. A missing haunt list is a
# quieter map, not an error.
REQUEST_TIMEOUT_S = 8.0

# Places do not move. A long TTL keeps us a courteous consumer of two free
# services, and the grid key below means a whole neighbourhood shares it.
CACHE_TTL_S = 24 * 3600

# ~0.01 deg latitude is roughly 1.1km. Coarse enough to protect the seeker,
# fine enough that results still feel local.
QUERY_GRID_DEG = 0.01

DEFAULT_RADIUS_M = 3000
MAX_RADIUS_M = 10000
MAX_RESULTS = 24

# Crime-flavoured entries must predate this. Chosen so that the events on
# the map are historical record rather than living memory — see the module
# docstring. Historic sites (battlefields, gaols, plague pits) are not
# subject to it; this gates *crime* framing specifically.
HISTORICAL_CUTOFF_YEAR = 1950

# Residential sites are reported no more precisely than this, so the map
# never points at a specific front door.
RESIDENTIAL_PRECISION_M = 250

# OSM tags that make a place worth a seeker's attention. Each maps to the
# flavour we present it as.
OSM_KINDS: dict[tuple[str, str], str] = {
    ("historic", "battlefield"): "battlefield",
    ("historic", "ruins"): "ruin",
    ("historic", "memorial"): "memorial",
    ("historic", "monument"): "memorial",
    ("historic", "wayside_cross"): "memorial",
    ("historic", "archaeological_site"): "old ground",
    ("historic", "castle"): "old ground",
    ("historic", "manor"): "old ground",
    ("landuse", "cemetery"): "burial ground",
    ("amenity", "grave_yard"): "burial ground",
    ("amenity", "prison"): "gaol",
    ("historic", "prison"): "gaol",
    ("building", "chapel"): "chapel",
    ("historic", "church"): "chapel",
}

# Strong hints are unambiguous on their own: a place described with any of
# these belongs on a haunted map.
STRONG_HINTS = (
    "cemetery", "graveyard", "burial ground", "crypt", "catacomb", "mausoleum",
    "plague", "asylum", "sanatorium", "workhouse", "gallows", "execution",
    "executed", "hanged", "beheaded", "massacre", "battlefield", "siege",
    "haunted", "ghost", "apparition", "poltergeist", "folklore",
    "abbey", "priory", "monastery", "nunnery", "ruins", "castle", "dungeon",
    "gaol", "witch trial", "witchcraft", "shipwreck", "crematorium", "tomb",
)

# Weak hints are ambiguous alone — a railway station's article mentions
# "fire" and a modern clinic mentions "hospital". Verified against the live
# API: "Fenchurch Street railway station" was matching the old flat list and
# landing on the map. Two or more weak hints are required to qualify.
WEAK_HINTS = (
    "hospital", "infirmary", "prison", "jail", "battle", "disaster", "fire",
    "famine", "legend", "church", "chapel", "monument", "memorial", "burial",
    "murder", "killing", "witch", "trial", "grave", "death", "died",
)

# Hints that specifically carry crime framing — these are the ones the
# historical cutoff applies to.
CRIME_HINTS = ("murder", "killing", "massacre", "execution", "hanged", "beheaded", "witch")


def snap_to_grid(value: float, grid: float = QUERY_GRID_DEG) -> float:
    """Round a coordinate to the query grid, so an exact position never
    leaves this process. Symmetric around zero so southern/western
    hemispheres are not biased."""
    return round(value / grid) * grid


def haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Great-circle distance in metres."""
    r = 6371000.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dp = math.radians(lat2 - lat1)
    dl = math.radians(lon2 - lon1)
    a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
    return 2 * r * math.asin(math.sqrt(a))


def looks_like_lore(text: str) -> bool:
    """Does this article/place belong on a haunted map at all?

    One strong hint qualifies; weak hints need corroboration. A flat
    any-keyword match put a railway station on the map during live testing,
    because its article happened to mention a fire.
    """
    lowered = text.lower()
    if any(hint in lowered for hint in STRONG_HINTS):
        return True
    return sum(1 for hint in WEAK_HINTS if hint in lowered) >= 2


def carries_crime_framing(text: str) -> bool:
    return any(hint in text.lower() for hint in CRIME_HINTS)


def extract_years(text: str) -> list[int]:
    """Every plausible 3-4 digit year mentioned. Used only to decide whether
    a crime-framed entry is historical enough to show."""
    years: list[int] = []
    token = ""
    for ch in text + " ":
        if ch.isdigit():
            token += ch
        else:
            if 3 <= len(token) <= 4:
                value = int(token)
                if 1000 <= value <= 2100 or 100 <= value <= 999:
                    years.append(value)
            token = ""
    return years


def passes_historical_gate(text: str, cutoff: int = HISTORICAL_CUTOFF_YEAR) -> bool:
    """Crime-framed entries must be demonstrably historical.

    The rule is deliberately conservative in the ambiguous direction: an
    entry that reads as a crime but carries no legible date is EXCLUDED
    rather than assumed old. Being wrong in the other direction means
    pointing strangers at a recent victim's address, which is exactly what
    this gate exists to prevent.
    """
    if not carries_crime_framing(text):
        return True  # not crime framing; the gate does not apply
    years = [y for y in extract_years(text) if y >= 1000]
    if not years:
        return False
    return max(years) < cutoff


def is_residential(tags: dict) -> bool:
    """Does this OSM element look like somewhere people live?"""
    if tags.get("building") in ("house", "residential", "apartments", "detached", "semidetached_house"):
        return True
    return tags.get("landuse") == "residential"


def blur_for_residential(lat: float, lon: float, residential: bool) -> tuple[float, float]:
    """Snap residential coordinates to ~RESIDENTIAL_PRECISION_M so the map
    shows a street, never a door."""
    if not residential:
        return lat, lon
    grid_deg = RESIDENTIAL_PRECISION_M / 111_320.0
    return round(lat / grid_deg) * grid_deg, round(lon / grid_deg) * grid_deg


def osm_kind(tags: dict) -> str | None:
    for (key, value), kind in OSM_KINDS.items():
        if tags.get(key) == value:
            return kind
    return None


def _dedupe(haunts: list[dict]) -> list[dict]:
    """Wikipedia and OSM frequently describe the same site. Collapse by
    name, keeping the richer entry (the one carrying lore text)."""
    best: dict[str, dict] = {}
    for h in haunts:
        key = (h.get("name") or "").strip().lower()
        if not key:
            continue
        existing = best.get(key)
        if existing is None or (len(h.get("lore") or "") > len(existing.get("lore") or "")):
            best[key] = h
    return list(best.values())


class HauntCache:
    """Grid-keyed cache over both upstreams.

    Same contract as GeomagneticCache: never raises, never blocks a séance,
    and a failed refresh keeps serving whatever was last known good.
    """

    def __init__(self, ttl_s: float = CACHE_TTL_S):
        self._ttl = ttl_s
        self._entries: dict[tuple[float, float, int], tuple[float, list[dict]]] = {}
        self._locks: dict[tuple[float, float, int], asyncio.Lock] = {}

    def _key(self, lat: float, lon: float, radius_m: int) -> tuple[float, float, int]:
        return (snap_to_grid(lat), snap_to_grid(lon), radius_m)

    def cached(self, lat: float, lon: float, radius_m: int) -> list[dict] | None:
        entry = self._entries.get(self._key(lat, lon, radius_m))
        if entry is None:
            return None
        fetched_at, haunts = entry
        if time.monotonic() - fetched_at >= self._ttl:
            return None
        return haunts

    async def get(self, lat: float, lon: float, radius_m: int = DEFAULT_RADIUS_M) -> list[dict]:
        radius_m = max(200, min(MAX_RADIUS_M, int(radius_m)))
        key = self._key(lat, lon, radius_m)
        fresh = self.cached(lat, lon, radius_m)
        if fresh is not None:
            return fresh

        lock = self._locks.setdefault(key, asyncio.Lock())
        async with lock:
            fresh = self.cached(lat, lon, radius_m)
            if fresh is not None:
                return fresh

            # The snapped coordinate is what actually leaves this process.
            q_lat, q_lon = key[0], key[1]
            wiki, osm = await asyncio.gather(
                fetch_wikipedia(q_lat, q_lon, radius_m),
                fetch_overpass(q_lat, q_lon, radius_m),
                return_exceptions=True,
            )
            haunts: list[dict] = []
            for result in (wiki, osm):
                if isinstance(result, list):
                    haunts.extend(result)

            # Distances are measured from the real position so ordering is
            # honest, even though the query itself was snapped.
            for h in haunts:
                h["distance_m"] = round(haversine_m(lat, lon, h["lat"], h["lon"]))
            haunts = _dedupe(haunts)
            haunts.sort(key=lambda h: h["distance_m"])
            haunts = haunts[:MAX_RESULTS]

            if haunts or key not in self._entries:
                self._entries[key] = (time.monotonic(), haunts)
            return self._entries[key][1]


async def fetch_wikipedia(lat: float, lon: float, radius_m: int) -> list[dict]:
    """Nearby articles, filtered to things that belong on a haunted map."""
    params = {
        "action": "query",
        "format": "json",
        "generator": "geosearch",
        "ggscoord": f"{lat}|{lon}",
        "ggsradius": str(min(10000, radius_m)),
        "ggslimit": "40",
        "prop": "extracts|coordinates",
        "exintro": "1",
        "explaintext": "1",
        "exsentences": "3",
    }
    try:
        async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_S) as client:
            response = await client.get(
                WIKI_GEOSEARCH_URL,
                params=params,
                headers={"User-Agent": "Quantumancy/1.0 (haunted-places map)"},
            )
            response.raise_for_status()
            payload = response.json()
    except Exception:
        return []

    out: list[dict] = []
    pages = (payload.get("query") or {}).get("pages") or {}
    for page in pages.values():
        title = page.get("title") or ""
        extract = (page.get("extract") or "").strip()
        blob = f"{title} {extract}"
        if not looks_like_lore(blob) or not passes_historical_gate(blob):
            continue
        coords = (page.get("coordinates") or [{}])[0]
        p_lat, p_lon = coords.get("lat"), coords.get("lon")
        if p_lat is None or p_lon is None:
            continue
        out.append(
            {
                "name": title,
                "kind": "recorded history",
                "lat": float(p_lat),
                "lon": float(p_lon),
                "lore": extract[:400],
                "source": "Wikipedia",
                "url": f"https://en.wikipedia.org/?curid={page.get('pageid')}",
                "precise": True,
            }
        )
    return out


def build_overpass_query(lat: float, lon: float, radius_m: int) -> str:
    """Overpass QL for the physical furniture of a haunted map.

    Written as one regex-alternated clause per tag KEY rather than one
    clause per key/value pair. The naive form (14 kinds x 2 element types =
    28 separate `around:` searches) makes the public Overpass instance do 28
    spatial lookups and it answers 504 — verified against the live service.
    Grouping by key collapses that to four, and `nwr` covers node/way/
    relation in a single pass instead of enumerating element types.
    """
    by_key: dict[str, list[str]] = {}
    for (key, value) in OSM_KINDS:
        by_key.setdefault(key, []).append(value)
    clauses = "".join(
        f'nwr["{key}"~"^({"|".join(sorted(set(values)))})$"](around:{radius_m},{lat},{lon});'
        for key, values in sorted(by_key.items())
    )
    return f"[out:json][timeout:25];({clauses});out center {MAX_RESULTS * 3};"


async def fetch_overpass(lat: float, lon: float, radius_m: int) -> list[dict]:
    try:
        async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_S) as client:
            response = await client.post(
                OVERPASS_URL,
                data={"data": build_overpass_query(lat, lon, radius_m)},
                headers={"User-Agent": "Quantumancy/1.0 (haunted-places map)"},
            )
            response.raise_for_status()
            payload = response.json()
    except Exception:
        return []

    out: list[dict] = []
    for element in payload.get("elements") or []:
        tags = element.get("tags") or {}
        name = tags.get("name")
        if not name:
            continue  # unnamed furniture is noise on a map
        kind = osm_kind(tags)
        if kind is None:
            continue
        centre = element.get("center") or element
        e_lat, e_lon = centre.get("lat"), centre.get("lon")
        if e_lat is None or e_lon is None:
            continue
        residential = is_residential(tags)
        e_lat, e_lon = blur_for_residential(float(e_lat), float(e_lon), residential)
        out.append(
            {
                "name": name,
                "kind": kind,
                "lat": e_lat,
                "lon": e_lon,
                "lore": (tags.get("description") or tags.get("inscription") or "")[:400],
                "source": "OpenStreetMap",
                "url": f"https://www.openstreetmap.org/{element.get('type')}/{element.get('id')}",
                "precise": not residential,
            }
        )
    return out


haunt_cache = HauntCache()