""" Direct booking engine API — hotel management, discovery, scrape control, rate data. """ import asyncio import json import logging from datetime import date, timedelta from typing import Optional from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks from pydantic import BaseModel from sqlalchemy import text from database import AsyncSessionLocal, SyncSessionLocal from auth import get_current_user, require_cap from services.direct_profiles import PROFILES, detect_profile, get_profile from services.direct_scraper import run_discovery, get_discovery_status, run_scrape router = APIRouter() log = logging.getLogger(__name__) # ─── Pydantic models ───────────────────────────────────────────────────────── class HotelCreate(BaseModel): name: str profile_name: str params: dict class HotelUpdate(BaseModel): name: Optional[str] = None room_labels: Optional[dict] = None rate_labels: Optional[dict] = None room_order: Optional[list] = None benchmark_room: Optional[str] = None benchmark_rate: Optional[str] = None tier_base_room: Optional[str] = None tier_offsets: Optional[dict] = None scrape_enabled: Optional[bool] = None params: Optional[dict] = None class DetectRequest(BaseModel): url: str # ─── Profiles ──────────────────────────────────────────────────────────────── @router.get("/profiles") async def list_profiles(user=Depends(get_current_user)): require_cap(user, "view_direct_rates") return [ { "name": name, "label": cls.label, "required_params": cls.required_params, } for name, cls in PROFILES.items() ] @router.post("/profiles/detect") async def detect_engine(req: DetectRequest, user=Depends(get_current_user)): require_cap(user, "manage_hotels") result = detect_profile(req.url) if not result: raise HTTPException(status_code=422, detail="Could not detect booking engine from URL") return result # ─── Hotel CRUD ─────────────────────────────────────────────────────────────── @router.get("/hotels") async def list_hotels(user=Depends(get_current_user)): require_cap(user, "view_direct_rates") async with AsyncSessionLocal() as db: result = await db.execute( text("""SELECT h.id, h.name, h.profile_name, h.params, h.room_labels, h.rate_labels, h.room_order, h.benchmark_room, h.benchmark_rate, h.tier_base_room, h.tier_offsets, h.scrape_enabled, h.last_scraped_at, COUNT(DISTINCT r.stay_date) AS scraped_dates, MAX(r.scraped_at) AS last_rate_at FROM direct_competitor_hotels h LEFT JOIN direct_rates r ON r.hotel_id = h.id GROUP BY h.id ORDER BY h.id""") ) rows = result.mappings().all() return [dict(r) for r in rows] @router.post("/hotels", status_code=201) async def create_hotel(body: HotelCreate, user=Depends(get_current_user)): require_cap(user, "manage_hotels") if body.profile_name not in PROFILES: raise HTTPException(status_code=400, detail=f"Unknown profile: {body.profile_name}") async with AsyncSessionLocal() as db: result = await db.execute( text("""INSERT INTO direct_competitor_hotels (name, profile_name, params) VALUES (:name, :profile, :params) RETURNING id"""), {"name": body.name, "profile": body.profile_name, "params": json.dumps(body.params)} ) new_id = result.fetchone()[0] await db.commit() return {"id": new_id, "name": body.name, "profile_name": body.profile_name} @router.put("/hotels/{hotel_id}") async def update_hotel(hotel_id: int, body: HotelUpdate, user=Depends(get_current_user)): require_cap(user, "manage_hotels") updates = {} if body.name is not None: updates["name"] = body.name if body.scrape_enabled is not None: updates["scrape_enabled"] = body.scrape_enabled if body.benchmark_room is not None: updates["benchmark_room"] = body.benchmark_room if body.benchmark_rate is not None: updates["benchmark_rate"] = body.benchmark_rate if body.tier_base_room is not None: updates["tier_base_room"] = body.tier_base_room if body.params is not None: updates["params"] = json.dumps(body.params) if body.room_labels is not None: updates["room_labels"] = json.dumps(body.room_labels) if body.rate_labels is not None: updates["rate_labels"] = json.dumps(body.rate_labels) if body.room_order is not None: updates["room_order"] = json.dumps(body.room_order) if body.tier_offsets is not None: updates["tier_offsets"] = json.dumps(body.tier_offsets) if not updates: raise HTTPException(status_code=400, detail="No fields to update") set_clause = ", ".join(f"{k} = :{k}" for k in updates) updates["hotel_id"] = hotel_id async with AsyncSessionLocal() as db: await db.execute( text(f"UPDATE direct_competitor_hotels SET {set_clause} WHERE id = :hotel_id"), updates ) await db.commit() return {"ok": True} @router.delete("/hotels/{hotel_id}", status_code=204) async def delete_hotel(hotel_id: int, user=Depends(get_current_user)): require_cap(user, "manage_hotels") async with AsyncSessionLocal() as db: await db.execute( text("DELETE FROM direct_competitor_hotels WHERE id = :id"), {"id": hotel_id} ) await db.commit() # ─── Discovery ─────────────────────────────────────────────────────────────── @router.post("/hotels/{hotel_id}/discover") async def trigger_discovery(hotel_id: int, background_tasks: BackgroundTasks, user=Depends(get_current_user)): require_cap(user, "manage_hotels") async with AsyncSessionLocal() as db: row = await db.execute( text("SELECT profile_name, params FROM direct_competitor_hotels WHERE id = :id"), {"id": hotel_id} ) hotel = row.mappings().fetchone() if not hotel: raise HTTPException(status_code=404, detail="Hotel not found") background_tasks.add_task( asyncio.get_event_loop().run_until_complete, run_discovery(hotel_id, hotel["profile_name"], hotel["params"]) ) return {"status": "discovery started", "hotel_id": hotel_id} @router.get("/hotels/{hotel_id}/discovery-status") async def discovery_status(hotel_id: int, user=Depends(get_current_user)): require_cap(user, "view_direct_rates") return get_discovery_status(hotel_id) # ─── Manual scrape trigger ──────────────────────────────────────────────────── @router.post("/hotels/{hotel_id}/scrape") async def trigger_scrape(hotel_id: int, background_tasks: BackgroundTasks, user=Depends(get_current_user)): require_cap(user, "manage_scraper") async with AsyncSessionLocal() as db: row = await db.execute( text("SELECT profile_name, params FROM direct_competitor_hotels WHERE id = :id"), {"id": hotel_id} ) hotel = row.mappings().fetchone() if not hotel: raise HTTPException(status_code=404, detail="Hotel not found") def _bg(): run_scrape(hotel_id, hotel["profile_name"], hotel["params"]) background_tasks.add_task(_bg) return {"status": "scrape started", "hotel_id": hotel_id} # ─── Rate data ──────────────────────────────────────────────────────────────── @router.get("/hotels/{hotel_id}/dates") async def hotel_dates( hotel_id: int, from_date: date = None, to_date: date = None, user=Depends(get_current_user) ): require_cap(user, "view_direct_rates") if from_date is None: from_date = date.today() if to_date is None: to_date = date.today() + timedelta(days=89) async with AsyncSessionLocal() as db: # Hotel config for labels/tier offsets cfg_row = await db.execute( text("SELECT name, room_labels, rate_labels, tier_offsets, benchmark_room, benchmark_rate, tier_base_room FROM direct_competitor_hotels WHERE id = :id"), {"id": hotel_id} ) hotel = cfg_row.mappings().fetchone() if not hotel: raise HTTPException(status_code=404, detail="Hotel not found") # Latest snapshot per date: cheapest available price_incl rates_result = await db.execute( text(""" SELECT r.stay_date, MIN(r.price_incl) FILTER (WHERE r.availability > 0) AS cheapest_rate, BOOL_OR(r.availability > 0) AS has_availability, BOOL_OR(r.min_stay_nights IS NOT NULL AND r.min_stay_nights > 1) AS has_min_stay, MAX(r.scraped_at) AS scraped_at FROM ( SELECT DISTINCT ON (room_id, rate_id) stay_date, room_id, rate_id, availability, price_incl, min_stay_nights, scraped_at FROM direct_rates WHERE hotel_id = :hid AND stay_date BETWEEN :fd AND :td ORDER BY room_id, rate_id, scraped_at DESC ) r GROUP BY r.stay_date ORDER BY r.stay_date """), {"hid": hotel_id, "fd": from_date, "td": to_date} ) dates = [dict(r) for r in rates_result.mappings().all()] return { "hotel_id": hotel_id, "hotel_name": hotel["name"], "from_date": str(from_date), "to_date": str(to_date), "dates": dates, } @router.get("/hotels/{hotel_id}/date/{rate_date}/rooms") async def hotel_date_rooms( hotel_id: int, rate_date: date, user=Depends(get_current_user) ): require_cap(user, "view_direct_rates") async with AsyncSessionLocal() as db: cfg_row = await db.execute( text("SELECT room_labels, rate_labels, tier_offsets, tier_base_room, benchmark_room, benchmark_rate FROM direct_competitor_hotels WHERE id = :id"), {"id": hotel_id} ) hotel = cfg_row.mappings().fetchone() if not hotel: raise HTTPException(status_code=404, detail="Hotel not found") # Latest snapshot per room/rate for this date rooms_result = await db.execute( text(""" SELECT DISTINCT ON (room_id, rate_id) room_id, rate_id, availability, price_excl, price_incl, currency, min_stay_nights, scraped_at FROM direct_rates WHERE hotel_id = :hid AND stay_date = :sd ORDER BY room_id, rate_id, scraped_at DESC """), {"hid": hotel_id, "sd": rate_date} ) rooms = [dict(r) for r in rooms_result.mappings().all()] # Resolve tier-normalised benchmark rates room_labels = hotel["room_labels"] or {} rate_labels = hotel["rate_labels"] or {} tier_offsets = hotel["tier_offsets"] or {} tier_base_room = hotel["tier_base_room"] benchmark_room = hotel["benchmark_room"] benchmark_rate = hotel["benchmark_rate"] # Find benchmark price bench_price = None if benchmark_room and benchmark_rate: bench_match = next( (r for r in rooms if r["room_id"] == benchmark_room and r["rate_id"] == benchmark_rate and r["availability"] > 0 and r["price_incl"]), None ) if bench_match: bench_price = float(bench_match["price_incl"]) elif tier_base_room and tier_offsets: base_match = next( (r for r in rooms if r["room_id"] == tier_base_room and r["availability"] > 0 and r["price_incl"]), None ) if base_match: base_price = float(base_match["price_incl"]) bench_offset = tier_offsets.get(benchmark_room, 0) base_offset = tier_offsets.get(tier_base_room, 0) bench_price = base_price - base_offset + bench_offset enriched = [] for r in rooms: r["room_label"] = room_labels.get(r["room_id"], r["room_id"]) r["rate_label"] = rate_labels.get(r["rate_id"], r["rate_id"]) # Derive bench_rate for this room from tier offsets r["bench_rate"] = None if bench_price is not None and tier_offsets and tier_base_room: room_offset = tier_offsets.get(r["room_id"]) bench_offset = tier_offsets.get(benchmark_room, 0) if room_offset is not None: r["bench_rate"] = round(bench_price + (room_offset - bench_offset), 2) enriched.append(r) return { "hotel_id": hotel_id, "date": str(rate_date), "bench_price": bench_price, "rooms": enriched, }