diff --git a/backend/api/competitors.py b/backend/api/competitors.py index deb67c2..957c215 100644 --- a/backend/api/competitors.py +++ b/backend/api/competitors.py @@ -79,6 +79,7 @@ class ScraperStatusResponse(BaseModel): location_configured: bool location_name: Optional[str] last_scrape: Optional[dict] + lock_held_seconds: Optional[int] = None # ============================================ @@ -133,12 +134,16 @@ async def get_scraper_status( 'error_message': last_scrape_row.error_message, } + from services.booking_scraper import get_lock_status + lock = get_lock_status() + return ScraperStatusResponse( enabled=config.get('booking_scraper_enabled', 'false') == 'true', backend=config.get('booking_scraper_backend', 'playwright_local'), location_configured=location_row is not None, location_name=location_row.location_name if location_row else None, - last_scrape=last_scrape + last_scrape=last_scrape, + lock_held_seconds=lock["held_seconds"], ) @@ -408,8 +413,8 @@ async def trigger_manual_scrape( # Best-effort early 409: the background task re-acquires the lock and will # no-op (logging "another scrape is running") if it loses a millisecond- # window race, so no double-run can slip through here. - from services.booking_scraper import SCRAPE_LOCK - if SCRAPE_LOCK.locked(): + from services.booking_scraper import get_lock_status + if get_lock_status()["locked"]: raise HTTPException(status_code=409, detail="A scrape is already running. Try again when it finishes.") # Start background task @@ -423,6 +428,25 @@ async def trigger_manual_scrape( } +@router.post("/scrape/reset") +async def reset_scraper( + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +): + """Force-release the scrape lock and mark any stuck running batches as interrupted. + Use when the scraper is stuck and the 409 won't clear on its own.""" + if not (current_user.get('is_admin') or 'manage_scraper' in (current_user.get('caps') or [])): + raise HTTPException(status_code=403, detail="manage_scraper capability required") + + from services.booking_scraper import force_reset_scraper + sync_db = SyncSessionLocal() + try: + result = force_reset_scraper(sync_db) + finally: + sync_db.close() + return result + + # ============================================ # HOTELS MANAGEMENT # ============================================ diff --git a/backend/scheduler.py b/backend/scheduler.py index 6ddfef9..e8c64a1 100644 --- a/backend/scheduler.py +++ b/backend/scheduler.py @@ -68,6 +68,22 @@ async def run_scheduled_parity_check(): await loop.run_in_executor(None, run_parity_check) +async def run_scrape_watchdog(): + """Auto-release the scrape lock if held for >3 hours (hung Playwright browser).""" + from services.booking_scraper import get_lock_status, force_reset_scraper + status = get_lock_status() + held = status.get("held_seconds") + if held and held > 3 * 3600: + logger.warning(f"Scrape watchdog: lock held for {held}s — force releasing") + db = SyncSessionLocal() + try: + force_reset_scraper(db) + except Exception as e: + logger.error(f"Scrape watchdog reset failed: {e}") + finally: + db.close() + + async def run_scheduled_booking_scrape_async(): from jobs.scrape_booking_rates import run_scheduled_booking_scrape import asyncio @@ -131,8 +147,17 @@ def start_scheduler(): replace_existing=True, ) + # Scrape lock watchdog — every 30 min; force-releases if held >3 hours + from apscheduler.triggers.interval import IntervalTrigger + scheduler.add_job( + run_scrape_watchdog, + IntervalTrigger(minutes=30), + id='scrape_watchdog', + replace_existing=True, + ) + scheduler.start() - logger.info(f"Scheduler started: booking scrape at {scrape_hour:02d}:{scrape_minute:02d}, rates fetch at {rates_hour:02d}:{rates_minute:02d}, direct scrape at 06:00, parity check at 06:45") + logger.info(f"Scheduler started: booking scrape at {scrape_hour:02d}:{scrape_minute:02d}, rates fetch at {rates_hour:02d}:{rates_minute:02d}, direct scrape at 06:00, parity check at 06:45, watchdog every 30m") def shutdown_scheduler(): diff --git a/backend/services/booking_scraper.py b/backend/services/booking_scraper.py index 56462d9..d83d3a9 100644 --- a/backend/services/booking_scraper.py +++ b/backend/services/booking_scraper.py @@ -14,6 +14,7 @@ Features: import asyncio import logging import threading +import time import uuid from datetime import date, timedelta from decimal import Decimal @@ -32,6 +33,48 @@ logger = logging.getLogger(__name__) # other of memory (page timeouts → partial results) and double the request # rate at Booking.com. Guards both manual scrapes and queue processing. SCRAPE_LOCK = threading.Lock() +_lock_acquired_at: Optional[float] = None # monotonic timestamp when lock was acquired + + +def _acquire_scrape_lock() -> bool: + global _lock_acquired_at + if SCRAPE_LOCK.acquire(blocking=False): + _lock_acquired_at = time.monotonic() + return True + return False + + +def _release_scrape_lock(): + global _lock_acquired_at + _lock_acquired_at = None + try: + SCRAPE_LOCK.release() + except RuntimeError: + pass # already released — race between watchdog and normal release + + +def get_lock_status() -> dict: + """Return lock state and how long it has been held (seconds).""" + locked = SCRAPE_LOCK.locked() + held = int(time.monotonic() - _lock_acquired_at) if (locked and _lock_acquired_at) else None + return {"locked": locked, "held_seconds": held} + + +def force_reset_scraper(db: Session) -> dict: + """Force-release the scrape lock and tidy up any stuck DB state. + Safe to call even when the lock is not held.""" + was_locked = SCRAPE_LOCK.locked() + _release_scrape_lock() + # Mark any batch left in 'running' state as interrupted + db.execute(text(""" + UPDATE booking_scrape_log + SET status = 'interrupted', completed_at = NOW(), + error_message = 'Force-reset by admin' + WHERE status = 'running' + """)) + db.commit() + logger.warning(f"Scrape lock force-reset (was_locked={was_locked})") + return {"was_locked": was_locked, "reset": True} def get_scraper_backend(db: Session) -> ScraperBackend: @@ -509,7 +552,7 @@ async def run_manual_scrape( if to_date is None: to_date = from_date - if not SCRAPE_LOCK.acquire(blocking=False): + if not _acquire_scrape_lock(): logger.warning(f"Manual scrape {from_date}..{to_date} refused — another scrape is running") return { 'success': False, @@ -519,7 +562,7 @@ async def run_manual_scrape( try: return await _run_manual_scrape_locked(db, from_date, to_date) finally: - SCRAPE_LOCK.release() + _release_scrape_lock() async def _run_manual_scrape_locked( @@ -694,7 +737,7 @@ async def process_queue(db: Session) -> Dict[str, Any]: Returns: Dict with processing results """ - if not SCRAPE_LOCK.acquire(blocking=False): + if not _acquire_scrape_lock(): logger.warning("Queue processing skipped — another scrape is running") return { 'success': False, @@ -704,7 +747,7 @@ async def process_queue(db: Session) -> Dict[str, Any]: try: return await _process_queue_locked(db) finally: - SCRAPE_LOCK.release() + _release_scrape_lock() async def _process_queue_locked(db: Session) -> Dict[str, Any]: diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx index 62b59db..c27cecb 100644 --- a/frontend/src/pages/Settings.tsx +++ b/frontend/src/pages/Settings.tsx @@ -1,7 +1,7 @@ import { useState, useEffect } from 'react' import { useParams, useNavigate } from 'react-router-dom' import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query' -import { Save, RefreshCw, Database, Clock, BedDouble, ChevronUp, ChevronDown, Network, ShieldCheck } from 'lucide-react' +import { Save, RefreshCw, Database, Clock, BedDouble, ChevronUp, ChevronDown, Network, ShieldCheck, AlertTriangle } from 'lucide-react' import api from '../api' const TABS = [ @@ -488,6 +488,83 @@ function ProxyTab() { + + + + ) +} + +// ─── Scraper Lock Status Card ───────────────────────────────────────────────── + +function ScraperLockCard() { + const qc = useQueryClient() + + const { data: status, isLoading } = useQuery<{ lock_held_seconds: number | null }>({ + queryKey: ['scraper-status-lock'], + queryFn: () => api.get('/competitors/status').then(r => r.data), + refetchInterval: 15000, + }) + + const resetM = useMutation({ + mutationFn: () => api.post('/competitors/scrape/reset').then(r => r.data), + onSuccess: () => qc.invalidateQueries({ queryKey: ['scraper-status-lock'] }), + }) + + const held = status?.lock_held_seconds ?? null + const isStuck = held !== null && held > 0 + const heldStr = held + ? held >= 3600 + ? `${Math.floor(held / 3600)}h ${Math.floor((held % 3600) / 60)}m` + : held >= 60 + ? `${Math.floor(held / 60)}m ${held % 60}s` + : `${held}s` + : null + + return ( +
+
Scraper Lock
+
+
+ + {isStuck + ? <> Locked ({heldStr}) + : <> Idle} + + {isStuck && ( + + A scrape may be stuck. The watchdog will auto-release after 3 hours. + + )} +
+

+ Force-release the scrape lock if it is stuck (e.g. a hung Playwright browser). + Any in-progress scrape batch will be marked interrupted; the queue remains intact + and will retry on the next scheduled run or manual trigger. +

+ + {resetM.isSuccess && ( + + ✓ Reset — was {(resetM.data as any)?.was_locked ? 'locked' : 'already idle'} + + )} +
) }