ik
97660833ab
Bug: main() called both enforce_fetch_interval() and mark_fetch(), which both now call check_and_reserve(). On cold start: 1. enforce_fetch_interval() → check_and_reserve() → SUCCESS (reserves) 2. mark_fetch() → check_and_reserve() → FAILS (cooldown now active) This prevented HTTP from ever being called on cold start. Fix: - Removed duplicate calls to enforce_fetch_interval() and mark_fetch() - Single check_and_reserve() call before fetch_html() - enforce_fetch_interval() and mark_fetch() remain as legacy wrappers Verification: - All 18 community_cli tests pass - Code analysis confirms single check_and_reserve() call in main() - check_and_reserve() is atomic with exclusive lock for check+reserve
291 lines
11 KiB
Python
291 lines
11 KiB
Python
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_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:
|
|
"""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.
|
|
"""
|
|
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
|
|
try:
|
|
with open(state_file, "r") as sf:
|
|
state = json.loads(sf.read()) or {}
|
|
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)
|
|
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())
|