feat: check source links during scheduled fetches

apps/api/app/community_scheduler.py

@@ -14,6 +14,7 @@ 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
@@ -102,11 +103,27 @@ def run_source(source_system: str, *, now: datetime | None = None) -> bool:
             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})
+            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
 

apps/api/app/source_lifecycle.py

@@ -2,7 +2,9 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 from typing import Literal
+from urllib.error import HTTPError
 
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from .models import ExternalObservation, ModerationStatus
@@ -11,6 +13,16 @@ from .models import ExternalObservation, ModerationStatus
 SourceCheckStatus = Literal["available", "missing", "temporary_error", "blocked"]
 
 
+def classify_source_failure(exc: Exception) -> SourceCheckStatus:
+    """Classify the result of the scheduled request without retrying it."""
+    if isinstance(exc, HTTPError):
+        if exc.code in {404, 410}:
+            return "missing"
+        if exc.code in {401, 403, 429}:
+            return "blocked"
+    return "temporary_error"
+
+
 def record_source_check(
     session: Session,
     observation: ExternalObservation,
@@ -24,7 +36,18 @@ def record_source_check(
     data. Transient errors and access blocks remain diagnostic and never remove
     an observation from activity.
     """
-    current = checked_at or datetime.now(timezone.utc)
+    _apply_source_check(observation, status, checked_at or datetime.now(timezone.utc))
+    session.commit()
+    return observation
+
+
+def _apply_source_check(
+    observation: ExternalObservation,
+    status: SourceCheckStatus,
+    checked_at: datetime,
+) -> None:
+    """Mutate one observation; the caller owns the transaction boundary."""
+    current = checked_at
     observation.source_check_status = status
     observation.source_checked_at = current
     if status == "missing" and observation.status != "withdrawn":
@@ -34,5 +57,27 @@ def record_source_check(
         observation.review_note = "Source record missing; withdrawn pending moderator review"
         observation.reviewed_at = current
         observation.moderation_version += 1
+
+
+def record_scheduled_source_check(
+    session: Session,
+    *,
+    source_system: str,
+    source_url: str,
+    status: SourceCheckStatus,
+    checked_at: datetime | None = None,
+) -> int:
+    """Apply one scheduled request result only to observations with that exact URL.
+
+    Aggregate pages cannot prove that an omitted record was deleted, so absence
+    from a parsed listing is deliberately ignored.
+    """
+    observations = list(session.scalars(select(ExternalObservation).where(
+        ExternalObservation.source_system == source_system,
+        ExternalObservation.source_url == source_url,
+    )))
+    current = checked_at or datetime.now(timezone.utc)
+    for observation in observations:
+        _apply_source_check(observation, status, current)
     session.commit()
-    return observation
+    return len(observations)

apps/api/tests/test_community_importer.py

@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
 
 from app.community_importer import CommunityImportError, stage_observations
 from app.community_review import ExternalReviewError, map_observation, publish_observation, suggest_aliases
-from app.source_lifecycle import record_source_check
+from app.source_lifecycle import record_scheduled_source_check, record_source_check
 from app.database import Base
 from app.models import CatchReport, DataSource, ExternalEntityAlias, ExternalObservation, Fish, Waterbody
 from rf4_research.community_sources import parse_rf4db_catches, parse_rf4map_point, parse_rf4posts_spot
@@ -186,6 +186,35 @@ def test_non_authoritative_source_failures_do_not_withdraw(db: Session, status:
     assert item.source_check_status == status
 
 
+def test_scheduled_failure_only_affects_exact_source_url(db: Session) -> None:
+    stage_observations(db, [
+        record(external_id="matching"),
+        record(external_id="other") | {"source_url": "https://rf4db.com/catches/other"},
+    ])
+
+    affected = record_scheduled_source_check(
+        db,
+        source_system="rf4db",
+        source_url="https://rf4db.com/ru/catches/matching",
+        status="missing",
+    )
+    items = {item.source_external_id: item for item in db.scalars(select(ExternalObservation))}
+
+    assert affected == 1
+    assert items["matching"].status == "withdrawn"
+    assert items["other"].status != "withdrawn"
+    assert items["other"].source_check_status == "available"
+
+    record_scheduled_source_check(
+        db,
+        source_system="rf4db",
+        source_url="https://rf4db.com/ru/catches/matching",
+        status="available",
+    )
+    assert items["matching"].source_check_status == "available"
+    assert items["matching"].status == "withdrawn"
+
+
 def test_auto_publication_requires_enabled_source(db: Session) -> None:
     source = DataSource(key="rf4db", name="RF4DB", base_url="https://rf4db.com", default_confidence=70, enabled=False)
     fish = Fish(slug="pike", name_ru="Щука")

apps/api/tests/test_community_scheduler.py

@@ -1,10 +1,12 @@
 import pytest
 from pydantic import ValidationError
+from urllib.error import HTTPError
 
 from datetime import datetime, timedelta, timezone
 
 from app.community_scheduler import MAX_BACKOFF_SECONDS, configured_sources, _static_registry, oldest_site_source, retry_delay
 from app.config import Settings
+from app.source_lifecycle import classify_source_failure
 
 
 def test_all_authorized_sources_are_scheduled() -> None:
@@ -23,6 +25,16 @@ def test_failed_runs_back_off_but_success_resets_delay() -> None:
     assert retry_delay(["success", "failed"]) == 1800
 
 
+@pytest.mark.parametrize(("code", "expected"), [(404, "missing"), (410, "missing"), (403, "blocked"), (429, "blocked"), (500, "temporary_error")])
+def test_source_http_failure_classification(code: int, expected: str) -> None:
+    error = HTTPError("https://rf4.example/source", code, "failure", {}, None)
+    assert classify_source_failure(error) == expected
+
+
+def test_non_http_source_failure_is_temporary() -> None:
+    assert classify_source_failure(TimeoutError("timeout")) == "temporary_error"
+
+
 def test_same_site_endpoints_rotate_by_oldest_attempt() -> None:
     now = datetime.now(timezone.utc)
     all_keys = {"rf4db", "rf4stat-fishing", "rf4stat-post", "rf4map", "rf4posts-spot"}