"""ArcGIS FeatureServer and MapServer query helpers."""

from __future__ import annotations

import csv
import io
import urllib.parse
from typing import Any

from raleighlib import core


def service_metadata(url: str) -> dict[str, Any]:
    """Fetch service or layer metadata from an ArcGIS REST endpoint."""
    sep = "&" if "?" in url else "?"
    return core.json_request(f"{url}{sep}f=json")


def _looks_like_layer_url(url: str) -> bool:
    """Return True if url already ends with a layer id."""
    import re

    return bool(re.search(r"/(FeatureServer|MapServer)/\d+/?$", url.rstrip("/")))


def _looks_like_service_root(url: str) -> bool:
    import re

    return bool(re.search(r"/(FeatureServer|MapServer)/?$", url.rstrip("/")))


def resolve_queryable_layer(url: str) -> str:
    """Return a queryable layer URL, resolving service roots to their first layer."""
    if _looks_like_layer_url(url):
        return url.rstrip("/")
    if not _looks_like_service_root(url):
        # Not a recognized service root or layer; return as-is and let the caller fail clearly.
        return url
    meta = service_metadata(url)
    layers = meta.get("layers", []) or meta.get("tables", []) or []
    for candidate in layers:
        if candidate.get("subLayerIds") is None:
            return f"{url.rstrip('/')}/{candidate['id']}"
    # Fallback to first layer if all are group layers.
    if layers:
        return f"{url.rstrip('/')}/{layers[0]['id']}"
    raise ValueError(f"No queryable layers found at {url}")


def layer_has_geometry(url: str) -> bool:
    """Return True if the layer advertises a geometry type."""
    meta = service_metadata(url)
    if meta.get("type") == "Table":
        return False
    return meta.get("geometryType") is not None


def layer_fields(url: str) -> list[dict[str, Any]]:
    """Return the field definitions advertised by a layer or table."""
    meta = service_metadata(url)
    return meta.get("fields", [])


def sample_features(url: str, limit: int = 3) -> list[dict[str, Any]]:
    """Query a small sample of features from a layer."""
    return query_all_pages(url, max_records=limit, return_geometry=True)


def query_layer(
    url: str,
    where: str = "1=1",
    out_fields: str = "*",
    return_geometry: bool = True,
    result_record_count: int | None = None,
    result_offset: int = 0,
    order_by_fields: str | None = None,
    f: str = "json",
    out_sr: int | None = 4326,
) -> dict[str, Any]:
    """Query a single page from an ArcGIS layer."""
    core.require_positive_limit(result_record_count, allow_none=True)
    if result_offset < 0:
        raise ValueError("result_offset must be nonnegative")
    base = url.rstrip("/") + "/query"
    params: dict[str, Any] = {
        "where": where,
        "outFields": out_fields,
        "returnGeometry": "true" if return_geometry else "false",
        "f": f,
        "outSR": out_sr,
    }
    if result_record_count is not None:
        params["resultRecordCount"] = result_record_count
    if result_offset:
        params["resultOffset"] = result_offset
    if order_by_fields:
        params["orderByFields"] = order_by_fields
    if out_sr is not None:
        params["outSR"] = out_sr
    full_url = f"{base}?{urllib.parse.urlencode(params)}"
    response = core.json_request(full_url)
    core.raise_for_arcgis_error(response, "ArcGIS query")
    return response


def query_all_pages(
    url: str,
    where: str = "1=1",
    out_fields: str = "*",
    return_geometry: bool = True,
    max_records: int | None = None,
    offset: int = 0,
    order_by_fields: str | None = None,
    f: str = "json",
    out_sr: int | None = 4326,
    page_size: int = 1000,
    max_pages: int = 100,
) -> list[dict[str, Any]]:
    """Paginate through an ArcGIS query and return all feature records."""
    if page_size < 1 or max_pages < 1:
        raise ValueError("page_size and max_pages must be positive")
    core.require_positive_limit(max_records, allow_none=True)
    if offset < 0:
        raise ValueError("offset must be nonnegative")
    records: list[dict[str, Any]] = []
    total_offset = offset
    remaining = max_records
    seen_pages: set[str] = set()
    for _page_number in range(max_pages):
        limit = min(page_size, remaining) if remaining is not None else page_size
        page = query_layer(
            url,
            where=where,
            out_fields=out_fields,
            return_geometry=return_geometry,
            result_record_count=limit,
            result_offset=total_offset,
            order_by_fields=order_by_fields,
            f=f,
            out_sr=out_sr,
        )
        features = page.get("features", [])
        if not isinstance(features, list):
            raise ValueError("ArcGIS query returned invalid features")
        if not features:
            return records
        signature = repr(features)
        if signature in seen_pages:
            raise ValueError("ArcGIS query repeated a page")
        seen_pages.add(signature)
        raw_count = len(features)
        accepted = features[:remaining] if remaining is not None else features
        records.extend(accepted)
        total_offset += raw_count
        if remaining is not None:
            remaining -= len(accepted)
            if remaining <= 0:
                return records
        if raw_count < limit:
            return records
        # Some servers signal pagination via exceededTransferLimit.
        if not page.get("exceededTransferLimit", True):
            return records
    raise ValueError(f"ArcGIS query exceeded {max_pages} pages")


def _signed_area(ring: list[list[float]]) -> float:
    """Return the signed area of a linear ring using the shoelace formula."""
    area = 0.0
    n = len(ring)
    for i in range(n):
        if len(ring[i]) < 2 or len(ring[(i + 1) % n]) < 2:
            raise ValueError("ArcGIS polygon coordinate must contain x and y")
        x1, y1 = ring[i][0], ring[i][1]
        x2, y2 = ring[(i + 1) % n][0], ring[(i + 1) % n][1]
        area += (x1 * y2) - (x2 * y1)
    return area / 2.0


def _normalize_ring(ring: list[list[float]]) -> list[list[float]]:
    """Return a closed ring (append first point if missing)."""
    if not ring:
        return ring
    if ring[0] != ring[-1]:
        return ring + [ring[0]]
    return ring


def _rings_to_geojson(rings: list[list[list[float]]]) -> dict[str, Any]:
    """Convert ArcGIS rings to a GeoJSON Polygon or MultiPolygon."""
    if not rings:
        return {"type": "Polygon", "coordinates": []}
    normalized = [_normalize_ring(ring) for ring in rings]
    # ArcGIS uses clockwise exterior rings and counterclockwise holes. GeoJSON
    # recommends the opposite winding, so reverse both while converting.
    exteriors = [ring for ring in normalized if _signed_area(ring) < 0]
    holes = [ring for ring in normalized if _signed_area(ring) >= 0]
    if not exteriors:
        raise ValueError("ArcGIS polygon has no clockwise exterior ring")

    def contains(ring: list[list[float]], point: list[float]) -> bool:
        if len(point) < 2:
            raise ValueError("ArcGIS polygon coordinate must contain x and y")
        x, y = point[0], point[1]
        inside = False
        for first, second in zip(ring, ring[1:]):
            if len(first) < 2 or len(second) < 2:
                raise ValueError("ArcGIS polygon coordinate must contain x and y")
            x1, y1 = first[0], first[1]
            x2, y2 = second[0], second[1]
            if (y1 > y) != (y2 > y):
                crossing = (x2 - x1) * (y - y1) / (y2 - y1) + x1
                if x < crossing:
                    inside = not inside
        return inside

    polygons: list[list[list[list[float]]]] = [[list(reversed(ring))] for ring in exteriors]
    for hole in holes:
        containing = [
            (abs(_signed_area(exterior)), index)
            for index, exterior in enumerate(exteriors)
            if contains(exterior, hole[0])
        ]
        if not containing:
            raise ValueError("ArcGIS polygon contains a hole outside every exterior ring")
        _, index = min(containing)
        polygons[index].append(list(reversed(hole)))

    if len(polygons) == 1:
        return {"type": "Polygon", "coordinates": polygons[0]}
    return {"type": "MultiPolygon", "coordinates": polygons}


def _paths_to_geojson(paths: list[list[list[float]]]) -> dict[str, Any]:
    """Convert ArcGIS paths to a GeoJSON LineString or MultiLineString."""
    if not paths:
        return {"type": "LineString", "coordinates": []}
    if len(paths) == 1:
        return {"type": "LineString", "coordinates": paths[0]}
    return {"type": "MultiLineString", "coordinates": paths}


def geometry_from_record(record: dict[str, Any]) -> dict[str, Any] | None:
    """Extract and normalize geometry from a feature record."""
    geom = record.get("geometry")
    if not geom:
        return None
    if "x" in geom and "y" in geom:
        return {"type": "Point", "coordinates": [geom["x"], geom["y"]]}
    if "points" in geom:
        return {"type": "MultiPoint", "coordinates": geom["points"]}
    if "rings" in geom:
        return _rings_to_geojson(geom["rings"])
    if "paths" in geom:
        return _paths_to_geojson(geom["paths"])
    return geom


def _attribute_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [r.get("attributes", {}) for r in records]


def csv_safe_value(value: Any) -> str:
    """Return a CSV-safe string, prefixing formula triggers to prevent injection."""
    if value is None:
        return ""
    text = str(value)
    stripped = text.lstrip().lstrip("\ufeff").lstrip()
    if text.startswith(("\t", "\r", "\n", "\v", "\f")) or (
        stripped and stripped[0] in "=+-@"
    ):
        return "'" + text
    return text


def csv_from_records(records: list[dict[str, Any]]) -> str:
    """Convert feature records to a CSV string with unioned keys and formula injection protection."""
    rows = _attribute_rows(records)
    if not rows:
        return ""
    # Preserve key order of the first row while unioning keys from all rows.
    fieldnames: list[str] = []
    seen: set[str] = set()
    for row in rows:
        for key in row.keys():
            if key not in seen:
                seen.add(key)
                fieldnames.append(key)
    safe_rows = [
        {key: csv_safe_value(row.get(key, "")) for key in fieldnames}
        for row in rows
    ]
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow([csv_safe_value(field) for field in fieldnames])
    for row in safe_rows:
        writer.writerow([row.get(field, "") for field in fieldnames])
    return buf.getvalue()


def geojson_from_records(records: list[dict[str, Any]]) -> dict[str, Any]:
    """Convert feature records to a GeoJSON FeatureCollection."""
    features: list[dict[str, Any]] = []
    for record in records:
        geom = geometry_from_record(record)
        features.append(
            {
                "type": "Feature",
                "properties": record.get("attributes", {}),
                "geometry": geom,
            }
        )
    return {"type": "FeatureCollection", "features": features}

"""RaleighNC.gov public civic-content adapter (JSON:API + RSS)."""

from __future__ import annotations

import urllib.parse
import xml.etree.ElementTree as ET
from datetime import date, datetime
from typing import Any

from raleighlib import core


JSONAPI_ROOT = "https://raleighnc.gov/jsonapi"
RSS_FEED = "https://raleighnc.gov/rss.xml"

# Explicit allowlist of public content resource types.
ALLOWED_RESOURCE_TYPES = frozenset({
    "node--news",
    "node--event",
    "node--event_series",
    "node--project",
    "node--place",
    "node--service",
    "node--service_core",
    "node--directory_entry",
    "node--organizational_unit",
    "node--alert",
    "node--alert_update",
    "node--status_alert",
})


class ResourceError(Exception):
    """Raised when a disallowed resource type is requested."""


def _resource_type_to_parts(resource_type: str) -> tuple[str, str]:
    """Split a resource type into entity_type and bundle."""
    if resource_type not in ALLOWED_RESOURCE_TYPES:
        raise ResourceError(f"Resource type is not allowlisted: {resource_type}")
    parts = resource_type.split("--")
    if len(parts) != 2:
        raise ResourceError(f"Unsupported resource type format: {resource_type}")
    return parts[0], parts[1]


def _discover_resource_paths() -> dict[str, str]:
    """Discover canonical JSON:API paths from the index and cache them."""
    cached = core.read_cache("jsonapi-paths.json", max_age_seconds=86400)
    if isinstance(cached, dict) and all(
        isinstance(key, str) and isinstance(value, str)
        for key, value in cached.items()
    ):
        return cached
    try:
        index = core.json_request(JSONAPI_ROOT)
    except Exception:
        return {}
    if not isinstance(index, dict) or not isinstance(index.get("links"), dict):
        return {}
    paths: dict[str, str] = {}
    for key, value in index["links"].items():
        if "--" not in key:
            continue
        href = _extract_href(value)
        if href:
            paths[key] = href
    core.write_cache("jsonapi-paths.json", paths)
    return paths


def _resource_type_to_path(resource_type: str) -> str:
    """Map a resource type to its canonical JSON:API collection URL."""
    entity_type, bundle = _resource_type_to_parts(resource_type)
    discovered = _discover_resource_paths()
    if resource_type in discovered:
        url = urllib.parse.urljoin(JSONAPI_ROOT + "/", discovered[resource_type])
        parsed = urllib.parse.urlparse(url)
        expected_path = f"/jsonapi/{entity_type}/{bundle}"
        try:
            port = parsed.port
        except ValueError as exc:
            raise ResourceError("Raleigh JSON:API resource path has an invalid port") from exc
        if (
            parsed.scheme != "https"
            or parsed.hostname != "raleighnc.gov"
            or parsed.username is not None
            or parsed.password is not None
            or port not in (None, 443)
            or parsed.path.rstrip("/") != expected_path
        ):
            raise ResourceError(
                f"Raleigh JSON:API index returned an invalid path for {resource_type}"
            )
        return url
    # Fallback to conventional Drupal path.
    return f"{JSONAPI_ROOT}/{entity_type}/{bundle}"


def _extract_href(value: Any) -> str | None:
    """Return the href string from a JSON:API link object or string."""
    if isinstance(value, str):
        return value
    if isinstance(value, dict):
        return value.get("href")
    return None


def _validated_jsonapi_next(value: Any, current_url: str = JSONAPI_ROOT + "/") -> str | None:
    """Return a pagination URL only when it stays on the JSON:API origin/path."""
    href = _extract_href(value)
    if href is None:
        return None
    url = urllib.parse.urljoin(current_url, href)
    parsed = urllib.parse.urlparse(url)
    current = urllib.parse.urlparse(current_url)
    if (
        parsed.scheme != "https"
        or parsed.hostname != "raleighnc.gov"
        or parsed.username is not None
        or parsed.password is not None
        or parsed.port not in (None, 443)
        or parsed.path.rstrip("/") != current.path.rstrip("/")
    ):
        raise ResourceError("Raleigh JSON:API pagination link left the expected origin or path")
    return url


def list_resource_types() -> list[str]:
    """Return the allowlisted public resource types."""
    return sorted(ALLOWED_RESOURCE_TYPES)


def _canonical_url(item: dict[str, Any]) -> str | None:
    """Return the canonical public page URL for a JSON:API item."""
    attrs = item.get("attributes", {})
    path_alias = attrs.get("path", {}).get("alias")
    if path_alias:
        return f"https://raleighnc.gov{path_alias}"
    links = item.get("links", {})
    return _extract_href(links.get("canonical")) or _extract_href(links.get("self"))


def _normalize_jsonapi_item(item: dict[str, Any]) -> dict[str, Any]:
    attrs = item.get("attributes", {})
    relationships: dict[str, Any] = {}
    for name, relationship in item.get("relationships", {}).items():
        if isinstance(relationship, dict):
            relationships[name] = relationship.get("data")
    return {
        "id": item.get("id"),
        "type": item.get("type"),
        "title": attrs.get("title") or attrs.get("label") or attrs.get("name"),
        "created": attrs.get("created"),
        "changed": attrs.get("changed"),
        "status": attrs.get("status"),
        "url": _canonical_url(item),
        "attributes": attrs,
        "relationships": relationships,
    }


def _matches_search(record: dict[str, Any], search: str | None) -> bool:
    if not search:
        return True
    term = search.lower()
    text = " ".join(
        str(record.get(k, "")) for k in ("title", "attributes")
    ).lower()
    return term in text


def _matches_date_range(
    record: dict[str, Any],
    date_from: str | None,
    date_to: str | None,
    date_field: str | None,
) -> bool:
    if not date_from and not date_to:
        return True
    attrs = record.get("attributes", {})
    value = attrs.get(date_field) if date_field else None
    if isinstance(value, dict):
        value = value.get("value")
    if not value and "created" in attrs:
        value = attrs["created"]
    if not value:
        return False
    value_str = str(value).strip()
    try:
        if "T" in value_str or " " in value_str:
            normalized = value_str.replace("Z", "+00:00")
            value_date = datetime.fromisoformat(normalized).date()
        else:
            value_date = date.fromisoformat(value_str)
    except ValueError as exc:
        raise ResourceError(f"Raleigh JSON:API returned an invalid date: {value_str}") from exc
    if date_from and value_date < date.fromisoformat(date_from):
        return False
    if date_to and value_date > date.fromisoformat(date_to):
        return False
    return True


def _matches_relationship(record: dict[str, Any], relationship: str | None) -> bool:
    """Match a client-side JSON:API relationship expression, FIELD=ID."""
    if not relationship:
        return True
    if "=" not in relationship:
        raise ValueError("relationship must use FIELD=ID")
    field, target_id = (part.strip() for part in relationship.split("=", 1))
    if not field or not target_id:
        raise ValueError("relationship must use FIELD=ID")
    data = record.get("relationships", {}).get(field)
    related = data if isinstance(data, list) else [data]
    return any(
        isinstance(item, dict) and str(item.get("id") or "") == target_id
        for item in related
    )


def _build_collection_url(base: str, page_limit: int) -> str:
    query = {"filter[status]": "1", "page[limit]": page_limit}
    return f"{base}?{urllib.parse.urlencode(query)}"


def fetch_jsonapi(
    resource_type: str,
    limit: int = 20,
    search: str | None = None,
    date_from: str | None = None,
    date_to: str | None = None,
    date_field: str | None = None,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch paginated JSON:API records for an allowlisted resource type.

    Only ``filter[status]=1`` is sent to the server. Search and date filters are
    applied client-side because the Raleigh JSON:API implementation does not
    expose a verified server-side fulltext or date filter.
    """
    core.require_positive_limit(limit)
    if relationship:
        if "=" not in relationship or not all(
            part.strip() for part in relationship.split("=", 1)
        ):
            raise ValueError("relationship must use FIELD=ID")
    base = _resource_type_to_path(resource_type)
    # Client-side filters may need to inspect many upstream records before
    # finding ``limit`` matches. Use a bounded service page size independent of
    # the requested output count so selective date/search filters remain usable.
    url = _build_collection_url(base, page_limit=max(50, min(limit, 100)))
    records: list[dict[str, Any]] = []
    seen_ids: set[str] = set()
    pages = 0
    max_pages = 100
    while url and pages < max_pages:
        pages += 1
        data = core.json_request(url)
        if not isinstance(data, dict):
            raise ResourceError("Raleigh JSON:API returned a non-object document")
        items = data.get("data", [])
        if not isinstance(items, list):
            raise ResourceError("Raleigh JSON:API returned invalid data")
        for item in items:
            if not isinstance(item, dict):
                raise ResourceError("Raleigh JSON:API returned a non-object resource")
            attributes = item.get("attributes")
            if not isinstance(attributes, dict) or attributes.get("status") is not True:
                continue
            record = _normalize_jsonapi_item(item)
            rid = record.get("id")
            if rid in seen_ids:
                continue
            seen_ids.add(rid)
            if (
                _matches_search(record, search)
                and _matches_date_range(record, date_from, date_to, date_field)
                and _matches_relationship(record, relationship)
            ):
                records.append(record)
                if len(records) >= limit:
                    return records
        links = data.get("links", {})
        if not isinstance(links, dict):
            raise ResourceError("Raleigh JSON:API returned invalid links")
        url = _validated_jsonapi_next(links.get("next"), url)
    if url:
        raise ResourceError(f"Raleigh JSON:API exceeded {max_pages} pages")
    return records


def fetch_news(
    limit: int = 20,
    search: str | None = None,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch news items."""
    return fetch_jsonapi("node--news", limit=limit, search=search, relationship=relationship)


def fetch_events(
    limit: int = 20,
    date_from: str | None = None,
    date_to: str | None = None,
    search: str | None = None,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch events filtered by date range."""
    return fetch_jsonapi(
        "node--event",
        limit=limit,
        search=search,
        date_from=date_from,
        date_to=date_to,
        date_field="field_event_date",
        relationship=relationship,
    )


def fetch_projects(
    limit: int = 20,
    search: str | None = None,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch projects."""
    return fetch_jsonapi("node--project", limit=limit, search=search, relationship=relationship)


def fetch_places(
    limit: int = 20,
    search: str | None = None,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch places."""
    return fetch_jsonapi("node--place", limit=limit, search=search, relationship=relationship)


def fetch_services(
    limit: int = 20,
    search: str | None = None,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch services."""
    return fetch_jsonapi("node--service", limit=limit, search=search, relationship=relationship)


def fetch_directory(
    limit: int = 20,
    search: str | None = None,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch directory entries."""
    return fetch_jsonapi(
        "node--directory_entry", limit=limit, search=search, relationship=relationship
    )


def fetch_alerts(
    limit: int = 20,
    relationship: str | None = None,
) -> list[dict[str, Any]]:
    """Fetch public alerts."""
    return fetch_jsonapi("node--alert", limit=limit, relationship=relationship)


def fetch_rss(limit: int = 20, new_only: bool = False) -> list[dict[str, Any]]:
    """Fetch and parse the RSS feed."""
    core.require_positive_limit(limit)
    data = core.raw_request(RSS_FEED)
    root = ET.fromstring(data)
    items: list[dict[str, Any]] = []
    previous = set(core.read_cache("rss-seen.json") or []) if new_only else set()
    seen: set[str] = set()
    for channel in root.findall("channel"):
        for item in channel.findall("item"):
            title = item.findtext("title", default="").strip()
            link = item.findtext("link", default="").strip()
            pub_date = item.findtext("pubDate", default="").strip()
            description = item.findtext("description", default="").strip()
            guid = item.findtext("guid", default="").strip()
            identity = guid or link or f"{title}\0{pub_date}"
            if identity in seen or identity in previous:
                continue
            seen.add(identity)
            items.append(
                {
                    "title": title,
                    "url": link,
                    "pub_date": pub_date,
                    "description": description,
                }
            )
            if len(items) >= limit:
                break
        if len(items) >= limit:
            break
    if new_only:
        core.write_cache("rss-seen.json", sorted(previous | seen))
    return items

"""Command-line interface for the Raleigh civic-data skill."""

from __future__ import annotations

import argparse
import csv
import io
import json
import math
import os
import re
import sys
import urllib.error
import urllib.parse
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Any

from raleighlib import core
from raleighlib import hub
from raleighlib import arcgis
from raleighlib import imagery
from raleighlib import geocode
from raleighlib import transit
from raleighlib import development
from raleighlib import civic
from raleighlib import meetings
from raleighlib import police
from raleighlib import fire
from raleighlib import fire_protection
from raleighlib import rfd_reports
from raleighlib import public_safety_stats
from raleighlib import incidents


def _output_json(data: Any) -> None:
    json.dump(data, sys.stdout, indent=2)
    sys.stdout.write("\n")


def _output_table(headers: list[str], rows: list[list[str]]) -> None:
    if not rows:
        return
    widths = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(str(cell)))
    fmt = "  ".join(f"{{:{w}}}" for w in widths)
    print(fmt.format(*headers))
    print(fmt.format(*["-" * w for w in widths]))
    for row in rows:
        print(fmt.format(*(str(c) for c in row)))


def cli_error(message: str) -> SystemExit:
    """Return a SystemExit with concise stderr message."""
    return SystemExit(f"Error: {message}")


def _positive_int(value: str) -> int:
    number = int(value)
    if not 1 <= number <= 100_000:
        raise argparse.ArgumentTypeError("value must be between 1 and 100000")
    return number


def _csaid_value(value: str) -> int:
    number = int(value)
    if not 1 <= number <= 99_999_999:
        raise argparse.ArgumentTypeError("CSAID must be between 1 and 99999999")
    return number


def _nonnegative_int(value: str) -> int:
    number = int(value)
    if not 0 <= number <= 100_000_000:
        raise argparse.ArgumentTypeError("value must be between 0 and 100000000")
    return number


def _timeout_seconds(value: str) -> int:
    number = int(value)
    if not 1 <= number <= 600:
        raise argparse.ArgumentTypeError("timeout must be between 1 and 600 seconds")
    return number


def _year(value: str) -> int:
    number = int(value)
    if not 1900 <= number <= 2100:
        raise argparse.ArgumentTypeError("year must be between 1900 and 2100")
    return number


def _score(value: str) -> float:
    number = float(value)
    if not math.isfinite(number) or not 0 <= number <= 100:
        raise argparse.ArgumentTypeError("score must be finite and between 0 and 100")
    return number


def _latitude(value: str) -> float:
    number = float(value)
    if not math.isfinite(number) or not -90 <= number <= 90:
        raise argparse.ArgumentTypeError("latitude must be finite and between -90 and 90")
    return number


def _longitude(value: str) -> float:
    number = float(value)
    if not math.isfinite(number) or not -180 <= number <= 180:
        raise argparse.ArgumentTypeError("longitude must be finite and between -180 and 180")
    return number


def _iso_date(value: str) -> str:
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
        raise argparse.ArgumentTypeError("date must use YYYY-MM-DD")
    try:
        parsed = date.fromisoformat(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("date must use YYYY-MM-DD") from exc
    return parsed.isoformat()


def _nonempty_text(value: str) -> str:
    value = value.strip()
    if not value:
        raise argparse.ArgumentTypeError("value must not be empty")
    if len(value) > 200 or any(ord(char) < 32 for char in value):
        raise argparse.ArgumentTypeError("value is invalid")
    return value


def _service_date(value: str) -> str:
    try:
        datetime.strptime(value, "%Y%m%d")
    except ValueError as exc:
        raise argparse.ArgumentTypeError("date must use YYYYMMDD") from exc
    return value


def _bbox_value(value: str) -> str:
    _parse_bbox(value)
    return value


def _point_value(value: str) -> str:
    _parse_point(value)
    return value


def _latlon_value(value: str) -> str:
    parts = value.split(",")
    if len(parts) != 2:
        raise argparse.ArgumentTypeError("coordinate must be lat,lon")
    _latitude(parts[0].strip())
    _longitude(parts[1].strip())
    return value


def _size_value(value: str) -> str:
    try:
        _parse_size(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(str(exc)) from exc
    return value


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="raleigh",
        description="CLI for the City of Raleigh civic data and services.",
    )
    parser.add_argument(
        "--json", action="store_true", help="Emit machine-readable JSON output."
    )
    parser.add_argument(
        "--cache-dir", help="Override the default cache directory."
    )
    parser.add_argument(
        "--refresh", action="store_true", help="Bypass cache for catalog operations."
    )
    parser.add_argument(
        "--timeout", type=_timeout_seconds, default=30, help="HTTP timeout in seconds."
    )
    sub = parser.add_subparsers(dest="command", help="Available commands")

    # Legacy / core dataset commands.
    cat_p = sub.add_parser("catalog", help="List datasets from the live catalog.")
    cat_p.add_argument("--limit", type=_positive_int, default=100, help="Maximum datasets to display.")
    cat_p.add_argument("--category", help="Filter by category substring.")
    cat_p.add_argument("--search", help="Filter by search term.")
    search_p = sub.add_parser("search", help="Search the live catalog.")
    search_p.add_argument("query", help="Search terms.")
    search_p.add_argument("--limit", type=_positive_int, default=20, help="Maximum results.")

    info_p = sub.add_parser("info", help="Show dataset info.")
    info_p.add_argument("dataset", help="Dataset title or ID.")

    query_p = sub.add_parser("query", help="Query records from a dataset.")
    query_p.add_argument("dataset", help="Dataset title or ID.")
    query_p.add_argument("--where", default="1=1", help="SQL WHERE clause.")
    query_p.add_argument("--limit", type=_positive_int, default=10, help="Maximum records.")
    query_p.add_argument("--offset", type=_nonnegative_int, default=0, help="Records to skip.")
    query_p.add_argument("--out-fields", default="*", help="Comma-separated fields.")
    query_p.add_argument("--no-geometry", action="store_true", help="Omit geometry.")
    query_p.add_argument("--order-by", help="ORDER BY fields.")
    query_p.add_argument("--csv", action="store_true", help="Emit CSV output.")

    download_p = sub.add_parser("download", help="Download records from a dataset.")
    download_p.add_argument("dataset", help="Dataset title or ID.")
    download_p.add_argument("--where", default="1=1", help="SQL WHERE clause.")
    download_p.add_argument("--limit", type=_positive_int, default=1000, help="Maximum records.")
    download_p.add_argument("--all", action="store_true", help="Fetch all records; overrides --limit.")
    download_p.add_argument("--out-fields", default="*", help="Comma-separated fields.")
    download_p.add_argument("--no-geometry", action="store_true", help="Omit geometry.")
    download_p.add_argument("-f", "--format", default="csv", choices=["csv", "geojson", "json"])
    download_p.add_argument("-o", "--output", required=True, help="Output file path.")
    download_p.add_argument("--force", action="store_true", help="Overwrite an existing output file.")

    sub.add_parser("categories", help="List catalog categories.")

    # Imagery commands.
    img = sub.add_parser("imagery", help="ArcGIS ImageServer imagery operations.")
    img_sub = img.add_subparsers(dest="imagery_command")
    img_cat = img_sub.add_parser("catalog", help="List imagery services.")
    img_cat.add_argument("--limit", type=_positive_int, default=100, help="Maximum services to display.")
    img_info = img_sub.add_parser("info", help="Show imagery service metadata.")
    img_info.add_argument("service", help="Service name or URL.")
    img_export = img_sub.add_parser("export", help="Export a bounded image.")
    img_export.add_argument("service", help="Service name or URL.")
    img_export.add_argument("--bbox", required=True, type=_bbox_value, help="xmin,ymin,xmax,ymax")
    img_export.add_argument("--size", type=_size_value, help="width,height")
    img_export.add_argument("--format", default="jpgpng", dest="format_")
    img_export.add_argument("--in-sr", type=_positive_int, default=4326)
    img_export.add_argument("--out-sr", type=_positive_int, default=4326)
    img_export.add_argument("-o", "--output", required=True)
    img_export.add_argument("--force", action="store_true")
    img_identify = img_sub.add_parser("identify", help="Identify pixel at a point.")
    img_identify.add_argument("service", help="Service name or URL.")
    img_identify.add_argument("--point", required=True, type=_point_value, help="lon,lat")
    img_stats = img_sub.add_parser("statistics", help="Compute statistics for an extent.")
    img_stats.add_argument("service", help="Service name or URL.")
    img_stats.add_argument("--bbox", required=True, type=_bbox_value, help="xmin,ymin,xmax,ymax")

    # Geocoding commands.
    geo = sub.add_parser("geocode", help="Forward geocode an address.")
    geo.add_argument("address", help="Address string.")
    geo.add_argument("--min-score", type=_score, default=None)
    geo.add_argument("--max", type=_positive_int, default=10, dest="max_locations")

    rev_geo = sub.add_parser("reverse-geocode", help="Reverse geocode a coordinate.")
    rev_geo.add_argument("--lat", required=True, type=_latitude)
    rev_geo.add_argument("--lon", required=True, type=_longitude)

    suggest_p = sub.add_parser("suggest", help="Address autocomplete.")
    suggest_p.add_argument("text", help="Partial address.")
    suggest_p.add_argument("--max", "--limit", type=_positive_int, default=10, dest="max_suggestions")

    batch_geo = sub.add_parser("geocode-batch", help="Batch geocode addresses from CSV.")
    batch_geo.add_argument("input", help="Input CSV path.")
    batch_geo.add_argument("--address-field", default="address")
    batch_geo.add_argument("-o", "--output", required=True)
    batch_geo.add_argument("--force", action="store_true")

    # Transit commands.
    transit_p = sub.add_parser("transit", help="GoRaleigh transit feeds.")
    transit_sub = transit_p.add_subparsers(dest="transit_command")
    transit_routes = transit_sub.add_parser("routes", help="List routes.")
    transit_routes.add_argument("--limit", type=_positive_int, default=100)
    transit_stops = transit_sub.add_parser("stops", help="List stops.")
    transit_stops.add_argument("--near", type=_latlon_value, help="lat,lon")
    transit_stops.add_argument("--limit", type=_positive_int, default=20)
    transit_sched = transit_sub.add_parser("schedule", help="Schedule for a route.")
    transit_sched.add_argument("--route", required=True)
    transit_sched.add_argument("--date", type=_service_date, help="YYYYMMDD")
    transit_sched.add_argument("--limit", type=_positive_int, default=100)
    transit_arr = transit_sub.add_parser("arrivals", help="Arrivals for a stop.")
    transit_arr.add_argument("--stop", required=True)
    transit_arr.add_argument("--limit", type=_positive_int, default=100)
    transit_veh = transit_sub.add_parser("vehicles", help="Live vehicle positions.")
    transit_veh.add_argument("--route")
    transit_veh.add_argument("--limit", type=_positive_int, default=20)
    transit_alerts = transit_sub.add_parser("alerts", help="Service alerts.")
    transit_alerts.add_argument("--limit", type=_positive_int, default=20)
    transit_trips = transit_sub.add_parser("trip-updates", help="Live trip updates.")
    transit_trips.add_argument("--route")
    transit_trips.add_argument("--limit", type=_positive_int, default=20)
    transit_download = transit_sub.add_parser("download-gtfs", help="Download static GTFS.")
    transit_download.add_argument("-o", "--output", default="goraleigh_gtfs.zip")
    transit_download.add_argument("--force", action="store_true")

    # Development / EnerGov commands.
    dev = sub.add_parser("development", help="Permit and Development Portal (guest read-only).")
    dev_sub = dev.add_subparsers(dest="development_command")
    dev_search = dev_sub.add_parser("search", help="Search public records.")
    dev_search.add_argument("type", nargs="?", default=None, help="Record type")
    dev_search.add_argument("--type", dest="type_opt", default=None, help=argparse.SUPPRESS)
    dev_search.add_argument("--query")
    dev_search.add_argument("--limit", type=_positive_int, default=20)
    dev_permit = dev_sub.add_parser("permit", help="Show permit details.")
    dev_permit.add_argument("record")
    dev_insp = dev_sub.add_parser("inspections", help="Inspections for a record.")
    dev_insp.add_argument("--record", required=True)
    dev_insp.add_argument("--limit", type=_positive_int, default=10)
    dev_cc = dev_sub.add_parser("code-cases", help="Search code cases.")
    dev_cc.add_argument("--query")
    dev_cc.add_argument("--limit", type=_positive_int, default=20)
    dev_lic = dev_sub.add_parser("licenses", help="Search licenses.")
    dev_lic.add_argument("--query")
    dev_lic.add_argument("--limit", type=_positive_int, default=20)

    # Civic content commands (top-level aliases).
    news_p = sub.add_parser("news", help="RaleighNC.gov news.")
    news_p.add_argument("--limit", type=_positive_int, default=20)
    news_p.add_argument("--search")

    events_p = sub.add_parser("events", help="RaleighNC.gov events.")
    events_p.add_argument("--limit", type=_positive_int, default=20)
    events_p.add_argument("--from", dest="date_from", type=_iso_date)
    events_p.add_argument("--to", dest="date_to", type=_iso_date)
    events_p.add_argument("--search")

    projects_p = sub.add_parser("projects", help="RaleighNC.gov projects.")
    projects_p.add_argument("--limit", type=_positive_int, default=20)
    projects_p.add_argument("--search")

    places_p = sub.add_parser("places", help="RaleighNC.gov places.")
    places_p.add_argument("--limit", type=_positive_int, default=20)
    places_p.add_argument("--search")

    services_p = sub.add_parser("services", help="RaleighNC.gov services.")
    services_p.add_argument("--limit", type=_positive_int, default=20)
    services_p.add_argument("--search")

    directory_p = sub.add_parser("directory", help="RaleighNC.gov directory entries.")
    directory_p.add_argument("--limit", type=_positive_int, default=20)
    directory_p.add_argument("--search")

    alerts_p = sub.add_parser("alerts", help="RaleighNC.gov public alerts.")
    alerts_p.add_argument("--limit", type=_positive_int, default=20)

    for civic_parser in (
        news_p, events_p, projects_p, places_p, services_p, directory_p, alerts_p
    ):
        civic_parser.add_argument(
            "--relationship",
            metavar="FIELD=ID",
            help="Filter by a JSON:API relationship identifier.",
        )

    rss_p = sub.add_parser("rss", help="RaleighNC.gov RSS feed.")
    rss_p.add_argument("--limit", type=_positive_int, default=20)
    rss_p.add_argument(
        "--new-only", action="store_true", help="Return only entries not seen by prior --new-only runs."
    )

    # Meetings commands.
    mtg = sub.add_parser("meetings", help="eSCRIBE public meetings.")
    mtg_sub = mtg.add_subparsers(dest="meetings_command")
    mtg_upcoming = mtg_sub.add_parser("upcoming", help="Upcoming meetings.")
    mtg_upcoming.add_argument("--limit", type=_positive_int, default=20)
    mtg_list = mtg_sub.add_parser("list", help="List meetings.")
    mtg_list.add_argument("--body")
    mtg_list.add_argument("--year", type=_year)
    mtg_list.add_argument("--limit", type=_positive_int, default=20)
    mtg_search = mtg_sub.add_parser("search", help="Search upcoming or historical meetings.")
    mtg_search.add_argument("query")
    mtg_search.add_argument("--body")
    mtg_search.add_argument("--year", type=_year)
    mtg_search.add_argument("--limit", type=_positive_int, default=20)
    mtg_show = mtg_sub.add_parser("show", help="Show meeting details.")
    mtg_show.add_argument("id")
    mtg_agenda = mtg_sub.add_parser("download-agenda", help="Download agenda.")
    mtg_agenda.add_argument("id")
    mtg_agenda.add_argument("-o", "--output", required=True)
    mtg_agenda.add_argument("--force", action="store_true")
    mtg_minutes = mtg_sub.add_parser("download-minutes", help="Download minutes.")
    mtg_minutes.add_argument("id")
    mtg_minutes.add_argument("-o", "--output", required=True)
    mtg_minutes.add_argument("--force", action="store_true")

    # Catalog-check command for CI/validation.
    check_p = sub.add_parser("catalog-check", help="Validate catalog endpoints.")
    check_p.add_argument("--sample", type=_positive_int, default=10, help="Number of items to sample (default 10).")
    check_p.add_argument("--snapshot", action="store_true", help="Use the cached/live catalog snapshot instead of fetching fresh.")
    check_p.add_argument("--type", help="Check only datasets of this type (FeatureServer, MapServer, ImageServer).")
    check_p.add_argument("--full", action="store_true", help="Check every catalog item (overrides --sample).")

    # Police incident commands.
    police_p = sub.add_parser("police", help="Raleigh Police Department incident data.")
    police_sub = police_p.add_subparsers(dest="police_command")

    police_incidents = police_sub.add_parser("incidents", help="Query NIBRS incidents (June 2014–present).")
    police_incidents.add_argument("--since", type=_duration_value, help="Time range, e.g. 7d, 24h, 2w.")
    police_incidents.add_argument("--category", help="Filter by crime description substring.")
    police_incidents.add_argument("--district", help="Filter by police district substring.")
    police_incidents.add_argument("--limit", type=_positive_int, default=20)
    police_incidents.add_argument("--offset", type=_nonnegative_int, default=0)

    police_recent = police_sub.add_parser("recent", help="Query CrimeMapper past-90-day incidents.")
    police_recent.add_argument("--days", type=_positive_int, default=90, help="Days back (max 90).")
    police_recent.add_argument("--category", help="Filter by crime description substring.")
    police_recent.add_argument("--district", help="Filter by police district substring.")
    police_recent.add_argument("--limit", type=_positive_int, default=20)
    police_recent.add_argument("--offset", type=_nonnegative_int, default=0)

    police_prev = police_sub.add_parser("previous-day", help="Query previous-day incidents.")
    police_prev.add_argument("--category", help="Filter by crime description substring.")
    police_prev.add_argument("--district", help="Filter by police district substring.")
    police_prev.add_argument("--limit", type=_positive_int, default=20)
    police_prev.add_argument("--offset", type=_nonnegative_int, default=0)

    police_history = police_sub.add_parser("history", help="Query historical incidents (SRS or NIBRS).")
    police_history.add_argument("--reporting-system", choices=["srs", "nibrs"], default="nibrs",
                                help="Reporting system (default: nibrs).")
    police_history.add_argument("--since", type=_duration_value, help="Time range, e.g. 30d, 4w.")
    police_history.add_argument("--category", help="Filter by crime description substring.")
    police_history.add_argument("--district", help="Filter by police district substring.")
    police_history.add_argument("--limit", type=_positive_int, default=20)
    police_history.add_argument("--offset", type=_nonnegative_int, default=0)

    police_stats = police_sub.add_parser("stats", help="Official published crime statistics and availability.")
    police_stats.add_argument("--year", type=_year, help="Published year; omit to enumerate available years.")
    police_reports = police_sub.add_parser("reports", help="Official annual and quarterly crime report links.")
    police_reports.add_argument("--year", type=_year, help="Published year; omit to enumerate all reports.")
    police_reports.add_argument("--quarter", type=int, choices=range(1, 5), help="Quarter 1-4 (requires --year).")

    # Fire incident commands.
    fire_p = sub.add_parser("fire", help="Raleigh Fire Department incident data.")
    fire_sub = fire_p.add_subparsers(dest="fire_command")

    fire_incidents = fire_sub.add_parser("incidents", help="Query RFD incidents (full history 2007–present or past month).")
    fire_incidents.add_argument("--source", choices=["full-history", "past-month"], default="full-history",
                                help="Dataset to query (default: full-history).")
    fire_incidents.add_argument("--since", type=_duration_value, help="Time range, e.g. 30d, 24h, 2w, 1y.")
    fire_incidents.add_argument("--station", type=_positive_int, help="Filter by station number.")
    fire_incidents.add_argument("--platoon", help="Filter by platoon (e.g. A, B, C).")
    fire_incidents.add_argument("--group", help="Filter by incident group substring (populated only for 2026+ records).")
    fire_incidents.add_argument("--type", dest="incident_type", help="Filter by incident type name/description substring or legacy NFIRS code.")
    fire_incidents.add_argument("--limit", type=_positive_int, default=20)
    fire_incidents.add_argument("--offset", type=_nonnegative_int, default=0)

    fire_rt = fire_sub.add_parser("response-times", help="Compute response durations from dispatch/arrival/cleared timestamps.")
    fire_rt.add_argument("--source", choices=["full-history", "past-month"], default="full-history",
                         help="Dataset to query (default: full-history).")
    fire_rt.add_argument("--since", type=_duration_value, help="Time range, e.g. 30d, 24h, 2w, 1y.")
    fire_rt.add_argument("--station", type=_positive_int, help="Filter by station number.")
    fire_rt.add_argument("--platoon", help="Filter by platoon (e.g. A, B, C).")
    fire_rt.add_argument("--group", help="Filter by incident group substring (populated only for 2026+ records).")
    fire_rt.add_argument("--type", dest="incident_type", help="Filter by incident type name/description substring or legacy NFIRS code.")
    fire_rt.add_argument("--limit", type=_positive_int, default=20)
    fire_rt.add_argument("--offset", type=_nonnegative_int, default=0)

    fire_prot = fire_sub.add_parser("protection", help="Wake County MAR fire-protection proximity lookup.")
    fire_prot_group = fire_prot.add_mutually_exclusive_group(required=True)
    fire_prot_group.add_argument("--address", help="Address to geocode and resolve to a CSAID.")
    fire_prot_group.add_argument("--csaid", type=_csaid_value, help="Canonical site-address identifier (CSAID).")

    fire_stats = fire_sub.add_parser("stats", help="Official published incident and sprinkler-save statistics.")
    fire_stats.add_argument("--year", type=_year, help="Published year; omit to enumerate available years.")

    fire_reports = fire_sub.add_parser("reports", help="Published aggregate reports or exact incident-report lookup.")
    fire_reports_group = fire_reports.add_mutually_exclusive_group()
    fire_reports_group.add_argument("--date", type=_iso_date, help="Exact dispatch date (YYYY-MM-DD).")
    fire_reports_group.add_argument("--incident-number", type=_nonempty_text, help="Exact incident number.")
    fire_reports.add_argument("--year", type=_year, help="Published aggregate-report year; omit to enumerate all reports.")
    fire_reports.add_argument("--quarter", type=int, choices=range(1, 5), help="Published quarter 1-4 (requires --year).")
    fire_reports.add_argument("--allow-rfd-fallback", action="store_true", help="Use the RFD date form only when ArcGIS returns no records.")
    fire_reports.add_argument("--include-narrative", action="store_true", help="Fetch one exact incident narrative (requires --incident-number).")
    fire_reports.add_argument("--acknowledge-insecure-rfd", action="store_true", help="Acknowledge that RFD data crosses unencrypted HTTP for this invocation.")

    fire_inspections = fire_sub.add_parser("inspections", help="Search fragile RFD inspection records over acknowledged HTTP.")
    fire_inspections_group = fire_inspections.add_mutually_exclusive_group(required=True)
    fire_inspections_group.add_argument("--business", type=_nonempty_text, help="Nonempty business-name search.")
    fire_inspections_group.add_argument("--address", type=_nonempty_text, help="Nonempty address search.")
    fire_inspections.add_argument("--acknowledge-insecure-rfd", action="store_true", help="Acknowledge that the search crosses unencrypted HTTP for this invocation.")

    incidents_p = sub.add_parser("incidents", help="Raleigh-Wake ECC active incident feed (undocumented).")
    incidents_sub = incidents_p.add_subparsers(dest="incidents_command")
    incidents_active = incidents_sub.add_parser("active", help="Currently active incidents.")
    incidents_active.add_argument("--agency", help="Filter by agency substring (e.g. raleigh-fire, raleigh-police).")
    incidents_active.add_argument("--type", dest="incident_type", help="Filter by incident type substring.")
    incidents_active.add_argument("--limit", type=_positive_int, default=50)
    incidents_active.add_argument("--no-cache", action="store_true", help="Bypass the short-lived cache.")

    return parser


def _resolve_imagery_url(service: str) -> str:
    if service.startswith("http"):
        return service
    encoded_service = urllib.parse.quote(service.strip("/"), safe="/")
    return f"{imagery.IMAGE_ROOT}/{encoded_service}/ImageServer"


def _parse_bbox(value: str) -> tuple[float, float, float, float]:
    try:
        parts = [float(x.strip()) for x in value.split(",")]
    except ValueError as exc:
        raise argparse.ArgumentTypeError("bbox must be xmin,ymin,xmax,ymax") from exc
    if len(parts) != 4:
        raise argparse.ArgumentTypeError("bbox must be xmin,ymin,xmax,ymax")
    if not all(math.isfinite(value) for value in parts):
        raise argparse.ArgumentTypeError("bbox coordinates must be finite")
    if parts[0] >= parts[2] or parts[1] >= parts[3]:
        raise argparse.ArgumentTypeError("bbox minimums must be less than maximums")
    return tuple(parts)  # type: ignore[return-value]


def _parse_point(value: str) -> tuple[float, float]:
    try:
        parts = [float(x.strip()) for x in value.split(",")]
    except ValueError as exc:
        raise argparse.ArgumentTypeError("point must be lon,lat") from exc
    if len(parts) != 2:
        raise argparse.ArgumentTypeError("point must be lon,lat")
    _longitude(str(parts[0]))
    _latitude(str(parts[1]))
    return tuple(parts)  # type: ignore[return-value]


def _parse_latlon(value: str) -> tuple[float, float]:
    parts = value.split(",")
    if len(parts) != 2:
        raise argparse.ArgumentTypeError("coordinate must be lat,lon")
    return _latitude(parts[0].strip()), _longitude(parts[1].strip())


def _parse_size(value: str) -> tuple[int, int]:
    try:
        parts = [int(item.strip()) for item in value.split(",")]
    except ValueError as exc:
        raise ValueError("size must be width,height using positive integers") from exc
    if len(parts) != 2 or any(part <= 0 for part in parts):
        raise ValueError("size must be width,height using positive integers")
    return parts[0], parts[1]


def _ensure_catalog(args: argparse.Namespace) -> list[dict[str, Any]]:
    if args.refresh:
        return hub.fetch_catalog()
    return hub.catalog_from_cache_or_live()


def cmd_catalog(args: argparse.Namespace) -> int:
    catalog = _ensure_catalog(args)
    if args.category:
        catalog = [i for i in catalog if args.category.lower() in i.get("category", "").lower()]
    if args.search:
        catalog = [i for i in catalog if args.search.lower() in " ".join([i.get("title", ""), i.get("description", ""), " ".join(i.get("tags", []))]).lower()]
    limit = args.limit
    if args.json:
        _output_json([{"id": i["id"], "title": i["title"], "type": i["type"], "url": i.get("url")} for i in catalog[:limit]])
    else:
        _output_table(["TYPE", "TITLE"], [[i["type"], i["title"]] for i in catalog[:limit]])
        if len(catalog) > limit:
            print(
                f"... and {len(catalog) - limit} more "
                f"(use --limit {len(catalog)} --json for the full list)"
            )
    return 0


def cmd_search(args: argparse.Namespace) -> int:
    catalog = _ensure_catalog(args)
    results = hub.search_catalog(args.query, catalog=catalog, limit=args.limit)
    if args.json:
        _output_json(results)
    else:
        _output_table(["TYPE", "TITLE", "CATEGORY"], [[r["type"], r["title"], r["category"]] for r in results])
    return 0


def cmd_info(args: argparse.Namespace) -> int:
    catalog = _ensure_catalog(args)
    item = hub.resolve_item(args.dataset, catalog=catalog)
    url = item.get("url", "")
    resolved_url: str | None = None
    fields: list[dict[str, Any]] = []
    sample: dict[str, Any] = {"type": "FeatureCollection", "features": []}
    sample_error: str | None = None
    if url and item.get("type") in ("FeatureServer", "MapServer"):
        try:
            resolved_url = arcgis.resolve_queryable_layer(url)
            fields = arcgis.layer_fields(resolved_url)
            sample_records = arcgis.sample_features(resolved_url, limit=3)
            sample = arcgis.geojson_from_records(sample_records)
        except Exception as exc:
            sample_error = str(exc)
    output = {
        "dataset": item,
        "fields": fields,
        "sample": sample,
    }
    if sample_error:
        output["sample_error"] = sample_error
    if args.json:
        _output_json(output)
        return 1 if sample_error else 0
    else:
        print(f"Title: {item['title']}")
        print(f"Type: {item['type']}")
        print(f"Category: {item['category']}")
        print(f"URL: {item['url']}")
        if resolved_url:
            print(f"Resolved layer: {resolved_url}")
        print(f"Fields: {len(fields)}")
        print(f"Sample features: {len(sample['features'])}")
        if sample_error:
            print(f"Sample error: {sample_error}", file=sys.stderr)
            return 1
        print(f"Tags: {', '.join(item['tags'])}")
    return 0


def _query_dataset(
    args: argparse.Namespace,
    max_records: int | None = None,
    offset: int = 0,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    catalog = _ensure_catalog(args)
    item = hub.resolve_item(args.dataset, catalog=catalog)
    url = item.get("url")
    if not url:
        raise cli_error("Dataset has no queryable URL.")
    resolved = arcgis.resolve_queryable_layer(url)
    item["resolved_url"] = resolved
    has_geometry = arcgis.layer_has_geometry(resolved)
    item["has_geometry"] = has_geometry
    return_geometry = not args.no_geometry and has_geometry
    records = arcgis.query_all_pages(
        resolved,
        where=args.where,
        out_fields=args.out_fields,
        return_geometry=return_geometry,
        max_records=max_records,
        offset=offset,
        order_by_fields=getattr(args, "order_by", None),
    )
    return records, item


def cmd_query(args: argparse.Namespace) -> int:
    records, _ = _query_dataset(args, max_records=args.limit, offset=args.offset)
    if args.json:
        _output_json(arcgis.geojson_from_records(records))
    elif args.csv:
        sys.stdout.write(arcgis.csv_from_records(records))
    elif records:
        headers = list(records[0].get("attributes", {}).keys())[:8]
        rows = [[str(r.get("attributes", {}).get(h, "")) for h in headers] for r in records]
        _output_table(headers, rows)
    return 0


def cmd_download(args: argparse.Namespace) -> int:
    max_records = None if args.all else args.limit
    records, _ = _query_dataset(args, max_records=max_records)
    fmt = args.format
    if fmt == "csv":
        output = arcgis.csv_from_records(records)
    elif fmt == "geojson" or fmt == "json":
        output = json.dumps(arcgis.geojson_from_records(records), indent=2)
    core.safe_write(Path(args.output), output, force=getattr(args, "force", False))
    if not args.json:
        print(f"Wrote {len(records)} records to {args.output}")
    return 0


def cmd_categories(args: argparse.Namespace) -> int:
    catalog = _ensure_catalog(args)
    counts: dict[str, int] = {}
    for item in catalog:
        counts[item.get("category", "Other")] = counts.get(item.get("category", "Other"), 0) + 1
    categories = sorted(counts.items(), key=lambda x: x[0])
    if args.json:
        _output_json([{"category": c, "count": n} for c, n in categories])
    else:
        _output_table(["CATEGORY", "COUNT"], [[c, str(n)] for c, n in categories])
    return 0


def cmd_imagery_catalog(args: argparse.Namespace) -> int:
    services, restricted_folders = imagery.list_services()
    limit = args.limit
    if args.json:
        _output_json(services[:limit])
        if restricted_folders:
            print(
                f"note: {len(restricted_folders)} imagery folder(s) require a token "
                f"and were skipped: {', '.join(restricted_folders)}",
                file=sys.stderr,
            )
    else:
        _output_table(["NAME", "TYPE"], [[s.get("name", ""), s.get("type", "")] for s in services[:limit]])
        if restricted_folders:
            print(
                f"Note: {len(restricted_folders)} imagery folder(s) require a token "
                f"and were skipped: {', '.join(restricted_folders)}"
            )
    return 0


def cmd_imagery_info(args: argparse.Namespace) -> int:
    url = _resolve_imagery_url(args.service)
    info = imagery.service_info(url)
    if args.json:
        _output_json(info)
    else:
        print(f"Service: {info.get('serviceDescription') or args.service}")
        print(f"Capabilities: {info.get('capabilities')}")
        print(f"Band count: {info.get('bandCount')}")
    return 0


def cmd_imagery_export(args: argparse.Namespace) -> int:
    url = _resolve_imagery_url(args.service)
    bbox = _parse_bbox(args.bbox)
    size = _parse_size(args.size) if args.size else None
    data = imagery.export_image(url, bbox=bbox, size=size, format_=args.format_, in_sr=args.in_sr, out_sr=args.out_sr)
    core.safe_write(Path(args.output), data, force=args.force)
    if not args.json:
        print(f"Wrote {len(data)} bytes to {args.output}")
    return 0


def cmd_imagery_identify(args: argparse.Namespace) -> int:
    url = _resolve_imagery_url(args.service)
    point = _parse_point(args.point)
    result = imagery.identify(url, point=point)
    if args.json:
        _output_json(result)
    else:
        print(json.dumps(result, indent=2))
    return 0


def cmd_imagery_statistics(args: argparse.Namespace) -> int:
    url = _resolve_imagery_url(args.service)
    bbox = _parse_bbox(args.bbox)
    result = imagery.compute_statistics(url, bbox=bbox)
    if args.json:
        _output_json(result)
    else:
        print(json.dumps(result, indent=2))
    return 0


def cmd_geocode(args: argparse.Namespace) -> int:
    candidates = geocode.find_address_candidates(
        args.address,
        max_locations=args.max_locations,
        min_score=args.min_score,
    )
    if args.json:
        _output_json(candidates)
    else:
        if not candidates:
            print("No match.")
            return 0
        _output_table(
            ["SCORE", "ADDRESS", "LON", "LAT"],
            [
                [str(c.get("score", "")), c.get("address", ""), str(c.get("location", {}).get("x", "")), str(c.get("location", {}).get("y", ""))]
                for c in candidates
            ],
        )
    return 0


def cmd_reverse_geocode(args: argparse.Namespace) -> int:
    result = geocode.reverse_geocode(args.lat, args.lon)
    if args.json:
        _output_json(result)
    else:
        print(json.dumps(result, indent=2))
    return 0


def cmd_suggest(args: argparse.Namespace) -> int:
    suggestions = geocode.suggest(args.text, max_suggestions=args.max_suggestions)
    if args.json:
        _output_json(suggestions)
    else:
        _output_table(["SUGGESTION"], [[s.get("text", "")] for s in suggestions])
    return 0


def cmd_geocode_batch(args: argparse.Namespace) -> int:
    records: list[dict[str, Any]] = []
    with open(args.input, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        source_fields = list(reader.fieldnames or [])
        for row in reader:
            source = dict(row)
            source.setdefault("SingleLine", row.get(args.address_field, ""))
            records.append(source)
    results = geocode.geocode_addresses(records)
    result_fields = ["input_id", "match_address", "score", "lat", "lon", "status"]
    result_columns = {
        field: field if field not in source_fields else f"geocode_{field}"
        for field in result_fields
    }
    output_rows = []
    for result in results:
        source = dict(result.get("source") or {})
        source.update({
            result_columns["input_id"]: result.get("input_id"),
            result_columns["match_address"]: result.get("address"),
            result_columns["score"]: result.get("score"),
            result_columns["lat"]: result.get("lat"),
            result_columns["lon"]: result.get("lon"),
            result_columns["status"]: result.get("status"),
        })
        output_rows.append({key: arcgis.csv_safe_value(value) for key, value in source.items()})
    output = io.StringIO()
    fieldnames = source_fields + list(result_columns.values())
    writer = csv.writer(output)
    writer.writerow([arcgis.csv_safe_value(field) for field in fieldnames])
    for row in output_rows:
        writer.writerow([row.get(field, "") for field in fieldnames])
    core.safe_write(Path(args.output), output.getvalue(), force=args.force)
    if not args.json:
        print(f"Geocoded {len(results)} addresses to {args.output}")
    return 0


def cmd_transit_routes(args: argparse.Namespace) -> int:
    routes = transit.get_routes()
    routes = routes[: args.limit]
    if args.json:
        _output_json(routes)
    else:
        _output_table(
            ["SHORT", "LONG NAME"],
            [[r.get("route_short_name", ""), r.get("route_long_name", "")] for r in routes],
        )
    return 0


def cmd_transit_stops(args: argparse.Namespace) -> int:
    stops = transit.get_stops()
    if args.near:
        lat, lon = _parse_latlon(args.near)
        # Simple Euclidean sort for nearest; sufficient for CLI filtering.
        stops = sorted(
            stops,
            key=lambda s: ((float(s.get("stop_lat", 0)) - lat) ** 2 + (float(s.get("stop_lon", 0)) - lon) ** 2),
        )
    stops = stops[: args.limit]
    if args.json:
        _output_json(stops)
    else:
        _output_table(
            ["STOP", "NAME", "LAT", "LON"],
            [
                [s.get("stop_id", ""), s.get("stop_name", ""), s.get("stop_lat", ""), s.get("stop_lon", "")]
                for s in stops
            ],
        )
    return 0


def cmd_transit_schedule(args: argparse.Namespace) -> int:
    schedule = transit.get_schedule_for_route(args.route, target_date=args.date)
    schedule = schedule[: args.limit]
    if args.json:
        _output_json(schedule)
    else:
        _output_table(
            ["TRIP", "STOP", "ARRIVAL"],
            [[s["trip_id"], s["stop_id"], s["arrival_time"]] for s in schedule],
        )
    return 0


def cmd_transit_arrivals(args: argparse.Namespace) -> int:
    arrivals = transit.get_arrivals_for_stop(args.stop)
    arrivals = arrivals[: args.limit]
    if args.json:
        _output_json(arrivals)
    else:
        _output_table(
            ["TRIP", "ARRIVAL"],
            [[a["trip_id"], a["arrival_time"]] for a in arrivals],
        )
    return 0


def cmd_transit_vehicles(args: argparse.Namespace) -> int:
    entities = transit.get_vehicle_positions(route=args.route, limit=args.limit)
    if args.json:
        _output_json(entities)
    else:
        print(json.dumps(entities, indent=2))
    return 0


def cmd_transit_alerts(args: argparse.Namespace) -> int:
    entities = transit.get_alerts(limit=args.limit)
    if args.json:
        _output_json(entities)
    else:
        print(json.dumps(entities, indent=2))
    return 0


def cmd_transit_trip_updates(args: argparse.Namespace) -> int:
    entities = transit.get_trip_updates(route=args.route, limit=args.limit)
    if args.json:
        _output_json(entities)
    else:
        print(json.dumps(entities, indent=2))
    return 0


def cmd_transit_download_gtfs(args: argparse.Namespace) -> int:
    data = transit.download_gtfs()
    transit.parse_gtfs_zip(data)
    core.safe_write(Path(args.output), data, force=args.force)
    if not args.json:
        print(f"Downloaded {len(data)} bytes to {args.output}")
    return 0


def cmd_development_search(args: argparse.Namespace) -> int:
    record_type = args.type or args.type_opt
    if not record_type:
        print("Error: record type is required (e.g. permit, plan, inspection)", file=sys.stderr)
        return 2
    type_map = {
        "permit": "permit",
        "permits": "permit",
        "plan": "plan",
        "plans": "plan",
        "inspection": "inspection",
        "inspections": "inspection",
        "code-case": "code-case",
        "code-cases": "code-case",
        "request": "request",
        "requests": "request",
        "license": "license",
        "licenses": "license",
        "project": "project",
        "projects": "project",
    }
    if record_type not in type_map:
        print(f"Error: invalid record type '{record_type}'", file=sys.stderr)
        return 2
    result = development.public_search(type_map[record_type], query=args.query, limit=args.limit)
    if args.json:
        _output_json(result)
    else:
        print(json.dumps(result, indent=2))
    return 0


def cmd_development_permit(args: argparse.Namespace) -> int:
    try:
        result = development.permit_detail(args.record)
    except development.UnsupportedEndpointError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1
    if args.json:
        _output_json(result)
    else:
        print(json.dumps(result, indent=2))
    return 0


def cmd_development_inspections(args: argparse.Namespace) -> int:
    try:
        results = development.inspections_for_record(args.record, limit=args.limit)
    except development.UnsupportedEndpointError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1
    if args.json:
        _output_json(results)
    else:
        print(json.dumps(results, indent=2))
    return 0


def cmd_development_code_cases(args: argparse.Namespace) -> int:
    results = development.code_cases(query=args.query, limit=args.limit)
    if args.json:
        _output_json(results)
    else:
        print(json.dumps(results, indent=2))
    return 0


def cmd_development_licenses(args: argparse.Namespace) -> int:
    results = development.licenses(query=args.query, limit=args.limit)
    if args.json:
        _output_json(results)
    else:
        print(json.dumps(results, indent=2))
    return 0


def cmd_news(args: argparse.Namespace) -> int:
    results = civic.fetch_news(
        limit=args.limit, search=args.search, relationship=args.relationship
    )
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_events(args: argparse.Namespace) -> int:
    if args.date_from and args.date_to and args.date_from > args.date_to:
        raise cli_error("--from must not be later than --to")
    results = civic.fetch_events(
        limit=args.limit,
        date_from=args.date_from,
        date_to=args.date_to,
        search=args.search,
        relationship=args.relationship,
    )
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_projects(args: argparse.Namespace) -> int:
    results = civic.fetch_projects(
        limit=args.limit, search=args.search, relationship=args.relationship
    )
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_places(args: argparse.Namespace) -> int:
    results = civic.fetch_places(
        limit=args.limit, search=args.search, relationship=args.relationship
    )
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_services(args: argparse.Namespace) -> int:
    results = civic.fetch_services(
        limit=args.limit, search=args.search, relationship=args.relationship
    )
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_directory(args: argparse.Namespace) -> int:
    results = civic.fetch_directory(
        limit=args.limit, search=args.search, relationship=args.relationship
    )
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_alerts(args: argparse.Namespace) -> int:
    results = civic.fetch_alerts(limit=args.limit, relationship=args.relationship)
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_rss(args: argparse.Namespace) -> int:
    results = civic.fetch_rss(limit=args.limit, new_only=args.new_only)
    if args.json:
        _output_json(results)
    else:
        _output_table(["TITLE", "URL"], [[r["title"], r["url"]] for r in results])
    return 0


def cmd_meetings_upcoming(args: argparse.Namespace) -> int:
    results = meetings.list_upcoming(limit=args.limit)
    if args.json:
        _output_json(results)
    else:
        _output_table(["DATE", "BODY", "TITLE"], [[r["date"], r["body"], r["title"]] for r in results])
    return 0


def cmd_meetings_list(args: argparse.Namespace) -> int:
    results = meetings.list_meetings(body=args.body, year=args.year, limit=args.limit)
    if args.json:
        _output_json(results)
    else:
        _output_table(["DATE", "BODY", "TITLE"], [[r["date"], r["body"], r["title"]] for r in results])
    return 0


def cmd_meetings_search(args: argparse.Namespace) -> int:
    results = meetings.search_meetings(
        args.query,
        body=args.body,
        year=args.year,
        limit=args.limit,
    )
    if args.json:
        _output_json(results)
    else:
        _output_table(
            ["ID", "Title", "Date", "Body"],
            [[str(r.get("id", "")), str(r.get("title", "")), str(r.get("date", "")), str(r.get("body", ""))] for r in results],
        )
    return 0


def cmd_meetings_show(args: argparse.Namespace) -> int:
    result = meetings.meeting_detail(args.id)
    if args.json:
        _output_json(result)
    else:
        print(json.dumps(result, indent=2))
    return 0


def cmd_meetings_download_agenda(args: argparse.Namespace) -> int:
    detail = meetings.meeting_detail(args.id)
    url = detail.get("agenda")
    if not url:
        print("No agenda available.", file=sys.stderr)
        return 1
    meetings.download_document(url, args.output, force=args.force)
    if not args.json:
        print(f"Downloaded agenda to {args.output}")
    return 0


def cmd_meetings_download_minutes(args: argparse.Namespace) -> int:
    detail = meetings.meeting_detail(args.id)
    url = detail.get("minutes")
    if not url:
        print("No minutes available.", file=sys.stderr)
        return 1
    meetings.download_document(url, args.output, force=args.force)
    if not args.json:
        print(f"Downloaded minutes to {args.output}")
    return 0


def cmd_catalog_check(args: argparse.Namespace) -> int:
    if args.snapshot:
        catalog = hub.catalog_from_cache_or_live()
    else:
        catalog = hub.fetch_catalog()
    if args.type:
        catalog = [i for i in catalog if i.get("type") == args.type]
    # Only ArcGIS service records have a metadata contract this command can
    # validate. Documents, web applications, and external links remain visible
    # in catalog output but are intentionally not dereferenced here.
    supported_types = {"FeatureServer", "MapServer", "ImageServer"}
    catalog = [i for i in catalog if i.get("type") in supported_types]
    if args.full:
        sample = catalog
    else:
        sample = catalog[: args.sample]
    failures: list[dict[str, Any]] = []
    for item in sample:
        url = item.get("url", "")
        if not url:
            continue
        try:
            if item.get("type") == "ImageServer":
                meta = imagery.service_info(url)
            else:
                meta = arcgis.service_metadata(url)
            if "error" in meta:
                failures.append({"id": item["id"], "title": item["title"], "error": meta["error"]})
        except Exception as exc:
            failures.append({"id": item["id"], "title": item["title"], "error": str(exc)})
    if args.json:
        _output_json({"checked": len(sample), "failures": failures})
    else:
        print(f"Checked {len(sample)} endpoints; {len(failures)} failures")
        for f in failures:
            print(f"  {f['title']}: {f['error']}")
    return 0 if not failures else 1


def _duration_value(value: str) -> str:
    """Validate a duration string like '7d', '24h', '2w', '1y'."""
    if not re.fullmatch(r"\d+[dhwy]", value.strip()):
        raise argparse.ArgumentTypeError("duration must be a positive integer followed by d (days), h (hours), w (weeks), or y (years)")
    amount = int(value.strip()[:-1])
    if amount < 1:
        raise argparse.ArgumentTypeError("duration must be positive")
    return value.strip()


def _duration_to_ms(value: str) -> int:
    """Convert a validated duration string to a Unix-milliseconds timestamp in the past."""
    amount = int(value[:-1])
    unit = value[-1]
    multiplier = {"h": 3600, "d": 86400, "w": 604800, "y": 31536000}
    seconds_ago = amount * multiplier[unit]
    now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
    return now_ms - (seconds_ago * 1000)


def _police_output(result: dict[str, Any], args: argparse.Namespace) -> int:
    """Shared output logic for police commands."""
    if args.json:
        _output_json(result)
    else:
        features = result.get("features", [])
        if features:
            rows = []
            for f in features:
                props = f.get("properties", {})
                rows.append([
                    str(props.get("_source", "")),
                    str(props.get("crime_description") or props.get("LCR_DESC") or ""),
                    str(props.get("district") or props.get("DISTRICT") or ""),
                    str(props.get("reported_block_address") or ""),
                    str(props.get("_location_status", "")),
                ])
            _output_table(["SOURCE", "DESCRIPTION", "DISTRICT", "BLOCK ADDRESS", "LOC STATUS"], rows)
        else:
            print("No records returned.")
        print(police.LOCATION_CAVERAT, file=sys.stderr)
    return 0


def cmd_police_incidents(args: argparse.Namespace) -> int:
    since_ms = _duration_to_ms(args.since) if args.since else None
    if since_ms is not None and since_ms < police.NIBRS_EPOCH_MS:
        print(
            "Warning: --since predates NIBRS availability (June 2014); "
            "results will only include June 2014 onward. Use 'police history --reporting-system srs' for older data.",
            file=sys.stderr,
        )
    result = police.query_incidents(
        "nibrs", since_ms=since_ms, category=args.category, district=args.district,
        limit=args.limit, offset=args.offset,
    )
    return _police_output(result, args)


def cmd_police_recent(args: argparse.Namespace) -> int:
    if args.days > 90:
        raise cli_error("--days cannot exceed 90 (the CrimeMapper feed covers at most 90 days)")
    since_ms = int(datetime.now(timezone.utc).timestamp() * 1000) - (args.days * 86400 * 1000)
    result = police.query_incidents(
        "crimemapper-90d", since_ms=since_ms, category=args.category, district=args.district,
        limit=args.limit, offset=args.offset,
    )
    return _police_output(result, args)


def cmd_police_previous_day(args: argparse.Namespace) -> int:
    result = police.query_incidents(
        "previous-day", category=args.category, district=args.district,
        limit=args.limit, offset=args.offset,
    )
    return _police_output(result, args)


def cmd_police_history(args: argparse.Namespace) -> int:
    source_key = args.reporting_system
    if source_key == "nibrs":
        print("Note: defaulting to NIBRS; use --reporting-system srs for pre-2014 data.", file=sys.stderr)
    since_ms = _duration_to_ms(args.since) if args.since else None
    result = police.query_incidents(
        source_key, since_ms=since_ms, category=args.category, district=args.district,
        limit=args.limit, offset=args.offset,
    )
    return _police_output(result, args)


def _published_output(result: dict[str, Any], args: argparse.Namespace) -> int:
    if args.json:
        _output_json(result)
        return 0
    datasets = result.get("datasets", [])
    rows = []
    for dataset in datasets:
        for item in dataset.get("values", []):
            rows.append([
                str(dataset.get("year") or ""),
                str(dataset.get("kind") or ""),
                str(item.get("label") or ""),
                str(item.get("published_value") or item.get("value") or ""),
            ])
    if rows:
        _output_table(["YEAR", "KIND", "LABEL", "PUBLISHED VALUE"], rows)
    reports = result.get("reports", [])
    report_rows = [[
        str(item.get("year") or ""),
        f"Q{item['quarter']}" if item.get("quarter") else str(item.get("period") or ""),
        str(item.get("label") or ""),
        str(item.get("document_url") or ""),
    ] for item in reports]
    if report_rows:
        if rows:
            print()
        _output_table(["YEAR", "PERIOD", "LABEL", "DOCUMENT"], report_rows)
    if not rows and not report_rows:
        print("No published statistics returned.")
    print("Available years: " + ", ".join(str(year) for year in result.get("available_years", [])))
    for warning in result.get("warnings", []):
        print(f"Warning: {warning}", file=sys.stderr)
    return 0


def cmd_police_stats(args: argparse.Namespace) -> int:
    return _published_output(public_safety_stats.statistics("police", args.year), args)


def cmd_police_reports(args: argparse.Namespace) -> int:
    return _published_output(public_safety_stats.reports("police", args.year, args.quarter), args)


def _format_ms_datetime(value: Any) -> str:
    """Format a Unix-milliseconds timestamp for table output, or '' if unusable."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return ""
    try:
        return datetime.fromtimestamp(value / 1000, timezone.utc).strftime("%Y-%m-%d %H:%MZ")
    except (OverflowError, OSError, ValueError):
        return ""


def _fire_query_args(args: argparse.Namespace) -> tuple[int | None, dict[str, Any]]:
    """Shared filter extraction for fire commands, with source-specific notes."""
    since_ms = _duration_to_ms(args.since) if args.since else None
    if args.group and since_ms is not None and since_ms < fire.SCHEMA_TRANSITION_EPOCH_MS:
        print(
            "Note: incident_group_name is populated only for 2026+ records; "
            "older records will be excluded by the group filter.",
            file=sys.stderr,
        )
    if args.station and args.source == "full-history":
        print(
            "Note: station is unpopulated for most full-history records after "
            "early 2021; use --source past-month for current station data.",
            file=sys.stderr,
        )
    filters = {
        "station": args.station,
        "platoon": args.platoon,
        "group": args.group,
        "incident_type": args.incident_type,
        "limit": args.limit,
        "offset": args.offset,
    }
    return since_ms, filters


def _fire_output(result: dict[str, Any], args: argparse.Namespace) -> int:
    """Shared output logic for fire incidents."""
    if args.json:
        _output_json(result)
    else:
        features = result.get("features", [])
        if features:
            rows = []
            for f in features:
                props = f.get("properties", {})
                station = props.get("_station")
                rows.append([
                    str(props.get("incident_number") or ""),
                    str(props.get("_classification_era") or ""),
                    str(props.get("_incident_group") or ""),
                    str(props.get("_incident_type") or ""),
                    str(station if station is not None else ""),
                    str(props.get("platoon") or ""),
                    _format_ms_datetime(props.get("dispatch_date_time")),
                ])
            _output_table(["INCIDENT", "ERA", "GROUP", "TYPE", "STATION", "PLATOON", "DISPATCHED"], rows)
        else:
            print("No records returned.")
        print(fire.PRIVACY_CAVERAT, file=sys.stderr)
    return 0


def _fire_duration_cell(props: dict[str, Any], key: str, status_key: str) -> str:
    """Format a duration value for table output, blank unless the pair validated."""
    if props.get(status_key) != "ok":
        return ""
    value = props.get(key)
    return f"{value:.0f}" if isinstance(value, (int, float)) else ""


def _fire_response_times_output(result: dict[str, Any], args: argparse.Namespace) -> int:
    """Shared output logic for fire response-times."""
    features = result.get("features", [])
    valid = 0
    rejected = 0
    for f in features:
        status = f.get("properties", {}).get("_dispatch_to_arrive_status")
        if status == "ok":
            valid += 1
        elif status is not None:
            rejected += 1

    if args.json:
        _output_json(result)
    else:
        if features:
            rows = []
            for f in features:
                props = f.get("properties", {})
                rows.append([
                    str(props.get("incident_number") or ""),
                    str(props.get("_incident_group") or ""),
                    _fire_duration_cell(props, "_dispatch_to_arrive_seconds", "_dispatch_to_arrive_status"),
                    _fire_duration_cell(props, "_arrive_to_clear_seconds", "_arrive_to_clear_status"),
                    _fire_duration_cell(props, "_dispatch_to_clear_seconds", "_dispatch_to_clear_status"),
                    str(props.get("_dispatch_to_arrive_status") or ""),
                ])
            _output_table(
                ["INCIDENT", "GROUP", "DISPATCH->ARRIVE (S)", "ARRIVE->CLEAR (S)", "DISPATCH->CLEAR (S)", "STATUS"],
                rows,
            )
        else:
            print("No records returned.")
        print(
            f"{len(features)} records: {valid} with a valid dispatch-to-arrival "
            f"duration, {rejected} rejected (missing, malformed, or reversed timestamps).",
            file=sys.stderr,
        )
        print(fire.PRIVACY_CAVERAT, 

"""Shared HTTP, cache, and configuration utilities."""

from __future__ import annotations

import json
import os
import stat
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any, Callable


DEFAULT_TIMEOUT = 30
ENV_TIMEOUT = "RALEIGH_TIMEOUT"

# Fixed service hosts that the CLI may dereference. Any URL whose host is not in
# this allowlist is rejected before a network request is made. Hosts are stored
# without scheme/port and matched case-insensitively.
ALLOWED_HOSTS: frozenset[str] = frozenset({
    "data.raleighnc.gov",
    "ral.maps.arcgis.com",
    "services.arcgis.com",
    "maps.raleighnc.gov",
    "maps.wake.gov",
    "maps.wakegov.com",
    "services1.arcgis.com",
    "services3.arcgis.com",
    "utility.arcgis.com",
    "raleighnc-energovpub.tylerhost.net",
    "goraleigh.org",
    "www.goraleighlive.org",
    "www.goraleigh.org",
    "raleighnc.gov",
    "cityofraleigh0drupal.blob.core.usgovcloudapi.net",
    "pub-raleighnc.escribemeetings.com",
    "incidents.rwecc.com",
})


USER_AGENT = (
    "RaleighCivicDataCLI/2.0 (read-only public data; Python urllib)"
)

# Maximum number of HTTP redirects the CLI will follow automatically.
MAX_REDIRECTS = 5

# Default response size caps, in bytes.
DEFAULT_MAX_JSON_BYTES = 5 * 1024 * 1024
DEFAULT_MAX_RAW_BYTES = 10 * 1024 * 1024


class SecurityError(Exception):
    """Raised when a requested URL is outside the fixed allowlist."""


class RequestPolicyError(SecurityError):
    """Raised when the HTTP method or path is not allowed."""


class ResponseTooLargeError(Exception):
    """Raised when a response exceeds the endpoint-appropriate size cap."""


def _get_timeout() -> int:
    """Return the active timeout, preferring the environment override."""
    env = os.environ.get(ENV_TIMEOUT)
    if env:
        try:
            return int(env)
        except ValueError:
            pass
    return DEFAULT_TIMEOUT


def is_allowed_host(url: str) -> bool:
    """Return True if url uses HTTPS and its final host is in the allowlist."""
    try:
        parsed = urllib.parse.urlparse(url)
    except Exception:
        return False
    if parsed.scheme != "https":
        return False
    try:
        port = parsed.port
    except ValueError:
        return False
    if port not in (None, 443) or parsed.username is not None or parsed.password is not None:
        return False
    host = parsed.hostname or ""
    return host.lower() in {h.lower() for h in ALLOWED_HOSTS}


def _origin(url: str) -> tuple[str, str, int]:
    """Return a normalized origin tuple for an already-validated HTTPS URL."""
    parsed = urllib.parse.urlparse(url)
    return (parsed.scheme.lower(), (parsed.hostname or "").lower(), parsed.port or 443)


def _request_headers(extra: dict[str, str] | None = None) -> dict[str, str]:
    headers = {"User-Agent": USER_AGENT}
    if extra:
        headers.update(extra)
    return headers


# Service-specific headers that must be stripped when a redirect crosses hosts.
_SENSITIVE_HEADERS = frozenset({
    "tenantid",
    "tenantname",
    "tyler-tenanturl",
    "tyler-tenant-culture",
    "authorization",
    "cookie",
})


class AllowlistRedirectHandler(urllib.request.HTTPRedirectHandler):
    """Follow redirects only to HTTPS allowlisted hosts.

    Service-specific headers are stripped on cross-origin redirects, and the
    total number of redirects is bounded.
    """

    max_redirections = MAX_REDIRECTS

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        if not is_allowed_host(newurl):
            raise SecurityError(f"Redirect led to a non-allowlisted host: {newurl}")
        final_url_validator = getattr(req, "_raleigh_final_url_validator", None)
        if final_url_validator is not None:
            final_url_validator(newurl)
        # Strip sensitive headers when crossing origins.
        old_origin = _origin(req.full_url)
        new_origin = _origin(newurl)
        new_headers = dict(req.headers)
        if old_origin != new_origin:
            for name in list(new_headers):
                if name.lower() in _SENSITIVE_HEADERS:
                    del new_headers[name]
        # Preserve method/body only for 307/308; otherwise downgrade to GET.
        if req.get_method() == "HEAD":
            method = "HEAD"
            data = None
        elif code not in (307, 308):
            new_headers.pop("Content-Length", None)
            new_headers.pop("Content-Type", None)
            method = "GET"
            data = None
        else:
            method = req.get_method()
            data = req.data
        if data is not None and old_origin != new_origin:
            raise SecurityError("Cross-origin redirects cannot preserve a request body")
        _enforce_method_policy(method, newurl)
        redirected = urllib.request.Request(
            newurl,
            headers=new_headers,
            method=method,
            data=data,
            origin_req_host=req.origin_req_host,
            unverifiable=True,
        )
        if final_url_validator is not None:
            setattr(redirected, "_raleigh_final_url_validator", final_url_validator)
        return redirected


# Prebuilt opener with the bounded allowlisted redirect handler.
_OPENER = urllib.request.build_opener(AllowlistRedirectHandler)


# Regex patterns for verified ArcGIS read-only POST endpoints.
_ARCGIS_POST_PATTERNS = (
    r"/arcgis/rest/services/.*/(FeatureServer|MapServer)/\d+/query$",
    r"/server/rest/services/.*/(FeatureServer|MapServer)/\d+/query$",
    r"/arcgis/rest/services/.*/GeocodeServer/geocodeAddresses$",
    r"/server/rest/services/.*/GeocodeServer/geocodeAddresses$",
)

_ARCGIS_POST_HOSTS = frozenset({
    "maps.raleighnc.gov",
    "maps.wake.gov",
    "maps.wakegov.com",
    "services.arcgis.com",
    "services1.arcgis.com",
    "services3.arcgis.com",
    "utility.arcgis.com",
})

_HOST_SCOPED_POST_PATHS: dict[str, frozenset[str]] = {
    "raleighnc-energovpub.tylerhost.net": frozenset({
        "/apps/selfservice/api/energov/search/search",
        "/apps/selfservice/api/energov/entity/inspections/search/search",
    }),
    "pub-raleighnc.escribemeetings.com": frozenset({
        "/MeetingsCalendarView.aspx/PastMeetings",
    }),
}


def _is_allowed_post(url: str) -> bool:
    """Return True if url is a known read-only POST endpoint."""
    import re
    parsed = urllib.parse.urlparse(url)
    path = parsed.path.rstrip("/")
    host = (parsed.hostname or "").lower()
    host_paths = _HOST_SCOPED_POST_PATHS.get(host)
    if host_paths and path in host_paths:
        return True
    if host in _ARCGIS_POST_HOSTS:
        for pattern in _ARCGIS_POST_PATTERNS:
            if re.search(pattern, path, re.IGNORECASE):
                return True
    return False


def _enforce_method_policy(method: str | None, url: str) -> None:
    """Reject disallowed methods before any network I/O."""
    method = (method or "GET").upper()
    if method in {"GET", "HEAD"}:
        return
    if method == "POST" and _is_allowed_post(url):
        return
    raise RequestPolicyError(
        f"HTTP {method} is not allowed for {url}"
    )


def require_object(payload: Any, operation: str = "JSON request") -> dict[str, Any]:
    """Return an object-shaped JSON value or raise an explicit protocol error."""
    if not isinstance(payload, dict):
        raise ValueError(f"{operation} returned a non-object JSON document")
    return payload


def require_object_list(payload: Any, operation: str) -> list[dict[str, Any]]:
    """Return a list of objects or raise an explicit protocol error."""
    if not isinstance(payload, list) or any(not isinstance(item, dict) for item in payload):
        raise ValueError(f"{operation} returned an invalid object list")
    return payload


def require_positive_limit(limit: int | None, *, allow_none: bool = False) -> int | None:
    """Validate a public library limit consistently with the CLI contract."""
    if limit is None and allow_none:
        return None
    if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1:
        raise ValueError("limit must be a positive integer")
    return limit


def raise_for_arcgis_error(payload: Any, operation: str = "ArcGIS request") -> None:
    """Raise a concise error for ArcGIS HTTP-200 error envelopes."""
    payload = require_object(payload, operation)
    error = payload.get("error")
    if not error:
        return
    message = error.get("message") if isinstance(error, dict) else str(error)
    details = error.get("details", []) if isinstance(error, dict) else []
    suffix = f": {'; '.join(str(item) for item in details)}" if details else ""
    raise ValueError(f"{operation} failed: {message or 'unknown error'}{suffix}")


def _check_content_length(headers: dict[str, list[str]] | None, max_bytes: int) -> None:
    """Raise if the declared Content-Length exceeds the cap."""
    if headers is None:
        return
    # HTTPMessage.get can return a string or None.
    cl = headers.get("Content-Length")
    if cl:
        try:
            length = int(cl)
        except (TypeError, ValueError):
            return
        if length > max_bytes:
            raise ResponseTooLargeError(
                f"Content-Length {length} exceeds maximum {max_bytes}"
            )


def _read_limited(resp, max_bytes: int) -> bytes:
    """Read up to max_bytes from a response, raising if the cap is exceeded."""
    _check_content_length(resp.headers, max_bytes)
    chunks: list[bytes] = []
    total = 0
    while total < max_bytes:
        chunk = resp.read(min(65536, max_bytes - total))
        if not chunk:
            break
        chunks.append(chunk)
        total += len(chunk)
    else:
        # We reached the cap; ensure there is no more data.
        extra = resp.read(1)
        if extra:
            raise ResponseTooLargeError(
                f"Response exceeds maximum {max_bytes} bytes"
            )
    return b"".join(chunks)


def json_request(
    url: str,
    timeout: int | None = None,
    method: str | None = None,
    data: bytes | None = None,
    headers: dict[str, str] | None = None,
    max_bytes: int = DEFAULT_MAX_JSON_BYTES,
    final_url_validator: Callable[[str], None] | None = None,
) -> dict[str, Any]:
    """Fetch JSON from an allowlisted URL and return the parsed body."""
    if not is_allowed_host(url):
        raise SecurityError(f"URL host is not allowlisted: {url}")
    effective_method = method or ("POST" if data is not None else "GET")
    _enforce_method_policy(effective_method, url)
    req_headers = _request_headers({"Accept": "application/json"})
    if headers:
        req_headers.update(headers)
    req = urllib.request.Request(
        url, headers=req_headers, method=effective_method, data=data
    )
    if final_url_validator is not None:
        setattr(req, "_raleigh_final_url_validator", final_url_validator)
    with _OPENER.open(req, timeout=timeout or _get_timeout()) as resp:
        final_url = resp.geturl()
        if not is_allowed_host(final_url):
            raise SecurityError(f"Redirect led to a non-allowlisted host: {final_url}")
        if final_url_validator is not None:
            final_url_validator(final_url)
        body = _read_limited(resp, max_bytes)
        if not body:
            return {}
        payload = json.loads(body.decode("utf-8"))
        return require_object(payload)


def raw_request(
    url: str,
    timeout: int | None = None,
    method: str | None = None,
    data: bytes | None = None,
    headers: dict[str, str] | None = None,
    max_bytes: int = DEFAULT_MAX_RAW_BYTES,
) -> bytes:
    """Fetch raw bytes from an allowlisted URL."""
    if not is_allowed_host(url):
        raise SecurityError(f"URL host is not allowlisted: {url}")
    effective_method = method or ("POST" if data is not None else "GET")
    _enforce_method_policy(effective_method, url)
    req_headers = _request_headers()
    if headers:
        req_headers.update(headers)
    req = urllib.request.Request(
        url, headers=req_headers, method=effective_method, data=data
    )
    with _OPENER.open(req, timeout=timeout or _get_timeout()) as resp:
        final_url = resp.geturl()
        if not is_allowed_host(final_url):
            raise SecurityError(f"Redirect led to a non-allowlisted host: {final_url}")
        return _read_limited(resp, max_bytes)


def probe_url(
    url: str,
    timeout: int | None = None,
    final_url_validator: Callable[[str], None] | None = None,
) -> str:
    """Verify that an allowlisted HTTPS resource is available without reading it."""
    if not is_allowed_host(url):
        raise SecurityError(f"URL host is not allowlisted: {url}")
    _enforce_method_policy("HEAD", url)
    request = urllib.request.Request(
        url, headers=_request_headers(), method="HEAD"
    )
    if final_url_validator is not None:
        setattr(request, "_raleigh_final_url_validator", final_url_validator)
    with _OPENER.open(request, timeout=timeout or _get_timeout()) as response:
        final_url = response.geturl()
        if not is_allowed_host(final_url):
            raise SecurityError(f"Redirect led to a non-allowlisted host: {final_url}")
        if final_url_validator is not None:
            final_url_validator(final_url)
        return final_url


def cache_dir() -> Path:
    """Return the cache directory."""
    base = os.environ.get("RALEIGH_CACHE")
    if base:
        return Path(base)
    return Path.home() / ".cache" / "raleigh"


def cache_path(key: str) -> Path:
    """Return a cache file path for the given key."""
    return cache_dir() / key


def read_cache(key: str, max_age_seconds: int | None = None) -> Any | None:
    """Read a cached JSON value if present and fresh."""
    path = cache_path(key)
    try:
        fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size > DEFAULT_MAX_JSON_BYTES:
            os.close(fd)
            return None
        if max_age_seconds is not None and time.time() - info.st_mtime > max_age_seconds:
            os.close(fd)
            return None
        with os.fdopen(fd, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None


def read_cache_bytes(
    key: str,
    *,
    max_age_seconds: int | None = None,
    max_bytes: int = DEFAULT_MAX_RAW_BYTES,
) -> bytes | None:
    """Read a bounded binary cache value if present, regular, and fresh."""
    path = cache_path(key)
    try:
        fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size > max_bytes:
            os.close(fd)
            return None
        if max_age_seconds is not None and time.time() - info.st_mtime > max_age_seconds:
            os.close(fd)
            return None
        with os.fdopen(fd, "rb") as stream:
            data = stream.read(max_bytes + 1)
        return data if len(data) <= max_bytes else None
    except OSError:
        return None


def _atomic_write(path: Path, data: bytes | str, *, replace: bool = True) -> None:
    """Atomically write through a unique, exclusively-created sibling file."""
    fd, temp_name = tempfile.mkstemp(
        dir=path.parent,
        prefix=f".{path.name}.",
        suffix=".tmp",
    )
    temp = Path(temp_name)
    try:
        if isinstance(data, bytes):
            with os.fdopen(fd, "wb") as stream:
                stream.write(data)
                stream.flush()
                os.fsync(stream.fileno())
        else:
            with os.fdopen(fd, "w", encoding="utf-8") as stream:
                stream.write(data)
                stream.flush()
                os.fsync(stream.fileno())
        if replace:
            os.replace(temp, path)
        else:
            os.link(temp, path)
            temp.unlink()
    except Exception:
        try:
            temp.unlink()
        except FileNotFoundError:
            pass
        raise


def write_cache(key: str, value: Any) -> None:
    """Write a JSON value to the cache atomically."""
    path = cache_path(key)
    path.parent.mkdir(parents=True, exist_ok=True)
    _atomic_write(path, json.dumps(value, indent=2))


def write_cache_bytes(key: str, value: bytes) -> None:
    """Write a binary cache value atomically."""
    path = cache_path(key)
    path.parent.mkdir(parents=True, exist_ok=True)
    _atomic_write(path, value)


def clear_cache() -> None:
    """Remove all cached files."""
    cd = cache_dir()
    if not cd.exists():
        return
    for entry in cd.iterdir():
        if entry.is_file():
            entry.unlink()
        elif entry.is_dir():
            import shutil

            shutil.rmtree(entry)


def safe_write(path: Path, data: bytes | str, *, force: bool = False) -> None:
    """Write data atomically, rejecting symlinks and existing files by default."""
    path = Path(path)
    if path.is_symlink() or path.exists() and not force:
        raise FileExistsError(f"Destination exists (use --force to overwrite): {path}")
    path.parent.mkdir(parents=True, exist_ok=True)
    _atomic_write(path, data, replace=force)

"""Read-only guest-public EnerGov Permit and Development Portal adapter."""

from __future__ import annotations

import json
import os
from typing import Any

from raleighlib import core


BASE_URL = "https://raleighnc-energovpub.tylerhost.net/apps/selfservice"
SEARCH_URL = f"{BASE_URL}/api/energov/search/search"
CRITERIA_URL = f"{BASE_URL}/api/energov/search/criteria"
PERMIT_URL = f"{BASE_URL}/api/energov/permits"
INSPECTION_SEARCH_URL = f"{BASE_URL}/api/energov/entity/inspections/search/search"


class UnsupportedEndpointError(ValueError):
    """Raised for operations that have no verified guest-public endpoint."""


def _ensure_enabled() -> None:
    value = os.environ.get("RALEIGH_DISABLE_DEVELOPMENT", "").strip().casefold()
    if value in {"1", "true", "yes", "on"}:
        raise UnsupportedEndpointError(
            "Permit and Development Portal adapter is disabled by RALEIGH_DISABLE_DEVELOPMENT"
        )


def _energov_headers(extra: dict[str, str] | None = None) -> dict[str, str]:
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "tenantId": "1",
        "tenantName": "RaleighNCProd",
        "Tyler-TenantUrl": "RaleighNCProd",
        "Tyler-Tenant-Culture": "en-US",
    }
    if extra:
        headers.update(extra)
    return headers


def fetch_criteria() -> dict[str, Any]:
    """Fetch the guest-public search criteria contract."""
    _ensure_enabled()
    data = core.json_request(CRITERIA_URL, headers=_energov_headers())
    if not isinstance(data, dict) or data.get("Success") is False:
        raise UnsupportedEndpointError("EnerGov criteria endpoint returned an error")
    if not isinstance(data.get("Result"), dict):
        raise UnsupportedEndpointError("EnerGov criteria schema is incompatible: Result is not an object")
    return data


def _criteria_field(record_type: str) -> str:
    mapping = {
        "permit": "PermitCriteria",
        "plan": "PlanCriteria",
        "inspection": "InspectionCriteria",
        "code-case": "CodeCaseCriteria",
        "request": "RequestCriteria",
        "license": "LicenseCriteria",
        "project": "ProjectCriteria",
    }
    return mapping.get(record_type.lower(), "PermitCriteria")


def supported_record_types(criteria: dict[str, Any]) -> set[str]:
    """Discover guest-public record types advertised by the criteria contract."""
    result = criteria.get("Result")
    if not isinstance(result, dict):
        raise UnsupportedEndpointError("EnerGov criteria schema is incompatible: Result is not an object")
    candidates = {"permit", "plan", "inspection", "code-case", "request", "license", "project"}
    return {kind for kind in candidates if isinstance(result.get(_criteria_field(kind)), dict)}


def _filter_module(record_type: str) -> int:
    # EnerGov FilterModule enum values used by the public search controller.
    mapping = {
        "permit": 2,
        "plan": 3,
        "inspection": 4,
        "code-case": 5,
        "request": 6,
        "license": 10,
        "project": 11,
    }
    return mapping.get(record_type.lower(), 1)


def _public_scalar(value: Any, *nested_keys: str) -> Any:
    """Return a guest-display scalar, never an upstream container."""
    if isinstance(value, dict):
        for key in nested_keys:
            candidate = value.get(key)
            if candidate is None or isinstance(candidate, (dict, list, tuple, set)):
                continue
            return candidate
        return None
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    return None


def _first_public(record: dict[str, Any], keys: tuple[str, ...], *nested_keys: str) -> Any:
    for key in keys:
        value = _public_scalar(record.get(key), *nested_keys)
        if value not in (None, ""):
            return value
    return None


def _normalize_search_record(record: dict[str, Any]) -> dict[str, Any]:
    """Return only fields represented by the guest-public search cards."""
    address = _first_public(record, ("Address",), "FullAddress", "AddressLine", "DisplayText")
    return {
        "RecordId": _first_public(record, ("CaseId", "Id", "id"), "Id", "Value"),
        "RecordNumber": _first_public(record, ("CaseNumber", "RecordNumber", "PermitNumber"), "Number", "Value"),
        "RecordType": _first_public(record, ("CaseType", "RecordType"), "Name", "DisplayText", "Value"),
        "WorkClass": _first_public(record, ("CaseWorkclass", "WorkClass"), "Name", "DisplayText", "Value"),
        "Status": _first_public(record, ("CaseStatus", "Status"), "Name", "DisplayText", "Value"),
        "ProjectName": _first_public(record, ("ProjectName",), "Name", "DisplayText"),
        "IssueDate": _public_scalar(record.get("IssueDate")),
        "ApplyDate": _public_scalar(record.get("ApplyDate")),
        "ExpireDate": _public_scalar(record.get("ExpireDate")),
        "FinalDate": _public_scalar(record.get("FinalDate")),
        "Address": address or _public_scalar(record.get("AddressDisplay")),
        "ParcelNumber": _first_public(record, ("MainParcel",), "ParcelNumber", "Number", "Value"),
        "Description": _public_scalar(record.get("Description")),
    }


def public_search(
    record_type: str,
    query: str | None = None,
    limit: int = 20,
) -> dict[str, Any]:
    """Search guest-visible records by type using the verified POST contract.

    Returns a dict with ``results`` and ``total`` so callers can paginate or
    report counts.
    """
    core.require_positive_limit(limit)
    criteria = fetch_criteria()
    normalized_type = record_type.lower()
    supported = supported_record_types(criteria)
    if normalized_type not in supported:
        raise UnsupportedEndpointError(
            f"EnerGov does not advertise guest-public {normalized_type} search criteria"
        )
    result = dict(criteria["Result"])
    result["Keyword"] = query or ""
    result["ExactMatch"] = bool(query)
    result["SearchModule"] = 1  # Global public search
    result["FilterModule"] = _filter_module(record_type)
    result["SearchMainAddress"] = False
    result["PageNumber"] = 1
    result["PageSize"] = limit
    result["SortBy"] = None
    result["SortAscending"] = True
    data = core.json_request(
        SEARCH_URL,
        method="POST",
        data=json.dumps(result).encode("utf-8"),
        headers=_energov_headers(),
    )
    data = core.require_object(data, "EnerGov search")
    search_result = data.get("Result")
    if not isinstance(search_result, dict):
        message = (
            data.get("ErrorMessage")
            or data.get("ValidationErrorMessage")
            or "EnerGov search returned no result"
        )
        raise UnsupportedEndpointError(message)
    entity_results = search_result.get("EntityResults")
    total_found = search_result.get("TotalFound")
    if not isinstance(entity_results, list) or any(not isinstance(item, dict) for item in entity_results):
        raise UnsupportedEndpointError("EnerGov search schema is incompatible: EntityResults is not an object list")
    if not isinstance(total_found, int) or isinstance(total_found, bool) or total_found < 0:
        raise UnsupportedEndpointError("EnerGov search schema is incompatible: TotalFound is invalid")
    if len(entity_results) > total_found:
        raise UnsupportedEndpointError(
            "EnerGov search schema is incompatible: EntityResults exceeds TotalFound"
        )
    return {
        "results": [
            _normalize_search_record(record)
            for record in entity_results[:limit]
        ],
        "total": total_found,
    }


def _is_uuid(value: str) -> bool:
    """Return True if value looks like an EnerGov record UUID."""
    import uuid
    try:
        uuid.UUID(value)
        return True
    except ValueError:
        return False


def _resolve_uuid(record: str) -> str:
    """Resolve a record identifier to a UUID, searching by record number if needed."""
    record = record.strip()
    if _is_uuid(record):
        return record
    results = public_search("permit", query=record, limit=20)
    record_folded = record.casefold()
    matches = [
        r for r in results.get("results", [])
        if str(r.get("RecordNumber") or r.get("PermitNumber") or r.get("CaseNumber") or "").casefold()
        == record_folded
    ]
    if not matches:
        raise ValueError(f"No permit found for record number: {record}")
    if len(matches) > 1:
        raise ValueError(f"Ambiguous record number '{record}'; supply the UUID")
    resolved = (
        matches[0].get("RecordId")
        or matches[0].get("Id")
        or matches[0].get("id")
        or matches[0].get("CaseId")
    )
    if not resolved:
        raise ValueError("Resolved record has no UUID")
    return resolved


def permit_detail(record: str) -> dict[str, Any]:
    """Fetch guest-public permit details by UUID or record number."""
    _ensure_enabled()
    uuid = _resolve_uuid(record)
    url = f"{PERMIT_URL}/{uuid}"
    data = core.json_request(url, headers=_energov_headers())
    if not isinstance(data, dict):
        raise UnsupportedEndpointError("EnerGov permit detail schema is incompatible")
    if data.get("Success") is False:
        raise ValueError(data.get("ErrorMessage") or "EnerGov permit detail failed")
    result = data.get("Result", data)
    if not isinstance(result, dict):
        raise UnsupportedEndpointError("EnerGov permit detail Result must be an object")
    projected = {
        "PermitId": _first_public(result, ("PermitId",), "Id", "Value"),
        "PermitNumber": _first_public(result, ("PermitNumber",), "Number", "Value"),
        "PermitType": _first_public(result, ("PermitType",), "Name", "DisplayText", "Value"),
        "PermitStatus": _first_public(result, ("PermitStatus",), "Name", "DisplayText", "Value"),
        "IssueDate": _public_scalar(result.get("IssueDate")),
        "ExpireDate": _public_scalar(result.get("ExpireDate")),
        "FinalizeDate": _public_scalar(result.get("FinalizeDate")),
        "ApplyDate": _public_scalar(result.get("ApplyDate")),
        "WorkClassName": _first_public(result, ("WorkClassName",), "Name", "DisplayText", "Value"),
        "Description": _public_scalar(result.get("Description")),
        "IVRNumber": _public_scalar(result.get("IVRNumber")),
        "MainAddress": _first_public(result, ("MainAddress",), "FullAddress", "AddressLine", "DisplayText"),
        "MainParcelNumber": _first_public(result, ("MainParcelNumber",), "ParcelNumber", "Number", "Value"),
        "ProjectName": _first_public(result, ("ProjectName",), "Name", "DisplayText"),
        "DistrictName": _first_public(result, ("DistrictName",), "Name", "DisplayText"),
        "SquareFeet": _public_scalar(result.get("SquareFeet")),
        "Value": _public_scalar(result.get("Value")),
    }
    if not projected["PermitId"] and not projected["PermitNumber"]:
        raise UnsupportedEndpointError(
            "EnerGov permit detail Result has no permit identifier"
        )
    return projected


def inspections_for_record(record: str, limit: int = 10) -> list[dict[str, Any]]:
    """Fetch guest-visible inspections for a permit UUID or record number."""
    if limit < 1:
        raise ValueError("limit must be at least 1")
    _ensure_enabled()
    uuid = _resolve_uuid(record)
    payload = {
        "PageNumber": 1,
        "PageSize": limit,
        "SortField": "",
        "IsSortedInAscendingOrder": True,
        "ModuleId": 1,
        "EntityId": uuid,
        "IsExistingInspection": True,
        "IsOptionalInspection": False,
        "IsFailed": False,
    }
    data = core.json_request(
        INSPECTION_SEARCH_URL,
        method="POST",
        data=json.dumps(payload).encode("utf-8"),
        headers=_energov_headers(),
    )
    if not isinstance(data, dict):
        raise UnsupportedEndpointError("EnerGov inspection schema is incompatible")
    if data.get("Success") is False:
        raise ValueError(data.get("ErrorMessage") or "EnerGov inspection search failed")
    result = data.get("Result", data)
    if isinstance(result, dict):
        if "results" in result:
            result = result["results"]
        elif "Results" in result:
            result = result["Results"]
        else:
            raise UnsupportedEndpointError(
                "EnerGov inspection Result has no recognized result list"
            )
    if not isinstance(result, list):
        raise UnsupportedEndpointError("EnerGov inspection Result must be a list")
    rows: list[dict[str, Any]] = []
    for inspection in result[:limit]:
        if not isinstance(inspection, dict):
            raise UnsupportedEndpointError(
                "EnerGov inspection Result contains a non-object item"
            )
        rows.append({
            "InspectionId": _first_public(inspection, ("InspectionId",), "Id", "Value"),
            "InspectionNumber": _first_public(inspection, ("InspectionNumber",), "Number", "Value"),
            "InspectionType": _first_public(inspection, ("InspectionType",), "Name", "DisplayText", "Value"),
            "InspectionStatus": _first_public(inspection, ("InspectionStatus",), "Name", "DisplayText", "Value"),
            "RequestedDate": _public_scalar(inspection.get("RequestedDate")),
            "ScheduledStartDate": _public_scalar(inspection.get("ScheduledStartDate")),
            "ActualDate": _public_scalar(inspection.get("ActualDate")),
            "PrimaryInspector": _first_public(inspection, ("PrimaryInspector",), "Name", "DisplayName"),
            "IsReinspectionDisplayText": _public_scalar(inspection.get("IsReinspectionDisplayText")),
        })
    return rows


def code_cases(query: str | None = None, limit: int = 20) -> list[dict[str, Any]]:
    """Search guest-visible code cases."""
    return public_search("code-case", query=query, limit=limit)


def licenses(query: str | None = None, limit: int = 20) -> list[dict[str, Any]]:
    """Search guest-visible licenses."""
    return public_search("license", query=query, limit=limit)

"""Raleigh Fire Department incident data access.

Resolves stable ArcGIS item IDs to live FeatureServer URLs and provides
source-aware queries across two RFD datasets: the full public history
(2007–present) and the past-month feed. Normalizes the 2026 classification
schema transition into stable output keys without fabricating cross-era
mappings, and computes response durations only from valid timestamps.
"""

from __future__ import annotations

import math
import re
import sys
import urllib.parse
from datetime import date, datetime, timedelta, timezone
from typing import Any

from raleighlib import arcgis
from raleighlib import core

ITEM_RESOLUTION_URL = "https://ral.maps.arcgis.com/sharing/rest/content/items/{item_id}"

RFD_SOURCES: dict[str, dict[str, str]] = {
    "full-history": {
        "item_id": "ea466e39e9ca4448b645c33a0d6c60ad",
        "label": "Fire Incidents (full public history, 2007–present)",
        "caveats": (
            "Legacy classification fields are null for 2026+ records; "
            "station is unpopulated for many records."
        ),
    },
    "past-month": {
        "item_id": "c983765e304a41d19087c8d95aa46d54",
        "label": "Fire Incidents (past month)",
        "caveats": (
            "Rolling past-month feed; carries station_name and the current "
            "classification fields only."
        ),
    },
}

_FIELD_MAPS: dict[str, dict[str, str]] = {
    "full-history": {
        "date": "dispatch_date_time",
        "station": "station",
        "platoon": "platoon",
        "group": "incident_group_name",
        "type_name": "incident_type_name",
        "type_legacy": "incident_type_description",
        "type_code": "incident_type",
    },
    "past-month": {
        "date": "dispatch_date_time",
        "station": "station_name",
        "platoon": "platoon",
        "group": "incident_group_name",
        "type_name": "incident_type_name",
    },
}

SCHEMA_TRANSITION_EPOCH_MS = 1767225600000

PRIVACY_CAVERAT = (
    "RFD excludes incident types 300–399 and 661 from this public feed for "
    "EMS/privacy reasons. This is not a complete record of all fire department "
    "responses and must not be used for emergency response."
)

REPORT_FIELDS = (
    "incident_number",
    "dispatch_date_time",
    "arrive_date_time",
    "cleared_date_time",
    "address",
    "station_name",
    "platoon",
    "incident_group_name",
    "incident_subgroup_code",
    "incident_type_name",
)
REPORT_LIMIT = 200

_STATION_NAME_RE = re.compile(r"^station\s*0*(\d+)\s*$", re.IGNORECASE)


class FireError(Exception):
    """Raised for RFD data resolution or query failures."""


def resolve_item_url(item_id: str) -> str:
    """Resolve an ArcGIS item ID to its FeatureServer/MapServer URL."""
    url = ITEM_RESOLUTION_URL.format(item_id=item_id)
    params = {"f": "json"}
    full_url = f"{url}?{urllib.parse.urlencode(params)}"
    meta = core.json_request(full_url)
    service_url = meta.get("url")
    if not service_url:
        raise FireError(f"Item {item_id} has no service URL")
    return service_url


def resolve_layer_url(source_key: str) -> str:
    """Resolve a source key to a queryable layer URL."""
    source = RFD_SOURCES.get(source_key)
    if not source:
        raise FireError(f"Unknown source: {source_key}")
    service_url = resolve_item_url(source["item_id"])
    return arcgis.resolve_queryable_layer(service_url)


def _escape_sql_value(value: str) -> str:
    """Escape a string value for use inside single quotes in an ArcGIS WHERE clause."""
    return value.replace("'", "''")


def _escape_like_value(value: str) -> str:
    """Escape LIKE wildcards and quotes for use in a LIKE pattern."""
    return value.replace("'", "''").replace("%", "\\%").replace("_", "\\_")


def _discover_fields(layer_url: str) -> set[str]:
    """Return the set of field names advertised by a layer."""
    fields = arcgis.layer_fields(layer_url)
    return {f.get("name", "") for f in fields if isinstance(f, dict)}


def _ms_to_timestamp_literal(ms: int) -> str:
    """Format Unix milliseconds as an ArcGIS TIMESTAMP literal in UTC.

    The RFD layers reject bare epoch-millisecond literals in date comparisons
    and require TIMESTAMP 'YYYY-MM-DD HH:MM:SS'.
    """
    try:
        dt = datetime.fromtimestamp(ms / 1000, timezone.utc)
    except (OverflowError, OSError, ValueError) as exc:
        raise FireError(f"date range out of bounds: {ms}") from exc
    return "TIMESTAMP '" + dt.strftime("%Y-%m-%d %H:%M:%S") + "'"


def _station_patterns(station: int) -> list[str]:
    """Return case-insensitive station_name match patterns for a station number."""
    padded = f"STATION {station:02d}"
    bare = f"STATION {station}"
    return [padded, bare] if padded != bare else [padded]


def build_where_clause(
    source_key: str,
    available_fields: set[str],
    since_ms: int | None = None,
    station: int | None = None,
    platoon: str | None = None,
    group: str | None = None,
    incident_type: str | None = None,
) -> str:
    """Build an ArcGIS WHERE clause from durable filters.

    Field names are validated against the supplied field set. If a filter
    field is missing, the filter is skipped with a stderr warning.
    """
    field_map = _FIELD_MAPS.get(source_key)
    if not field_map:
        raise FireError(f"No field map for source: {source_key}")

    clauses: list[str] = ["1=1"]

    if since_ms is not None:
        date_field = field_map["date"]
        if date_field in available_fields:
            clauses.append(f"{date_field} >= {_ms_to_timestamp_literal(since_ms)}")
        else:
            print(
                f"Warning: date field '{date_field}' not found in {source_key}; skipping date filter",
                file=sys.stderr,
            )

    if station is not None:
        station_field = field_map["station"]
        if station_field in available_fields:
            if source_key == "past-month":
                patterns = " OR ".join(
                    f"UPPER({station_field}) LIKE '{pattern}'"
                    for pattern in _station_patterns(station)
                )
                clauses.append(f"({patterns})")
            else:
                clauses.append(f"{station_field} = {int(station)}")
        else:
            print(
                f"Warning: station field '{station_field}' not found in {source_key}; skipping station filter",
                file=sys.stderr,
            )

    if platoon:
        platoon_field = field_map["platoon"]
        if platoon_field in available_fields:
            escaped = _escape_sql_value(platoon.strip().upper())
            clauses.append(f"UPPER({platoon_field}) = '{escaped}'")
        else:
            print(
                f"Warning: platoon field '{platoon_field}' not found in {source_key}; skipping platoon filter",
                file=sys.stderr,
            )

    if group:
        group_field = field_map["group"]
        if group_field in available_fields:
            escaped = _escape_like_value(group.upper())
            clauses.append(f"UPPER({group_field}) LIKE '%{escaped}%'")
        else:
            print(
                f"Warning: group field '{group_field}' not found in {source_key}; skipping group filter",
                file=sys.stderr,
            )

    if incident_type:
        type_fields = [
            field_map["type_name"],
            field_map.get("type_legacy"),
        ]
        present = [f for f in type_fields if f and f in available_fields]
        if present:
            escaped = _escape_like_value(incident_type.upper())
            like_terms = [f"UPPER({field}) LIKE '%{escaped}%'" for field in present]
            code_field = field_map.get("type_code")
            if code_field and code_field in available_fields and incident_type.strip().isdigit():
                like_terms.append(f"{code_field} = {int(incident_type.strip())}")
            clauses.append(f"({' OR '.join(like_terms)})")
        else:
            print(
                f"Warning: incident type fields not found in {source_key}; skipping type filter",
                file=sys.stderr,
            )

    return " AND ".join(clauses)


def _present(value: Any) -> Any:
    """Return the value if it carries content, else None. Empty strings are missing."""
    if value is None:
        return None
    if isinstance(value, str) and not value.strip():
        return None
    return value


def normalize_classification(attrs: dict[str, Any]) -> dict[str, Any]:
    """Map legacy and current RFD classification fields to stable keys.

    RFD deprecated incident_type and incident_type_description for records
    after 2026-01-01, replacing them with incident_group_name,
    incident_subgroup_code, and incident_type_name. This never fabricates
    cross-era mappings: legacy NFIRS codes are not translated into current
    group names. If both field sets are populated, the replacement fields
    win and the era is reported as 'current'.
    """
    group = _present(attrs.get("incident_group_name"))
    subgroup = _present(attrs.get("incident_subgroup_code"))
    type_name = _present(attrs.get("incident_type_name"))
    legacy_desc = _present(attrs.get("incident_type_description"))
    legacy_code = _present(attrs.get("incident_type"))

    has_current = any(v is not None for v in (group, subgroup, type_name))
    has_legacy = any(v is not None for v in (legacy_desc, legacy_code))

    if has_current:
        era = "current"
    elif has_legacy:
        era = "legacy"
    else:
        era = "unknown"

    return {
        "_classification_era": era,
        "_incident_group": group,
        "_incident_subgroup": subgroup,
        "_incident_type": type_name if type_name is not None else legacy_desc,
        "_incident_code": legacy_code,
    }


def normalize_station(attrs: dict[str, Any]) -> int | None:
    """Return a station number from either source's station field.

    The full-history feed carries an integer 'station'; the past-month feed
    carries a 'station_name' string such as 'Station 09'. Returns None when
    neither carries a usable value; never guesses.
    """
    raw = attrs.get("station")
    if isinstance(raw, (int, float)) and not isinstance(raw, bool):
        if math.isfinite(raw) and raw == int(raw) and int(raw) > 0:
            return int(raw)
    name = _present(attrs.get("station_name"))
    if isinstance(name, str):
        match = _STATION_NAME_RE.match(name)
        if match:
            return int(match.group(1))
    return None


def _valid_timestamp_ms(value: Any) -> int | None:
    """Return the timestamp as integer milliseconds if it is a finite non-negative number."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    if not math.isfinite(value):
        return None
    if value < 0:
        return None
    return int(value)


def _duration_seconds(start: Any, end: Any) -> tuple[float | None, str]:
    """Compute a duration in seconds between two timestamps with a status.

    Statuses: ok, missing_timestamp, malformed_timestamp, reversed_timestamps.
    """
    if start is None or end is None:
        return None, "missing_timestamp"
    start_ms = _valid_timestamp_ms(start)
    end_ms = _valid_timestamp_ms(end)
    if start_ms is None or end_ms is None:
        return None, "malformed_timestamp"
    if end_ms < start_ms:
        return None, "reversed_timestamps"
    return (end_ms - start_ms) / 1000.0, "ok"


def compute_response_times(attrs: dict[str, Any]) -> dict[str, Any]:
    """Compute incident durations in seconds from dispatch/arrival/cleared timestamps.

    Each pair is validated independently; invalid pairs yield None with a
    status explaining why. No duration is invented.
    """
    dispatch = attrs.get("dispatch_date_time")
    arrive = attrs.get("arrive_date_time")
    cleared = attrs.get("cleared_date_time")

    pairs = [
        ("_dispatch_to_arrive_seconds", "_dispatch_to_arrive_status", dispatch, arrive),
        ("_arrive_to_clear_seconds", "_arrive_to_clear_status", arrive, cleared),
        ("_dispatch_to_clear_seconds", "_dispatch_to_clear_status", dispatch, cleared),
    ]
    out: dict[str, Any] = {}
    for value_key, status_key, start, end in pairs:
        seconds, status = _duration_seconds(start, end)
        out[value_key] = seconds
        out[status_key] = status
    return out


def _normalize_geometry(record: dict[str, Any]) -> dict[str, Any] | None:
    """Return GeoJSON geometry, suppressing missing and null-island placeholder points."""
    geom = record.get("geometry")
    if not geom:
        return None
    if geom.get("x") == 0 and geom.get("y") == 0:
        return None
    return arcgis.geometry_from_record(record)


def query_incidents(
    source_key: str,
    since_ms: int | None = None,
    station: int | None = None,
    platoon: str | None = None,
    group: str | None = None,
    incident_type: str | None = None,
    limit: int = 20,
    offset: int = 0,
    include_response_times: bool = False,
) -> dict[str, Any]:
    """Query a single RFD source and return an enriched GeoJSON FeatureCollection."""
    source = RFD_SOURCES.get(source_key)
    if not source:
        raise FireError(f"Unknown source: {source_key}")

    layer_url = resolve_layer_url(source_key)
    available_fields = _discover_fields(layer_url)
    where = build_where_clause(
        source_key,
        available_fields,
        since_ms=since_ms,
        station=station,
        platoon=platoon,
        group=group,
        incident_type=incident_type,
    )

    records = arcgis.query_all_pages(
        layer_url,
        where=where,
        return_geometry=True,
        max_records=limit,
        offset=offset,
    )

    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    features: list[dict[str, Any]] = []
    for record in records:
        attrs = dict(record.get("attributes", {}))
        attrs["_source"] = source_key
        attrs["_item_id"] = source["item_id"]
        attrs["_retrieved_at"] = now
        attrs.update(normalize_classification(attrs))
        attrs["_station"] = normalize_station(attrs)
        if include_response_times:
            attrs.update(compute_response_times(attrs))
        features.append({
            "type": "Feature",
            "properties": attrs,
            "geometry": _normalize_geometry(record),
        })

    return {
        "type": "FeatureCollection",
        "_sources": [
            {
                "item_id": source["item_id"],
                "label": source["label"],
                "caveats": source["caveats"],
            }
        ],
        "features": features,
    }


def query_reports(
    *,
    report_date: str | None = None,
    incident_number: str | None = None,
) -> dict[str, Any]:
    """Query the authoritative past-month layer by one exact bounded selector."""
    if bool(report_date) == bool(incident_number):
        raise FireError("provide exactly one report date or incident number")

    layer_url = resolve_layer_url("past-month")
    available_fields = _discover_fields(layer_url)
    missing = set(REPORT_FIELDS) - available_fields
    if missing:
        raise FireError(
            "Fire report source schema drift; missing fields: "
            + ", ".join(sorted(missing))
        )

    query: dict[str, str | None] = {
        "date": report_date,
        "incident_number": incident_number,
    }
    if report_date:
        try:
            start = date.fromisoformat(report_date)
        except ValueError as exc:
            raise FireError("report date must use YYYY-MM-DD") from exc
        end = start + timedelta(days=1)
        where = (
            f"dispatch_date_time >= TIMESTAMP '{start.isoformat()} 00:00:00' AND "
            f"dispatch_date_time < TIMESTAMP '{end.isoformat()} 00:00:00'"
        )
    else:
        number = (incident_number or "").strip()
        if not number:
            raise FireError("incident number must not be empty")
        where = f"incident_number = '{_escape_sql_value(number)}'"

    response = arcgis.query_layer(
        layer_url,
        where=where,
        out_fields=",".join(REPORT_FIELDS),
        return_geometry=False,
        result_record_count=REPORT_LIMIT,
        order_by_fields="dispatch_date_time ASC,incident_number ASC",
    )
    records = response.get("features")
    if not isinstance(records, list):
        raise FireError("Fire report source returned invalid features")
    if response.get("exceededTransferLimit"):
        raise FireError(f"Fire report query exceeded the {REPORT_LIMIT}-record safety limit")

    retrieved_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    reports: list[dict[str, Any]] = []
    for record in records:
        if not isinstance(record, dict) or not isinstance(record.get("attributes"), dict):
            raise FireError("Fire report source returned a malformed record")
        attrs = record["attributes"]
        if not _present(attrs.get("incident_number")):
            raise FireError("Fire report source returned a record without incident_number")
        reports.append({
            "source": "arcgis",
            "source_fragility": "authoritative-structured",
            **{field: attrs.get(field) for field in REPORT_FIELDS},
        })

    return {
        "query": query,
        "reports": reports,
        "sources": [{
            "source": "arcgis",
            "item_id": RFD_SOURCES["past-month"]["item_id"],
            "url": layer_url,
            "retrieved_at": retrieved_at,
        }],
        "warnings": [
            "The rolling ArcGIS feed may lag the RFD Report System and is not proof of completeness."
        ],
    }

"""Wake County MAR Fire Protection proximity data access.

Resolves the Wake County MAR Fire Protection Data table (item
8ab8c4f1a8eb473bacfcc1a1c1980b6c) and provides read-only lookups of
source-provided station rankings, road-network distances, ISO ratings,
and nearest-hydrant distances by canonical site-address identifier (CSAID).

Address input is composed through the official Raleigh locator and the
Wake County MAR Addresses layer to resolve a CSAID. The CLI never
calculates its own station routing, hydrant distance, or ISO rating.
"""

from __future__ import annotations

import json
import urllib.parse
from datetime import datetime, timezone
from typing import Any

from raleighlib import arcgis
from raleighlib import core
from raleighlib import geocode

FIRE_PROTECTION_ITEM_ID = "8ab8c4f1a8eb473bacfcc1a1c1980b6c"
FIRE_PROTECTION_ITEM_URL = (
    "https://ral.maps.arcgis.com/sharing/rest/content/items/" + FIRE_PROTECTION_ITEM_ID
)

MAR_ADDRESSES_LAYER_URL = (
    "https://services1.arcgis.com/a7CWfuGP5ZnLYE7I/arcgis/rest/services/"
    "Wake_County_MAR_Address_Data_Public/FeatureServer/0"
)

MIN_GEOCODE_SCORE = 90.0
MAR_ENVELOPE_DELTA = 0.0001
REQUIRED_PROTECTION_FIELDS = frozenset(
    {
        "CSAID",
        "STATION_RANK",
        "STATION_DISTANCE",
        "STATIONID",
        "STATION_ISO",
        "Hydrant_Distance",
    }
)

PROTECTION_CAVERAT = (
    "Source-provided fire protection proximity data. Distances are "
    "road-network values from the Wake County MAR table; the source does "
    "not advertise distance units. This is not live emergency response "
    "data and must not be used for emergency dispatch."
)


class FireProtectionError(Exception):
    """Raised for fire-protection resolution or query failures."""


def _resolve_fire_protection_layer() -> str:
    """Resolve the fire-protection item ID to a queryable layer URL."""
    params = {"f": "json"}
    url = f"{FIRE_PROTECTION_ITEM_URL}?{urllib.parse.urlencode(params)}"
    meta = core.json_request(url)
    service_url = meta.get("url")
    if not service_url:
        raise FireProtectionError(f"Item {FIRE_PROTECTION_ITEM_ID} has no service URL")
    return arcgis.resolve_queryable_layer(service_url)


def resolve_csaid_from_address(address: str) -> dict[str, Any]:
    """Geocode an address and resolve it to a Wake County MAR CSAID.

    Returns a dict with keys: csaid, match_address, score, lat, lon.
    Raises FireProtectionError on no match, ambiguous match, or when
    the geocoded location cannot be mapped to a MAR record.
    """
    candidates = geocode.find_address_candidates(
        address,
        out_fields="StAddr,SubAddr,Match_addr",
        min_score=MIN_GEOCODE_SCORE,
        max_locations=5,
    )
    if not candidates:
        raise FireProtectionError(
            f"No geocode match for '{address}' at score >= {MIN_GEOCODE_SCORE:.0f}"
        )

    top_score = max(candidate.get("score", 0) or 0 for candidate in candidates)
    top_candidates = [
        candidate
        for candidate in candidates
        if (candidate.get("score", 0) or 0) == top_score
    ]
    identities = {
        (
            candidate.get("address"),
            candidate.get("location", {}).get("x"),
            candidate.get("location", {}).get("y"),
        )
        for candidate in top_candidates
    }
    if len(identities) != 1:
        raise FireProtectionError(
            f"Address '{address}' has {len(identities)} equally ranked geocode "
            "matches; refine the address or supply --csaid directly"
        )

    best = top_candidates[0]
    location = best.get("location", {})
    lat = location.get("y")
    lon = location.get("x")
    if lat is None or lon is None:
        raise FireProtectionError(f"Geocode match for '{address}' has no coordinates")

    match_address = best.get("address", "")
    score = best.get("score")
    attributes = best.get("attributes", {})
    if not isinstance(attributes, dict):
        attributes = {}
    street_address = attributes.get("StAddr")
    subaddress = attributes.get("SubAddr")
    if isinstance(street_address, str) and street_address.strip():
        mar_match_address = " ".join(
            value.strip()
            for value in (street_address, subaddress)
            if isinstance(value, str) and value.strip()
        )
    else:
        mar_match_address = match_address.split(",", 1)[0]

    csaid = _spatial_resolve_csaid(lon, lat, mar_match_address)
    if csaid is None:
        raise FireProtectionError(
            f"Address '{match_address}' geocoded but no unique Wake County "
            "MAR record found nearby; supply --csaid directly"
        )

    return {
        "csaid": csaid,
        "match_address": match_address,
        "score": score,
        "lat": lat,
        "lon": lon,
    }


def _spatial_resolve_csaid(lon: float, lat: float, match_address: str) -> int | None:
    """Query the MAR Addresses layer near a point and return a unique CSAID.

    Uses a small envelope around the geocoded point and prefers an exact
    matched-address record. If none exists, it prefers base addresses over
    unit-level records. Returns None when zero or multiple distinct CSAIDs
    remain after filtering.
    """
    delta = MAR_ENVELOPE_DELTA
    geometry = json.dumps(
        {
            "xmin": lon - delta,
            "ymin": lat - delta,
            "xmax": lon + delta,
            "ymax": lat + delta,
            "spatialReference": {"wkid": 4326},
        }
    )
    params: dict[str, Any] = {
        "geometry": geometry,
        "geometryType": "esriGeometryEnvelope",
        "inSR": 4326,
        "spatialRel": "esriSpatialRelIntersects",
        "outFields": "CSAID,ADDRESS,SUBADDR_TYPE",
        "returnGeometry": "false",
        "f": "json",
    }
    url = f"{MAR_ADDRESSES_LAYER_URL}/query?{urllib.parse.urlencode(params)}"
    data = core.json_request(url)
    core.raise_for_arcgis_error(data, "MAR address lookup")
    features = data.get("features", [])
    if not isinstance(features, list):
        return None

    base_csaids: set[int] = set()
    all_csaids: set[int] = set()
    exact_csaids: set[int] = set()
    normalized_match = match_address.strip().casefold()
    for feature in features:
        attrs = feature.get("attributes", {})
        csaid = attrs.get("CSAID")
        if not isinstance(csaid, int) or isinstance(csaid, bool):
            continue
        all_csaids.add(csaid)
        mar_address = attrs.get("ADDRESS")
        if (
            isinstance(mar_address, str)
            and mar_address.strip().casefold() == normalized_match
        ):
            exact_csaids.add(csaid)
        subaddr = attrs.get("SUBADDR_TYPE")
        if subaddr is None or (isinstance(subaddr, str) and not subaddr.strip()):
            base_csaids.add(csaid)

    candidates = exact_csaids or base_csaids or all_csaids
    if len(candidates) == 1:
        return candidates.pop()
    return None


def query_fire_protection(csaid: int) -> dict[str, Any]:
    """Query the MAR Fire Protection table for a given CSAID.

    Returns a dict with source metadata and the ranked station records.
    """
    layer_url = _resolve_fire_protection_layer()
    fields = arcgis.layer_fields(layer_url)
    field_names = {field.get("name") for field in fields if isinstance(field, dict)}
    missing_fields = REQUIRED_PROTECTION_FIELDS - field_names
    if missing_fields:
        missing = ", ".join(sorted(missing_fields))
        raise FireProtectionError(
            f"Fire protection source schema drift: missing fields: {missing}"
        )

    where = f"CSAID = {int(csaid)}"
    records = arcgis.query_all_pages(
        layer_url,
        where=where,
        return_geometry=False,
        max_records=10,
        order_by_fields="STATION_RANK ASC",
    )

    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    stations: list[dict[str, Any]] = []
    hydrant_distances: set[Any] = set()

    for record in records:
        attrs = record.get("attributes", {})
        rank = attrs.get("STATION_RANK")
        station: dict[str, Any] = {
            "rank": rank,
            "station_id": attrs.get("STATIONID"),
            "distance": attrs.get("STATION_DISTANCE"),
            "iso": attrs.get("STATION_ISO"),
        }
        stations.append(station)
        if attrs.get("Hydrant_Distance") is not None:
            hydrant_distances.add(attrs.get("Hydrant_Distance"))

    if len(hydrant_distances) > 1:
        raise FireProtectionError(
            f"Fire protection source returned inconsistent hydrant distances "
            f"for CSAID {csaid}"
        )
    hydrant_distance = next(iter(hydrant_distances), None)

    return {
        "csaid": csaid,
        "item_id": FIRE_PROTECTION_ITEM_ID,
        "retrieved_at": now,
        "stations": stations,
        "hydrant_distance": hydrant_distance,
        "distance_units": None,
        "caveats": PROTECTION_CAVERAT,
    }

"""Raleigh public geocoding adapter."""

from __future__ import annotations

import json
import urllib.parse
from typing import Any

from raleighlib import core


GEOCODE_URL = "https://maps.raleighnc.gov/arcgis/rest/services/Locators/Locator/GeocodeServer"


def find_address_candidates(
    address: str,
    out_fields: str | None = None,
    out_sr: int = 4326,
    max_locations: int = 10,
    min_score: float | None = None,
) -> list[dict[str, Any]]:
    """Forward geocode a single-line address."""
    core.require_positive_limit(max_locations)
    params: dict[str, Any] = {
        "SingleLine": address,
        "outSR": out_sr,
        "maxLocations": max_locations,
        "f": "json",
    }
    if out_fields:
        params["outFields"] = out_fields
    url = f"{GEOCODE_URL}/findAddressCandidates?{urllib.parse.urlencode(params)}"
    data = core.json_request(url)
    core.raise_for_arcgis_error(data, "Forward geocoding")
    candidates = core.require_object_list(data.get("candidates", []), "Forward geocoding")
    return filter_candidates(candidates, min_score=min_score)


def reverse_geocode(
    lat: float,
    lon: float,
    out_sr: int = 4326,
    distance: int | None = None,
) -> dict[str, Any]:
    """Reverse geocode a coordinate."""
    params: dict[str, Any] = {
        "location": json.dumps({
            "x": lon,
            "y": lat,
            "spatialReference": {"wkid": 4326},
        }),
        "outSR": out_sr,
        "f": "json",
    }
    if distance is not None:
        params["distance"] = distance
    url = f"{GEOCODE_URL}/reverseGeocode?{urllib.parse.urlencode(params)}"
    result = core.json_request(url)
    core.raise_for_arcgis_error(result, "Reverse geocoding")
    return result


def suggest(
    text: str,
    out_sr: int = 4326,
    max_suggestions: int = 10,
) -> list[dict[str, Any]]:
    """Return address suggestions for a partial string."""
    core.require_positive_limit(max_suggestions)
    params: dict[str, Any] = {
        "text": text,
        "outSR": out_sr,
        "maxSuggestions": max_suggestions,
        "f": "json",
    }
    url = f"{GEOCODE_URL}/suggest?{urllib.parse.urlencode(params)}"
    data = core.json_request(url)
    core.raise_for_arcgis_error(data, "Address suggestion")
    return core.require_object_list(data.get("suggestions", []), "Address suggestion")


MAX_BATCH_SIZE = 1000


def geocode_addresses(
    records: list[dict[str, Any]],
    out_sr: int = 4326,
    out_fields: str | None = None,
    max_batch: int = MAX_BATCH_SIZE,
) -> list[dict[str, Any]]:
    """Batch geocode address records using POST form data.

    Every input row is preserved in the output with a status of ``matched``,
    ``unmatched``, or ``error``. The batch size is capped to avoid oversized
    requests.
    """
    if not records:
        return []
    core.require_positive_limit(max_batch)
    if len(records) > max_batch:
        raise ValueError(f"Batch geocoding limit is {max_batch} addresses")
    prepared: list[tuple[Any, int, dict[str, Any]]] = []
    used_request_ids: set[int] = set()
    for ordinal, record in enumerate(records, start=1):
        source = dict(record)
        source_id = source.get("OBJECTID", ordinal)
        request_id = source_id if isinstance(source_id, int) and not isinstance(source_id, bool) else ordinal
        if request_id in used_request_ids:
            request_id = ordinal
            while request_id in used_request_ids:
                request_id += len(records)
        used_request_ids.add(request_id)
        prepared.append((source_id, request_id, source))
    wrapped_records = [{"attributes": {**source, "OBJECTID": request_id}} for _, request_id, source in prepared]
    params: dict[str, Any] = {
        "addresses": json.dumps({"records": wrapped_records}),
        "outSR": out_sr,
        "f": "json",
    }
    if out_fields:
        params["outFields"] = out_fields
    data = core.json_request(
        f"{GEOCODE_URL}/geocodeAddresses",
        method="POST",
        data=urllib.parse.urlencode(params).encode("utf-8"),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    core.raise_for_arcgis_error(data, "Batch geocoding")
    results: dict[str, dict[str, Any]] = {}
    locations = core.require_object_list(data.get("locations", []), "Batch geocoding")
    for loc in locations:
        attrs = loc.get("attributes", {})
        rid = attrs.get("ResultID")
        score = attrs.get("Score")
        matched = score is not None and score > 0
        results[str(rid)] = {
            "score": score,
            "address": attrs.get("Match_addr") or attrs.get("MatchAddress"),
            "lat": loc.get("location", {}).get("y"),
            "lon": loc.get("location", {}).get("x"),
            "status": "matched" if matched else "unmatched",
            "attributes": attrs,
        }
    outputs: list[dict[str, Any]] = []
    for source_id, request_id, source in prepared:
        result = results.get(str(request_id)) or results.get(str(source_id))
        if result:
            outputs.append({"input_id": source_id, "source": source, **result})
        else:
            outputs.append(
                {
                    "input_id": source_id,
                    "source": source,
                    "score": None,
                    "address": None,
                    "lat": None,
                    "lon": None,
                    "status": "unmatched",
                    "attributes": {},
                }
            )
    return outputs


def geocode_with_magic_key(
    text: str,
    magic_key: str,
    out_sr: int = 4326,
    max_locations: int = 1,
) -> list[dict[str, Any]]:
    """Geocode a suggestion using its magicKey."""
    core.require_positive_limit(max_locations)
    params: dict[str, Any] = {
        "SingleLine": text,
        "magicKey": magic_key,
        "outSR": out_sr,
        "maxLocations": max_locations,
        "f": "json",
    }
    url = f"{GEOCODE_URL}/findAddressCandidates?{urllib.parse.urlencode(params)}"
    data = core.json_request(url)
    core.raise_for_arcgis_error(data, "Magic-key geocoding")
    return core.require_object_list(data.get("candidates", []), "Magic-key geocoding")


def filter_candidates(
    candidates: list[dict[str, Any]],
    min_score: float | None = None,
) -> list[dict[str, Any]]:
    """Filter candidates by minimum score."""
    if min_score is None:
        return candidates
    return [c for c in candidates if (c.get("score") or 0) >= min_score]

# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: gtfs-realtime.proto
# Protobuf Python Version: 6.31.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
    _runtime_version.Domain.PUBLIC,
    6,
    31,
    1,
    '',
    'gtfs-realtime.proto'
)
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x13gtfs-realtime.proto\x12\x10transit_realtime\"y\n\x0b\x46\x65\x65\x64Message\x12,\n\x06header\x18\x01 \x02(\x0b\x32\x1c.transit_realtime.FeedHeader\x12,\n\x06\x65ntity\x18\x02 \x03(\x0b\x32\x1c.transit_realtime.FeedEntity*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xed\x01\n\nFeedHeader\x12\x1d\n\x15gtfs_realtime_version\x18\x01 \x02(\t\x12Q\n\x0eincrementality\x18\x02 \x01(\x0e\x32+.transit_realtime.FeedHeader.Incrementality:\x0c\x46ULL_DATASET\x12\x11\n\ttimestamp\x18\x03 \x01(\x04\x12\x14\n\x0c\x66\x65\x65\x64_version\x18\x04 \x01(\t\"4\n\x0eIncrementality\x12\x10\n\x0c\x46ULL_DATASET\x10\x00\x12\x10\n\x0c\x44IFFERENTIAL\x10\x01*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xe1\x02\n\nFeedEntity\x12\n\n\x02id\x18\x01 \x02(\t\x12\x19\n\nis_deleted\x18\x02 \x01(\x08:\x05\x66\x61lse\x12\x31\n\x0btrip_update\x18\x03 \x01(\x0b\x32\x1c.transit_realtime.TripUpdate\x12\x32\n\x07vehicle\x18\x04 \x01(\x0b\x32!.transit_realtime.VehiclePosition\x12&\n\x05\x61lert\x18\x05 \x01(\x0b\x32\x17.transit_realtime.Alert\x12&\n\x05shape\x18\x06 \x01(\x0b\x32\x17.transit_realtime.Shape\x12$\n\x04stop\x18\x07 \x01(\x0b\x32\x16.transit_realtime.Stop\x12?\n\x12trip_modifications\x18\x08 \x01(\x0b\x32#.transit_realtime.TripModifications*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xf6\x0b\n\nTripUpdate\x12.\n\x04trip\x18\x01 \x02(\x0b\x32 .transit_realtime.TripDescriptor\x12\x34\n\x07vehicle\x18\x03 \x01(\x0b\x32#.transit_realtime.VehicleDescriptor\x12\x45\n\x10stop_time_update\x18\x02 \x03(\x0b\x32+.transit_realtime.TripUpdate.StopTimeUpdate\x12\x11\n\ttimestamp\x18\x04 \x01(\x04\x12\r\n\x05\x64\x65lay\x18\x05 \x01(\x05\x12\x44\n\x0ftrip_properties\x18\x06 \x01(\x0b\x32+.transit_realtime.TripUpdate.TripProperties\x1ai\n\rStopTimeEvent\x12\r\n\x05\x64\x65lay\x18\x01 \x01(\x05\x12\x0c\n\x04time\x18\x02 \x01(\x03\x12\x13\n\x0buncertainty\x18\x03 \x01(\x05\x12\x16\n\x0escheduled_time\x18\x04 \x01(\x03*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\x1a\xb9\x07\n\x0eStopTimeUpdate\x12\x15\n\rstop_sequence\x18\x01 \x01(\r\x12\x0f\n\x07stop_id\x18\x04 \x01(\t\x12;\n\x07\x61rrival\x18\x02 \x01(\x0b\x32*.transit_realtime.TripUpdate.StopTimeEvent\x12=\n\tdeparture\x18\x03 \x01(\x0b\x32*.transit_realtime.TripUpdate.StopTimeEvent\x12U\n\x1a\x64\x65parture_occupancy_status\x18\x07 \x01(\x0e\x32\x31.transit_realtime.VehiclePosition.OccupancyStatus\x12j\n\x15schedule_relationship\x18\x05 \x01(\x0e\x32@.transit_realtime.TripUpdate.StopTimeUpdate.ScheduleRelationship:\tSCHEDULED\x12\\\n\x14stop_time_properties\x18\x06 \x01(\x0b\x32>.transit_realtime.TripUpdate.StopTimeUpdate.StopTimeProperties\x1a\xff\x02\n\x12StopTimeProperties\x12\x18\n\x10\x61ssigned_stop_id\x18\x01 \x01(\t\x12\x15\n\rstop_headsign\x18\x02 \x01(\t\x12\x65\n\x0bpickup_type\x18\x03 \x01(\x0e\x32P.transit_realtime.TripUpdate.StopTimeUpdate.StopTimeProperties.DropOffPickupType\x12g\n\rdrop_off_type\x18\x04 \x01(\x0e\x32P.transit_realtime.TripUpdate.StopTimeUpdate.StopTimeProperties.DropOffPickupType\"X\n\x11\x44ropOffPickupType\x12\x0b\n\x07REGULAR\x10\x00\x12\x08\n\x04NONE\x10\x01\x12\x10\n\x0cPHONE_AGENCY\x10\x02\x12\x1a\n\x16\x43OORDINATE_WITH_DRIVER\x10\x03*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"P\n\x14ScheduleRelationship\x12\r\n\tSCHEDULED\x10\x00\x12\x0b\n\x07SKIPPED\x10\x01\x12\x0b\n\x07NO_DATA\x10\x02\x12\x0f\n\x0bUNSCHEDULED\x10\x03*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\x1a\x9b\x01\n\x0eTripProperties\x12\x0f\n\x07trip_id\x18\x01 \x01(\t\x12\x12\n\nstart_date\x18\x02 \x01(\t\x12\x12\n\nstart_time\x18\x03 \x01(\t\x12\x10\n\x08shape_id\x18\x04 \x01(\t\x12\x15\n\rtrip_headsign\x18\x05 \x01(\t\x12\x17\n\x0ftrip_short_name\x18\x06 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xdf\t\n\x0fVehiclePosition\x12.\n\x04trip\x18\x01 \x01(\x0b\x32 .transit_realtime.TripDescriptor\x12\x34\n\x07vehicle\x18\x08 \x01(\x0b\x32#.transit_realtime.VehicleDescriptor\x12,\n\x08position\x18\x02 \x01(\x0b\x32\x1a.transit_realtime.Position\x12\x1d\n\x15\x63urrent_stop_sequence\x18\x03 \x01(\r\x12\x0f\n\x07stop_id\x18\x07 \x01(\t\x12Z\n\x0e\x63urrent_status\x18\x04 \x01(\x0e\x32\x33.transit_realtime.VehiclePosition.VehicleStopStatus:\rIN_TRANSIT_TO\x12\x11\n\ttimestamp\x18\x05 \x01(\x04\x12K\n\x10\x63ongestion_level\x18\x06 \x01(\x0e\x32\x31.transit_realtime.VehiclePosition.CongestionLevel\x12K\n\x10occupancy_status\x18\t \x01(\x0e\x32\x31.transit_realtime.VehiclePosition.OccupancyStatus\x12\x1c\n\x14occupancy_percentage\x18\n \x01(\r\x12Q\n\x16multi_carriage_details\x18\x0b \x03(\x0b\x32\x31.transit_realtime.VehiclePosition.CarriageDetails\x1a\xd9\x01\n\x0f\x43\x61rriageDetails\x12\n\n\x02id\x18\x01 \x01(\t\x12\r\n\x05label\x18\x02 \x01(\t\x12^\n\x10occupancy_status\x18\x03 \x01(\x0e\x32\x31.transit_realtime.VehiclePosition.OccupancyStatus:\x11NO_DATA_AVAILABLE\x12 \n\x14occupancy_percentage\x18\x04 \x01(\x05:\x02-1\x12\x19\n\x11\x63\x61rriage_sequence\x18\x05 \x01(\r*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"G\n\x11VehicleStopStatus\x12\x0f\n\x0bINCOMING_AT\x10\x00\x12\x0e\n\nSTOPPED_AT\x10\x01\x12\x11\n\rIN_TRANSIT_TO\x10\x02\"}\n\x0f\x43ongestionLevel\x12\x1c\n\x18UNKNOWN_CONGESTION_LEVEL\x10\x00\x12\x14\n\x10RUNNING_SMOOTHLY\x10\x01\x12\x0f\n\x0bSTOP_AND_GO\x10\x02\x12\x0e\n\nCONGESTION\x10\x03\x12\x15\n\x11SEVERE_CONGESTION\x10\x04\"\xd9\x01\n\x0fOccupancyStatus\x12\t\n\x05\x45MPTY\x10\x00\x12\x18\n\x14MANY_SEATS_AVAILABLE\x10\x01\x12\x17\n\x13\x46\x45W_SEATS_AVAILABLE\x10\x02\x12\x16\n\x12STANDING_ROOM_ONLY\x10\x03\x12\x1e\n\x1a\x43RUSHED_STANDING_ROOM_ONLY\x10\x04\x12\x08\n\x04\x46ULL\x10\x05\x12\x1c\n\x18NOT_ACCEPTING_PASSENGERS\x10\x06\x12\x15\n\x11NO_DATA_AVAILABLE\x10\x07\x12\x11\n\rNOT_BOARDABLE\x10\x08*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xf1\x0b\n\x05\x41lert\x12\x36\n\ractive_period\x18\x01 \x03(\x0b\x32\x1b.transit_realtime.TimeRangeB\x02\x18\x01\x12\x39\n\x14\x63ommunication_period\x18\x02 \x03(\x0b\x32\x1b.transit_realtime.TimeRange\x12\x32\n\rimpact_period\x18\x03 \x03(\x0b\x32\x1b.transit_realtime.TimeRange\x12\x39\n\x0finformed_entity\x18\x05 \x03(\x0b\x32 .transit_realtime.EntitySelector\x12;\n\x05\x63\x61use\x18\x06 \x01(\x0e\x32\x1d.transit_realtime.Alert.Cause:\rUNKNOWN_CAUSE\x12>\n\x06\x65\x66\x66\x65\x63t\x18\x07 \x01(\x0e\x32\x1e.transit_realtime.Alert.Effect:\x0eUNKNOWN_EFFECT\x12/\n\x03url\x18\x08 \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x37\n\x0bheader_text\x18\n \x01(\x0b\x32\".transit_realtime.TranslatedString\x12<\n\x10\x64\x65scription_text\x18\x0b \x01(\x0b\x32\".transit_realtime.TranslatedString\x12;\n\x0ftts_header_text\x18\x0c \x01(\x0b\x32\".transit_realtime.TranslatedString\x12@\n\x14tts_description_text\x18\r \x01(\x0b\x32\".transit_realtime.TranslatedString\x12O\n\x0eseverity_level\x18\x0e \x01(\x0e\x32%.transit_realtime.Alert.SeverityLevel:\x10UNKNOWN_SEVERITY\x12\x30\n\x05image\x18\x0f \x01(\x0b\x32!.transit_realtime.TranslatedImage\x12\x42\n\x16image_alternative_text\x18\x10 \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x38\n\x0c\x63\x61use_detail\x18\x11 \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x39\n\reffect_detail\x18\x12 \x01(\x0b\x32\".transit_realtime.TranslatedString\"\xeb\x01\n\x05\x43\x61use\x12\x11\n\rUNKNOWN_CAUSE\x10\x01\x12\x0f\n\x0bOTHER_CAUSE\x10\x02\x12\x15\n\x11TECHNICAL_PROBLEM\x10\x03\x12\n\n\x06STRIKE\x10\x04\x12\x11\n\rDEMONSTRATION\x10\x05\x12\x0c\n\x08\x41\x43\x43IDENT\x10\x06\x12\x0b\n\x07HOLIDAY\x10\x07\x12\x0b\n\x07WEATHER\x10\x08\x12\x0f\n\x0bMAINTENANCE\x10\t\x12\x10\n\x0c\x43ONSTRUCTION\x10\n\x12\x13\n\x0fPOLICE_ACTIVITY\x10\x0b\x12\x15\n\x11MEDICAL_EMERGENCY\x10\x0c\x12\x11\n\rSPECIAL_EVENT\x10\r\"\xdd\x01\n\x06\x45\x66\x66\x65\x63t\x12\x0e\n\nNO_SERVICE\x10\x01\x12\x13\n\x0fREDUCED_SERVICE\x10\x02\x12\x16\n\x12SIGNIFICANT_DELAYS\x10\x03\x12\n\n\x06\x44\x45TOUR\x10\x04\x12\x16\n\x12\x41\x44\x44ITIONAL_SERVICE\x10\x05\x12\x14\n\x10MODIFIED_SERVICE\x10\x06\x12\x10\n\x0cOTHER_EFFECT\x10\x07\x12\x12\n\x0eUNKNOWN_EFFECT\x10\x08\x12\x0e\n\nSTOP_MOVED\x10\t\x12\r\n\tNO_EFFECT\x10\n\x12\x17\n\x13\x41\x43\x43\x45SSIBILITY_ISSUE\x10\x0b\"H\n\rSeverityLevel\x12\x14\n\x10UNKNOWN_SEVERITY\x10\x01\x12\x08\n\x04INFO\x10\x02\x12\x0b\n\x07WARNING\x10\x03\x12\n\n\x06SEVERE\x10\x04*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"7\n\tTimeRange\x12\r\n\x05start\x18\x01 \x01(\x04\x12\x0b\n\x03\x65nd\x18\x02 \x01(\x04*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"q\n\x08Position\x12\x10\n\x08latitude\x18\x01 \x02(\x02\x12\x11\n\tlongitude\x18\x02 \x02(\x02\x12\x0f\n\x07\x62\x65\x61ring\x18\x03 \x01(\x02\x12\x10\n\x08odometer\x18\x04 \x01(\x01\x12\r\n\x05speed\x18\x05 \x01(\x02*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xb7\x04\n\x0eTripDescriptor\x12\x0f\n\x07trip_id\x18\x01 \x01(\t\x12\x10\n\x08route_id\x18\x05 \x01(\t\x12\x14\n\x0c\x64irection_id\x18\x06 \x01(\r\x12\x12\n\nstart_time\x18\x02 \x01(\t\x12\x12\n\nstart_date\x18\x03 \x01(\t\x12T\n\x15schedule_relationship\x18\x04 \x01(\x0e\x32\x35.transit_realtime.TripDescriptor.ScheduleRelationship\x12L\n\rmodified_trip\x18\x07 \x01(\x0b\x32\x35.transit_realtime.TripDescriptor.ModifiedTripSelector\x1a\x82\x01\n\x14ModifiedTripSelector\x12\x18\n\x10modifications_id\x18\x01 \x01(\t\x12\x18\n\x10\x61\x66\x66\x65\x63ted_trip_id\x18\x02 \x01(\t\x12\x12\n\nstart_time\x18\x03 \x01(\t\x12\x12\n\nstart_date\x18\x04 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\x8a\x01\n\x14ScheduleRelationship\x12\r\n\tSCHEDULED\x10\x00\x12\r\n\x05\x41\x44\x44\x45\x44\x10\x01\x1a\x02\x08\x01\x12\x0f\n\x0bUNSCHEDULED\x10\x02\x12\x0c\n\x08\x43\x41NCELED\x10\x03\x12\x0f\n\x0bREPLACEMENT\x10\x05\x12\x0e\n\nDUPLICATED\x10\x06\x12\x0b\n\x07\x44\x45LETED\x10\x07\x12\x07\n\x03NEW\x10\x08*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xa3\x02\n\x11VehicleDescriptor\x12\n\n\x02id\x18\x01 \x01(\t\x12\r\n\x05label\x18\x02 \x01(\t\x12\x15\n\rlicense_plate\x18\x03 \x01(\t\x12\x61\n\x15wheelchair_accessible\x18\x04 \x01(\x0e\x32\x38.transit_realtime.VehicleDescriptor.WheelchairAccessible:\x08NO_VALUE\"i\n\x14WheelchairAccessible\x12\x0c\n\x08NO_VALUE\x10\x00\x12\x0b\n\x07UNKNOWN\x10\x01\x12\x19\n\x15WHEELCHAIR_ACCESSIBLE\x10\x02\x12\x1b\n\x17WHEELCHAIR_INACCESSIBLE\x10\x03*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xb0\x01\n\x0e\x45ntitySelector\x12\x11\n\tagency_id\x18\x01 \x01(\t\x12\x10\n\x08route_id\x18\x02 \x01(\t\x12\x12\n\nroute_type\x18\x03 \x01(\x05\x12.\n\x04trip\x18\x04 \x01(\x0b\x32 .transit_realtime.TripDescriptor\x12\x0f\n\x07stop_id\x18\x05 \x01(\t\x12\x14\n\x0c\x64irection_id\x18\x06 \x01(\r*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xa6\x01\n\x10TranslatedString\x12\x43\n\x0btranslation\x18\x01 \x03(\x0b\x32..transit_realtime.TranslatedString.Translation\x1a=\n\x0bTranslation\x12\x0c\n\x04text\x18\x01 \x02(\t\x12\x10\n\x08language\x18\x02 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xc1\x01\n\x0fTranslatedImage\x12I\n\x0flocalized_image\x18\x01 \x03(\x0b\x32\x30.transit_realtime.TranslatedImage.LocalizedImage\x1aS\n\x0eLocalizedImage\x12\x0b\n\x03url\x18\x01 \x02(\t\x12\x12\n\nmedia_type\x18\x02 \x02(\t\x12\x10\n\x08language\x18\x03 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"C\n\x05Shape\x12\x10\n\x08shape_id\x18\x01 \x01(\t\x12\x18\n\x10\x65ncoded_polyline\x18\x02 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\x84\x05\n\x04Stop\x12\x0f\n\x07stop_id\x18\x01 \x01(\t\x12\x35\n\tstop_code\x18\x02 \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x35\n\tstop_name\x18\x03 \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x39\n\rtts_stop_name\x18\x04 \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x35\n\tstop_desc\x18\x05 \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x10\n\x08stop_lat\x18\x06 \x01(\x02\x12\x10\n\x08stop_lon\x18\x07 \x01(\x02\x12\x0f\n\x07zone_id\x18\x08 \x01(\t\x12\x34\n\x08stop_url\x18\t \x01(\x0b\x32\".transit_realtime.TranslatedString\x12\x16\n\x0eparent_station\x18\x0b \x01(\t\x12\x15\n\rstop_timezone\x18\x0c \x01(\t\x12O\n\x13wheelchair_boarding\x18\r \x01(\x0e\x32).transit_realtime.Stop.WheelchairBoarding:\x07UNKNOWN\x12\x10\n\x08level_id\x18\x0e \x01(\t\x12\x39\n\rplatform_code\x18\x0f \x01(\x0b\x32\".transit_realtime.TranslatedString\"C\n\x12WheelchairBoarding\x12\x0b\n\x07UNKNOWN\x10\x00\x12\r\n\tAVAILABLE\x10\x01\x12\x11\n\rNOT_AVAILABLE\x10\x02*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"\xdf\x04\n\x11TripModifications\x12I\n\x0eselected_trips\x18\x01 \x03(\x0b\x32\x31.transit_realtime.TripModifications.SelectedTrips\x12\x13\n\x0bstart_times\x18\x02 \x03(\t\x12\x15\n\rservice_dates\x18\x03 \x03(\t\x12G\n\rmodifications\x18\x04 \x03(\x0b\x32\x30.transit_realtime.TripModifications.Modification\x1a\xb4\x02\n\x0cModification\x12;\n\x13start_stop_selector\x18\x01 \x01(\x0b\x32\x1e.transit_realtime.StopSelector\x12\x39\n\x11\x65nd_stop_selector\x18\x02 \x01(\x0b\x32\x1e.transit_realtime.StopSelector\x12(\n\x1dpropagated_modification_delay\x18\x03 \x01(\x05:\x01\x30\x12<\n\x11replacement_stops\x18\x04 \x03(\x0b\x32!.transit_realtime.ReplacementStop\x12\x18\n\x10service_alert_id\x18\x05 \x01(\t\x12\x1a\n\x12last_modified_time\x18\x06 \x01(\x04*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\x1a\x43\n\rSelectedTrips\x12\x10\n\x08trip_ids\x18\x01 \x03(\t\x12\x10\n\x08shape_id\x18\x02 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"F\n\x0cStopSelector\x12\x15\n\rstop_sequence\x18\x01 \x01(\r\x12\x0f\n\x07stop_id\x18\x02 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90N\"O\n\x0fReplacementStop\x12\x1b\n\x13travel_time_to_stop\x18\x01 \x01(\x05\x12\x0f\n\x07stop_id\x18\x02 \x01(\t*\x06\x08\xe8\x07\x10\xd0\x0f*\x06\x08\xa8\x46\x10\x90NB\x1d\n\x1b\x63om.google.transit.realtime')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'gtfs_realtime_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
  _globals['DESCRIPTOR']._loaded_options = None
  _globals['DESCRIPTOR']._serialized_options = b'\n\033com.google.transit.realtime'
  _globals['_ALERT'].fields_by_name['active_period']._loaded_options = None
  _globals['_ALERT'].fields_by_name['active_period']._serialized_options = b'\030\001'
  _globals['_TRIPDESCRIPTOR_SCHEDULERELATIONSHIP'].values_by_name["ADDED"]._loaded_options = None
  _globals['_TRIPDESCRIPTOR_SCHEDULERELATIONSHIP'].values_by_name["ADDED"]._serialized_options = b'\010\001'
  _globals['_FEEDMESSAGE']._serialized_start=41
  _globals['_FEEDMESSAGE']._serialized_end=162
  _globals['_FEEDHEADER']._serialized_start=165
  _globals['_FEEDHEADER']._serialized_end=402
  _globals['_FEEDHEADER_INCREMENTALITY']._serialized_start=334
  _globals['_FEEDHEADER_INCREMENTALITY']._serialized_end=386
  _globals['_FEEDENTITY']._serialized_start=405
  _globals['_FEEDENTITY']._serialized_end=758
  _globals['_TRIPUPDATE']._serialized_start=761
  _globals['_TRIPUPDATE']._serialized_end=2287
  _globals['_TRIPUPDATE_STOPTIMEEVENT']._serialized_start=1052
  _globals['_TRIPUPDATE_STOPTIMEEVENT']._serialized_end=1157
  _globals['_TRIPUPDATE_STOPTIMEUPDATE']._serialized_start=1160
  _globals['_TRIPUPDATE_STOPTIMEUPDATE']._serialized_end=2113
  _globals['_TRIPUPDATE_STOPTIMEUPDATE_STOPTIMEPROPERTIES']._serialized_start=1632
  _globals['_TRIPUPDATE_STOPTIMEUPDATE_STOPTIMEPROPERTIES']._serialized_end=2015
  _globals['_TRIPUPDATE_STOPTIMEUPDATE_STOPTIMEPROPERTIES_DROPOFFPICKUPTYPE']._serialized_start=1911
  _globals['_TRIPUPDATE_STOPTIMEUPDATE_STOPTIMEPROPERTIES_DROPOFFPICKUPTYPE']._serialized_end=1999
  _globals['_TRIPUPDATE_STOPTIMEUPDATE_SCHEDULERELATIONSHIP']._serialized_start=2017
  _globals['_TRIPUPDATE_STOPTIMEUPDATE_SCHEDULERELATIONSHIP']._serialized_end=2097
  _globals['_TRIPUPDATE_TRIPPROPERTIES']._serialized_start=2116
  _globals['_TRIPUPDATE_TRIPPROPERTIES']._serialized_end=2271
  _globals['_VEHICLEPOSITION']._serialized_start=2290
  _globals['_VEHICLEPOSITION']._serialized_end=3537
  _globals['_VEHICLEPOSITION_CARRIAGEDETAILS']._serialized_start=2884
  _globals['_VEHICLEPOSITION_CARRIAGEDETAILS']._serialized_end=3101
  _globals['_VEHICLEPOSITION_VEHICLESTOPSTATUS']._serialized_start=3103
  _globals['_VEHICLEPOSITION_VEHICLESTOPSTATUS']._serialized_end=3174
  _globals['_VEHICLEPOSITION_CONGESTIONLEVEL']._serialized_start=3176
  _globals['_VEHICLEPOSITION_CONGESTIONLEVEL']._serialized_end=3301
  _globals['_VEHICLEPOSITION_OCCUPANCYSTATUS']._serialized_start=3304
  _globals['_VEHICLEPOSITION_OCCUPANCYSTATUS']._serialized_end=3521
  _globals['_ALERT']._serialized_start=3540
  _globals['_ALERT']._serialized_end=5061
  _globals['_ALERT_CAUSE']._serialized_start=4512
  _globals['_ALERT_CAUSE']._serialized_end=4747
  _globals['_ALERT_EFFECT']._serialized_start=4750
  _globals['_ALERT_EFFECT']._serialized_end=4971
  _globals['_ALERT_SEVERITYLEVEL']._serialized_start=4973
  _globals['_ALERT_SEVERITYLEVEL']._serialized_end=5045
  _globals['_TIMERANGE']._serialized_start=5063
  _globals['_TIMERANGE']._serialized_end=5118
  _globals['_POSITION']._serialized_start=5120
  _globals['_POSITION']._serialized_end=5233
  _globals['_TRIPDESCRIPTOR']._serialized_start=5236
  _globals['_TRIPDESCRIPTOR']._serialized_end=5803
  _globals['_TRIPDESCRIPTOR_MODIFIEDTRIPSELECTOR']._serialized_start=5516
  _globals['_TRIPDESCRIPTOR_MODIFIEDTRIPSELECTOR']._serialized_end=5646
  _globals['_TRIPDESCRIPTOR_SCHEDULERELATIONSHIP']._serialized_start=5649
  _globals['_TRIPDESCRIPTOR_SCHEDULERELATIONSHIP']._serialized_end=5787
  _globals['_VEHICLEDESCRIPTOR']._serialized_start=5806
  _globals['_VEHICLEDESCRIPTOR']._serialized_end=6097
  _globals['_VEHICLEDESCRIPTOR_WHEELCHAIRACCESSIBLE']._serialized_start=5976
  _globals['_VEHICLEDESCRIPTOR_WHEELCHAIRACCESSIBLE']._serialized_end=6081
  _globals['_ENTITYSELECTOR']._serialized_start=6100
  _globals['_ENTITYSELECTOR']._serialized_end=6276
  _globals['_TRANSLATEDSTRING']._serialized_start=6279
  _globals['_TRANSLATEDSTRING']._serialized_end=6445
  _globals['_TRANSLATEDSTRING_TRANSLATION']._serialized_start=6368
  _globals['_TRANSLATEDSTRING_TRANSLATION']._serialized_end=6429
  _globals['_TRANSLATEDIMAGE']._serialized_start=6448
  _globals['_TRANSLATEDIMAGE']._serialized_end=6641
  _globals['_TRANSLATEDIMAGE_LOCALIZEDIMAGE']._serialized_start=6542
  _globals['_TRANSLATEDIMAGE_LOCALIZEDIMAGE']._serialized_end=6625
  _globals['_SHAPE']._serialized_start=6643
  _globals['_SHAPE']._serialized_end=6710
  _globals['_STOP']._serialized_start=6713
  _globals['_STOP']._serialized_end=7357
  _globals['_STOP_WHEELCHAIRBOARDING']._serialized_start=7274
  _globals['_STOP_WHEELCHAIRBOARDING']._serialized_end=7341
  _globals['_TRIPMODIFICATIONS']._serialized_start=7360
  _globals['_TRIPMODIFICATIONS']._serialized_end=7967
  _globals['_TRIPMODIFICATIONS_MODIFICATION']._serialized_start=7574
  _globals['_TRIPMODIFICATIONS_MODIFICATION']._serialized_end=7882
  _globals['_TRIPMODIFICATIONS_SELECTEDTRIPS']._serialized_start=7884
  _globals['_TRIPMODIFICATIONS_SELECTEDTRIPS']._serialized_end=7951
  _globals['_STOPSELECTOR']._serialized_start=7969
  _globals['_STOPSELECTOR']._serialized_end=8039
  _globals['_REPLACEMENTSTOP']._serialized_start=8041
  _globals['_REPLACEMENTSTOP']._serialized_end=8120
# @@protoc_insertion_point(module_scope)

"""ArcGIS Hub catalog discovery and normalization."""

from __future__ import annotations

import urllib.parse
from typing import Any

from raleighlib import core


HUB_SEARCH_URL = "https://data.raleighnc.gov/api/search/v1/collections/{collection}/items"


class CatalogError(Exception):
    """Raised for catalog discovery or resolution failures."""


def fetch_collection(collection: str, start_index: int = 1, num: int = 100) -> dict[str, Any]:
    """Fetch a page from a Hub search collection."""
    if start_index < 1:
        raise ValueError("start_index must be positive")
    core.require_positive_limit(num)
    url = HUB_SEARCH_URL.format(collection=collection)
    params = {"limit": num, "startindex": start_index}
    url = f"{url}?{urllib.parse.urlencode(params)}"
    return core.json_request(url)


def fetch_all_records(
    collection: str,
    max_records: int | None = None,
    max_pages: int = 100,
) -> list[dict[str, Any]]:
    """Paginate through a Hub collection and return all records."""
    if max_pages < 1:
        raise ValueError("max_pages must be at least 1")
    core.require_positive_limit(max_records, allow_none=True)
    records: list[dict[str, Any]] = []
    start_index = 1
    page_size = 100
    seen_pages: set[tuple[str, ...]] = set()
    expected_total: int | None = None
    for _page_number in range(max_pages):
        page = fetch_collection(collection, start_index=start_index, num=page_size)
        if not isinstance(page, dict):
            raise CatalogError(f"Hub collection {collection!r} returned a non-object page")
        matched = page.get("numberMatched")
        if matched is not None:
            if isinstance(matched, bool) or not isinstance(matched, int) or matched < 0:
                raise CatalogError(f"Hub collection {collection!r} returned invalid numberMatched")
            if expected_total is None:
                expected_total = matched
            elif matched != expected_total:
                raise CatalogError(f"Hub collection {collection!r} changed numberMatched")
        features = page.get("features", [])
        if not isinstance(features, list):
            raise CatalogError(f"Hub collection {collection!r} returned invalid features")
        if not features:
            if expected_total is not None and len(records) < expected_total:
                raise CatalogError(f"Hub collection {collection!r} ended before numberMatched")
            return records
        if any(not isinstance(feature, dict) for feature in features):
            raise CatalogError(f"Hub collection {collection!r} returned a non-object feature")
        page_ids: list[str] = []
        for feature in features:
            properties = feature.get("properties", {})
            if not isinstance(properties, dict):
                raise CatalogError(
                    f"Hub collection {collection!r} returned invalid feature properties"
                )
            page_ids.append(str(feature.get("id") or properties.get("id") or ""))
        page_key = tuple(page_ids)
        if any(not item for item in page_key):
            raise CatalogError(f"Hub collection {collection!r} returned a feature without an id")
        if page_key in seen_pages:
            raise CatalogError(f"Hub collection {collection!r} repeated a page")
        seen_pages.add(page_key)
        records.extend(features)
        if expected_total is not None and len(records) > expected_total:
            raise CatalogError(f"Hub collection {collection!r} exceeded numberMatched")
        start_index += len(features)
        if matched is not None and start_index > matched:
            return records
        if max_records is not None and len(records) >= max_records:
            return records[:max_records]
        # Only break on partial page when the server did not report a total.
        if matched is None and len(features) < page_size:
            return records
    raise CatalogError(f"Hub collection {collection!r} exceeded {max_pages} pages")


def _first(props: dict[str, Any], *keys: str) -> Any:
    for key in keys:
        if key in props and props[key] not in (None, ""):
            return props[key]
    return None


def _normalize_type(record_type: str | None) -> str:
    if not record_type:
        return "Unknown"
    record_type = record_type.strip()
    mapping = {
        "Feature Service": "FeatureServer",
        "Map Service": "MapServer",
        "Image Service": "ImageServer",
        "Table": "Table",
    }
    return mapping.get(record_type, record_type)


def normalize_record(record: dict[str, Any]) -> dict[str, Any]:
    """Normalize a Hub search record into a stable catalog item."""
    props = record.get("properties", {})
    record_type = _normalize_type(props.get("type") or record.get("type"))
    raw_url = props.get("url") or props.get("serviceUrl") or ""

    # ImageServer URLs are root services; never append a layer suffix.
    url = raw_url
    if record_type == "ImageServer" and url.endswith("/0"):
        url = url[:-2]

    tags = props.get("tags") or []
    categories = props.get("categories") or []
    if isinstance(categories, str):
        categories = [categories]
    category = categories[0] if categories else "Other"

    return {
        "id": record.get("id", ""),
        "title": _first(props, "title", "name") or "",
        "description": props.get("description") or "",
        "type": record_type,
        "url": url,
        "tags": tags,
        "categories": categories,
        "category": category,
        "owner": props.get("owner") or props.get("source") or "",
        "access": str(props.get("access") or "").lower(),
        "license": props.get("license") or "",
        "extent": props.get("extent"),
        "has_geometry": None,  # Resolved from layer metadata at query time.
        "created": props.get("created"),
        "modified": props.get("modified"),
    }


def fetch_catalog(max_records: int | None = None) -> list[dict[str, Any]]:
    """Return the live catalog across all curated collections."""
    core.require_positive_limit(max_records, allow_none=True)
    collections = ["dataset", "document", "appAndMap"]
    catalog: list[dict[str, Any]] = []
    for collection in collections:
        records = fetch_all_records(collection, max_records=max_records)
        normalized = (normalize_record(r) for r in records)
        catalog.extend(item for item in normalized if item.get("access") == "public")
    return catalog


def catalog_from_cache_or_live(
    max_age_seconds: int = 3600,
    allow_stale: bool = True,
) -> list[dict[str, Any]]:
    """Return cached catalog if fresh; otherwise fetch live and cache.

    On transient failure, fall back to a stale cache entry and label it.
    """
    cached = core.read_cache("hub-catalog.json", max_age_seconds=max_age_seconds)
    if cached is not None:
        return [item for item in cached if item.get("access") == "public"]

    try:
        catalog = fetch_catalog()
    except Exception as exc:
        if not allow_stale:
            raise CatalogError(f"Catalog refresh failed and stale cache is disabled: {exc}") from exc
        stale = core.read_cache("hub-catalog.json", max_age_seconds=None)
        if stale is None:
            raise CatalogError(f"Catalog refresh failed and no stale cache is available: {exc}") from exc
        for item in stale:
            item["_stale"] = True
            item["_stale_reason"] = str(exc)
        return [item for item in stale if item.get("access") == "public"]

    core.write_cache("hub-catalog.json", catalog)
    return catalog


def search_catalog(
    query: str,
    catalog: list[dict[str, Any]] | None = None,
    limit: int = 20,
) -> list[dict[str, Any]]:
    """Search catalog records by title, description, tags, and categories."""
    core.require_positive_limit(limit)
    if catalog is None:
        catalog = catalog_from_cache_or_live()
    terms = query.lower().split()
    scored: list[tuple[int, dict[str, Any]]] = []
    for item in catalog:
        if item.get("access") != "public":
            continue
        text = " ".join(
            str(x)
            for x in [
                item.get("title", ""),
                item.get("description", ""),
                " ".join(item.get("tags", [])),
                " ".join(item.get("categories", [])),
            ]
        ).lower()
        if all(term in text for term in terms):
            score = 0
            title_lower = item.get("title", "").lower()
            if query.lower() in title_lower:
                score += 10
            if any(term in title_lower for term in terms):
                score += 5
            scored.append((score, item))
    scored.sort(key=lambda x: (x[0], x[1].get("title", "").lower()), reverse=True)
    return [item for _, item in scored[:limit]]


def resolve_item(
    identifier: str,
    catalog: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Resolve a catalog item by stable ID or exact title."""
    if catalog is None:
        catalog = catalog_from_cache_or_live()
    catalog = [item for item in catalog if item.get("access") == "public"]
    by_id: dict[str, dict[str, Any]] = {}
    by_title: dict[str, dict[str, Any]] = {}
    for item in catalog:
        by_id[item.get("id", "")] = item
        by_title[item.get("title", "").lower()] = item

    if identifier in by_id:
        return by_id[identifier]

    key = identifier.lower()
    exact_matches = [item for item in catalog if item.get("title", "").lower() == key]
    if len(exact_matches) == 1:
        return exact_matches[0]
    if len(exact_matches) > 1:
        raise CatalogError(
            f"'{identifier}' matches multiple catalog items; supply the stable item ID"
        )

    # Fuzzy title match as fallback.
    matches = [item for item in catalog if key in item.get("title", "").lower()]
    if len(matches) == 1:
        return matches[0]
    if len(matches) > 1:
        titles = [m.get("title", "") for m in matches]
        raise CatalogError(
            f"'{identifier}' is ambiguous; matches: {', '.join(titles[:5])}"
        )
    raise CatalogError(f"'{identifier}' was not found in the catalog")

"""ArcGIS ImageServer imagery adapter."""

from __future__ import annotations

import json
import urllib.parse
from typing import Any

from raleighlib import core


IMAGE_ROOT = "https://maps.raleighnc.gov/images/rest/services"
MAX_IMAGE_FOLDERS = 50
MAX_IMAGE_SERVICES = 1000


class CapabilityError(Exception):
    """Raised when an operation is requested that the service does not support."""


def _checked_json(data: Any, operation: str) -> dict[str, Any]:
    data = core.require_object(data, operation)
    core.raise_for_arcgis_error(data, operation)
    return data


def _is_token_required_error(exc: Exception) -> bool:
    """Return True when an ArcGIS error indicates the resource needs a token."""
    return "token required" in str(exc).lower()


def list_services(
    root_url: str = IMAGE_ROOT,
    max_folders: int = MAX_IMAGE_FOLDERS,
    max_services: int = MAX_IMAGE_SERVICES,
) -> tuple[list[dict[str, Any]], list[str]]:
    """Recursively discover ImageServer services from the REST directory.

    Returns ``(services, restricted_folders)`` where ``restricted_folders``
    names folders whose listing requires a token. Those folders are skipped:
    this tool only reads publicly accessible services, and a token-gated folder
    is not a public-data contract violation.
    """
    if max_folders < 0 or max_services < 1:
        raise ValueError("imagery discovery bounds are invalid")
    sep = "&" if "?" in root_url else "?"
    data = _checked_json(
        core.json_request(f"{root_url}{sep}f=pjson"), "Image service listing"
    )
    root_services = data.get("services", [])
    folders = data.get("folders", [])
    if not isinstance(root_services, list) or any(
        not isinstance(service, dict) for service in root_services
    ):
        raise CapabilityError("Image service listing returned invalid services")
    if not isinstance(folders, list) or any(
        not isinstance(folder, str) or not folder.strip() for folder in folders
    ):
        raise CapabilityError("Image service listing returned invalid folders")
    if len(folders) > max_folders:
        raise CapabilityError(
            f"Image service listing exceeded {max_folders} folders"
        )
    if len(root_services) > max_services:
        raise CapabilityError(
            f"Image service listing exceeded {max_services} services"
        )
    services = list(root_services)
    restricted_folders: list[str] = []
    for folder in folders:
        folder_url = f"{root_url}/{urllib.parse.quote(folder, safe='')}"
        sep = "&" if "?" in folder_url else "?"
        try:
            folder_data = _checked_json(
                core.json_request(f"{folder_url}{sep}f=pjson"), "Image folder listing"
            )
        except ValueError as exc:
            if _is_token_required_error(exc):
                restricted_folders.append(folder)
                continue
            raise
        folder_services = folder_data.get("services", [])
        if not isinstance(folder_services, list) or any(
            not isinstance(service, dict) for service in folder_services
        ):
            raise CapabilityError("Image folder listing returned invalid services")
        if len(services) + len(folder_services) > max_services:
            raise CapabilityError(
                f"Image service listing exceeded {max_services} services"
            )
        for svc in folder_services:
            svc = dict(svc)
            svc["folder"] = folder
            services.append(svc)
    return services, restricted_folders


def service_info(url: str) -> dict[str, Any]:
    """Fetch ImageServer service metadata."""
    sep = "&" if "?" in url else "?"
    return _checked_json(
        core.json_request(f"{url}{sep}f=pjson"), "Image service metadata"
    )


def supports_capability(info: dict[str, Any], capability: str) -> bool:
    """Return True if the service info advertises the given capability."""
    caps = info.get("capabilities", "")
    return capability.lower() in [c.strip().lower() for c in str(caps).split(",")]


def _bbox_str(bbox: tuple[float, float, float, float]) -> str:
    return ",".join(str(round(c, 6)) for c in bbox)


def export_image(
    url: str,
    bbox: tuple[float, float, float, float],
    size: tuple[int, int] | None = None,
    format_: str = "jpgpng",
    in_sr: int = 4326,
    out_sr: int = 4326,
    **kwargs: Any,
) -> bytes:
    """Export a bounded image from an ImageServer."""
    info = service_info(url)
    if not supports_capability(info, "Image"):
        raise CapabilityError(f"{url} does not advertise Image capability")
    max_width = info.get("maxImageWidth", 4000)
    max_height = info.get("maxImageHeight", 4000)
    if size:
        if size[0] > max_width or size[1] > max_height:
            raise CapabilityError(
                f"Requested image size {size} exceeds server maximum {max_width}x{max_height}"
            )
    base = url.rstrip("/") + "/exportImage"
    params: dict[str, Any] = {
        "bbox": _bbox_str(bbox),
        "bboxSR": in_sr,
        "imageSR": out_sr,
        "format": format_,
        "f": "image",
    }
    if size:
        params["size"] = f"{size[0]},{size[1]}"
    for key, value in kwargs.items():
        params[key] = value
    full_url = f"{base}?{urllib.parse.urlencode(params)}"
    body = core.raw_request(full_url)
    if body.lstrip().startswith(b"{"):
        try:
            payload = json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CapabilityError("Image export returned malformed JSON instead of an image") from exc
        if isinstance(payload, dict):
            core.raise_for_arcgis_error(payload, "Image export")
        raise CapabilityError("Image export returned JSON instead of an image")
    image_signatures = (b"\x89PNG\r\n\x1a\n", b"\xff\xd8\xff", b"GIF87a", b"GIF89a", b"II*\x00", b"MM\x00*", b"BM")
    if not any(body.startswith(signature) for signature in image_signatures):
        raise CapabilityError("Image export returned an unrecognized non-image response")
    return body


def identify(
    url: str,
    point: tuple[float, float],
    in_sr: int = 4326,
    out_sr: int = 4326,
    **kwargs: Any,
) -> dict[str, Any]:
    """Identify pixel value at a point on an ImageServer."""
    info = service_info(url)
    if not supports_capability(info, "Image"):
        raise CapabilityError(f"{url} does not advertise Image capability")
    base = url.rstrip("/") + "/identify"
    params: dict[str, Any] = {
        "geometry": json.dumps({"x": point[0], "y": point[1]}),
        "geometryType": "esriGeometryPoint",
        "inSR": in_sr,
        "outSR": out_sr,
        "f": "json",
    }
    for key, value in kwargs.items():
        params[key] = value
    full_url = f"{base}?{urllib.parse.urlencode(params)}"
    return _checked_json(core.json_request(full_url), "Image identify")


def compute_statistics(
    url: str,
    bbox: tuple[float, float, float, float],
    in_sr: int = 4326,
    out_sr: int = 4326,
    **kwargs: Any,
) -> dict[str, Any]:
    """Compute statistics for an extent on an ImageServer."""
    info = service_info(url)
    if not supports_capability(info, "Image"):
        raise CapabilityError(f"{url} does not advertise Image capability")
    base = url.rstrip("/") + "/computeStatisticsHistograms"
    params: dict[str, Any] = {
        "geometry": json.dumps(
            {
                "xmin": bbox[0],
                "ymin": bbox[1],
                "xmax": bbox[2],
                "ymax": bbox[3],
            }
        ),
        "geometryType": "esriGeometryEnvelope",
        "inSR": in_sr,
        "outSR": out_sr,
        "f": "json",
    }
    for key, value in kwargs.items():
        params[key] = value
    full_url = f"{base}?{urllib.parse.urlencode(params)}"
    return _checked_json(core.json_request(full_url), "Image statistics")

"""Read-only adapter for the Raleigh-Wake Emergency Communications Center active incident feed.

This is an UNDOCUMENTED public application endpoint, not a versioned API.
The adapter is isolated and can be disabled independently via
RALEIGH_DISABLE_INCIDENTS=1.
"""

from __future__ import annotations

import os
from datetime import datetime, timezone
from typing import Any

from raleighlib import core


FEED_URL = "https://incidents.rwecc.com/getdata"

CACHE_KEY = "incidents-rwecc-active.json"
CACHE_TTL_SECONDS = 90

SOURCE_LABEL = (
    "Filtered active public incident feed from incidents.rwecc.com "
    "(undocumented endpoint). This is NOT all 911 calls and NOT "
    "authoritative emergency status."
)

KNOWN_AGENCIES = frozenset({
    "raleigh police department",
    "raleigh fire department",
    "raleigh police",
    "raleigh fire",
})


class IncidentFeedError(ValueError):
    """Raised when the incident feed is unavailable, malformed, or drifted."""


def _ensure_enabled() -> None:
    value = os.environ.get("RALEIGH_DISABLE_INCIDENTS", "").strip().casefold()
    if value in {"1", "true", "yes", "on"}:
        raise IncidentFeedError(
            "Incident feed adapter is disabled by RALEIGH_DISABLE_INCIDENTS"
        )


def _validate_record(record: Any, index: int) -> dict[str, Any] | None:
    if not isinstance(record, dict):
        return None
    jurisdiction = record.get("jurisdiction")
    problem = record.get("problem")
    address = record.get("address")
    timestamp = record.get("timestamp")
    if not isinstance(jurisdiction, str) or not jurisdiction.strip():
        return None
    if not isinstance(problem, str) or not problem.strip():
        return None
    lat = record.get("lat")
    lon = record.get("long")
    if not isinstance(lat, (int, float)) or isinstance(lat, bool):
        lat = None
    if not isinstance(lon, (int, float)) or isinstance(lon, bool):
        lon = None
    if lat is not None and not (-90 <= lat <= 90):
        lat = None
    if lon is not None and not (-180 <= lon <= 180):
        lon = None
    return {
        "jurisdiction": jurisdiction.strip(),
        "problem": problem.strip(),
        "address": address.strip() if isinstance(address, str) else None,
        "lat": lat,
        "long": lon,
        "timestamp": timestamp if isinstance(timestamp, str) else None,
    }


def fetch_active(
    agency: str | None = None,
    incident_type: str | None = None,
    limit: int = 50,
    use_cache: bool = True,
) -> dict[str, Any]:
    """Fetch currently active incidents from the RWECC public feed.

    Returns a dict with ``incidents``, ``retrieved_at``, ``source``, and
    ``warnings`` keys.
    """
    _ensure_enabled()
    core.require_positive_limit(limit)

    warnings: list[str] = []

    if use_cache:
        cached = core.read_cache(CACHE_KEY, max_age_seconds=CACHE_TTL_SECONDS)
        if cached is not None and isinstance(cached, list):
            raw = cached
        else:
            raw = _fetch_raw()
    else:
        raw = _fetch_raw()

    if not isinstance(raw, list):
        raise IncidentFeedError(
            "Feed schema drift: expected a JSON list of incident records"
        )

    if len(raw) == 0:
        warnings.append(
            "Feed returned zero records. This does not prove no incidents "
            "are active; the feed may be stale or filtered upstream."
        )

    incidents: list[dict[str, Any]] = []
    skipped = 0
    seen_keys: set[str] = set()
    for i, record in enumerate(raw):
        validated = _validate_record(record, i)
        if validated is None:
            skipped += 1
            continue
        dedup_key = "|".join([
            validated["jurisdiction"].casefold(),
            validated["problem"].casefold(),
            str(validated.get("address") or "").casefold(),
            str(validated.get("timestamp") or ""),
        ])
        if dedup_key in seen_keys:
            continue
        seen_keys.add(dedup_key)
        incidents.append(validated)

    if skipped > 0:
        warnings.append(
            f"{skipped} record(s) skipped due to schema drift or missing required fields."
        )

    if agency:
        agency_lower = agency.strip().casefold().replace("-", " ")
        if agency_lower not in KNOWN_AGENCIES:
            warnings.append(
                f"Agency '{agency}' is not a known agency. "
                f"Known agencies: {', '.join(sorted(KNOWN_AGENCIES))}. "
                "Filtering by substring match anyway."
            )
        incidents = [
            r for r in incidents
            if agency_lower in r["jurisdiction"].casefold()
        ]

    if incident_type:
        type_lower = incident_type.strip().casefold()
        incidents = [
            r for r in incidents
            if type_lower in r["problem"].casefold()
        ]

    incidents = incidents[:limit]

    return {
        "incidents": incidents,
        "retrieved_at": datetime.now(timezone.utc).isoformat(),
        "source": SOURCE_LABEL,
        "warnings": warnings,
    }


def _fetch_raw() -> list[Any]:
    import json as _json
    try:
        body = core.raw_request(FEED_URL, max_bytes=2 * 1024 * 1024)
    except core.SecurityError:
        raise
    except Exception as exc:
        raise IncidentFeedError(
            f"Incident feed unavailable: {exc}"
        ) from exc
    if not body:
        return []
    try:
        data = _json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, _json.JSONDecodeError) as exc:
        raise IncidentFeedError(
            f"Incident feed returned malformed JSON: {exc}"
        ) from exc
    if isinstance(data, dict) and not data:
        return []
    if isinstance(data, dict):
        raise IncidentFeedError(
            "Feed schema drift: expected a JSON list, got an object. "
            "The undocumented endpoint contract may have changed."
        )
    if isinstance(data, list):
        core.write_cache(CACHE_KEY, data)
        return data
    raise IncidentFeedError(
        "Feed schema drift: unexpected response type"
    )

"""Read-only eSCRIBE public-meetings adapter."""

from __future__ import annotations

import html
import json
import re
import urllib.parse
from collections.abc import Iterator
from datetime import date, datetime
from pathlib import Path
from typing import Any

from raleighlib import core


BASE_URL = "https://pub-raleighnc.escribemeetings.com"
MEETING_VIEW = f"{BASE_URL}/?MeetingViewId=2"
PAST_MEETINGS_URL = f"{BASE_URL}/MeetingsCalendarView.aspx/PastMeetings"
MAX_PAST_PAGES_PER_TYPE = 50
MAX_PAST_MEETING_TYPES = 25
MAX_PAST_REQUESTS = 100


class CompatibilityError(Exception):
    """Raised when the eSCRIBE page structure has changed and cannot be parsed."""


def _absolute_url(path: str) -> str:
    if path.startswith("http://") or path.startswith("https://"):
        return path
    return urllib.parse.urljoin(BASE_URL, path)


def _fetch_html(url: str) -> str:
    if not core.is_allowed_host(url):
        raise core.SecurityError(f"URL host is not allowlisted: {url}")
    data = core.raw_request(url)
    return data.decode("utf-8", errors="replace")


def _extract_meeting_rows(html_text: str) -> list[dict[str, Any]]:
    """Extract meeting rows from listing HTML.

    eSCRIBE lists meetings with links like:
      Meeting.aspx?Id=<UUID>&lang=English
    The surrounding aria-label contains the meeting title and date.
    """
    rows: dict[str, dict[str, Any]] = {}
    for match in re.finditer(
        r"Meeting\.aspx\?Id=([0-9a-fA-F-]+)&lang=English",
        html_text,
    ):
        meeting_id = match.group(1).lower()
        if meeting_id in rows:
            continue
        # Read the label from this link's opening tag, not an earlier nearby
        # link. Historical pages contain many adjacent meeting anchors.
        start = html_text.rfind("<a", max(0, match.start() - 1000), match.start())
        end = html_text.find(">", match.end())
        opening_tag = html_text[start : end + 1] if start >= 0 and end >= 0 else ""
        label_match = re.search(r'aria-label="([^"]+)"', opening_tag)
        label = html.unescape(label_match.group(1)) if label_match else ""

        # Labels are like "Share <Title> <Weekday>, <Month> <Day>, <Year> @ <Time>"
        # or "Public Comment for <Title> <Weekday>, ...".
        title = label
        date_text = ""
        date_match = re.search(
            r"(Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday),\s+[^@]+@\s+\d{1,2}:\d{2}\s+(?:AM|PM)",
            label,
        )
        if date_match:
            date_text = date_match.group(0)
            title = label[: date_match.start()].strip()
            # Strip common prefixes.
            for prefix in ("Share ", "Public Comment for "):
                if title.startswith(prefix):
                    title = title[len(prefix) :].strip()

        body = ""
        body_match = re.search(
            r"(City Council|Board of Commissioners|Planning Commission|City Council Meeting|Boards and Commissions)",
            title,
            re.IGNORECASE,
        )
        if body_match:
            body = body_match.group(1)

        rows[meeting_id] = {
            "id": meeting_id,
            "title": title,
            "date": date_text,
            "body": body,
            "url": f"{BASE_URL}/Meeting.aspx?Id={meeting_id}&lang=English",
        }
    if "Meeting.aspx" in html_text and not any(
        r.get("title") or r.get("date") for r in rows.values()
    ):
        raise CompatibilityError(
            "eSCRIBE meeting page contains Meeting.aspx links but no semantic meetings could be parsed"
        )
    return list(rows.values())


def _date_from_row(row: dict[str, Any]) -> date | None:
    value = row.get("date", "")
    try:
        return datetime.strptime(value, "%A, %B %d, %Y @ %I:%M %p").date()
    except (TypeError, ValueError):
        return None


def _meeting_types(html_text: str) -> list[str]:
    """Extract unique public meeting-type names from the listing page."""
    found: list[str] = []
    for encoded in re.findall(r'MeetingType="([^"]+)"', html_text):
        meeting_type = html.unescape(encoded).strip()
        if meeting_type and meeting_type not in found:
            found.append(meeting_type)
    return found


def _normalize_past_meeting(item: dict[str, Any]) -> dict[str, Any]:
    meeting_id = str(item.get("Id") or "")
    return {
        "id": meeting_id,
        "title": item.get("MeetingType") or "",
        "date": item.get("FormattedStart") or "",
        "body": item.get("MeetingType") or "",
        "location": item.get("LocationName") or "",
        "cancelled": bool(item.get("Cancelled")),
        "url": f"{BASE_URL}/Meeting.aspx?Id={meeting_id}&lang=English",
    }


def _past_meeting_items(
    year: int,
    body: str | None = None,
) -> Iterator[dict[str, Any]]:
    """Yield raw historical meeting records from the public page method."""
    listing = _fetch_html(MEETING_VIEW)
    meeting_types = _meeting_types(listing)
    if not meeting_types:
        raise CompatibilityError("eSCRIBE listing exposed no historical meeting types")
    if len(meeting_types) != len(set(meeting_types)):
        raise CompatibilityError("eSCRIBE returned duplicate meeting types")
    if len(meeting_types) > MAX_PAST_MEETING_TYPES:
        raise CompatibilityError(f"eSCRIBE returned more than {MAX_PAST_MEETING_TYPES} meeting types")
    if body:
        needle = body.casefold()
        meeting_types = [item for item in meeting_types if needle in item.casefold()]
    requests_remaining = MAX_PAST_REQUESTS
    seen_meeting_ids: set[str] = set()
    url = f"{PAST_MEETINGS_URL}?MeetingViewId=2&Year={year}"
    for meeting_type in meeting_types:
        page_number = 1
        loaded_for_type = 0
        expected_total: int | None = None
        while page_number <= MAX_PAST_PAGES_PER_TYPE:
            if requests_remaining <= 0:
                raise CompatibilityError("eSCRIBE historical request budget exhausted")
            requests_remaining -= 1
            response = core.json_request(
                url,
                method="POST",
                data=json.dumps({"type": meeting_type, "pageNumber": page_number}).encode("utf-8"),
                headers={"Content-Type": "application/json; charset=utf-8"},
            )
            response = core.require_object(response, "eSCRIBE historical meetings")
            result = response.get("d", response)
            page = result.get("Meetings", []) if isinstance(result, dict) else []
            if not isinstance(result, dict) or "TotalCount" not in result:
                raise CompatibilityError("eSCRIBE historical response omitted TotalCount")
            total_value = result["TotalCount"]
            if isinstance(total_value, bool) or not isinstance(total_value, int) or total_value < 0:
                raise CompatibilityError("eSCRIBE historical response has invalid TotalCount")
            total = total_value
            if expected_total is None:
                expected_total = total
            elif total != expected_total:
                raise CompatibilityError("eSCRIBE historical TotalCount changed during pagination")
            if not isinstance(page, list):
                raise CompatibilityError("eSCRIBE historical response has invalid Meetings")
            if not page:
                if loaded_for_type < total:
                    raise CompatibilityError("eSCRIBE historical pagination ended before TotalCount")
                break
            if loaded_for_type + len(page) > total:
                raise CompatibilityError("eSCRIBE historical Meetings exceed TotalCount")
            for item in page:
                if not isinstance(item, dict):
                    raise CompatibilityError("eSCRIBE historical response contains a non-object meeting")
                meeting_id = str(item.get("Id") or "").strip().casefold()
                if not meeting_id:
                    raise CompatibilityError("eSCRIBE historical meeting omitted Id")
                if meeting_id in seen_meeting_ids:
                    raise CompatibilityError("eSCRIBE historical response repeated a meeting Id")
                seen_meeting_ids.add(meeting_id)
                yield item
            loaded_for_type += len(page)
            if loaded_for_type >= total:
                break
            page_number += 1
        else:
            raise CompatibilityError(
                f"eSCRIBE pagination exceeded {MAX_PAST_PAGES_PER_TYPE} pages for {meeting_type}"
            )


def _fetch_past_meetings(
    year: int,
    body: str | None = None,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """Fetch normalized historical meetings from eSCRIBE's public page method."""
    rows: list[dict[str, Any]] = []
    for item in _past_meeting_items(year, body=body):
        rows.append(_normalize_past_meeting(item))
        if limit is not None and len(rows) >= limit:
            break
    return rows


def _find_past_meeting_record(meeting_id: str, year: int) -> dict[str, Any] | None:
    for item in _past_meeting_items(year):
        if str(item.get("Id") or "").casefold() == meeting_id.casefold():
            return item
    return None


def list_upcoming(
    limit: int | None = None,
    today: date | None = None,
) -> list[dict[str, Any]]:
    """List upcoming meetings, optionally limited to the first N."""
    core.require_positive_limit(limit, allow_none=True)
    html_text = _fetch_html(MEETING_VIEW)
    rows = _extract_meeting_rows(html_text)
    if not rows:
        page_is_escribe = "eSCRIBE Published Meetings" in html_text
        explicit_empty = re.search(
            r"\b(no upcoming meetings|no meetings (?:found|available|scheduled))\b",
            html_text,
            re.IGNORECASE,
        )
        if not page_is_escribe:
            raise CompatibilityError(
                "eSCRIBE upcoming page identity was not recognized"
            )
        if not explicit_empty:
            raise CompatibilityError(
                "eSCRIBE upcoming page contained no meetings or recognized empty state"
            )
    cutoff = today or date.today()
    rows = [row for row in rows if (_date_from_row(row) or date.min) >= cutoff]
    if limit is not None:
        rows = rows[:limit]
    return rows


def list_meetings(body: str | None = None, year: int | None = None, limit: int | None = None) -> list[dict[str, Any]]:
    """List meetings, optionally filtered by body and year.

    When ``year`` is supplied, fetch the historical archive for that year
    instead of the upcoming-meetings view.
    """
    core.require_positive_limit(limit, allow_none=True)
    if year is not None:
        rows = _fetch_past_meetings(year, body=body, limit=limit)
    else:
        rows = list_upcoming(limit=limit)
    if body:
        body_lower = body.lower()
        rows = [r for r in rows if body_lower in r.get("body", "").lower() or body_lower in r.get("title", "").lower()]
    if limit is not None:
        rows = rows[:limit]
    return rows


def search_meetings(
    query: str,
    body: str | None = None,
    year: int | None = None,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """Search upcoming or historical meetings by keyword."""
    core.require_positive_limit(limit, allow_none=True)
    rows = list_meetings(body=body, year=year)
    query_lower = query.lower()
    rows = [
        row for row in rows
        if query_lower in " ".join(
            str(row.get(field, "")) for field in ("title", "body", "date")
        ).lower()
    ]
    if limit is not None:
        rows = rows[:limit]
    return rows


def meeting_detail(meeting_id: str) -> dict[str, Any]:
    """Fetch meeting details including documents and links."""
    url = f"{BASE_URL}/Meeting.aspx?Id={meeting_id}&lang=English"
    html_text = _fetch_html(url)

    def extract_text(pattern: str) -> str:
        match = re.search(pattern, html_text, re.DOTALL | re.IGNORECASE)
        if match:
            return re.sub(r"<[^>]+>", "", html.unescape(match.group(1))).strip()
        return ""

    title = extract_text(r'<[^>]+class="[^"]*AgendaMeetingName[^"]*"[^>]*>(.*?)</[^>]+>')
    title = title or extract_text(r'<h1[^>]*>(.*?)</h1>') or extract_text(r'<title[^>]*>(.*?)</title>')
    meeting_date = extract_text(r'<div[^>]*class="[^"]*(?:meeting-date|date)[^"]*"[^>]*>(.*?)</div>')
    location = extract_text(r'<div[^>]*class="[^"]*(?:meeting-location|location)[^"]*"[^>]*>(.*?)</div>')

    def link_by_label(label: str) -> str | None:
        match = re.search(
            rf"<a[^>]+href=\"([^\"]+)\"[^>]*>\s*{re.escape(label)}\s*</a>",
            html_text,
            re.DOTALL | re.IGNORECASE,
        )
        return _absolute_url(match.group(1)) if match else None

    agenda = link_by_label("Agenda")
    minutes = link_by_label("Minutes")
    agenda_package = link_by_label("Agenda Packet")

    # Video/stream links.
    video_match = re.search(
        r"<a[^>]+href=\"([^\"]+)\"[^>]*>\s*(?:Video|Watch|Stream|Live)\s*</a>",
        html_text,
        re.DOTALL | re.IGNORECASE,
    )
    video = _absolute_url(video_match.group(1)) if video_match else None

    # Attachments: links to files near the agenda section.
    attachments: list[dict[str, str]] = []
    for att_match in re.finditer(
        r"<a[^>]+href=\"([^\"]+\.(?:pdf|doc|docx|xlsx|pptx|zip))\"[^>]*>(.*?)</a>",
        html_text,
        re.DOTALL | re.IGNORECASE,
    ):
        href = _absolute_url(att_match.group(1))
        text = re.sub(r"<[^>]+>", "", html.unescape(att_match.group(2))).strip()
        attachments.append({"url": href, "title": text})

    # Historical meeting metadata and document links are exposed by the same
    # read-only page method used by the site's past-meetings accordion.
    year_match = re.search(
        r'datetime=["\'][^"\']*?(20\d{2})-\d{2}-\d{2}', html_text, re.IGNORECASE
    ) or re.search(r"\b(20\d{2})\b", meeting_date)
    if not year_match:
        raise CompatibilityError("eSCRIBE meeting detail exposed no semantic meeting year")
    record = _find_past_meeting_record(meeting_id, int(year_match.group(1)))
    if record:
        title = str(record.get("MeetingType") or title)
        meeting_date = str(record.get("FormattedStart") or meeting_date)
        location = str(record.get("LocationName") or location)
        links = record.get("MeetingLinks", [])
        if isinstance(links, list):
            for link in links:
                if not isinstance(link, dict) or not link.get("Url"):
                    continue
                href = _absolute_url(str(link["Url"]))
                label = str(link.get("Title") or link.get("AriaLabel") or "Document")
                link_type = str(link.get("Type") or "").casefold()
                link_format = str(link.get("Format") or "").casefold()
                if link_type == "agendacover" and not agenda:
                    agenda = href
                elif link_type == "agenda":
                    if not agenda_package or link_format == ".pdf":
                        agenda_package = href
                elif "minute" in link_type or "minute" in label.casefold():
                    if not minutes:
                        minutes = href
                if link.get("HasVideo") and not video:
                    video = href
                if not any(item["url"] == href for item in attachments):
                    attachments.append({"url": href, "title": label})
        video = str(record.get("VideoUrl") or video or "") or None

    return {
        "id": meeting_id,
        "title": title,
        "date": meeting_date,
        "location": location,
        "cancelled": bool(record.get("Cancelled")) if record else False,
        "url": url,
        "agenda": agenda,
        "minutes": minutes,
        "agenda_package": agenda_package,
        "video": video,
        "attachments": attachments,
    }


def download_document(url: str, dest: str, force: bool = False) -> str:
    """Download a meeting document to a local path."""
    if not core.is_allowed_host(url):
        raise core.SecurityError(f"URL host is not allowlisted: {url}")
    data = core.raw_request(url)
    core.safe_write(Path(dest), data, force=force)
    return str(dest)

"""Raleigh Police Department incident data access.

Resolves stable ArcGIS item IDs to live FeatureServer URLs and provides
source-aware queries across four RPD datasets: NIBRS, SRS, previous-day,
and CrimeMapper past-90-days.
"""

from __future__ import annotations

import sys
import urllib.parse
from datetime import datetime, timezone
from typing import Any

from raleighlib import arcgis
from raleighlib import core

ITEM_RESOLUTION_URL = "https://ral.maps.arcgis.com/sharing/rest/content/items/{item_id}"

RPD_SOURCES: dict[str, dict[str, str]] = {
    "nibrs": {
        "item_id": "24c0b37fa9bb4e16ba8bcaa7e806c615",
        "label": "NIBRS (June 2014–present)",
        "caveats": "Block-level locations; may be randomized or redacted.",
    },
    "srs": {
        "item_id": "09af62a32ae8436bae6eda74aa7f172b",
        "label": "SRS (2005–May 2014)",
        "caveats": "Legacy reporting system; schema differs from NIBRS.",
    },
    "previous-day": {
        "item_id": "693811eb361f4da286891eca1fae5943",
        "label": "Previous-day incidents",
        "caveats": "May lag by more than one day; empty on some days.",
    },
    "crimemapper-90d": {
        "item_id": "a1f2d9204a184404b5a4c7e0fdceb6d0",
        "label": "CrimeMapper past 90 days",
        "caveats": "Not in the curated Hub catalog; field schema may differ.",
    },
}

_FIELD_MAPS: dict[str, dict[str, str]] = {
    "nibrs": {"category": "crime_description", "district": "district", "date": "reported_date"},
    "crimemapper-90d": {"category": "crime_description", "district": "district", "date": "reported_date"},
    "previous-day": {"category": "crime_description", "district": "district", "date": "reported_date"},
    "srs": {"category": "LCR_DESC", "district": "DISTRICT", "date": "INC_DATETIME"},
}

RALEIGH_BBOX = (-78.8, 35.6, -78.4, 36.0)

NIBRS_EPOCH_MS = 1401580800000

LOCATION_CAVERAT = (
    "Locations are block-level and may be randomized or redacted. "
    "This data does not include arrests, convictions, or dispositions."
)


class PoliceError(Exception):
    """Raised for RPD data resolution or query failures."""


def resolve_item_url(item_id: str) -> str:
    """Resolve an ArcGIS item ID to its FeatureServer/MapServer URL."""
    url = ITEM_RESOLUTION_URL.format(item_id=item_id)
    params = {"f": "json"}
    full_url = f"{url}?{urllib.parse.urlencode(params)}"
    meta = core.json_request(full_url)
    service_url = meta.get("url")
    if not service_url:
        raise PoliceError(f"Item {item_id} has no service URL")
    return service_url


def resolve_layer_url(source_key: str) -> str:
    """Resolve a source key to a queryable layer URL."""
    source = RPD_SOURCES.get(source_key)
    if not source:
        raise PoliceError(f"Unknown source: {source_key}")
    service_url = resolve_item_url(source["item_id"])
    return arcgis.resolve_queryable_layer(service_url)


def _escape_sql_value(value: str) -> str:
    """Escape a string value for use inside single quotes in an ArcGIS WHERE clause."""
    return value.replace("'", "''")


def _escape_like_value(value: str) -> str:
    """Escape LIKE wildcards and quotes for use in a LIKE pattern."""
    return value.replace("'", "''").replace("%", "\\%").replace("_", "\\_")


def _discover_fields(layer_url: str) -> set[str]:
    """Return the set of field names advertised by a layer."""
    fields = arcgis.layer_fields(layer_url)
    return {f.get("name", "") for f in fields if isinstance(f, dict)}


def _ms_to_timestamp_literal(ms: int) -> str:
    """Format Unix milliseconds as an ArcGIS TIMESTAMP literal in UTC."""
    try:
        dt = datetime.fromtimestamp(ms / 1000, timezone.utc)
    except (OverflowError, OSError, ValueError) as exc:
        raise PoliceError(f"date range out of bounds: {ms}") from exc
    return "TIMESTAMP '" + dt.strftime("%Y-%m-%d %H:%M:%S") + "'"


def build_where_clause(
    source_key: str,
    available_fields: set[str],
    since_ms: int | None = None,
    category: str | None = None,
    district: str | None = None,
) -> str:
    """Build an ArcGIS WHERE clause from durable filters.

    Field names are validated against the supplied field set. If a filter
    field is missing, the filter is skipped with a stderr warning.
    """
    field_map = _FIELD_MAPS.get(source_key)
    if not field_map:
        raise PoliceError(f"No field map for source: {source_key}")

    clauses: list[str] = ["1=1"]

    if since_ms is not None:
        date_field = field_map["date"]
        if date_field in available_fields:
            clauses.append(f"{date_field} >= {_ms_to_timestamp_literal(since_ms)}")
        else:
            print(
                f"Warning: date field '{date_field}' not found in {source_key}; skipping date filter",
                file=sys.stderr,
            )

    if category:
        cat_field = field_map["category"]
        if cat_field in available_fields:
            escaped = _escape_like_value(category.upper())
            clauses.append(f"UPPER({cat_field}) LIKE '%{escaped}%'")
        else:
            print(
                f"Warning: category field '{cat_field}' not found in {source_key}; skipping category filter",
                file=sys.stderr,
            )

    if district:
        dist_field = field_map["district"]
        if dist_field in available_fields:
            escaped = _escape_like_value(district.upper())
            clauses.append(f"UPPER({dist_field}) LIKE '%{escaped}%'")
        else:
            print(
                f"Warning: district field '{dist_field}' not found in {source_key}; skipping district filter",
                file=sys.stderr,
            )

    return " AND ".join(clauses)


def _is_placeholder_point(geom: dict[str, Any]) -> bool:
    """Return True if the geometry is a null-island placeholder (0,0)."""
    return geom.get("x") == 0 and geom.get("y") == 0


def _location_status(record: dict[str, Any]) -> str:
    """Classify a record's location quality."""
    geom = record.get("geometry")
    if not geom:
        return "redacted"
    if "x" in geom and "y" in geom:
        if _is_placeholder_point(geom):
            return "redacted"
        x, y = geom["x"], geom["y"]
        if not (RALEIGH_BBOX[0] <= x <= RALEIGH_BBOX[2] and RALEIGH_BBOX[1] <= y <= RALEIGH_BBOX[3]):
            return "out_of_area"
        return "block_level"
    return "unknown"


def _normalize_geometry(record: dict[str, Any]) -> dict[str, Any] | None:
    """Return GeoJSON geometry, suppressing redacted/placeholder points."""
    geom = record.get("geometry")
    if not geom:
        return None
    if "x" in geom and "y" in geom:
        if _is_placeholder_point(geom):
            return None
        return {"type": "Point", "coordinates": [geom["x"], geom["y"]]}
    return arcgis.geometry_from_record(record)


def query_incidents(
    source_key: str,
    since_ms: int | None = None,
    category: str | None = None,
    district: str | None = None,
    limit: int = 20,
    offset: int = 0,
) -> dict[str, Any]:
    """Query a single RPD source and return an enriched GeoJSON FeatureCollection."""
    source = RPD_SOURCES.get(source_key)
    if not source:
        raise PoliceError(f"Unknown source: {source_key}")

    layer_url = resolve_layer_url(source_key)
    available_fields = _discover_fields(layer_url)
    where = build_where_clause(
        source_key, available_fields, since_ms=since_ms, category=category, district=district
    )

    records = arcgis.query_all_pages(
        layer_url,
        where=where,
        return_geometry=True,
        max_records=limit,
        offset=offset,
    )

    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    features: list[dict[str, Any]] = []
    for record in records:
        geom = _normalize_geometry(record)
        attrs = dict(record.get("attributes", {}))
        attrs["_source"] = source_key
        attrs["_item_id"] = source["item_id"]
        attrs["_retrieved_at"] = now
        attrs["_location_status"] = _location_status(record)
        features.append({
            "type": "Feature",
            "properties": attrs,
            "geometry": geom,
        })

    return {
        "type": "FeatureCollection",
        "_sources": [
            {
                "item_id": source["item_id"],
                "label": source["label"],
                "caveats": source["caveats"],
            }
        ],
        "features": features,
    }

"""Official RPD and RFD aggregate statistics published on RaleighNC.gov."""

from __future__ import annotations

import re
import http.client
import urllib.error
import urllib.parse
from datetime import datetime, timezone
from html.parser import HTMLParser
from typing import Any

from raleighlib import core


SOURCES = {
    "police": {
        "id": "40ebbee4-2477-4f7d-9623-257685345e3d",
        "title": "Raleigh's Crime Data",
        "page_url": "https://raleighnc.gov/police/services/raleighs-crime-data",
    },
    "fire": {
        "id": "f95a0f43-3dbf-4378-b7c7-b1bdda20eb24",
        "title": "View Raleigh Fire Statistics",
        "page_url": "https://raleighnc.gov/fire/services/view-raleigh-fire-statistics",
    },
}
MAX_PUBLISHED_REPORTS = 100


class PublishedStatisticsError(ValueError):
    """Raised when an official statistics page no longer matches its contract."""


_TERMINAL_CONTROLS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")


class _FragmentParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.links: list[dict[str, Any]] = []
        self.tables: list[list[list[str]]] = []
        self.text: list[str] = []
        self._year: int | None = None
        self._heading: list[str] | None = None
        self._link: dict[str, Any] | None = None
        self._table: list[list[str]] | None = None
        self._row: list[str] | None = None
        self._cell: list[str] | None = None

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        tag = tag.casefold()
        if tag == "h5":
            self._heading = []
        elif tag == "a":
            self._link = {"href": dict(attrs).get("href"), "text": [], "year": self._year}
        elif tag == "table":
            self._table = []
        elif tag == "tr" and self._table is not None:
            self._row = []
        elif tag in {"th", "td"} and self._row is not None:
            self._cell = []

    def handle_data(self, data: str) -> None:
        self.text.append(data)
        if self._heading is not None:
            self._heading.append(data)
        if self._link is not None:
            self._link["text"].append(data)
        if self._cell is not None:
            self._cell.append(data)

    def handle_endtag(self, tag: str) -> None:
        tag = tag.casefold()
        if tag == "h5" and self._heading is not None:
            value = _clean(" ".join(self._heading))
            self._year = int(value) if re.fullmatch(r"20\d{2}", value) else None
            self._heading = None
        elif tag == "a" and self._link is not None:
            self._link["text"] = _clean(" ".join(self._link["text"]))
            self.links.append(self._link)
            self._link = None
        elif tag in {"th", "td"} and self._cell is not None and self._row is not None:
            self._row.append(_clean(" ".join(self._cell)))
            self._cell = None
        elif tag == "tr" and self._row is not None and self._table is not None:
            if self._row:
                self._table.append(self._row)
            self._row = None
        elif tag == "table" and self._table is not None:
            self.tables.append(self._table)
            self._table = None


def _clean(value: str) -> str:
    return " ".join(_TERMINAL_CONTROLS_RE.sub("", value).split())


def _source(agency: str) -> dict[str, str]:
    try:
        return SOURCES[agency]
    except KeyError as exc:
        raise PublishedStatisticsError(f"unknown public-safety agency: {agency}") from exc


def _document_url(href: Any, page_url: str, agency: str) -> str:
    if not isinstance(href, str) or not href.strip():
        raise PublishedStatisticsError("published report link is missing")
    if _TERMINAL_CONTROLS_RE.search(href):
        raise PublishedStatisticsError("published report link contains control characters")
    url = urllib.parse.urljoin(page_url, href)
    parsed = urllib.parse.urlparse(url)
    try:
        port = parsed.port
    except ValueError as exc:
        raise PublishedStatisticsError("published report link has an invalid port") from exc
    if (
        parsed.scheme != "https"
        or parsed.username is not None
        or parsed.password is not None
        or port not in (None, 443)
        or parsed.params
        or parsed.query
        or parsed.fragment
    ):
        raise PublishedStatisticsError("published report link left the approved HTTPS contract")
    host = (parsed.hostname or "").casefold()
    decoded_segments = urllib.parse.unquote(parsed.path).split("/")
    if any(segment in {".", ".."} for segment in decoded_segments):
        raise PublishedStatisticsError("published report link contains a path traversal segment")
    if agency == "fire" and host == "raleighnc.gov" and parsed.path.startswith("/fire/news/"):
        return url
    if (
        host == "cityofraleigh0drupal.blob.core.usgovcloudapi.net"
        and re.fullmatch(
            rf"/drupal-prod/{'COR23' if agency == 'police' else 'COR18'}/[^/]+\.[Pp][Dd][Ff]",
            parsed.path,
        )
    ):
        return url
    raise PublishedStatisticsError("published report link left the approved document origins")


def _fetch_page(agency: str) -> tuple[dict[str, Any], dict[str, str]]:
    source = _source(agency)
    url = (
        f"https://raleighnc.gov/jsonapi/node/service/{source['id']}"
        "?include=field_content_primary"
    )

    def validate_final_url(final_url: str) -> None:
        if final_url != url:
            raise PublishedStatisticsError("published statistics source left its exact JSON:API endpoint")

    try:
        response = core.json_request(url, final_url_validator=validate_final_url)
    except (
        core.SecurityError,
        urllib.error.HTTPError,
        urllib.error.URLError,
        OSError,
        http.client.HTTPException,
    ) as exc:
        raise PublishedStatisticsError(f"published statistics source is unavailable: {exc}") from exc
    payload = core.require_object(response, "published statistics request")
    data = payload.get("data")
    included = payload.get("included")
    if not isinstance(data, dict) or not isinstance(included, list):
        raise PublishedStatisticsError("published statistics source returned invalid JSON:API data")
    attrs = data.get("attributes")
    path = attrs.get("path") if isinstance(attrs, dict) else None
    if (
        data.get("type") != "node--service"
        or data.get("id") != source["id"]
        or not isinstance(attrs, dict)
        or not isinstance(path, dict)
        or attrs.get("status") is not True
        or attrs.get("title") != source["title"]
        or path.get("alias") != urllib.parse.urlparse(source["page_url"]).path
    ):
        raise PublishedStatisticsError("published statistics page identity changed")
    changed = attrs.get("changed")
    if not isinstance(changed, str) or not changed.strip():
        raise PublishedStatisticsError("published statistics page revision timestamp is missing")
    try:
        changed_at = datetime.fromisoformat(changed.replace("Z", "+00:00"))
    except ValueError as exc:
        raise PublishedStatisticsError("published statistics page revision timestamp is invalid") from exc
    if changed_at.tzinfo is None:
        raise PublishedStatisticsError("published statistics page revision timestamp has no timezone")

    relationships = data.get("relationships")
    if not isinstance(relationships, dict):
        raise PublishedStatisticsError("published statistics content relationships are invalid")
    relationship = relationships.get("field_content_primary", {})
    related = relationship.get("data") if isinstance(relationship, dict) else None
    if not isinstance(related, list):
        raise PublishedStatisticsError("published statistics content relationship is missing")
    referenced: set[tuple[str, str]] = set()
    for item in related:
        if (
            not isinstance(item, dict)
            or not isinstance(item.get("type"), str)
            or not isinstance(item.get("id"), str)
            or not item["type"]
            or not item["id"]
        ):
            raise PublishedStatisticsError("published statistics relationship identifiers are invalid")
        key = (item["type"], item["id"])
        if key in referenced:
            raise PublishedStatisticsError("published statistics relationship identifiers are duplicated")
        referenced.add(key)
    fragments: dict[str, str] = {}
    included_identifiers: set[tuple[str, str]] = set()
    for item in included:
        if (
            not isinstance(item, dict)
            or not isinstance(item.get("type"), str)
            or not isinstance(item.get("id"), str)
            or not item["type"]
            or not item["id"]
        ):
            raise PublishedStatisticsError("published statistics included resource identifiers are invalid")
        key = (item["type"], item["id"])
        if key in included_identifiers:
            raise PublishedStatisticsError("published statistics included resource identifiers are duplicated")
        included_identifiers.add(key)
        if item["type"] != "paragraph--stories_text":
            continue
        if key not in referenced:
            raise PublishedStatisticsError("published statistics included an unreferenced content section")
        item_attrs = item.get("attributes")
        if not isinstance(item_attrs, dict) or item_attrs.get("status") is not True:
            continue
        heading = item_attrs.get("field_heading")
        formatted = item_attrs.get("field_stories_text_formatted")
        html = formatted.get("value") if isinstance(formatted, dict) else None
        if isinstance(heading, str) and isinstance(html, str):
            if heading in fragments:
                raise PublishedStatisticsError(f"published statistics section is duplicated: {heading}")
            fragments[heading] = html
    metadata = {
        "url": source["page_url"],
        "retrieved_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "published_changed_at": changed,
    }
    return {"fragments": fragments}, metadata


def _parse_police(fragments: dict[str, str], page_url: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    html = fragments.get("Summary of Crime Statistics by Year")
    if html is None:
        raise PublishedStatisticsError("police report index section is missing")
    parser = _FragmentParser()
    parser.feed(html)
    reports = []
    for link in parser.links:
        label = link["text"]
        year = link["year"]
        quarter_match = re.search(r"\bQ([1-4])\b", label, re.IGNORECASE)
        if not isinstance(year, int):
            raise PublishedStatisticsError("police report is not grouped under a year")
        if "annual" in label.casefold():
            period, quarter = "annual", None
        elif quarter_match:
            period, quarter = "quarterly", int(quarter_match.group(1))
        else:
            raise PublishedStatisticsError(f"unrecognized police publication label: {label}")
        reports.append({
            "agency": "police",
            "year": year,
            "period": period,
            "quarter": quarter,
            "label": label,
            "document_url": _document_url(link["href"], page_url, "police"),
        })
    if not reports:
        raise PublishedStatisticsError("police report index returned no publications")
    if len(reports) > MAX_PUBLISHED_REPORTS:
        raise PublishedStatisticsError(f"police report index exceeded the {MAX_PUBLISHED_REPORTS}-report limit")
    return [], reports


def _parse_fire(fragments: dict[str, str], page_url: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    annual_html = fragments.get("Previous Years Statistics")
    quarterly_html = fragments.get("Quarterly Report")
    if annual_html is None or quarterly_html is None:
        raise PublishedStatisticsError("fire report index section is missing")

    reports: list[dict[str, Any]] = []
    annual_parser = _FragmentParser()
    annual_parser.feed(annual_html)
    for link in annual_parser.links:
        if not re.fullmatch(r"20\d{2}", link["text"]):
            raise PublishedStatisticsError(f"unrecognized fire annual publication label: {link['text']}")
        year = int(link["text"])
        reports.append({
            "agency": "fire", "year": year, "period": "annual", "quarter": None,
            "label": link["text"], "document_url": _document_url(link["href"], page_url, "fire"),
        })

    quarterly_parser = _FragmentParser()
    quarterly_parser.feed(quarterly_html)
    for link in quarterly_parser.links:
        if not link["text"]:
            raise PublishedStatisticsError("fire quarterly report publication label is missing")
        url = _document_url(link["href"], page_url, "fire")
        period = re.search(r"(?:^|[-_/])q([1-4])-(20\d{2})(?:[-_/]|$)", url, re.IGNORECASE)
        if period is None:
            raise PublishedStatisticsError("fire quarterly report URL has no stable quarter and year")
        reports.append({
            "agency": "fire", "year": int(period.group(2)), "period": "quarterly",
            "quarter": int(period.group(1)), "label": link["text"], "document_url": url,
        })

    statistic_headings = [
        heading for heading in fragments
        if re.fullmatch(r"20\d{2} Statistics", heading)
    ]
    if not statistic_headings:
        raise PublishedStatisticsError("fire incident statistics section is missing")

    if len(reports) > MAX_PUBLISHED_REPORTS:
        raise PublishedStatisticsError(f"fire report index exceeded the {MAX_PUBLISHED_REPORTS}-report limit")

    datasets: list[dict[str, Any]] = []
    for heading, html in fragments.items():
        match = re.fullmatch(r"(20\d{2}) Statistics", heading)
        if match is None:
            continue
        parser = _FragmentParser()
        parser.feed(html)
        if len(parser.tables) != 1 or len(parser.tables[0]) < 2:
            raise PublishedStatisticsError(f"fire statistics table is missing for {match.group(1)}")
        rows = parser.tables[0]
        if rows[0] != ["Incident Type", "Totals"]:
            raise PublishedStatisticsError("fire statistics table headers changed")
        values = []
        for row in rows[1:]:
            if len(row) != 2 or not row[0]:
                raise PublishedStatisticsError("fire statistics table contains a malformed row")
            valid_number = re.fullmatch(r"(?:0|[1-9]\d*|[1-9]\d{0,2}(?:,\d{3})+)", row[1])
            if valid_number is None:
                raise PublishedStatisticsError("fire statistics table contains a malformed row")
            values.append({"label": row[0], "value": int(row[1].replace(",", "")), "published_value": row[1]})
        datasets.append({"year": int(match.group(1)), "kind": "incident_totals", "values": values})

    sprinkler_html = fragments.get("Sprinkler Saves Stats")
    if sprinkler_html is None:
        raise PublishedStatisticsError("fire sprinkler statistics section is missing")
    parser = _FragmentParser()
    parser.feed(sprinkler_html)
    year_match = re.search(r"\b(20\d{2}) Sprinkler Saves Statistics\b", _clean(" ".join(parser.text)))
    if year_match is None or len(parser.tables) != 1 or len(parser.tables[0]) < 2:
        raise PublishedStatisticsError("fire sprinkler statistics contract changed")
    rows = parser.tables[0]
    if rows[0] != ["Type", "Description", "Statistic", "Percentage"]:
        raise PublishedStatisticsError("fire sprinkler statistics table headers changed")
    values = []
    for row in rows[1:]:
        if len(row) != 4 or not row[1] or not row[2]:
            raise PublishedStatisticsError("fire sprinkler statistics table contains a malformed row")
        values.append({"type": row[0] or None, "label": row[1], "published_value": row[2], "published_percentage": row[3]})
    datasets.append({"year": int(year_match.group(1)), "kind": "sprinkler_saves", "values": values})

    if not reports or not datasets:
        raise PublishedStatisticsError("fire statistics source returned no publications or structured totals")
    return datasets, reports


def _published(agency: str) -> dict[str, Any]:
    page, metadata = _fetch_page(agency)
    fragments = page["fragments"]
    if agency == "police":
        datasets, reports = _parse_police(fragments, metadata["url"])
    else:
        datasets, reports = _parse_fire(fragments, metadata["url"])
    return {"datasets": datasets, "reports": reports, "source": metadata}


def _assert_available(items: list[dict[str, Any]]) -> None:
    if len(items) > MAX_PUBLISHED_REPORTS:
        raise PublishedStatisticsError(f"publication selection exceeded the {MAX_PUBLISHED_REPORTS}-report limit")
    unique = {(item["document_url"], item["agency"]) for item in items}
    for url, agency in unique:
        def validate_final_url(final_url: str) -> None:
            _document_url(final_url, url, agency)

        try:
            final_url = core.probe_url(url, final_url_validator=validate_final_url)
            validate_final_url(final_url)
        except (
            core.SecurityError,
            urllib.error.HTTPError,
            urllib.error.URLError,
            OSError,
            http.client.HTTPException,
        ) as exc:
            raise PublishedStatisticsError(f"published document is unavailable: {url}: {exc}") from exc


def statistics(agency: str, year: int | None = None) -> dict[str, Any]:
    published = _published(agency)
    years = sorted({item["year"] for item in published["datasets"] + published["reports"]}, reverse=True)
    if year is not None and year not in years:
        raise PublishedStatisticsError(f"no published {agency} statistics found for {year}")
    datasets = [item for item in published["datasets"] if year is None or item["year"] == year]
    reports = [item for item in published["reports"] if year is None or item["year"] == year]
    _assert_available(reports)
    warnings = []
    if year is not None and not datasets:
        warnings.append("Published totals are document-only for this year; PDF contents were not parsed.")
    if agency == "fire":
        warnings.append(
            "Published medical totals are aggregate-only and must not be joined to or used to infer excluded incident records."
        )
    return {
        "agency": agency,
        "classification": "official_published_statistics",
        "year": year,
        "available_years": years,
        "datasets": datasets,
        "reports": reports,
        "source": published["source"],
        "warnings": warnings,
    }


def reports(agency: str, year: int | None = None, quarter: int | None = None) -> dict[str, Any]:
    if quarter is not None and year is None:
        raise PublishedStatisticsError("--quarter requires --year")
    published = _published(agency)
    available_years = sorted({item["year"] for item in published["reports"]}, reverse=True)
    if year is not None and year not in available_years:
        raise PublishedStatisticsError(f"no published {agency} reports found for {year}")
    selected = [
        item for item in published["reports"]
        if (year is None or item["year"] == year) and (quarter is None or item["quarter"] == quarter)
    ]
    if not selected:
        period = f" Q{quarter}" if quarter is not None else ""
        raise PublishedStatisticsError(f"no published {agency} report found for {year}{period}")
    _assert_available(selected)
    return {
        "agency": agency,
        "classification": "official_published_reports",
        "year": year,
        "quarter": quarter,
        "available_years": available_years,
        "reports": selected,
        "source": published["source"],
        "warnings": ["Document links are preserved from the official index; PDF contents were not parsed."],
    }

"""Guarded adapter for the fragile plain-HTTP RFD Report System."""

from __future__ import annotations

import html
import re
import urllib.error
import urllib.parse
import urllib.request
from datetime import date, datetime, timezone
from html.parser import HTMLParser
from typing import Any

from raleighlib import core

BASE_URL = "http://rfdreports.net"
DATE_PATH = "/fd_date.php"
NARRATIVE_PATH = "/fd_incidentreport.php"
BUSINESS_PATH = "/fd_inspection_business_name.php"
ADDRESS_PATH = "/fd_inspection_business_address.php"
MAX_HTML_BYTES = 2 * 1024 * 1024
INSECURE_WARNING = (
    "RFD Report System uses unencrypted HTTP; search terms and returned data "
    "can be observed or altered in transit."
)

_REPORT_HEADERS = [
    "Incident Data", "Incident Type", "Incident #", "Dispatch Time",
    "Arrive Time", "Clear Time", "Address", "Unit", "Cross Street", "View Report",
]
_INSPECTION_HEADERS = [
    "Occupancy Name", "Address", "Inspection Type", "Data Completed",
    "View Report", "View Invoice",
]
_TERMINAL_CONTROLS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")


class RFDReportError(ValueError):
    """Raised when an RFD request or HTML contract is unsafe or incompatible."""


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise RFDReportError("RFD redirects are not allowed")


_OPENER = urllib.request.build_opener(_NoRedirect)


class _TableParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.rows: list[list[dict[str, Any]]] = []
        self._row: list[dict[str, Any]] | None = None
        self._cell: dict[str, Any] | None = None

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        tag = tag.casefold()
        if tag == "tr":
            if self._row:
                self.rows.append(self._row)
            self._row = []
        elif tag in {"th", "td"} and self._row is not None:
            self._cell = {"text": [], "links": []}
        elif tag == "a" and self._cell is not None:
            href = dict(attrs).get("href")
            if href:
                self._cell["links"].append(href)

    def handle_data(self, data: str) -> None:
        if self._cell is not None:
            self._cell["text"].append(data)

    def handle_endtag(self, tag: str) -> None:
        tag = tag.casefold()
        if tag in {"th", "td"} and self._cell is not None and self._row is not None:
            text = _safe_text(" ".join(self._cell["text"]))
            self._row.append({"text": html.unescape(text), "links": self._cell["links"]})
            self._cell = None
        elif tag == "tr" and self._row is not None:
            if self._row:
                self.rows.append(self._row)
            self._row = None


def _require_text(value: str, label: str) -> str:
    value = value.strip()
    if not value:
        raise RFDReportError(f"{label} must not be empty")
    if len(value) > 200 or any(ord(char) < 32 for char in value):
        raise RFDReportError(f"{label} is invalid")
    return value


def _safe_text(value: str) -> str:
    """Remove terminal controls from untrusted HTTP text and normalize whitespace."""
    return " ".join(_TERMINAL_CONTROLS_RE.sub("", value).split())


def _validate_contract(path: str, params: dict[str, str], method: str) -> str:
    contracts = {
        DATE_PATH: ("POST", {"date"}),
        NARRATIVE_PATH: ("GET", {"incidentnumber", "incidentdate"}),
        BUSINESS_PATH: ("POST", {"fd_business"}),
        ADDRESS_PATH: ("POST", {"fd_address"}),
    }
    expected = contracts.get(path)
    if expected != (method, set(params)):
        raise RFDReportError("unsupported RFD request contract")
    return BASE_URL + path


def _request(path: str, params: dict[str, str], method: str, *, acknowledged: bool) -> str:
    if not acknowledged:
        raise RFDReportError("RFD access requires --acknowledge-insecure-rfd")
    url = _validate_contract(path, params, method)
    encoded = urllib.parse.urlencode(params)
    data = encoded.encode("utf-8") if method == "POST" else None
    if method == "GET":
        url = f"{url}?{encoded}"
    headers = {"User-Agent": core.USER_AGENT, "Accept": "text/html"}
    if data is not None:
        headers["Content-Type"] = "application/x-www-form-urlencoded"
    request = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with _OPENER.open(request, timeout=core._get_timeout()) as response:
            final = urllib.parse.urlparse(response.geturl())
            if final.scheme != "http" or final.hostname != "rfdreports.net" or final.port not in (None, 80):
                raise RFDReportError("RFD response left the fixed insecure origin")
            body = core._read_limited(response, MAX_HTML_BYTES)
    except urllib.error.HTTPError as exc:
        raise RFDReportError(f"RFD returned HTTP {exc.code}") from exc
    text = body.decode("utf-8", errors="replace")
    lowered = text.casefold()
    if "internal server error" in lowered or "service unavailable" in lowered:
        raise RFDReportError("RFD returned an error page")
    return text


def _table_rows(html_text: str, expected_headers: list[str]) -> list[list[dict[str, Any]]]:
    parser = _TableParser()
    parser.feed(html_text)
    if not parser.rows:
        if re.search(r"\b(no (?:records|results|reports|inspections) (?:found|available))\b", html_text, re.I):
            return []
        raise RFDReportError("RFD HTML contract drift: no result table")
    headers = [cell["text"] for cell in parser.rows[0]]
    if headers != expected_headers:
        raise RFDReportError("RFD HTML contract drift: unexpected table headers")
    rows = parser.rows[1:]
    for row in rows:
        if len(row) != len(expected_headers):
            raise RFDReportError("RFD HTML contract drift: malformed result row")
    return rows


def _canonical_link(href: str, path: str, required: set[str]) -> str:
    parsed = urllib.parse.urlparse(urllib.parse.urljoin(BASE_URL + "/", href))
    params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
    if parsed.scheme != "http" or parsed.hostname != "rfdreports.net" or parsed.path != path:
        raise RFDReportError("RFD HTML contract drift: unexpected result link")
    if set(params) != required or any(len(values) != 1 for values in params.values()):
        raise RFDReportError("RFD HTML contract drift: malformed result link")
    return urllib.parse.urlunparse(("http", "rfdreports.net", path, "", urllib.parse.urlencode({k: v[0] for k, v in params.items()}), ""))


def _inspection_link(href: str, number: str, address: str, name: str) -> str:
    """Validate the fixed report link, then repair upstream's unescaped # values."""
    parsed = urllib.parse.urlparse(urllib.parse.urljoin(BASE_URL + "/", href))
    params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
    if parsed.scheme != "http" or parsed.hostname != "rfdreports.net" or parsed.path != "/fd_report.php":
        raise RFDReportError("RFD HTML contract drift: unexpected inspection link")
    if params.get("inspection_number") != [number]:
        raise RFDReportError("RFD HTML contract drift: inspection link identifier mismatch")
    return BASE_URL + "/fd_report.php?" + urllib.parse.urlencode({
        "inspection_number": number,
        "address": address,
        "name": name,
    })


def search_date(report_date: str, *, acknowledged: bool) -> list[dict[str, Any]]:
    report_date = _require_text(report_date, "date")
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", report_date):
        raise RFDReportError("date must use YYYY-MM-DD")
    try:
        requested_date = date.fromisoformat(report_date)
    except ValueError as exc:
        raise RFDReportError("date must use YYYY-MM-DD") from exc
    html_text = _request(DATE_PATH, {"date": report_date}, "POST", acknowledged=acknowledged)
    reports: list[dict[str, Any]] = []
    for row in _table_rows(html_text, _REPORT_HEADERS):
        values = [cell["text"] for cell in row]
        links = row[9]["links"]
        if not values[2] or len(links) != 1:
            raise RFDReportError("RFD HTML contract drift: report identifier or link missing")
        source_url = _canonical_link(
            links[0], NARRATIVE_PATH, {"incidentnumber", "incidentdate"}
        )
        try:
            row_date = datetime.strptime(values[0], "%m/%d/%Y").date()
        except ValueError as exc:
            raise RFDReportError("RFD HTML contract drift: invalid incident date") from exc
        source_params = urllib.parse.parse_qs(urllib.parse.urlparse(source_url).query)
        if row_date != requested_date:
            raise RFDReportError("RFD response included a report outside the requested date")
        if source_params.get("incidentnumber") != [values[2]] or source_params.get("incidentdate") != [report_date]:
            raise RFDReportError("RFD report row and source link do not match")
        reports.append({
            "source": "rfd-html",
            "source_fragility": "fragile-html-over-http",
            "incident_date": values[0],
            "incident_type_name": values[1],
            "incident_number": values[2],
            "dispatch_time": values[3],
            "arrive_time": values[4],
            "clear_time": values[5],
            "address": values[6],
            "unit": values[7],
            "cross_street": values[8],
            "source_url": source_url,
        })
    return reports


def search_inspections(*, business: str | None = None, address: str | None = None, acknowledged: bool) -> dict[str, Any]:
    if bool(business) == bool(address):
        raise RFDReportError("provide exactly one business name or address")
    if business is not None:
        query = _require_text(business, "business name")
        path, params = BUSINESS_PATH, {"fd_business": query}
        mode = "business"
    else:
        query = _require_text(address or "", "address")
        path, params = ADDRESS_PATH, {"fd_address": query}
        mode = "address"
    html_text = _request(path, params, "POST", acknowledged=acknowledged)
    inspections: list[dict[str, Any]] = []
    for row in _table_rows(html_text, _INSPECTION_HEADERS):
        values = [cell["text"] for cell in row]
        report_links = row[4]["links"]
        if not values[4] or len(report_links) != 1:
            raise RFDReportError("RFD HTML contract drift: inspection identifier or link missing")
        source_url = _inspection_link(
            report_links[0], values[4], values[1], values[0]
        )
        inspections.append({
            "source": "rfd-html",
            "source_fragility": "fragile-html-over-http",
            "business_name": values[0],
            "address": values[1],
            "inspection_type": values[2],
            "completed_date": values[3],
            "inspection_number": values[4],
            "source_url": source_url,
        })
    return {
        "query": {mode: query},
        "inspections": inspections,
        "source": {
            "source": "rfd-html",
            "url": BASE_URL + path,
            "retrieved_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        },
        "warnings": [INSECURE_WARNING],
    }


def fetch_narrative(incident_number: str, incident_date: str, *, acknowledged: bool) -> dict[str, Any]:
    incident_number = _require_text(incident_number, "incident number")
    incident_date = _require_text(incident_date, "incident date")
    html_text = _request(
        NARRATIVE_PATH,
        {"incidentnumber": incident_number, "incidentdate": incident_date},
        "GET",
        acknowledged=acknowledged,
    )
    if "City of Raleigh Fire Department Basic Fire Report" not in html_text:
        raise RFDReportError("RFD narrative page identity was not recognized")
    number_match = re.search(r"Incident Number:</b>\s*([^<]+)", html_text, re.I)
    date_match = re.search(r"Alarm Date:</b>\s*([^<]+)", html_text, re.I)
    narrative_match = re.search(
        r"<b>\s*narrative(?:\&quot;|\")?\s*:\s*(.*?)</b>", html_text, re.I | re.S
    )
    if not number_match or not date_match or not narrative_match:
        raise RFDReportError("RFD HTML contract drift: narrative fields missing")
    returned_number = " ".join(html.unescape(number_match.group(1)).split())
    returned_date = " ".join(html.unescape(date_match.group(1)).split())
    if returned_number != incident_number or returned_date != incident_date:
        raise RFDReportError("RFD narrative response did not match the requested incident")
    narrative = _safe_text(re.sub(
        r'"?\s*}\]\s*$',
        "",
        html.unescape(narrative_match.group(1)).strip().lstrip('"').strip(),
    ))
    if not narrative:
        raise RFDReportError("RFD narrative was empty")
    return {
        "source": "rfd-html",
        "source_fragility": "fragile-html-over-http",
        "incident_number": returned_number,
        "incident_date": returned_date,
        "narrative": narrative,
        "source_url": BASE_URL + NARRATIVE_PATH + "?" + urllib.parse.urlencode({
            "incidentnumber": incident_number,
            "incidentdate": incident_date,
        }),
    }

"""GoRaleigh GTFS and GTFS-Realtime adapter."""

from __future__ import annotations

import csv
import copy
import hashlib
import io
import zipfile
from datetime import date, datetime, timezone
from typing import Any

from raleighlib import core


STATIC_FEED = "https://goraleigh.org/gr_gtfs"
REALTIME_BASE = "https://www.goraleighlive.org/gtfsrt"

# GTFS ZIP safety caps.
MAX_GTFS_ZIP_BYTES = 50 * 1024 * 1024
MAX_GTFS_MEMBERS = 100
MAX_GTFS_MEMBER_SIZE = 64 * 1024 * 1024
MAX_GTFS_TOTAL_UNCOMPRESSED = 200 * 1024 * 1024

REQUIRED_GTFS_FIELDS = {
    "agency": {"agency_name", "agency_url", "agency_timezone"},
    "stops": {"stop_id", "stop_name", "stop_lat", "stop_lon"},
    "routes": {"route_id", "route_type"},
    "trips": {"route_id", "service_id", "trip_id"},
    "stop_times": {"trip_id", "arrival_time", "departure_time", "stop_id", "stop_sequence"},
}
CALENDAR_FIELDS = {
    "service_id", "monday", "tuesday", "wednesday", "thursday", "friday",
    "saturday", "sunday", "start_date", "end_date",
}
CALENDAR_DATES_FIELDS = {"service_id", "date", "exception_type"}


def download_gtfs(url: str = STATIC_FEED, max_bytes: int = MAX_GTFS_ZIP_BYTES) -> bytes:
    """Download the static GTFS ZIP archive."""
    return core.raw_request(url, max_bytes=max_bytes)


def _parse_gtfs_zip(data: bytes) -> dict[str, list[dict[str, str]]]:
    """Parse required GTFS tables from a ZIP archive with bounded size checks."""
    if len(data) > MAX_GTFS_ZIP_BYTES:
        raise ValueError(f"GTFS ZIP archive exceeds {MAX_GTFS_ZIP_BYTES} bytes")
    feed: dict[str, list[dict[str, str]]] = {}
    total_uncompressed = 0
    with zipfile.ZipFile(io.BytesIO(data)) as zf:
        members = zf.infolist()
        if len(members) > MAX_GTFS_MEMBERS:
            raise ValueError(f"GTFS ZIP contains more than {MAX_GTFS_MEMBERS} members")
        for info in members:
            if info.file_size > MAX_GTFS_MEMBER_SIZE:
                raise ValueError(f"GTFS ZIP member {info.filename} exceeds {MAX_GTFS_MEMBER_SIZE} bytes")
            total_uncompressed += info.file_size
            if total_uncompressed > MAX_GTFS_TOTAL_UNCOMPRESSED:
                raise ValueError(f"GTFS ZIP uncompressed total exceeds {MAX_GTFS_TOTAL_UNCOMPRESSED} bytes")
        for name in zf.namelist():
            if not name.endswith(".txt"):
                continue
            table = name.replace(".txt", "")
            with zf.open(name) as f:
                reader = csv.DictReader(io.TextIOWrapper(f, encoding="utf-8-sig"))
                feed[table] = list(reader)
    return feed


def parse_gtfs_zip(data: bytes) -> dict[str, list[dict[str, str]]]:
    """Validate and parse a GTFS archive, normalizing malformed ZIP errors."""
    try:
        feed = _parse_gtfs_zip(data)
        _validate_gtfs_feed(feed)
        return feed
    except (zipfile.BadZipFile, csv.Error, UnicodeError) as exc:
        raise ValueError("GTFS ZIP archive is malformed") from exc


def _validate_table(
    feed: dict[str, list[dict[str, str]]],
    table: str,
    required_fields: set[str],
) -> None:
    rows = feed.get(table)
    if not isinstance(rows, list) or not rows:
        raise ValueError(f"GTFS archive is missing non-empty {table}.txt")
    for index, row in enumerate(rows, start=2):
        if not isinstance(row, dict) or None in row:
            raise ValueError(f"GTFS {table}.txt row {index} is malformed")
        missing = [field for field in required_fields if not str(row.get(field, "")).strip()]
        if missing:
            raise ValueError(
                f"GTFS {table}.txt row {index} is missing required values: {', '.join(sorted(missing))}"
            )


def _validate_gtfs_feed(feed: dict[str, list[dict[str, str]]]) -> None:
    """Require the semantic core of a usable GTFS Schedule dataset."""
    for table, fields in REQUIRED_GTFS_FIELDS.items():
        _validate_table(feed, table, fields)
    for index, row in enumerate(feed["routes"], start=2):
        if not (str(row.get("route_short_name", "")).strip() or str(row.get("route_long_name", "")).strip()):
            raise ValueError(
                f"GTFS routes.txt row {index} requires route_short_name or route_long_name"
            )
    if feed.get("calendar"):
        _validate_table(feed, "calendar", CALENDAR_FIELDS)
    if feed.get("calendar_dates"):
        _validate_table(feed, "calendar_dates", CALENDAR_DATES_FIELDS)
    if not feed.get("calendar") and not feed.get("calendar_dates"):
        raise ValueError("GTFS archive requires calendar.txt or calendar_dates.txt")


def _load_feed_with_cache() -> dict[str, list[dict[str, str]]]:
    archive = core.read_cache_bytes(
        "gtfs-feed.zip", max_age_seconds=86400, max_bytes=MAX_GTFS_ZIP_BYTES
    )
    metadata = core.read_cache("gtfs-feed-metadata.json", max_age_seconds=86400)
    if archive is not None and isinstance(metadata, dict):
        digest = hashlib.sha256(archive).hexdigest()
        if (
            metadata.get("source_url") == STATIC_FEED
            and metadata.get("validated") is True
            and metadata.get("sha256") == digest
            and metadata.get("archive_bytes") == len(archive)
        ):
            try:
                return parse_gtfs_zip(archive)
            except (ValueError, zipfile.BadZipFile):
                pass
    data = download_gtfs()
    feed = parse_gtfs_zip(data)
    core.write_cache_bytes("gtfs-feed.zip", data)
    core.write_cache("gtfs-feed-metadata.json", {
        "source_url": STATIC_FEED,
        "retrieved_at": datetime.now(timezone.utc).isoformat(),
        "archive_bytes": len(data),
        "sha256": hashlib.sha256(data).hexdigest(),
        "validated": True,
        "tables": sorted(feed),
        "feed_info": feed.get("feed_info", [])[:1],
    })
    return feed


def get_routes(feed: dict[str, list[dict[str, str]]] | None = None) -> list[dict[str, str]]:
    """List all routes from the static feed."""
    if feed is None:
        feed = _load_feed_with_cache()
    return feed.get("routes", [])


def get_stops(feed: dict[str, list[dict[str, str]]] | None = None) -> list[dict[str, str]]:
    """List all stops from the static feed."""
    if feed is None:
        feed = _load_feed_with_cache()
    return feed.get("stops", [])


def _today_date() -> str:
    return date.today().strftime("%Y%m%d")


def _service_ids_for_date(feed: dict[str, list[dict[str, str]]], target_date: str) -> set[str]:
    """Return active service IDs for a date (YYYYMMDD)."""
    weekday = datetime.strptime(target_date, "%Y%m%d").strftime("%A").lower()
    calendar = feed.get("calendar", [])
    services: set[str] = set()
    for row in calendar:
        if row.get("start_date", "") <= target_date <= row.get("end_date", ""):
            if row.get(weekday, "0") == "1":
                services.add(row["service_id"])
    for row in feed.get("calendar_dates", []):
        if row.get("date") == target_date:
            if row.get("exception_type") == "1":
                services.add(row["service_id"])
            elif row.get("exception_type") == "2":
                services.discard(row["service_id"])
    return services


def get_schedule_for_route(
    route_id: str,
    target_date: str | None = None,
    feed: dict[str, list[dict[str, str]]] | None = None,
) -> list[dict[str, Any]]:
    """Return scheduled trips and stop times for a route on a date."""
    if feed is None:
        feed = _load_feed_with_cache()
    if target_date is None:
        target_date = _today_date()
    services = _service_ids_for_date(feed, target_date)
    trips = [t for t in feed.get("trips", []) if t.get("route_id") == route_id and t.get("service_id") in services]
    trip_ids = {t["trip_id"] for t in trips}
    times = [s for s in feed.get("stop_times", []) if s.get("trip_id") in trip_ids]
    times.sort(key=lambda s: (s.get("trip_id", ""), int(s.get("stop_sequence", 0) or 0)))
    return [
        {
            "trip_id": t.get("trip_id"),
            "stop_id": t.get("stop_id"),
            "stop_sequence": t.get("stop_sequence"),
            "arrival_time": t.get("arrival_time"),
            "departure_time": t.get("departure_time"),
        }
        for t in times
    ]


def get_arrivals_for_stop(
    stop_id: str,
    feed: dict[str, list[dict[str, str]]] | None = None,
) -> list[dict[str, Any]]:
    """Return scheduled arrivals for a stop."""
    if feed is None:
        feed = _load_feed_with_cache()
    target_date = _today_date()
    services = _service_ids_for_date(feed, target_date)
    trip_ids = {t["trip_id"] for t in feed.get("trips", []) if t.get("service_id") in services}
    times = [
        s
        for s in feed.get("stop_times", [])
        if s.get("stop_id") == stop_id and s.get("trip_id") in trip_ids
    ]
    times.sort(key=lambda s: s.get("arrival_time", ""))
    return [
        {
            "trip_id": t.get("trip_id"),
            "arrival_time": t.get("arrival_time"),
            "departure_time": t.get("departure_time"),
        }
        for t in times
    ]


def _decode_realtime(data: bytes) -> dict[str, Any]:
    """Decode GTFS-Realtime protobuf using vendored gtfs_realtime_pb2.

    Requires the optional ``google.protobuf`` runtime (>= 6.31.1, < 7).
    The vendored descriptor provides message definitions but does not replace
    the runtime.
    """
    try:
        from raleighlib import gtfs_realtime_pb2
        from google.protobuf.message import DecodeError
    except Exception as exc:
        raise ValueError(
            "GTFS-Realtime decoding requires google.protobuf>=6.31.1,<7"
        ) from exc

    msg = gtfs_realtime_pb2.FeedMessage()
    try:
        msg.ParseFromString(data)
    except DecodeError as exc:
        raise ValueError("GTFS-Realtime feed is malformed") from exc
    if not msg.IsInitialized() or not msg.HasField("header"):
        raise ValueError("GTFS-Realtime feed is missing required fields")
    if not msg.header.IsInitialized() or not msg.header.gtfs_realtime_version.strip():
        raise ValueError("GTFS-Realtime feed has an invalid required header")
    for entity in msg.entity:
        payloads = [
            field.name
            for field in entity.DESCRIPTOR.fields
            if field.message_type is not None
            and not field.is_repeated
            and entity.HasField(field.name)
        ]
        if entity.is_deleted:
            if payloads:
                raise ValueError(
                    "GTFS-Realtime deleted entity must not include a payload"
                )
        elif len(payloads) != 1:
            raise ValueError(
                "GTFS-Realtime entity must include exactly one payload"
            )
    return _protobuf_to_dict(msg)


def _protobuf_to_dict(msg) -> dict[str, Any]:
    """Minimal recursive converter for protobuf messages."""
    from google.protobuf.descriptor import FieldDescriptor

    result: dict[str, Any] = {}
    for field in msg.DESCRIPTOR.fields:
        value = getattr(msg, field.name)
        is_repeated = field.is_repeated
        if not is_repeated and not msg.HasField(field.name):
            continue
        if field.type == FieldDescriptor.TYPE_MESSAGE:
            if is_repeated:
                result[field.name] = [_protobuf_to_dict(v) for v in value]
            else:
                result[field.name] = _protobuf_to_dict(value)
        else:
            if is_repeated:
                result[field.name] = list(value)
            else:
                result[field.name] = value
    return result


def fetch_realtime(kind: str) -> dict[str, Any]:
    """Fetch and decode a GTFS-Realtime feed (alerts, trips, or vehicles)."""
    if kind not in {"alerts", "trips", "vehicles"}:
        raise ValueError(f"Invalid realtime kind: {kind}")
    url = f"{REALTIME_BASE}/{kind}"
    data = core.raw_request(url)
    return _decode_realtime(data)


def _entity_staleness(entity: dict[str, Any]) -> float | None:
    """Return seconds since the entity's timestamp if available."""
    ts = None
    for key in ("vehicle", "trip_update", "alert"):
        if key in entity:
            ts = entity[key].get("timestamp")
            break
    if ts:
        return datetime.now(timezone.utc).timestamp() - ts
    return None


def enrich_realtime_with_static(
    entities: list[dict[str, Any]],
    feed: dict[str, list[dict[str, str]]],
    header_timestamp: int | None = None,
) -> list[dict[str, Any]]:
    """Add static route/trip/stop names, feed timestamp, and staleness to realtime entities."""
    routes = {r.get("route_id"): r for r in feed.get("routes", [])}
    trips = {t.get("trip_id"): t for t in feed.get("trips", [])}
    stops = {s.get("stop_id"): s for s in feed.get("stops", [])}
    now = datetime.now(timezone.utc).timestamp()
    enriched: list[dict[str, Any]] = []
    for entity in entities:
        item = copy.deepcopy(entity)
        trip_id = None
        route_id = None
        if "vehicle" in entity:
            trip_id = entity["vehicle"].get("trip", {}).get("trip_id")
            route_id = entity["vehicle"].get("trip", {}).get("route_id")
        elif "trip_update" in entity:
            trip_id = entity["trip_update"].get("trip", {}).get("trip_id")
            route_id = entity["trip_update"].get("trip", {}).get("route_id")
        if trip_id and trip_id in trips:
            item["trip_id"] = trip_id
            route_id = route_id or trips[trip_id].get("route_id")
            item["route_id"] = route_id
        if route_id and route_id in routes:
            item["route_short_name"] = routes[route_id].get("route_short_name")
            item["route_long_name"] = routes[route_id].get("route_long_name")
        if "trip_update" in entity:
            for update in item["trip_update"].get("stop_time_update", []):
                sid = update.get("stop_id")
                if sid and sid in stops:
                    update["stop_name"] = stops[sid].get("stop_name")
        if "alert" in item:
            for informed in item["alert"].get("informed_entity", []):
                route_id = informed.get("route_id")
                trip_id = informed.get("trip", {}).get("trip_id")
                stop_id = informed.get("stop_id")
                if route_id and route_id in routes:
                    informed["route_short_name"] = routes[route_id].get("route_short_name")
                    informed["route_long_name"] = routes[route_id].get("route_long_name")
                if trip_id and trip_id in trips:
                    informed["trip_headsign"] = trips[trip_id].get("trip_headsign")
                    informed["trip_route_id"] = trips[trip_id].get("route_id")
                if stop_id and stop_id in stops:
                    informed["stop_name"] = stops[stop_id].get("stop_name")
        item["feed_header_timestamp"] = header_timestamp
        entity_ts = _entity_staleness(entity)
        if header_timestamp is not None:
            item["staleness_seconds"] = now - header_timestamp
        elif entity_ts is not None:
            item["staleness_seconds"] = entity_ts
        else:
            item["staleness_seconds"] = None
        enriched.append(item)
    return enriched


def _realtime_envelope(
    feed: dict[str, Any], entities: list[dict[str, Any]]
) -> dict[str, Any]:
    header_ts = feed.get("header", {}).get("timestamp")
    now = datetime.now(timezone.utc).timestamp()
    return {
        "feed_timestamp": header_ts,
        "staleness_seconds": now - header_ts if header_ts else None,
        "entities": entities,
    }


def get_alerts(limit: int = 20) -> dict[str, Any]:
    """Fetch and return service alerts."""
    core.require_positive_limit(limit)
    feed = fetch_realtime("alerts")
    static = _load_feed_with_cache()
    header_ts = feed.get("header", {}).get("timestamp")
    entities = enrich_realtime_with_static(feed.get("entity", [])[:limit], static, header_ts)
    return _realtime_envelope(feed, entities)


def get_trip_updates(route: str | None = None, limit: int = 20) -> dict[str, Any]:
    """Fetch and return trip updates, optionally filtered by route."""
    core.require_positive_limit(limit)
    feed = fetch_realtime("trips")
    entities = feed.get("entity", [])
    if route:
        static = _load_feed_with_cache()
        trips = {t.get("trip_id"): t for t in static.get("trips", [])}
        entities = [
            e for e in entities
            if trips.get(e.get("trip_update", {}).get("trip", {}).get("trip_id"), {}).get("route_id") == route
        ]
    static = _load_feed_with_cache()
    header_ts = feed.get("header", {}).get("timestamp")
    enriched = enrich_realtime_with_static(entities[:limit], static, header_ts)
    return _realtime_envelope(feed, enriched)


def filter_vehicle_positions(
    entities: list[dict[str, Any]],
    feed: dict[str, list[dict[str, str]]],
    route: str | None = None,
    stop: str | None = None,
    limit: int = 20,
) -> list[dict[str, Any]]:
    """Filter and limit vehicle positions by route/stop."""
    core.require_positive_limit(limit)
    trips = {t.get("trip_id"): t for t in feed.get("trips", [])}
    routes = {r.get("route_id"): r for r in feed.get("routes", [])}
    filtered: list[dict[str, Any]] = []
    for entity in entities:
        vehicle = entity.get("vehicle", {})
        trip_id = vehicle.get("trip", {}).get("trip_id")
        stop_id = vehicle.get("stop_id")
        trip = trips.get(trip_id) if trip_id else None
        route_id = trip.get("route_id") if trip else vehicle.get("trip", {}).get("route_id")
        if route and route_id != route:
            continue
        if stop and stop_id != stop:
            continue
        if trip_id:
            entity["trip_id"] = trip_id
        if route_id:
            entity["route_id"] = route_id
        if route_id and route_id in routes:
            entity["route_short_name"] = routes[route_id].get("route_short_name")
            entity["route_long_name"] = routes[route_id].get("route_long_name")
        filtered.append(entity)
        if len(filtered) >= limit:
            break
    return filtered


def get_vehicle_positions(route: str | None = None, limit: int = 20) -> dict[str, Any]:
    """Fetch and return vehicle positions, optionally filtered by route."""
    core.require_positive_limit(limit)
    feed = fetch_realtime("vehicles")
    entities = feed.get("entity", [])
    static = _load_feed_with_cache()
    filtered = filter_vehicle_positions(entities, static, route=route, limit=limit)
    header_ts = feed.get("header", {}).get("timestamp")
    enriched = enrich_realtime_with_static(filtered, static, header_ts)
    return _realtime_envelope(feed, enriched)

"""Raleigh CLI implementation package."""

__version__ = "2.0.0"

{"features": []}

{"features": [], "exceededTransferLimit": false}

{
  "features": [
    {
      "attributes": {
        "incident_number": "26-032170",
        "dispatch_date_time": 1784866612000,
        "arrive_date_time": 1784866814000,
        "cleared_date_time": 1784868386000,
        "address": "505 FLORENCE ST",
        "station_name": "Station 01",
        "platoon": "A",
        "incident_group_name": "Hazardous Situation",
        "incident_subgroup_code": "Hazardous Materials",
        "incident_type_name": "Gas Leak / Gas Odor"
      }
    }
  ]
}

{
  "fields": [
    {"name": "incident_number"},
    {"name": "dispatch_date_time"}
  ]
}

{"error": {"code": 500, "message": "Service unavailable", "details": []}}

{
  "data": {
    "type": "node--service",
    "id": "f95a0f43-3dbf-4378-b7c7-b1bdda20eb24",
    "attributes": {
      "status": true,
      "title": "View Raleigh Fire Statistics",
      "changed": "2026-04-23T13:21:39+00:00",
      "path": {"alias": "/fire/services/view-raleigh-fire-statistics"}
    },
    "relationships": {"field_content_primary": {"data": [
      {"type": "paragraph--stories_text", "id": "current"},
      {"type": "paragraph--stories_text", "id": "history"},
      {"type": "paragraph--stories_text", "id": "quarter"},
      {"type": "paragraph--stories_text", "id": "sprinkler"}
    ]}}
  },
  "included": [
    {
      "type": "paragraph--stories_text",
      "id": "current",
      "attributes": {
        "status": true,
        "field_heading": "2026 Statistics",
        "field_stories_text_formatted": {"value": "<table><tr><td><strong>Incident Type</strong></td><td><strong>Totals</strong></td></tr><tr><td><strong>Fire</strong></td><td>401</td></tr><tr><td><strong>Medical</strong></td><td>7,882</td></tr></table>"}
      }
    },
    {
      "type": "paragraph--stories_text",
      "id": "history",
      "attributes": {
        "status": true,
        "field_heading": "Previous Years Statistics",
        "field_stories_text_formatted": {"value": "<p><a href=\"https://cityofraleigh0drupal.blob.core.usgovcloudapi.net/drupal-prod/COR18/2026.pdf\">2026</a> | <a href=\"https://cityofraleigh0drupal.blob.core.usgovcloudapi.net/drupal-prod/COR18/2025.pdf\">2025</a></p>"}
      }
    },
    {
      "type": "paragraph--stories_text",
      "id": "quarter",
      "attributes": {
        "status": true,
        "field_heading": "Quarterly Report",
        "field_stories_text_formatted": {"value": "<p>View <a href=\"/fire/news/keeping-score-q1-2025-fire-statistics\">fire statistics from the previous quarter</a>.</p>"}
      }
    },
    {
      "type": "paragraph--stories_text",
      "id": "sprinkler",
      "attributes": {
        "status": true,
        "field_heading": "Sprinkler Saves Stats",
        "field_stories_text_formatted": {"value": "<p><strong>2024 Sprinkler Saves Statistics</strong></p><table><tr><th>Type</th><th>Description</th><th>Statistic</th><th>Percentage</th></tr><tr><td>Fire Sprinkler Protected Properties</td><td>Total Property Value</td><td>$365,426,749</td><td>n/a</td></tr><tr><td></td><td>Total Number of Sprinkler Saves 2024 (YTD)</td><td>20</td><td>n/a</td></tr></table>"}
      }
    }
  ]
}

{
  "data": {
    "type": "node--service",
    "id": "40ebbee4-2477-4f7d-9623-257685345e3d",
    "attributes": {
      "status": true,
      "title": "Raleigh's Crime Data",
      "changed": "2026-07-10T13:59:08+00:00",
      "path": {"alias": "/police/services/raleighs-crime-data"}
    },
    "relationships": {"field_content_primary": {"data": [{"type": "paragraph--stories_text", "id": "summary"}]}}
  },
  "included": [
    {
      "type": "paragraph--stories_text",
      "id": "summary",
      "attributes": {
        "status": true,
        "field_heading": "Summary of Crime Statistics by Year",
        "field_stories_text_formatted": {
          "value": "<h5><span>2026</span></h5><ul><li><a href=\"https://cityofraleigh0drupal.blob.core.usgovcloudapi.net/drupal-prod/COR23/rpd-crime-data-26q1.pdf\">Q1 stats</a> (PDF)</li></ul><h5>2025</h5><ul><li><a href=\"https://cityofraleigh0drupal.blob.core.usgovcloudapi.net/drupal-prod/COR23/annual-crime-data-2025.pdf\">Annual Crime Data</a> (PDF)</li><li><a href=\"https://cityofraleigh0drupal.blob.core.usgovcloudapi.net/drupal-prod/COR23/raleigh-crime-data-q4-2025.pdf\"><span>Q4 stats</span></a> (PDF)</li></ul>"
        }
      }
    }
  ]
}

[
  {
    "jurisdiction": "Raleigh Police Department",
    "problem": "MVC - Fatal",
    "address": "Blue Ridge Rd / Macon Pond Rd",
    "lat": 35.81933,
    "long": -78.704862,
    "timestamp": "2026-07-24 22:28:54.000"
  },
  {
    "jurisdiction": "Raleigh Fire Department",
    "problem": "Structure Fire",
    "address": "100 Block Example St",
    "lat": 35.78,
    "long": -78.64,
    "timestamp": "2026-07-24 21:15:00.000"
  },
  {
    "jurisdiction": "Raleigh Police Department",
    "problem": "Larceny",
    "address": "200 Block Sample Ave",
    "lat": 35.79,
    "long": -78.65,
    "timestamp": "2026-07-24 20:45:00.000"
  },
  {
    "jurisdiction": "Raleigh Fire Department",
    "problem": "EMS Call",
    "address": "300 Block Test Blvd",
    "lat": null,
    "long": null,
    "timestamp": "2026-07-24 20:30:00.000"
  },
  {
    "jurisdiction": "Raleigh Police Department",
    "problem": "MVC - Fatal",
    "address": "Blue Ridge Rd / Macon Pond Rd",
    "lat": 35.81933,
    "long": -78.704862,
    "timestamp": "2026-07-24 22:28:54.000"
  }
]

[
  {"jurisdiction": "Raleigh Police Department", "problem": "Test", "address": "1 St", "lat": 35.7, "long": -78.6, "timestamp": "2026-07-24 10:00:00.000"},
  {"jurisdiction": "", "problem": "Bad", "address": "2 St", "lat": 35.7, "long": -78.6, "timestamp": "2026-07-24 10:01:00.000"},
  {"problem": "No Jurisdiction", "address": "3 St", "lat": 35.7, "long": -78.6, "timestamp": "2026-07-24 10:02:00.000"},
  "not-a-record",
  {"jurisdiction": "Raleigh Fire Department", "problem": "Valid Fire", "address": "4 St", "lat": 999, "long": -78.6, "timestamp": "2026-07-24 10:03:00.000"}
]

{"status": "ok", "data": []}