Add data generation scripts and public holidays service
- Add scripts for generating booking data, building metadata, and updating holidays - Implement public_holidays.py service with holiday lookup functionality - Update CSV data with expanded hotel booking records - Add holidays.json dataset for public holiday dates - Enhance LLM report service with improved formatting - Update metadata domain model and API dependencies - Add test coverage for dataset and public holidays
This commit is contained in:
@@ -1,48 +1,50 @@
|
||||
from functools import lru_cache
|
||||
|
||||
from fastapi import Depends
|
||||
|
||||
from nf_hotel_api.core.config import Settings, get_settings
|
||||
from nf_hotel_api.domain.metadata import HotelMetadata
|
||||
from nf_hotel_api.repositories.metadata_repository import JsonHotelMetadataRepository
|
||||
from nf_hotel_api.services.cleaning import DataCleaningService
|
||||
from nf_hotel_api.services.llm_report import LLMReportService
|
||||
from nf_hotel_api.services.report import ReportService
|
||||
from nf_hotel_api.services.statistics import DescriptiveStatsService
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_hotel_metadata() -> HotelMetadata:
|
||||
return JsonHotelMetadataRepository(get_settings().metadata_path).load()
|
||||
|
||||
|
||||
def get_cleaning_service(
|
||||
metadata: HotelMetadata = Depends(get_hotel_metadata),
|
||||
) -> DataCleaningService:
|
||||
return DataCleaningService(metadata)
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_stats_service() -> DescriptiveStatsService:
|
||||
return DescriptiveStatsService()
|
||||
|
||||
|
||||
def get_llm_service(
|
||||
settings: Settings = Depends(get_settings),
|
||||
metadata: HotelMetadata = Depends(get_hotel_metadata),
|
||||
) -> LLMReportService:
|
||||
return LLMReportService(
|
||||
metadata=metadata,
|
||||
base_url=settings.llm_base_url,
|
||||
api_key=settings.llm_api_key,
|
||||
model=settings.llm_model,
|
||||
timeout_seconds=settings.llm_timeout_seconds,
|
||||
)
|
||||
|
||||
|
||||
def get_report_service(
|
||||
cleaning_service: DataCleaningService = Depends(get_cleaning_service),
|
||||
stats_service: DescriptiveStatsService = Depends(get_stats_service),
|
||||
llm_service: LLMReportService = Depends(get_llm_service),
|
||||
) -> ReportService:
|
||||
return ReportService(cleaning_service, stats_service, llm_service)
|
||||
from functools import lru_cache
|
||||
|
||||
from fastapi import Depends
|
||||
|
||||
from nf_hotel_api.core.config import Settings, get_settings
|
||||
from nf_hotel_api.domain.metadata import HotelMetadata
|
||||
from nf_hotel_api.repositories.metadata_repository import JsonHotelMetadataRepository
|
||||
from nf_hotel_api.services.cleaning import DataCleaningService
|
||||
from nf_hotel_api.services.llm_report import LLMReportService
|
||||
from nf_hotel_api.services.public_holidays import PublicHolidayCalendar
|
||||
from nf_hotel_api.services.report import ReportService
|
||||
from nf_hotel_api.services.statistics import DescriptiveStatsService
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_hotel_metadata() -> HotelMetadata:
|
||||
return JsonHotelMetadataRepository(get_settings().metadata_path).load()
|
||||
|
||||
|
||||
def get_cleaning_service(
|
||||
metadata: HotelMetadata = Depends(get_hotel_metadata),
|
||||
) -> DataCleaningService:
|
||||
return DataCleaningService(metadata)
|
||||
|
||||
|
||||
@lru_cache
|
||||
def get_stats_service() -> DescriptiveStatsService:
|
||||
return DescriptiveStatsService()
|
||||
|
||||
|
||||
def get_llm_service(
|
||||
settings: Settings = Depends(get_settings),
|
||||
metadata: HotelMetadata = Depends(get_hotel_metadata),
|
||||
) -> LLMReportService:
|
||||
return LLMReportService(
|
||||
metadata=metadata,
|
||||
holiday_calendar=PublicHolidayCalendar(),
|
||||
base_url=settings.llm_base_url,
|
||||
api_key=settings.llm_api_key,
|
||||
model=settings.llm_model,
|
||||
timeout_seconds=settings.llm_timeout_seconds,
|
||||
)
|
||||
|
||||
|
||||
def get_report_service(
|
||||
cleaning_service: DataCleaningService = Depends(get_cleaning_service),
|
||||
stats_service: DescriptiveStatsService = Depends(get_stats_service),
|
||||
llm_service: LLMReportService = Depends(get_llm_service),
|
||||
) -> ReportService:
|
||||
return ReportService(cleaning_service, stats_service, llm_service)
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
from pydantic import BaseModel, Field
|
||||
from datetime import date
|
||||
|
||||
from pydantic import BaseModel, Field, model_validator
|
||||
|
||||
|
||||
class RoomType(BaseModel):
|
||||
@@ -11,19 +13,59 @@ class RoomType(BaseModel):
|
||||
room_count: int | None = Field(default=None, ge=0)
|
||||
|
||||
|
||||
class Address(BaseModel):
|
||||
street: str
|
||||
city: str
|
||||
country: str
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.street}, {self.city}, {self.country}"
|
||||
|
||||
|
||||
class NearbyEvent(BaseModel):
|
||||
"""An event in Phnom Penh close enough to the hotel to affect bookings."""
|
||||
|
||||
name: str
|
||||
start_date: date
|
||||
end_date: date
|
||||
venue: str
|
||||
source_url: str
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _end_not_before_start(self) -> "NearbyEvent":
|
||||
if self.end_date < self.start_date:
|
||||
raise ValueError(f"{self.name}: end_date is before start_date")
|
||||
return self
|
||||
|
||||
def covers(self, day: date) -> bool:
|
||||
return self.start_date <= day <= self.end_date
|
||||
|
||||
def describe(self) -> str:
|
||||
if self.start_date == self.end_date:
|
||||
return f"{self.start_date.isoformat()}: {self.name} ({self.venue})"
|
||||
return (
|
||||
f"{self.start_date.isoformat()} to {self.end_date.isoformat()}: "
|
||||
f"{self.name} ({self.venue})"
|
||||
)
|
||||
|
||||
|
||||
class HotelMetadata(BaseModel):
|
||||
"""Reference data about the hotel that is not part of the booking records.
|
||||
|
||||
``room_types`` is keyed by the ``assigned_room_type`` code used in the CSV.
|
||||
Public holidays are deliberately not stored here; see
|
||||
``services.public_holidays.PublicHolidayCalendar``.
|
||||
"""
|
||||
|
||||
hotel: str
|
||||
address: Address | None = None
|
||||
currency: str = "USD"
|
||||
room_types: dict[str, RoomType]
|
||||
nearby_events: list[NearbyEvent] = []
|
||||
|
||||
def describe(self) -> str:
|
||||
"""Plain-text summary of the room catalogue, for the LLM prompt."""
|
||||
lines = []
|
||||
"""Plain-text summary of the hotel, its rooms and events, for the LLM prompt."""
|
||||
lines = [f"Hotel: {self.hotel}" + (f", {self.address}" if self.address else "")]
|
||||
for code, room in self.room_types.items():
|
||||
count = "unknown" if room.room_count is None else str(room.room_count)
|
||||
lines.append(
|
||||
@@ -31,4 +73,7 @@ class HotelMetadata(BaseModel):
|
||||
f"{room.standard_price_per_night:g} {self.currency} per night, "
|
||||
f"number of rooms in hotel: {count}"
|
||||
)
|
||||
if self.nearby_events:
|
||||
lines.append("Events in Phnom Penh near the hotel:")
|
||||
lines.extend(f"- {event.describe()}" for event in self.nearby_events)
|
||||
return "\n".join(lines)
|
||||
|
||||
@@ -1,86 +1,107 @@
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from nf_hotel_api.domain.metadata import HotelMetadata
|
||||
|
||||
_PROMPT_TEMPLATE = """You work as a data analyst and in marketing to optimize hotel operations.
|
||||
User cannot interact with you so do not ask questions.
|
||||
Respond in markdown format.
|
||||
{hotel_context}We have extract descriptive analysis:\n
|
||||
{descriptive_analysis_data}"""
|
||||
|
||||
_HOTEL_CONTEXT_TEMPLATE = """Hotel reference data (room types the bookings refer to):
|
||||
{room_catalogue}
|
||||
The column prize_per_nigth is the price the customer actually paid per night.
|
||||
The column revenue is nights x prize_per_nigth for non-cancelled bookings.
|
||||
"""
|
||||
|
||||
# Reasoning models (e.g. Qwen3) may wrap their internal reasoning in
|
||||
# <think>...</think>; that content must never reach the API response.
|
||||
_THINK_BLOCK_PATTERN = re.compile(r"<think>.*?</think>", re.DOTALL | re.IGNORECASE)
|
||||
|
||||
|
||||
class LLMServiceError(RuntimeError):
|
||||
"""Raised when the LLM backend cannot produce a report."""
|
||||
|
||||
|
||||
class LLMReportService:
|
||||
"""Turns descriptive statistics into a marketing/ops narrative via an LLM."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
api_key: str,
|
||||
model: str,
|
||||
timeout_seconds: float,
|
||||
metadata: HotelMetadata | None = None,
|
||||
) -> None:
|
||||
self._metadata = metadata
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._api_key = api_key
|
||||
self._model = model
|
||||
self._timeout_seconds = timeout_seconds
|
||||
|
||||
async def generate_report(self, descriptive_stats: dict[str, Any]) -> str:
|
||||
hotel_context = (
|
||||
_HOTEL_CONTEXT_TEMPLATE.format(room_catalogue=self._metadata.describe())
|
||||
if self._metadata
|
||||
else ""
|
||||
)
|
||||
prompt = _PROMPT_TEMPLATE.format(
|
||||
hotel_context=hotel_context,
|
||||
descriptive_analysis_data=json.dumps(descriptive_stats, indent=2),
|
||||
)
|
||||
|
||||
payload = {
|
||||
"model": self._model,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
"temperature": 0.3,
|
||||
}
|
||||
headers = {"Authorization": f"Bearer {self._api_key}"}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self._timeout_seconds) as client:
|
||||
response = await client.post(
|
||||
f"{self._base_url}/chat/completions",
|
||||
json=payload,
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
except httpx.HTTPError as exc:
|
||||
raise LLMServiceError(f"LLM backend request failed: {exc}") from exc
|
||||
|
||||
data = response.json()
|
||||
try:
|
||||
raw_content = data["choices"][0]["message"]["content"]
|
||||
except (KeyError, IndexError) as exc:
|
||||
raise LLMServiceError(f"Unexpected LLM response shape: {data}") from exc
|
||||
|
||||
return self._strip_thinking(raw_content)
|
||||
|
||||
@staticmethod
|
||||
def _strip_thinking(content: str) -> str:
|
||||
return _THINK_BLOCK_PATTERN.sub("", content).strip()
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from nf_hotel_api.domain.metadata import HotelMetadata
|
||||
from nf_hotel_api.services.public_holidays import PublicHolidayCalendar
|
||||
|
||||
_PROMPT_TEMPLATE = """You work as a data analyst and in marketing to optimize hotel operations.
|
||||
User cannot interact with you so do not ask questions.
|
||||
Respond in markdown format.
|
||||
{hotel_context}We have extract descriptive analysis:\n
|
||||
{descriptive_analysis_data}"""
|
||||
|
||||
_HOTEL_CONTEXT_TEMPLATE = """Hotel reference data (room types the bookings refer to):
|
||||
{room_catalogue}
|
||||
The column prize_per_nigth is the price the customer actually paid per night.
|
||||
The column revenue is nights x prize_per_nigth for non-cancelled bookings.
|
||||
{holidays}"""
|
||||
|
||||
# Reasoning models (e.g. Qwen3) may wrap their internal reasoning in
|
||||
# <think>...</think>; that content must never reach the API response.
|
||||
_THINK_BLOCK_PATTERN = re.compile(r"<think>.*?</think>", re.DOTALL | re.IGNORECASE)
|
||||
|
||||
|
||||
class LLMServiceError(RuntimeError):
|
||||
"""Raised when the LLM backend cannot produce a report."""
|
||||
|
||||
|
||||
class LLMReportService:
|
||||
"""Turns descriptive statistics into a marketing/ops narrative via an LLM."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
api_key: str,
|
||||
model: str,
|
||||
timeout_seconds: float,
|
||||
metadata: HotelMetadata | None = None,
|
||||
holiday_calendar: PublicHolidayCalendar | None = None,
|
||||
) -> None:
|
||||
self._metadata = metadata
|
||||
self._holiday_calendar = holiday_calendar
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._api_key = api_key
|
||||
self._model = model
|
||||
self._timeout_seconds = timeout_seconds
|
||||
|
||||
async def generate_report(self, descriptive_stats: dict[str, Any]) -> str:
|
||||
prompt = self._build_prompt(descriptive_stats)
|
||||
|
||||
payload = {
|
||||
"model": self._model,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
"temperature": 0.3,
|
||||
}
|
||||
headers = {"Authorization": f"Bearer {self._api_key}"}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self._timeout_seconds) as client:
|
||||
response = await client.post(
|
||||
f"{self._base_url}/chat/completions",
|
||||
json=payload,
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
except httpx.HTTPError as exc:
|
||||
raise LLMServiceError(f"LLM backend request failed: {exc}") from exc
|
||||
|
||||
data = response.json()
|
||||
try:
|
||||
raw_content = data["choices"][0]["message"]["content"]
|
||||
except (KeyError, IndexError) as exc:
|
||||
raise LLMServiceError(f"Unexpected LLM response shape: {data}") from exc
|
||||
|
||||
return self._strip_thinking(raw_content)
|
||||
|
||||
def _build_prompt(self, descriptive_stats: dict[str, Any]) -> str:
|
||||
hotel_context = (
|
||||
_HOTEL_CONTEXT_TEMPLATE.format(
|
||||
room_catalogue=self._metadata.describe(),
|
||||
holidays=self._holiday_section(descriptive_stats),
|
||||
)
|
||||
if self._metadata
|
||||
else ""
|
||||
)
|
||||
return _PROMPT_TEMPLATE.format(
|
||||
hotel_context=hotel_context,
|
||||
descriptive_analysis_data=json.dumps(descriptive_stats, indent=2),
|
||||
)
|
||||
|
||||
def _holiday_section(self, descriptive_stats: dict[str, Any]) -> str:
|
||||
"""Public holidays for the years the bookings' arrival dates span."""
|
||||
if self._holiday_calendar is None:
|
||||
return ""
|
||||
arrival = descriptive_stats.get("arrival_date", {})
|
||||
try:
|
||||
first_year = int(str(arrival["min"])[:4])
|
||||
last_year = int(str(arrival["max"])[:4])
|
||||
except (KeyError, ValueError):
|
||||
return ""
|
||||
return f"Public holidays:\n{self._holiday_calendar.describe(first_year, last_year)}\n"
|
||||
|
||||
@staticmethod
|
||||
def _strip_thinking(content: str) -> str:
|
||||
return _THINK_BLOCK_PATTERN.sub("", content).strip()
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, timedelta
|
||||
|
||||
import holidays
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HolidayPeriod:
|
||||
"""A holiday, or a run of consecutive days with the same holiday name."""
|
||||
|
||||
name: str
|
||||
start_date: date
|
||||
end_date: date
|
||||
|
||||
def describe(self) -> str:
|
||||
if self.start_date == self.end_date:
|
||||
return f"{self.start_date.isoformat()}: {self.name}"
|
||||
return f"{self.start_date.isoformat()} to {self.end_date.isoformat()}: {self.name}"
|
||||
|
||||
|
||||
class PublicHolidayCalendar:
|
||||
"""Public holidays looked up in the ``holidays`` package (not stored in metadata).
|
||||
|
||||
Defaults to Cambodia (``KH``), where the hotel is. ``data/holidays.json`` is
|
||||
an export of :meth:`periods`, refreshed by ``scripts/update_holidays.py``.
|
||||
"""
|
||||
|
||||
def __init__(self, country_code: str = "KH") -> None:
|
||||
self._country_code = country_code
|
||||
|
||||
@property
|
||||
def country_code(self) -> str:
|
||||
return self._country_code
|
||||
|
||||
@staticmethod
|
||||
def package_version() -> str:
|
||||
return holidays.__version__
|
||||
|
||||
def for_years(self, first_year: int, last_year: int) -> dict[date, str]:
|
||||
"""Return ``{day: holiday name}`` for every holiday from first to last year."""
|
||||
calendar = holidays.country_holidays(
|
||||
self._country_code, years=range(first_year, last_year + 1), language="en_US"
|
||||
)
|
||||
return dict(sorted(calendar.items()))
|
||||
|
||||
def name_on(self, day: date) -> str | None:
|
||||
return holidays.country_holidays(self._country_code, years=day.year, language="en_US").get(day)
|
||||
|
||||
def periods(self, first_year: int, last_year: int) -> list[HolidayPeriod]:
|
||||
"""Consecutive days with the same name are merged into one period."""
|
||||
periods: list[HolidayPeriod] = []
|
||||
for day, name in self.for_years(first_year, last_year).items():
|
||||
last = periods[-1] if periods else None
|
||||
if last and last.name == name and day - last.end_date == timedelta(days=1):
|
||||
periods[-1] = HolidayPeriod(name, last.start_date, day)
|
||||
else:
|
||||
periods.append(HolidayPeriod(name, day, day))
|
||||
return periods
|
||||
|
||||
def describe(self, first_year: int, last_year: int) -> str:
|
||||
"""One line per holiday period, for the LLM prompt."""
|
||||
return "\n".join(f"- {p.describe()}" for p in self.periods(first_year, last_year))
|
||||
Reference in New Issue
Block a user