diff --git a/backend/api/competitors.py b/backend/api/competitors.py index 7439ca5..09b176d 100644 --- a/backend/api/competitors.py +++ b/backend/api/competitors.py @@ -4,7 +4,7 @@ Booking.com rate scraping, hotel management, and competitor comparison """ from typing import Optional, List, Dict, Any from datetime import date, datetime, timedelta -from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks +from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Query from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text from pydantic import BaseModel @@ -1042,6 +1042,18 @@ async def trigger_parity_check( return await loop.run_in_executor(None, run_parity_check) +@router.post("/parity/check-date") +async def trigger_parity_check_for_date( + rate_date: date = Query(..., description="Date to recheck (YYYY-MM-DD)"), + current_user: dict = Depends(get_current_user), +): + """Recheck parity for a single date using the latest scraped rates.""" + import asyncio + from jobs.check_rate_parity import run_parity_check_for_date + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, run_parity_check_for_date, rate_date) + + # ============================================ # PARITY ALERTS # ============================================ diff --git a/backend/jobs/check_rate_parity.py b/backend/jobs/check_rate_parity.py index aa23736..658659f 100644 --- a/backend/jobs/check_rate_parity.py +++ b/backend/jobs/check_rate_parity.py @@ -244,6 +244,93 @@ def gather_comparisons(db, start: date, end: date, cfg: dict) -> list: return out +def _fetch_latest_alerts(db, start: date, end: date) -> dict: + """Return {rate_date: {id, status, diff}} for the most-recent alert per date in [start, end].""" + rows = db.execute(text(""" + SELECT DISTINCT ON (rate_date) + id, rate_date, alert_status, difference_pct + FROM rate_parity_alerts + WHERE rate_date BETWEEN :fd AND :td + ORDER BY rate_date, created_at DESC + """), {"fd": start, "td": end}).fetchall() + return {r[1]: {"id": r[0], "status": r[2], "diff": float(r[3] or 0)} for r in rows} + + +def _upsert_alerts(db, comparisons: list, latest_alert: dict) -> tuple: + """Apply parity comparison results to the alerts table. + Returns (created, updated, resolved).""" + created = updated = resolved = 0 + for c in comparisons: + alert = latest_alert.get(c["date"]) + category_label = f"{c['newbook_tariff']} vs BC {c['booking_basis']}"[:255] + + if c["breach"]: + if alert and alert["status"] == "active": + if round(c["dev_pct"], 2) != round(alert["diff"], 2): + db.execute(text(""" + UPDATE rate_parity_alerts + SET newbook_rate = :nb, booking_com_rate = :bc, + difference_pct = :diff, alert_type = :atype, + room_category = :room + WHERE id = :id + """), { + "id": alert["id"], "nb": c["newbook_rate"], "bc": c["booking_rate"], + "diff": round(c["dev_pct"], 2), + "atype": "higher" if c["dev_pct"] > 0 else "lower", + "room": category_label, + }) + updated += 1 + elif alert and alert["status"] == "acknowledged": + pass # user has dealt with this date — don't nag + else: + db.execute(text(""" + INSERT INTO rate_parity_alerts + (rate_date, room_category, newbook_rate, booking_com_rate, + difference_pct, alert_type, alert_status) + VALUES (:date, :room, :nb, :bc, :diff, :atype, 'active') + """), { + "date": c["date"], "room": category_label, + "nb": c["newbook_rate"], "bc": c["booking_rate"], + "diff": round(c["dev_pct"], 2), + "atype": "higher" if c["dev_pct"] > 0 else "lower", + }) + created += 1 + else: + if alert and alert["status"] in ("active", "acknowledged"): + db.execute( + text("UPDATE rate_parity_alerts SET alert_status = 'resolved' WHERE id = :id"), + {"id": alert["id"]} + ) + resolved += 1 + + return created, updated, resolved + + +def run_parity_check_for_date(rate_date: date) -> dict: + """Run the parity check for a single date. + Called after a date is re-scraped, or triggered manually from the UI.""" + db = SyncSessionLocal() + try: + cfg = get_parity_config(db) + if not cfg["enabled"]: + return {"status": "disabled"} + + comparisons = gather_comparisons(db, rate_date, rate_date, cfg) + if not comparisons: + return {"status": "ok", "date": str(rate_date), "no_data": True} + + latest_alert = _fetch_latest_alerts(db, rate_date, rate_date) + created, updated, resolved = _upsert_alerts(db, comparisons, latest_alert) + db.commit() + return {"status": "ok", "date": str(rate_date), + "created": created, "updated": updated, "resolved": resolved} + except Exception: + db.rollback() + raise + finally: + db.close() + + def run_parity_check() -> dict: db = SyncSessionLocal() try: @@ -265,61 +352,8 @@ def run_parity_check() -> dict: """), {"today": today}) comparisons = gather_comparisons(db, start, end, cfg) - - # Latest alert per date in the horizon - alert_rows = db.execute(text(""" - SELECT DISTINCT ON (rate_date) - id, rate_date, alert_status, difference_pct - FROM rate_parity_alerts - WHERE rate_date BETWEEN :fd AND :td - ORDER BY rate_date, created_at DESC - """), {"fd": start, "td": end}).fetchall() - latest_alert = {r[1]: {"id": r[0], "status": r[2], "diff": float(r[3] or 0)} for r in alert_rows} - - created = updated = resolved = 0 - for c in comparisons: - alert = latest_alert.get(c["date"]) - # e.g. "DIRECT B&B PPAY vs BC prepaid/B&B" - category_label = f"{c['newbook_tariff']} vs BC {c['booking_basis']}"[:255] - - if c["breach"]: - if alert and alert["status"] == "active": - if round(c["dev_pct"], 2) != round(alert["diff"], 2): - db.execute(text(""" - UPDATE rate_parity_alerts - SET newbook_rate = :nb, booking_com_rate = :bc, - difference_pct = :diff, alert_type = :atype, - room_category = :room - WHERE id = :id - """), { - "id": alert["id"], "nb": c["newbook_rate"], "bc": c["booking_rate"], - "diff": round(c["dev_pct"], 2), - "atype": "higher" if c["dev_pct"] > 0 else "lower", - "room": category_label, - }) - updated += 1 - elif alert and alert["status"] == "acknowledged": - pass # user has dealt with this date — don't nag - else: - db.execute(text(""" - INSERT INTO rate_parity_alerts - (rate_date, room_category, newbook_rate, booking_com_rate, - difference_pct, alert_type, alert_status) - VALUES (:date, :room, :nb, :bc, :diff, :atype, 'active') - """), { - "date": c["date"], "room": category_label, - "nb": c["newbook_rate"], "bc": c["booking_rate"], - "diff": round(c["dev_pct"], 2), - "atype": "higher" if c["dev_pct"] > 0 else "lower", - }) - created += 1 - else: - if alert and alert["status"] in ("active", "acknowledged"): - db.execute( - text("UPDATE rate_parity_alerts SET alert_status = 'resolved' WHERE id = :id"), - {"id": alert["id"]} - ) - resolved += 1 + latest_alert = _fetch_latest_alerts(db, start, end) + created, updated, resolved = _upsert_alerts(db, comparisons, latest_alert) db.commit() summary = { diff --git a/backend/services/booking_scraper.py b/backend/services/booking_scraper.py index c080aa1..f0a7c5c 100644 --- a/backend/services/booking_scraper.py +++ b/backend/services/booking_scraper.py @@ -613,6 +613,7 @@ async def _scrape_hotels_concurrent( backend = PlaywrightHotelPageBackend(proxy_config=proxy_util.load_config(wdb)) try: for rate_date in date_shard: + date_rates = 0 for hotel in hotels_seen: try: result = await scrape_hotel_date(wdb, hotel, rate_date, backend, batch_id, adults) @@ -632,12 +633,16 @@ async def _scrape_hotels_concurrent( elif result['success']: acc['rates'] += result['rates_count'] acc['completed'] += 1 + date_rates += result['rates_count'] _increment_batch_progress(wdb, batch_id, delta_rates=result['rates_count'], delta_completed=1) else: acc['failed'] += 1 _increment_batch_progress(wdb, batch_id, delta_failed=1) + + if date_rates > 0: + _safe_parity_recheck(rate_date) finally: try: await backend.close() @@ -687,6 +692,15 @@ def _increment_batch_progress( pass +def _safe_parity_recheck(rate_date: date): + """Re-run parity check for one date after it has been successfully scraped. Never raises.""" + try: + from jobs.check_rate_parity import run_parity_check_for_date + run_parity_check_for_date(rate_date) + except Exception as e: + logger.warning(f"Post-scrape parity recheck for {rate_date} failed: {e}") + + def _safe_mark_queue(db: Session, queue_id: Optional[int], status: str, error: str = None): """mark_queue_item that never raises — a marking failure shouldn't kill a worker.""" if queue_id is None: @@ -749,6 +763,7 @@ async def _scrape_dates_concurrent( delta_hotels=result['hotels_count'], delta_rates=result['rates_count'], delta_completed=1) + _safe_parity_recheck(rate_date) else: acc['failed'] += 1 _safe_mark_queue(wdb, queue_id, 'failed', result.get('error')) diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index 7615d8e..d751187 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -23,7 +23,7 @@ export default function Layout({ children }: { children: ReactNode }) { const { data: alertCount } = useQuery({ queryKey: ['parity-alert-count'], queryFn: () => api.get('/competitors/parity/alerts?status=active').then(r => r.data.length), - refetchInterval: 60_000, + refetchInterval: 15_000, enabled: can(user, 'view_competitors'), }) diff --git a/frontend/src/pages/MarketView.tsx b/frontend/src/pages/MarketView.tsx index 306f91b..1fdb3f6 100644 --- a/frontend/src/pages/MarketView.tsx +++ b/frontend/src/pages/MarketView.tsx @@ -843,6 +843,15 @@ const ParityAlertsTab: React.FC = () => { }, }) + const recheckMutation = useMutation({ + mutationFn: async (rateDate: string) => + (await api.post(`/competitors/parity/check-date?rate_date=${rateDate}`)).data, + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ['parity-alerts'] }) + queryClient.invalidateQueries({ queryKey: ['parity-alert-count'] }) + }, + }) + const markupUnit = parityConfig?.['parity_markup_unit'] ?? 'pct' const toleranceUnit = parityConfig?.['parity_tolerance_unit'] ?? 'pct' const markupVal = parityConfig?.['parity_markup_value'] ?? parityConfig?.['parity_expected_markup_pct'] ?? '0' @@ -945,19 +954,29 @@ const ParityAlertsTab: React.FC = () => { {a.alert_status} - - {a.alert_status === 'active' && ( + +
+ {a.alert_status === 'active' && ( + + )} - )} - {a.alert_status === 'acknowledged' && a.acknowledged_by && ( - by {a.acknowledged_by} - )} + {a.alert_status === 'acknowledged' && a.acknowledged_by && ( + by {a.acknowledged_by} + )} +
))}