feat: deletion detection + fast scraper intervals + scraper reliability fixes

README.md

@@ -62,6 +62,59 @@ docker-compose logs scraper-worker
 
 ---
 
+## Scraper & Crawler Architecture
+
+The scraper worker runs two independent loops in parallel background threads:
+
+### CrawlerService — discovery loop
+
+Runs every `CRAWLER_INTERVAL_SECONDS` (default: 1s).
+
+1. BFS-crawls the source listing page to collect all live offer URLs.
+2. Creates a `CrawlUrl` record (status `PENDING`) for any new URL found.
+3. **Deletion detection** — any `CrawlUrl` no longer present in the live listing is bumped to `next_check_at = now`, so the scraper re-fetches it immediately and detects the 404.
+4. **Manual offer cleanup** — offers created manually in the admin (no `CrawlUrl`) whose link matches the source domain but is absent from the live listing are HEAD-checked directly; those returning 404/410 are archived.
+
+### UrlScraperService — scrape loop
+
+Runs every `SCRAPER_INTERVAL_SECONDS` (default: 1s).
+
+1. Claims all `CrawlUrl` records where `next_check_at <= now`.
+2. Fetches each URL (timeout: `SCRAPER_TIMEOUT_SECONDS`, default: 1s).
+3. **4xx response** → offer is immediately archived (`PUBLISHED`/`DRAFT` → `ARCHIVED`).
+4. **5xx response** → transient error; exponential backoff applied, retried later.
+5. **200 with empty/generic page** → if a linked offer exists, it is archived; otherwise the URL is skipped.
+6. **200 with valid content** → offer is created or updated.
+
+### Archiving behaviour
+
+Offers are **never hard-deleted** from the database. All removal triggers result in `status = ARCHIVED`. Archived offers are hidden from the public Vue frontend but remain visible in the admin dashboard.
+
+| Trigger | Result |
+|---|---|
+| URL returns 4xx (404, 410, etc.) | ARCHIVED immediately |
+| URL returns 5xx | Backoff, retried later |
+| URL returns 200 but page is empty/generic | ARCHIVED (if offer linked) |
+| URL disappears from source listing | CrawlUrl bumped → re-fetched → 404 → ARCHIVED |
+| Manually created offer URL gone from listing | HEAD-checked → 404/410 → ARCHIVED |
+
+### Scraper worker configuration
+
+| Variable | Default | Description |
+|---|---|---|
+| `CRAWLER_INTERVAL_SECONDS` | `1` | How often the crawler discovery loop runs |
+| `SCRAPER_INTERVAL_SECONDS` | `1` | How often the scraper drain loop runs |
+| `SCRAPER_TIMEOUT_SECONDS` | `1` | HTTP request timeout per URL |
+
+### LLM extraction
+
+Each scraped page is processed by one of two extraction methods:
+
+- **Ollama LLM (primary)** — structured JSON extraction with offer type classification. Configured per source in the admin panel.
+- **Deterministic fallback** — `<meta name="description">` as summary; TF-IDF cosine similarity for offer type classification. Used automatically when Ollama is not running or returns a connection error (no wait, no cooldown).
+
+---
+
 ## Useful Docker commands
 
 ```bash
@@ -146,8 +199,11 @@ Each scraped offer is automatically classified into one of the offer types store
 
 | Variable | Default | Description |
 |---|---|---|
-| `LLM_ENABLED` | `true` | Enable or disable the Ollama LLM |
+| `LLM_ENABLED` | `true` | Enable or disable the Ollama LLM globally |
 | `SCRAPER_CLASSIFIER_THRESHOLD` | `0.15` | Minimum cosine similarity score for the TF-IDF fallback to accept a classification |
+| `CRAWLER_INTERVAL_SECONDS` | `1` | Crawler discovery loop interval |
+| `SCRAPER_INTERVAL_SECONDS` | `1` | Scraper drain loop interval |
+| `SCRAPER_TIMEOUT_SECONDS` | `1` | HTTP request timeout per scraped URL |
 
 ### Adding or updating offer types
 

backend/.env

@@ -13,8 +13,9 @@ POSTGRES_HOST=localhost
 POSTGRES_PORT=5432
 POSTGRES_SCHEMA=content
 
-SCRAPER_TIMEOUT_SECONDS=30
-SCRAPER_INTERVAL_MINUTES=360
+SCRAPER_TIMEOUT_SECONDS=1
+CRAWLER_INTERVAL_SECONDS=1
+SCRAPER_INTERVAL_SECONDS=1
 SCRAPER_RUN_ON_START=true
 SCRAPER_LLM_FALLBACK_THRESHOLD=0.60
 SCRAPER_USER_AGENT=SUNRISE-OSS-Scraper/1.0

backend/.env.example

@@ -13,8 +13,9 @@ POSTGRES_HOST=localhost
 POSTGRES_PORT=5432
 POSTGRES_SCHEMA=content
 
-SCRAPER_TIMEOUT_SECONDS=30
-SCRAPER_INTERVAL_MINUTES=360
+SCRAPER_TIMEOUT_SECONDS=1
+CRAWLER_INTERVAL_SECONDS=1
+SCRAPER_INTERVAL_SECONDS=1
 SCRAPER_RUN_ON_START=true
 SCRAPER_LLM_FALLBACK_THRESHOLD=0.60
 SCRAPER_USER_AGENT=SUNRISE-OSS-Scraper/1.0

backend/content/management/commands/run_scraper_worker.py

@@ -1,13 +1,28 @@
+import logging
 import os
+import threading
+import time
 
-from apscheduler.schedulers.blocking import BlockingScheduler
+from apscheduler.schedulers.background import BackgroundScheduler
 from django.core.management.base import BaseCommand
 from django.utils import timezone
 
 from content.scrapers.queue_service import run_crawler, run_url_scraper_batch
 from content.scrapers.link_checker import check_offer_links
 from content.scrapers.translation_service import run_offer_translation_batch
 
+LOGGER = logging.getLogger(__name__)
+
+
+def _loop(fn: callable, interval: float, stop: threading.Event) -> None:
+    """Run fn, sleep interval seconds, repeat — guarantees no overlap."""
+    while not stop.is_set():
+        try:
+            fn()
+        except Exception:
+            LOGGER.exception("Worker loop error in %s", fn.__name__)
+        stop.wait(interval)
+
 
 def _archive_expired_offers():
     from django.utils import timezone as tz
@@ -36,8 +51,8 @@ def add_arguments(self, parser):
         )
 
     def handle(self, *args, **options):
-        crawler_interval = int(os.getenv("CRAWLER_INTERVAL_MINUTES", "360"))
-        scraper_interval = int(os.getenv("SCRAPER_INTERVAL_MINUTES", "5"))
+        crawler_interval = int(os.getenv("CRAWLER_INTERVAL_SECONDS", "1"))
+        scraper_interval = int(os.getenv("SCRAPER_INTERVAL_SECONDS", "1"))
         translation_interval = int(os.getenv("TRANSLATION_INTERVAL_MINUTES", "10"))
         translation_batch = int(os.getenv("TRANSLATION_BATCH_SIZE", "20"))
         run_on_start = os.getenv("SCRAPER_RUN_ON_START", "true").lower() == "true"
@@ -51,26 +66,27 @@ def handle(self, *args, **options):
             self.stdout.write(f"Scraper: {scraper_summary}")
             return
 
-        scheduler = BlockingScheduler(timezone=timezone.get_current_timezone_name())
+        stop = threading.Event()
 
-        scheduler.add_job(
-            run_crawler,
-            "interval",
-            id="crawl-discover-urls",
-            minutes=crawler_interval,
-            max_instances=1,
-            coalesce=True,
-            misfire_grace_time=300,
-        )
-        scheduler.add_job(
-            run_url_scraper_batch,
-            "interval",
-            id="scrape-url-queue",
-            minutes=scraper_interval,
-            max_instances=1,
-            coalesce=True,
-            misfire_grace_time=60,
-        )
+        if run_on_start:
+            self.stdout.write("Startup crawler run...")
+            self.stdout.write(f"Crawler: {run_crawler()}")
+            self.stdout.write("Startup scraper batch...")
+            self.stdout.write(f"Scraper: {run_url_scraper_batch()}")
+            self.stdout.write("Startup offer translation...")
+            self.stdout.write(f"Translation: {run_offer_translation_batch(limit=translation_batch)}")
+
+        # Crawler and scraper run in threads: each waits N seconds AFTER the
+        # previous run completes, so no overlap and no APScheduler skip warnings.
+        threading.Thread(
+            target=_loop, args=(run_crawler, crawler_interval, stop), daemon=True, name="crawler"
+        ).start()
+        threading.Thread(
+            target=_loop, args=(run_url_scraper_batch, scraper_interval, stop), daemon=True, name="scraper"
+        ).start()
+
+        # Cron jobs (infrequent) stay on APScheduler.
+        scheduler = BackgroundScheduler(timezone=timezone.get_current_timezone_name())
         scheduler.add_job(
             run_offer_translation_batch,
             "interval",
@@ -99,25 +115,17 @@ def handle(self, *args, **options):
             max_instances=1,
             coalesce=True,
         )
-
-        if run_on_start:
-            self.stdout.write("Startup crawler run...")
-            crawler_summary = run_crawler()
-            self.stdout.write(f"Crawler: {crawler_summary}")
-            self.stdout.write("Startup scraper batch...")
-            scraper_summary = run_url_scraper_batch()
-            self.stdout.write(f"Scraper: {scraper_summary}")
-            self.stdout.write("Startup offer translation...")
-            translation_summary = run_offer_translation_batch(limit=translation_batch)
-            self.stdout.write(f"Translation: {translation_summary}")
+        scheduler.start()
 
         self.stdout.write(
-            f"Worker running — crawler every {crawler_interval} min, "
-            f"scraper every {scraper_interval} min, "
+            f"Worker running — crawler every {crawler_interval}s after completion, "
+            f"scraper every {scraper_interval}s after completion, "
             f"translation every {translation_interval} min. Press Ctrl+C to stop."
         )
 
         try:
-            scheduler.start()
+            stop.wait()
         except (KeyboardInterrupt, SystemExit):
+            stop.set()
+            scheduler.shutdown(wait=False)
             self.stdout.write("Worker stopped.")

backend/content/scrapers/ollama_client.py

@@ -17,6 +17,11 @@
 # at the start of run N+1 before the cooldown has expired.
 _SHARED_MODEL_COOLDOWN: dict[str, float] = {}
 
+# Models whose cooldown was triggered by a connection error (server down),
+# not a 429 rate limit. These skip the cooldown wait — no point waiting
+# for a server that isn't running.
+_CONNECTION_ERROR_MODELS: set[str] = set()
+
 # Buffer for cooldown events emitted during _wait_for_available_model.
 # Flushed into run.log by ScrapeService after each LLM call.
 _COOLDOWN_LOG_BUFFER: list[dict] = []
@@ -58,6 +63,12 @@ def _wait_for_available_model(self) -> list[str]:
             return models
         if not _SHARED_MODEL_COOLDOWN:
             return []
+        # If every cooled-down model failed due to a connection error (server
+        # unreachable), skip the wait and fall through to deterministic immediately.
+        cooled_models = set(_SHARED_MODEL_COOLDOWN.keys())
+        if cooled_models and cooled_models.issubset(_CONNECTION_ERROR_MODELS):
+            LOGGER.debug("Ollama unreachable — skipping LLM, using deterministic fallback")
+            return []
         soonest = min(_SHARED_MODEL_COOLDOWN.values())
         wait = max(0.0, soonest - time.time())
         if wait > self.cooldown_max_wait_seconds:
@@ -109,7 +120,10 @@ def extract_fallback(
                 status_code = exc.status_code if isinstance(exc, ollama.ResponseError) else None
                 _SHARED_MODEL_COOLDOWN[model] = time.time() + self.model_cooldown_seconds
                 if status_code != 429:
+                    _CONNECTION_ERROR_MODELS.add(model)
                     LOGGER.warning("Ollama fallback unavailable for %s using %s: %s", source.key, model, exc)
+                else:
+                    _CONNECTION_ERROR_MODELS.discard(model)
                 continue
 
             text = response.response
@@ -122,6 +136,7 @@ def extract_fallback(
             confidence = parsed.get("confidence", deterministic_payload.confidence)
 
             self.last_switch_count = index
+            _CONNECTION_ERROR_MODELS.discard(model)
             LOGGER.info(
                 "LLM fallback success — source=%s model=%s conf=%.2f title=%r",
                 source.key, model, float(confidence), str(title)[:60],
@@ -179,7 +194,10 @@ def assess_and_extract(
                 status_code = exc.status_code if isinstance(exc, ollama.ResponseError) else None
                 _SHARED_MODEL_COOLDOWN[model] = time.time() + self.model_cooldown_seconds
                 if status_code != 429:
+                    _CONNECTION_ERROR_MODELS.add(model)
                     LOGGER.warning("Ollama relevance check unavailable for %s using %s: %s", source.key, model, exc)
+                else:
+                    _CONNECTION_ERROR_MODELS.discard(model)
                 continue
 
             text = response.response
@@ -200,6 +218,7 @@ def assess_and_extract(
             )
 
             self.last_switch_count = index
+            _CONNECTION_ERROR_MODELS.discard(model)
             LOGGER.info(
                 "LLM assess success — source=%s model=%s is_offer=%s offer_type=%s conf=%.2f reason=%r",
                 source.key, model, is_relevant, offer_type or "(will classify)", confidence, reason[:80] if reason else "",