from __future__ import annotations

import argparse
import fcntl
import json
import os
import sys
import time
import tempfile
import urllib.error
from dataclasses import asdict
from pathlib import Path
from urllib.parse import urlsplit, urljoin
from urllib.request import Request, urlopen, HTTPRedirectHandler, build_opener

from .community_sources import (
    parse_rf4db_catches,
    parse_rf4db_waterbodies,
    parse_rf4db_waterbody_detail,
    parse_rf4map_point,
    parse_rf4posts_spot,
    parse_rf4stat_fishing,
    parse_rf4stat_posts,
)


SOURCES = {
    "rf4db": ("https://download.rf4db.com/ru/catches", parse_rf4db_catches),
    "rf4db-waterbodies": ("https://rf4db.com/ru/maps", parse_rf4db_waterbodies),
    "rf4stat-fishing": ("https://rf4-stat.ru/fishing/", parse_rf4stat_fishing),
    "rf4stat-posts": ("https://rf4-stat.ru/posts/", parse_rf4stat_posts),
}
DETAIL_SOURCES = {
    "rf4db-waterbody": parse_rf4db_waterbody_detail,
    "rf4map-point": parse_rf4map_point,
    "rf4posts-spot": parse_rf4posts_spot,
}
USER_AGENT = "RF4-Spotter/0.1 (authorized data integration)"
MIN_FETCH_INTERVAL_SECONDS = 30 * 60
DEFAULT_STATE_FILE = Path(".cache/community-fetch-state.json")
ALLOWED_HOSTS = frozenset({
    "download.rf4db.com", "rf4db.com", "oss.rf4db.com",
    "rf4-stat.ru",
    "rf4map.ru", "gw.rf4map.ru", "hb.ru-msk.vkcloud-storage.ru",
    "rf4-posts.com",
    "rf4game.de", "rf4game.ru",
})
MAX_RESPONSE_BYTES = 5 * 1024 * 1024  # 5 MB


def _read_state(state_file: Path) -> dict:
    """Read state file with shared lock; return empty dict if missing/corrupt.

    Validates that the parsed JSON is a flat dict of string->number pairs.
    Returns {} for missing, corrupt, or structurally invalid files.
    """
    try:
        with open(state_file, "r") as f:
            fcntl.flock(f, fcntl.LOCK_SH)
            try:
                data = json.loads(f.read())
                if not isinstance(data, dict):
                    return {}
                # Validate structure: flat dict of string->number
                for key, value in data.items():
                    if not isinstance(key, str):
                        return {}
                    if not isinstance(value, (int, float)):
                        return {}
                return data
            finally:
                fcntl.flock(f, fcntl.LOCK_UN)
    except (FileNotFoundError, json.JSONDecodeError, ValueError, OSError):
        return {}


def _write_state(state_file: Path, state: dict) -> None:
    """A02: Atomic write with exclusive lock, flush before unlock.

    Writes to temp file first, then renames atomically. Lock is acquired
    before any file operations to prevent race conditions.
    """
    state_file.parent.mkdir(parents=True, exist_ok=True)
    temp_file = state_file.with_suffix(".tmp")
    with open(temp_file, "w") as f:
        fcntl.flock(f, fcntl.LOCK_EX)
        try:
            f.write(json.dumps(state, sort_keys=True))
            f.flush()
            os.fsync(f.fileno())
        finally:
            fcntl.flock(f, fcntl.LOCK_UN)
    temp_file.replace(state_file)


def fetch_site_key(url: str) -> str:
    """Return a stable cooldown key shared by all endpoints of one site.

    Normalizes hostname to a common key for related domains:
    - Strips common subdomains (www, download, api, cdn)
    - Keeps the base domain as the key
    """
    hostname = (urlsplit(url).hostname or "").lower()
    if hostname.startswith("www."):
        hostname = hostname[4:]
    if hostname.endswith(".rf4db.com"):
        hostname = "rf4db.com"
    elif hostname.startswith("download."):
        hostname = hostname[9:]
    elif hostname.startswith("api."):
        hostname = hostname[4:]
    elif hostname.startswith("cdn."):
        hostname = hostname[4:]
    if hostname == "gw.rf4map.ru":
        hostname = "rf4map.ru"
    if not hostname:
        raise ValueError("source URL must include a hostname")
    return hostname


def enforce_fetch_interval(
    source: str, *, state_file: Path, now: float | None = None,
) -> None:
    """Legacy: use check_and_reserve instead for atomic check-and-reserve."""
    check_and_reserve(source, state_file=state_file, now=now)


def mark_fetch(source: str, *, state_file: Path, now: float | None = None) -> None:
    """Legacy: use check_and_reserve instead for atomic check-and-reserve."""
    check_and_reserve(source, state_file=state_file, now=now)


def check_and_reserve(
    source: str, *, state_file: Path, now: float | None = None,
) -> None:
    """A02: Atomic check-and-reserve under a single exclusive lock.

    Opens state file with exclusive lock, reads state, checks cooldown,
    reserves if allowed — all in one critical section. Uses lockfile
    pattern for cross-process coordination.

    Handles corrupt/missing state files gracefully by treating them as empty.
    """
    if now is None:
        now = time.time()
    state_file.parent.mkdir(parents=True, exist_ok=True)
    lock_file = state_file.with_suffix(".lock")
    # Create lock file if not exists
    lock_file.touch(exist_ok=True)
    with open(lock_file, "w") as lf:
        fcntl.flock(lf, fcntl.LOCK_EX)
        try:
            # Read state under lock; validate structure
            try:
                with open(state_file, "r") as sf:
                    raw = sf.read()
                state = json.loads(raw) or {}
                if not isinstance(state, dict):
                    state = {}
                for key, value in state.items():
                    if not isinstance(key, str) or not isinstance(value, (int, float)):
                        state = {}
                        break
            except (FileNotFoundError, json.JSONDecodeError, ValueError, OSError):
                state = {}
            # Check cooldown
            last_fetch = state.get(source)
            if isinstance(last_fetch, (int, float)) and now - last_fetch < MIN_FETCH_INTERVAL_SECONDS:
                wait = int(MIN_FETCH_INTERVAL_SECONDS - (now - last_fetch))
                raise RuntimeError(f"source cooldown is active; retry in {wait} seconds")
            # Reserve
            state[source] = now
            # Write atomically via temp file
            temp_file = state_file.with_suffix(".tmp")
            with open(temp_file, "w") as sf:
                sf.write(json.dumps(state, sort_keys=True))
                sf.flush()
                os.fsync(sf.fileno())
            temp_file.replace(state_file)
        finally:
            fcntl.flock(lf, fcntl.LOCK_UN)


def _mark_fetch_only(source: str, state: dict, now: float) -> dict:
    """Internal: update state without cooldown check (for internal use)."""
    state[source] = now
    return state


MAX_REDIRECT_HOPS = 5
ALLOWED_PORTS = frozenset({443, 80})


class _StrictRedirectHandler(HTTPRedirectHandler):
    """A03: Raise on redirect instead of following automatically.

    Returns the HTTPError (3xx) so the caller can validate each hop
    before deciding whether to follow.
    """

    def http_error_302(self, req, fp, code, msg, headers):
        """A03: Intercept 302 — don't follow automatically."""
        return None

    def http_error_301(self, req, fp, code, msg, headers):
        """A03: Intercept 301 — don't follow automatically."""
        return None

    def http_error_303(self, req, fp, code, msg, headers):
        """A03: Intercept 303 — don't follow automatically."""
        return None

    def http_error_307(self, req, fp, code, msg, headers):
        """A03: Intercept 307 — don't follow automatically."""
        return None

    def http_error_308(self, req, fp, code, msg, headers):
        """A03: Intercept 308 — don't follow automatically."""
        return None


def _extract_redirect_url(headers) -> str | None:
    """A03: Extract Location header from redirect response."""
    # Try 'Location' first (RFC standard), then 'location' (lowercase)
    for key in ("Location", "location"):
        if key in headers:
            return headers[key]
    return None


def _validate_url_before_io(url: str) -> tuple[str, str]:
    """A03: Validate scheme, host, port before any network I/O.

    Returns (normalized_hostname, scheme). Raises ValueError for disallowed
    schemes, ports, or hosts.
    """
    parsed = urlsplit(url)
    # Validate scheme
    if parsed.scheme not in ("http", "https"):
        raise ValueError(f"URL scheme {parsed.scheme!r} not allowed")
    if parsed.scheme != "https":
        raise ValueError("Only HTTPS URLs are allowed")
    # Validate port
    if parsed.port is not None and parsed.port not in ALLOWED_PORTS:
        raise ValueError(f"Port {parsed.port} not in allowed ports {ALLOWED_PORTS}")
    # Validate hostname
    hostname = (parsed.hostname or "").lower()
    if hostname.startswith("www."):
        hostname = hostname[4:]
    if not hostname:
        raise ValueError("URL must include a valid hostname")
    if hostname not in ALLOWED_HOSTS:
        raise ValueError(f"URL hostname {hostname} not in allowlist")
    return hostname, parsed.scheme


def _validate_url_host(url: str) -> str:
    """Legacy alias for _validate_url_before_io (returns hostname only)."""
    hostname, _ = _validate_url_before_io(url)
    return hostname


def fetch_html(url: str, *, timeout: float = 30, _redirects: int = 0) -> str:
    """A03: Manual redirect control with per-hop validation.

    Each redirect hop is validated (scheme, host, port) before the request
    is made. urlopen's automatic redirect following is bypassed.
    """
    if _redirects > MAX_REDIRECT_HOPS:
        raise ValueError(f"Redirect chain exceeds {MAX_REDIRECT_HOPS} hops")
    # Validate BEFORE any network I/O
    hostname, scheme = _validate_url_before_io(url)
    # Build opener that does NOT follow redirects automatically
    opener = build_opener(_StrictRedirectHandler())
    request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"})

    try:
        with opener.open(request, timeout=timeout) as response:
            # If we get here, no redirect occurred (or final destination reached)
            data = response.read(MAX_RESPONSE_BYTES + 1)
            if len(data) > MAX_RESPONSE_BYTES:
                raise ValueError("response exceeded 5MB limit")
            content_type = response.headers.get_content_type() or ""
            if "text/html" not in content_type:
                raise ValueError(f"expected text/html, got {content_type}")
            return data.decode(response.headers.get_content_charset() or "utf-8")
    except urllib.error.HTTPError as exc:
        # Check if this is a redirect (3xx status)
        if exc.code in (301, 302, 303, 307, 308):
            redirect_url = _extract_redirect_url(exc.headers)
            if redirect_url is None:
                raise ValueError(f"Redirect {exc.code} with no Location header") from exc
            # Resolve relative URLs
            redirect_url = urljoin(url, redirect_url)
            # Validate this hop BEFORE following
            _validate_url_before_io(redirect_url)
            # Recursively follow with hop count
            return fetch_html(redirect_url, timeout=timeout, _redirects=_redirects + 1)
        else:
            raise


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Fetch one authorized RF4 community source page")
    parser.add_argument("source", choices=(*SOURCES, *DETAIL_SOURCES))
    parser.add_argument("--url", help="Override the configured public page URL")
    parser.add_argument("--limit", type=int, default=100, choices=range(1, 501), metavar="1..500")
    parser.add_argument(
        "--state-file", type=Path,
        default=Path(os.environ.get("RF4_COMMUNITY_FETCH_STATE", DEFAULT_STATE_FILE)),
        help="Persistent per-source cooldown state",
    )
    args = parser.parse_args(argv)
    if args.source in DETAIL_SOURCES and not args.url:
        parser.error(f"--url is required for {args.source}")
    default_url, parse = SOURCES.get(args.source, (None, DETAIL_SOURCES.get(args.source)))
    url = args.url or default_url
    try:
        site_key = fetch_site_key(url)
        # Single atomic check-and-reserve before network I/O: failed attempts count toward the limit too.
        check_and_reserve(site_key, state_file=args.state_file)
        html = fetch_html(url)
        parsed = (
            parse(html, source_url=url)
            if args.source in DETAIL_SOURCES or args.source == "rf4db-waterbodies"
            else parse(html)
        )
        records = parsed if args.source in DETAIL_SOURCES else parsed[:args.limit]
    except Exception as exc:
        print(f"community source failed: {exc}", file=sys.stderr)
        return 1
    payload = asdict(records) if args.source in DETAIL_SOURCES else [asdict(item) for item in records]
    print(json.dumps(payload, ensure_ascii=False, default=str))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())