139 lines
6.7 KiB
Python
139 lines
6.7 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import asdict
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select, text
|
|
|
|
from rf4_research.community_cli import SOURCES, fetch_html, fetch_site_key
|
|
from rf4_research.community_sources import parse_rf4map_point, parse_rf4posts_spot
|
|
from .community_importer import stage_observations
|
|
from .config import settings
|
|
from .database import SessionLocal
|
|
from .logging_config import configure_logging
|
|
from .models import CommunityImportRun, DataSource
|
|
from .source_lifecycle import classify_source_failure, record_scheduled_source_check
|
|
|
|
logger = logging.getLogger("rf4.community_scheduler")
|
|
MAX_BACKOFF_SECONDS = 24 * 60 * 60
|
|
|
|
def retry_delay(statuses: list[str]) -> int:
|
|
failures = 0
|
|
for status in statuses:
|
|
if status != "failed":
|
|
break
|
|
failures += 1
|
|
return min(settings.community_import_interval_seconds * (2 ** max(0, failures - 1)), MAX_BACKOFF_SECONDS)
|
|
|
|
|
|
def _static_registry() -> dict[str, tuple[str, callable]]:
|
|
"""Return the full static registry without DB access (for unit tests)."""
|
|
return {
|
|
"rf4db": SOURCES["rf4db"],
|
|
"rf4stat-fishing": SOURCES["rf4stat-fishing"],
|
|
"rf4stat-post": (SOURCES["rf4stat-posts"][0], SOURCES["rf4stat-posts"][1]),
|
|
"rf4map": (settings.rf4map_point_url, parse_rf4map_point),
|
|
"rf4posts-spot": (settings.rf4posts_spot_url, parse_rf4posts_spot),
|
|
}
|
|
|
|
|
|
def configured_sources(enabled_keys: set[str] | None = None) -> dict[str, tuple[str, callable]]:
|
|
"""Return enabled sources. When enabled_keys is None, query the DB."""
|
|
registry = _static_registry()
|
|
if enabled_keys is None:
|
|
with SessionLocal() as session:
|
|
enabled_keys = {
|
|
s.key for s in session.scalars(select(DataSource).where(DataSource.enabled.is_(True)))
|
|
}
|
|
return {k: v for k, v in registry.items() if k in enabled_keys}
|
|
|
|
|
|
def oldest_site_source(source_system: str, latest_by_source: dict[str, datetime], enabled_keys: set[str] | None = None) -> str:
|
|
"""Return the oldest candidate for the same site.
|
|
|
|
Uses the full registry (not just enabled) for cooldown history so that
|
|
disabling an endpoint does not reset the site-wide cooldown for its
|
|
neighbours. enabled_keys is used to filter candidates after the oldest
|
|
is found — if the oldest is disabled, the next oldest enabled is picked.
|
|
"""
|
|
registry = _static_registry()
|
|
site = fetch_site_key(registry[source_system][0])
|
|
candidates = [key for key, (url, _) in registry.items() if fetch_site_key(url) == site]
|
|
order = {key: index for index, key in enumerate(candidates)}
|
|
# Sort by (last_run, order) — pick oldest
|
|
sorted_candidates = sorted(candidates, key=lambda key: (latest_by_source.get(key, datetime.min.replace(tzinfo=timezone.utc)), order[key]))
|
|
# If enabled_keys is provided, prefer enabled; otherwise return oldest regardless
|
|
if enabled_keys:
|
|
enabled = [k for k in sorted_candidates if k in enabled_keys]
|
|
if enabled:
|
|
return enabled[0]
|
|
return sorted_candidates[0]
|
|
|
|
def run_source(source_system: str, *, now: datetime | None = None) -> bool:
|
|
current = now or datetime.now(timezone.utc)
|
|
with SessionLocal() as session:
|
|
enabled_keys = {s.key for s in session.scalars(select(DataSource).where(DataSource.enabled.is_(True)))}
|
|
registry = _static_registry()
|
|
url, parser = registry[source_system]
|
|
site_key = fetch_site_key(url)
|
|
site_sources = [key for key, (candidate_url, _) in registry.items() if fetch_site_key(candidate_url) == site_key]
|
|
with SessionLocal() as session:
|
|
source = session.get(DataSource, source_system)
|
|
if source is None or not source.enabled:
|
|
return False
|
|
# Lock before reading cooldown: committing the reservation makes it visible
|
|
# to the next contender before releasing this transaction lock.
|
|
if session.bind and session.bind.dialect.name == "postgresql" and not session.scalar(text("select pg_try_advisory_xact_lock(hashtext(:key))"), {"key": f"community-site:{site_key}"}):
|
|
return False
|
|
recent = list(session.scalars(select(CommunityImportRun).where(CommunityImportRun.source_system.in_(site_sources)).order_by(CommunityImportRun.started_at.desc()).limit(32)))
|
|
latest_by_source: dict[str, datetime] = {}
|
|
for previous in recent:
|
|
latest_by_source.setdefault(previous.source_system, previous.started_at if previous.started_at.tzinfo else previous.started_at.replace(tzinfo=timezone.utc))
|
|
if oldest_site_source(source_system, latest_by_source, enabled_keys) != source_system:
|
|
return False
|
|
latest = recent[0].started_at if recent else None
|
|
delay = retry_delay([run.status for run in recent])
|
|
if latest and (latest if latest.tzinfo else latest.replace(tzinfo=timezone.utc)) > current - timedelta(seconds=delay):
|
|
return False
|
|
run = CommunityImportRun(source_system=source_system, source_url=url, started_at=current, status="running")
|
|
session.add(run); session.commit()
|
|
try:
|
|
html = fetch_html(url)
|
|
records = parser(html, source_url=url) if source_system in {"rf4map", "rf4posts-spot"} else parser(html)
|
|
created, updated = stage_observations(session, [asdict(item) for item in records])
|
|
record_scheduled_source_check(
|
|
session, source_system=source_system, source_url=url,
|
|
status="available", checked_at=current,
|
|
)
|
|
run.status, run.rows_seen, run.rows_created, run.rows_updated = "success", len(records), created, updated
|
|
except Exception as exc:
|
|
session.rollback()
|
|
source_status = classify_source_failure(exc)
|
|
checked = datetime.now(timezone.utc)
|
|
affected = record_scheduled_source_check(
|
|
session,
|
|
source_system=source_system,
|
|
source_url=url,
|
|
status=source_status,
|
|
checked_at=checked,
|
|
)
|
|
run.status, run.error_summary = "failed", f"{type(exc).__name__}: {str(exc)[:500]}"
|
|
logger.exception("community import failed", extra={
|
|
"event":"community_import_failed", "source_system":source_system,
|
|
"source_check_status": source_status, "affected_observations": affected,
|
|
})
|
|
run.finished_at = datetime.now(timezone.utc); session.commit()
|
|
return True
|
|
|
|
def main() -> None:
|
|
configure_logging(settings.log_level)
|
|
while True:
|
|
for source_system in configured_sources():
|
|
run_source(source_system)
|
|
time.sleep(60)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|