ik/rf4-spotter
1
0
Fork 0
Code Issues Pull Requests Actions 8 Packages Projects Releases Wiki Activity

fix: R02 activity API contract — PaginatedActivity + all consumers (#1788956171115)

Browse Source
This commit is contained in:
ik
2026-09-09 20:58:42 +07:00
parent a37f9c4696
commit d962ba2f90
9 changed files with 85 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/community_scheduler.py
+28 -17
View File
@@ -26,23 +26,31 @@ def retry_delay(statuses: list[str]) -> int:
failures += 1
return min(settings.community_import_interval_seconds * (2 ** max(0, failures - 1)), MAX_BACKOFF_SECONDS)
def configured_sources():
with SessionLocal() as session:
enabled_keys = {
s.key for s in session.scalars(select(DataSource).where(DataSource.enabled.is_(True)))
}
def _static_registry() -> dict[str, tuple[str, callable]]:
"""Return the full static registry without DB access (for unit tests)."""
return {
k: v for k, v in {
"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),
}.items() if k in enabled_keys
"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 oldest_site_source(source_system: str, latest_by_source: dict[str, datetime]) -> str:
sources = configured_sources()
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:
sources = configured_sources(enabled_keys)
site = fetch_site_key(sources[source_system][0])
candidates = [key for key, (url, _) in sources.items() if fetch_site_key(url) == site]
order = {key: index for index, key in enumerate(candidates)}
@@ -50,9 +58,12 @@ def oldest_site_source(source_system: str, latest_by_source: dict[str, datetime]
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:
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 configured_sources().items() if fetch_site_key(candidate_url) == site_key]
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:
@@ -65,7 +76,7 @@ def run_source(source_system: str, *, now: datetime | None = None) -> bool:
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) != source_system:
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])