rf4-spotter/rf4_research/community_cli.py

from __future__ import annotations

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

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


SOURCES = {
    "rf4db": ("https://download.rf4db.com/ru/catches", parse_rf4db_catches),
    "rf4stat-fishing": ("https://rf4-stat.ru/fishing/", parse_rf4stat_fishing),
    "rf4stat-posts": ("https://rf4-stat.ru/posts/", parse_rf4stat_posts),
}
DETAIL_SOURCES = {
    "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",
    "rf4-stat.ru",
    "rf4map.ru",
    "rf4-posts.com",
})
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."""
    try:
        with open(state_file, "r") as f:
            fcntl.flock(f, fcntl.LOCK_SH)
            try:
                return json.loads(f.read())
            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."""
    hostname = (urlsplit(url).hostname or "").lower()
    if hostname.startswith("www."):
        hostname = hostname[4:]
    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:
    now = time.time() if now is None else now
    state = _read_state(state_file)
    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")


def mark_fetch(source: str, *, state_file: Path, now: float | None = None) -> None:
    now = time.time() if now is None else now
    state = _read_state(state_file)
    state[source] = now
    _write_state(state_file, state)


def _validate_url_host(url: str) -> str:
    """Validate URL hostname is in allowlist before making network call."""
    hostname = (urlsplit(url).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


def fetch_html(url: str, *, timeout: float = 30) -> str:
    # Validate host BEFORE network I/O to prevent SSRF to internal endpoints
    _validate_url_host(url)
    request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"})
    with urlopen(request, timeout=timeout) as response:
        final_url = response.url
        # Also validate redirect target
        final_hostname = (urlsplit(final_url).hostname or "").lower()
        if final_hostname.startswith("www."):
            final_hostname = final_hostname[4:]
        if final_hostname not in ALLOWED_HOSTS:
            raise ValueError(f"Redirect hostname {final_hostname} not in allowlist")
        if response.headers.get_content_type() != "text/html":
            raise ValueError(f"expected text/html, got {response.headers.get_content_type()}")
        data = response.read(MAX_RESPONSE_BYTES + 1)
        if len(data) > MAX_RESPONSE_BYTES:
            raise ValueError("response exceeded 5MB limit")
        return data.decode(response.headers.get_content_charset() or "utf-8")


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)
        enforce_fetch_interval(site_key, state_file=args.state_file)
        # Reserve before network I/O: failed attempts count toward the limit too.
        mark_fetch(site_key, state_file=args.state_file)
        html = fetch_html(url)
        records = (parse(html, source_url=url) if args.source in DETAIL_SOURCES else parse(html))[:args.limit]
    except Exception as exc:
        print(f"community source failed: {exc}", file=sys.stderr)
        return 1
    print(json.dumps([asdict(item) for item in records], ensure_ascii=False, default=str))
    return 0


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