"""Explicit read-only Geofabrik acquisition for approved pilot regions."""

from __future__ import annotations

import hashlib
import json
import os
import re
import stat
import tempfile
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable
from urllib.parse import urlsplit

from ..errors import PbfDownloadRejected

_GEOFABRIK_HOST = "download.geofabrik.de"
_CHUNK_SIZE = 1024 * 1024
_MAX_CHECKSUM_BYTES = 4096
_MD5_LINE = re.compile(r"^([0-9A-Fa-f]{32})(?:\s+[* ]?[^\s]+)?$")
_PBF_REDIRECT_CODE = 302
_DATED_PBF_NAME = re.compile(r"^[a-z0-9-]+-[0-9]{6}\.osm\.pbf$")


def _utc_now() -> datetime:
    return datetime.now(timezone.utc)


def _aware_utc(value: datetime, field_name: str) -> datetime:
    if not isinstance(value, datetime) or value.tzinfo is None:
        raise PbfDownloadRejected(f"{field_name} must be timezone-aware")
    return value.astimezone(timezone.utc)


def _approved_url(value: str, *, suffix: str) -> str:
    parsed = urlsplit(value)
    if (
        parsed.scheme != "https"
        or parsed.hostname != _GEOFABRIK_HOST
        or parsed.username is not None
        or parsed.password is not None
        or parsed.query
        or parsed.fragment
        or not parsed.path.endswith(suffix)
    ):
        raise PbfDownloadRejected("Geofabrik URL is outside the approved boundary")
    return value


@dataclass(frozen=True, slots=True)
class GeofabrikRegion:
    """One immutable, explicitly approved Geofabrik source region."""

    region_id: str
    display_name: str
    country_code: str
    calling_code: str
    pbf_url: str
    checksum_url: str
    max_size_bytes: int

    def __post_init__(self) -> None:
        if not isinstance(self.region_id, str) or not re.fullmatch(
            r"[a-z0-9][a-z0-9-]{0,127}", self.region_id
        ):
            raise PbfDownloadRejected("region_id has an invalid shape")
        if (
            not isinstance(self.display_name, str)
            or not self.display_name.strip()
            or len(self.display_name.strip()) > 256
        ):
            raise PbfDownloadRejected("display_name must be a bounded nonblank string")
        object.__setattr__(self, "display_name", self.display_name.strip())
        if not isinstance(self.country_code, str) or not re.fullmatch(
            r"[A-Z]{2}", self.country_code
        ):
            raise PbfDownloadRejected("country_code must be uppercase ISO alpha-2")
        if not isinstance(self.calling_code, str) or not re.fullmatch(
            r"[1-9][0-9]{0,3}", self.calling_code
        ):
            raise PbfDownloadRejected("calling_code must contain 1 to 4 digits")
        pbf_url = _approved_url(self.pbf_url, suffix="-latest.osm.pbf")
        checksum_url = _approved_url(self.checksum_url, suffix="-latest.osm.pbf.md5")
        if checksum_url != pbf_url + ".md5":
            raise PbfDownloadRejected("checksum URL must be the PBF URL plus .md5")
        if (
            isinstance(self.max_size_bytes, bool)
            or not isinstance(self.max_size_bytes, int)
            or not 1 <= self.max_size_bytes <= 4 * 1024 * 1024 * 1024
        ):
            raise PbfDownloadRejected("region size bound is invalid")

    def as_mapping(self) -> dict[str, object]:
        return {
            "calling_code": self.calling_code,
            "country_code": self.country_code,
            "display_name": self.display_name,
            "max_size_bytes": self.max_size_bytes,
            "pbf_url": self.pbf_url,
            "region_id": self.region_id,
        }


AU_ACT_PILOT = GeofabrikRegion(
    region_id="au-act",
    display_name="Australian Capital Territory",
    country_code="AU",
    calling_code="61",
    pbf_url="https://download.geofabrik.de/australia-oceania/australia/act-latest.osm.pbf",
    checksum_url=(
        "https://download.geofabrik.de/australia-oceania/australia/act-latest.osm.pbf.md5"
    ),
    max_size_bytes=32 * 1024 * 1024,
)


def approved_region(region_id: str) -> GeofabrikRegion:
    """Resolve only the current explicitly approved pilot region."""

    if region_id.strip().casefold() != AU_ACT_PILOT.region_id:
        raise PbfDownloadRejected("requested Geofabrik region is not approved")
    return AU_ACT_PILOT


def _approved_pbf_redirect(region: GeofabrikRegion, value: str) -> str:
    """Validate the single dated-file redirect used by Geofabrik latest links."""

    source = urlsplit(region.pbf_url)
    target = urlsplit(value)
    try:
        target_port = target.port
    except ValueError as exc:
        raise PbfDownloadRejected("Geofabrik redirect port is invalid") from exc
    if (
        target.scheme != "https"
        or target.hostname != _GEOFABRIK_HOST
        or target_port is not None
        or target.username is not None
        or target.password is not None
        or target.query
        or target.fragment
    ):
        raise PbfDownloadRejected("Geofabrik redirect is outside the approved boundary")

    source_directory, separator, source_name = source.path.rpartition("/")
    target_directory, target_separator, target_name = target.path.rpartition("/")
    latest_suffix = "-latest.osm.pbf"
    if (
        not separator
        or not target_separator
        or not source_name.endswith(latest_suffix)
        or source_directory != target_directory
        or not _DATED_PBF_NAME.fullmatch(target_name)
        or not target_name.startswith(source_name.removesuffix(latest_suffix) + "-")
    ):
        raise PbfDownloadRejected("Geofabrik redirect path is not an approved dated PBF")
    return value


class _RejectRedirects(urllib.request.HTTPRedirectHandler):
    def redirect_request(
        self,
        req: urllib.request.Request,
        fp: Any,
        code: int,
        msg: str,
        headers: Any,
        newurl: str,
    ) -> urllib.request.Request | None:
        del req, fp, code, msg, headers, newurl
        return None


def _default_opener() -> urllib.request.OpenerDirector:
    return urllib.request.build_opener(
        urllib.request.ProxyHandler({}),
        _RejectRedirects(),
    )


def _pbf_request(url: str) -> urllib.request.Request:
    return urllib.request.Request(
        url,
        headers={
            "Accept": "application/octet-stream",
            "Accept-Encoding": "identity",
            "User-Agent": "osm-lead-source-service/0.1",
        },
        method="GET",
    )


def _open_pbf_response(
    region: GeofabrikRegion,
    *,
    timeout_seconds: float,
    opener: Any,
) -> tuple[Any, str]:
    """Open the approved PBF URL, permitting exactly one constrained redirect."""

    try:
        return (
            opener.open(_pbf_request(region.pbf_url), timeout=timeout_seconds),
            region.pbf_url,
        )
    except urllib.error.HTTPError as exc:
        try:
            if exc.code != _PBF_REDIRECT_CODE:
                raise PbfDownloadRejected("Geofabrik PBF request failed") from exc
            location = exc.headers.get("Location") if exc.headers is not None else None
            if not isinstance(location, str) or not location.strip():
                raise PbfDownloadRejected("Geofabrik redirect has no Location header")
            redirect_url = _approved_pbf_redirect(region, location.strip())
        finally:
            if exc.fp is not None:
                exc.close()

    try:
        response = opener.open(_pbf_request(redirect_url), timeout=timeout_seconds)
    except urllib.error.HTTPError as exc:
        try:
            if exc.code in {301, 302, 303, 307, 308}:
                raise PbfDownloadRejected("Geofabrik PBF redirected more than once") from exc
            raise PbfDownloadRejected("Geofabrik PBF request failed") from exc
        finally:
            if exc.fp is not None:
                exc.close()
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        raise PbfDownloadRejected("Geofabrik PBF request failed") from exc
    return response, redirect_url


def _ensure_directory(path: Path, field_name: str) -> None:
    path.mkdir(mode=0o700, parents=True, exist_ok=True)
    try:
        info = path.lstat()
    except OSError as exc:
        raise PbfDownloadRejected(f"{field_name} cannot be inspected") from exc
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
        raise PbfDownloadRejected(f"{field_name} must be a regular directory")


def _regular_file(path: Path, field_name: str) -> os.stat_result:
    try:
        info = path.lstat()
    except OSError as exc:
        raise PbfDownloadRejected(f"{field_name} cannot be inspected") from exc
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
        raise PbfDownloadRejected(f"{field_name} must be a regular file")
    return info


def _hash_file(path: Path, maximum: int) -> tuple[int, str, str]:
    before = _regular_file(path, "cached PBF")
    if not 1 <= before.st_size <= maximum:
        raise PbfDownloadRejected("cached PBF size is outside the approved bound")
    sha256 = hashlib.sha256()
    md5 = hashlib.md5(usedforsecurity=False)
    total = 0
    try:
        with path.open("rb") as handle:
            while chunk := handle.read(_CHUNK_SIZE):
                total += len(chunk)
                if total > maximum:
                    raise PbfDownloadRejected("cached PBF exceeds the approved bound")
                sha256.update(chunk)
                md5.update(chunk)
    except OSError as exc:
        raise PbfDownloadRejected("cached PBF cannot be read") from exc
    after = _regular_file(path, "cached PBF")
    identity_before = (
        before.st_dev,
        before.st_ino,
        before.st_size,
        before.st_mtime_ns,
        before.st_ctime_ns,
    )
    identity_after = (
        after.st_dev,
        after.st_ino,
        after.st_size,
        after.st_mtime_ns,
        after.st_ctime_ns,
    )
    if identity_before != identity_after or total != before.st_size:
        raise PbfDownloadRejected("cached PBF changed during validation")
    return total, sha256.hexdigest(), md5.hexdigest()


def _atomic_json(path: Path, value: object) -> None:
    if path.exists() and path.is_symlink():
        raise PbfDownloadRejected("cache metadata cannot be a symbolic link")
    fd, name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
    temporary = Path(name)
    try:
        os.chmod(temporary, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(value, handle, sort_keys=True, separators=(",", ":"))
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    except OSError as exc:
        raise PbfDownloadRejected("cache metadata cannot be written") from exc
    finally:
        temporary.unlink(missing_ok=True)


def _read_metadata(path: Path) -> dict[str, object] | None:
    if not path.exists():
        return None
    _regular_file(path, "cache metadata")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise PbfDownloadRejected("cache metadata is invalid") from exc
    if not isinstance(value, dict):
        raise PbfDownloadRejected("cache metadata must be an object")
    return value


def _fetch_md5(
    region: GeofabrikRegion,
    *,
    timeout_seconds: float,
    opener: Any,
) -> str:
    request = urllib.request.Request(
        region.checksum_url,
        headers={
            "Accept": "text/plain",
            "Accept-Encoding": "identity",
            "User-Agent": "osm-lead-source-service/0.1",
        },
        method="GET",
    )
    try:
        with opener.open(request, timeout=timeout_seconds) as response:
            if response.status != 200 or response.geturl() != region.checksum_url:
                raise PbfDownloadRejected("Geofabrik checksum response changed URL or status")
            content = response.read(_MAX_CHECKSUM_BYTES + 1)
    except PbfDownloadRejected:
        raise
    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, OSError) as exc:
        raise PbfDownloadRejected("Geofabrik checksum request failed") from exc
    if len(content) > _MAX_CHECKSUM_BYTES:
        raise PbfDownloadRejected("Geofabrik checksum response exceeds the bounded size")
    try:
        text = content.decode("ascii").strip()
    except UnicodeDecodeError:
        raise PbfDownloadRejected("Geofabrik checksum response is not ASCII") from None
    match = _MD5_LINE.fullmatch(text)
    if match is None:
        raise PbfDownloadRejected("Geofabrik checksum response has an invalid shape")
    return match.group(1).casefold()


@dataclass(frozen=True, slots=True)
class AcquiredPbf:
    """Validated evidence for one downloaded or cached PBF."""

    region_id: str
    source_url: str
    local_path: Path
    content_sha256: str
    companion_md5: str
    size_bytes: int
    acquired_at: datetime
    validated_at: datetime
    cache_hit: bool

    def as_mapping(self) -> dict[str, object]:
        return {
            "acquired_at": self.acquired_at.isoformat(),
            "cache_hit": self.cache_hit,
            "companion_md5": self.companion_md5,
            "content_sha256": self.content_sha256,
            "local_path": str(self.local_path),
            "region_id": self.region_id,
            "size_bytes": self.size_bytes,
            "source_url": self.source_url,
            "validated_at": self.validated_at.isoformat(),
        }


def acquire_region(
    region: GeofabrikRegion,
    cache_dir: Path,
    *,
    timeout_seconds: float = 120.0,
    opener: Any | None = None,
    clock: Callable[[], datetime] = _utc_now,
) -> AcquiredPbf:
    """Fetch and validate an approved Geofabrik PBF without any Noco mutation."""

    if not isinstance(region, GeofabrikRegion):
        raise PbfDownloadRejected("region must be GeofabrikRegion")
    if (
        isinstance(timeout_seconds, bool)
        or not isinstance(timeout_seconds, (int, float))
        or not 5 <= float(timeout_seconds) <= 900
    ):
        raise PbfDownloadRejected("timeout_seconds is outside the approved range")
    root = cache_dir.expanduser()
    _ensure_directory(root, "cache directory")
    region_dir = root / region.region_id
    _ensure_directory(region_dir, "region cache directory")
    metadata_path = region_dir / "latest.json"

    active_opener = opener or _default_opener()
    expected_md5 = _fetch_md5(
        region,
        timeout_seconds=float(timeout_seconds),
        opener=active_opener,
    )
    now = _aware_utc(clock(), "validated_at")
    metadata = _read_metadata(metadata_path)
    if metadata is not None and metadata.get("companion_md5") == expected_md5:
        raw_sha256 = metadata.get("content_sha256")
        if not isinstance(raw_sha256, str) or not re.fullmatch(r"[0-9a-f]{64}", raw_sha256):
            raise PbfDownloadRejected("cache metadata has an invalid SHA-256")
        cached_path = region_dir / f"{raw_sha256}.osm.pbf"
        size, actual_sha256, actual_md5 = _hash_file(cached_path, region.max_size_bytes)
        if actual_sha256 != raw_sha256 or actual_md5 != expected_md5:
            raise PbfDownloadRejected("cached PBF does not match current checksum evidence")
        acquired_at = datetime.fromtimestamp(cached_path.stat().st_mtime, timezone.utc)
        raw_acquired_at = metadata.get("acquired_at")
        if isinstance(raw_acquired_at, str):
            try:
                acquired_at = datetime.fromisoformat(raw_acquired_at).astimezone(timezone.utc)
            except ValueError:
                pass
        return AcquiredPbf(
            region.region_id,
            region.pbf_url,
            cached_path,
            actual_sha256,
            actual_md5,
            size,
            acquired_at,
            now,
            True,
        )

    temporary_path: Path | None = None
    acquired_at = _aware_utc(clock(), "acquired_at")
    try:
        opened_response, expected_response_url = _open_pbf_response(
            region,
            timeout_seconds=float(timeout_seconds),
            opener=active_opener,
        )
        with opened_response as response:
            if response.status != 200 or response.geturl() != expected_response_url:
                raise PbfDownloadRejected("Geofabrik PBF response changed URL or status")
            content_encoding = response.headers.get("Content-Encoding")
            if content_encoding not in (None, "", "identity"):
                raise PbfDownloadRejected("HTTP content encoding is not accepted")
            raw_length = response.headers.get("Content-Length")
            try:
                content_length = int(raw_length) if raw_length is not None else 0
            except ValueError:
                raise PbfDownloadRejected("Geofabrik Content-Length is invalid") from None
            if not 1 <= content_length <= region.max_size_bytes:
                raise PbfDownloadRejected("Geofabrik Content-Length exceeds the pilot bound")

            sha256 = hashlib.sha256()
            md5 = hashlib.md5(usedforsecurity=False)
            total = 0
            with tempfile.NamedTemporaryFile(
                mode="wb",
                dir=region_dir,
                prefix=".download-",
                suffix=".part",
                delete=False,
            ) as temporary:
                temporary_path = Path(temporary.name)
                os.chmod(temporary_path, 0o600)
                while chunk := response.read(_CHUNK_SIZE):
                    total += len(chunk)
                    if total > content_length or total > region.max_size_bytes:
                        raise PbfDownloadRejected("Geofabrik download exceeds the approved size")
                    sha256.update(chunk)
                    md5.update(chunk)
                    temporary.write(chunk)
                temporary.flush()
                os.fsync(temporary.fileno())
        if total != content_length:
            raise PbfDownloadRejected("Geofabrik download size differs from Content-Length")
        if md5.hexdigest() != expected_md5:
            raise PbfDownloadRejected("Geofabrik PBF does not match companion checksum")

        content_sha256 = sha256.hexdigest()
        target = region_dir / f"{content_sha256}.osm.pbf"
        if target.exists():
            size, existing_sha256, existing_md5 = _hash_file(target, region.max_size_bytes)
            if size != total or existing_sha256 != content_sha256 or existing_md5 != expected_md5:
                raise PbfDownloadRejected("existing content-addressed cache entry is invalid")
        else:
            try:
                os.link(temporary_path, target, follow_symlinks=False)
            except FileExistsError:
                raise PbfDownloadRejected("cache target appeared concurrently") from None
        validated_at = _aware_utc(clock(), "validated_at")
        evidence = AcquiredPbf(
            region.region_id,
            region.pbf_url,
            target,
            content_sha256,
            expected_md5,
            total,
            acquired_at,
            validated_at,
            False,
        )
        _atomic_json(metadata_path, evidence.as_mapping())
        return evidence
    except PbfDownloadRejected:
        raise
    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, OSError) as exc:
        raise PbfDownloadRejected("Geofabrik PBF download failed") from exc
    finally:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)


__all__ = [
    "AU_ACT_PILOT",
    "AcquiredPbf",
    "GeofabrikRegion",
    "acquire_region",
    "approved_region",
]