Auto-recheck parity after scrape + manual Recheck button + faster badge
Parity logic: - Extract _fetch_latest_alerts() and _upsert_alerts() helpers so the alert upsert loop is no longer duplicated between the daily job and per-date runs - Add run_parity_check_for_date(date) which runs the full comparison + alert upsert for a single date Scraper integration: - _safe_parity_recheck(date) wrapper (never raises) called after each successful date in both search-results and hotel-page workers; hotel-page mode waits until all hotels for the date are done before rechecking API: - POST /competitors/parity/check-date?rate_date=YYYY-MM-DD for manual recheck Frontend: - Recheck button on every parity alert row (all statuses); invalidates alerts list and badge count on success - parity-alert-count badge polls every 15s (was 60s) so new alerts from the scheduled job or post-scrape rechecks appear quickly Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
8a80c66a3b
commit
cab1a39503
5 changed files with 146 additions and 66 deletions
|
|
@ -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
|
||||
# ============================================
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
|
|
@ -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'))
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ export default function Layout({ children }: { children: ReactNode }) {
|
|||
const { data: alertCount } = useQuery<number>({
|
||||
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'),
|
||||
})
|
||||
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
</span>
|
||||
</td>
|
||||
<td style={tdStyle}>
|
||||
{a.alert_status === 'active' && (
|
||||
<td style={mergeStyles(tdStyle, { whiteSpace: 'nowrap' })}>
|
||||
<div style={{ display: 'flex', gap: 6, alignItems: 'center' }}>
|
||||
{a.alert_status === 'active' && (
|
||||
<button
|
||||
style={buttonStyle('outline', 'small')}
|
||||
disabled={ackMutation.isPending || recheckMutation.isPending}
|
||||
onClick={() => ackMutation.mutate(a.id)}
|
||||
>
|
||||
Acknowledge
|
||||
</button>
|
||||
)}
|
||||
<button
|
||||
style={buttonStyle('outline', 'small')}
|
||||
disabled={ackMutation.isPending}
|
||||
onClick={() => ackMutation.mutate(a.id)}
|
||||
disabled={recheckMutation.isPending || ackMutation.isPending}
|
||||
onClick={() => recheckMutation.mutate(a.rate_date)}
|
||||
title="Re-run parity check using the latest scraped rates for this date"
|
||||
>
|
||||
Acknowledge
|
||||
Recheck
|
||||
</button>
|
||||
)}
|
||||
{a.alert_status === 'acknowledged' && a.acknowledged_by && (
|
||||
<span style={{ fontSize: 11, color: 'var(--text-mid)' }}>by {a.acknowledged_by}</span>
|
||||
)}
|
||||
{a.alert_status === 'acknowledged' && a.acknowledged_by && (
|
||||
<span style={{ fontSize: 11, color: 'var(--text-mid)' }}>by {a.acknowledged_by}</span>
|
||||
)}
|
||||
</div>
|
||||
</td>
|
||||
</tr>
|
||||
))}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue