63 lines
2.9 KiB
Python
63 lines
2.9 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
|
|
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
|
|
|
|
logger = logging.getLogger("rf4.community_scheduler")
|
|
|
|
def configured_sources():
|
|
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 run_source(source_system: str, *, now: datetime | None = None) -> bool:
|
|
current = now or datetime.now(timezone.utc)
|
|
url, parser = configured_sources()[source_system]
|
|
with SessionLocal() as session:
|
|
source = session.get(DataSource, source_system)
|
|
if source is None or not source.enabled:
|
|
return False
|
|
latest = session.scalar(select(CommunityImportRun.started_at).where(CommunityImportRun.source_system == source_system).order_by(CommunityImportRun.started_at.desc()).limit(1))
|
|
if latest and (latest if latest.tzinfo else latest.replace(tzinfo=timezone.utc)) > current - timedelta(seconds=settings.community_import_interval_seconds):
|
|
return False
|
|
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:{source_system}"}):
|
|
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])
|
|
run.status, run.rows_seen, run.rows_created, run.rows_updated = "success", len(records), created, updated
|
|
except Exception as exc:
|
|
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})
|
|
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()
|