"""Movie recommendation logic."""
from __future__ import annotations
from datetime import date, time
from src.models import Movie
from src.scraper import fetch_amsterdam_cinemas
def recommend_movies(
min_rating: float = 0.0,
preferred_times: list[time] | None = None,
preferred_cinemas: list[str] | None = None,
target_date: date | None = None,
) -> list[Movie]:
"""Recommend movies based on criteria."""
# This will be called from the MCP server which will handle async
# For now, we'll create a sync wrapper that the MCP server can use
raise NotImplementedError("Use async_recommend_movies instead")
async def async_recommend_movies(
min_rating: float = 0.0,
preferred_times: list[time] | None = None,
preferred_cinemas: list[str] | None = None,
target_date: date | None = None,
) -> list[Movie]:
"""Recommend movies based on criteria (async version)."""
# Fetch all cinemas
cinemas = await fetch_amsterdam_cinemas()
# Collect all unique movies
movie_map: dict[str, Movie] = {}
for cinema in cinemas:
for movie in cinema.movies:
if movie.title not in movie_map:
movie_map[movie.title] = Movie(
title=movie.title,
rating=movie.rating,
showtimes=[],
)
# Merge showtimes
movie_map[movie.title].showtimes.extend(movie.showtimes)
# Filter movies
filtered_movies: list[Movie] = []
for movie in movie_map.values():
# Filter by rating
if movie.rating is not None and movie.rating < min_rating:
continue
# Filter by date if specified
if target_date:
movie.showtimes = [st for st in movie.showtimes if st.showtime_date == target_date]
if not movie.showtimes:
continue
# Filter by preferred cinemas if specified
if preferred_cinemas:
movie.showtimes = [
st for st in movie.showtimes if st.cinema_name in preferred_cinemas
]
if not movie.showtimes:
continue
# Filter by preferred times if specified
if preferred_times:
movie.showtimes = [
st
for st in movie.showtimes
if any(
abs((st.showtime_time.hour * 60 + st.showtime_time.minute) - (pt.hour * 60 + pt.minute)) < 30
for pt in preferred_times
)
]
if not movie.showtimes:
continue
# Only include movies that have showtimes after filtering
if movie.showtimes:
filtered_movies.append(movie)
# Sort by rating (highest first), then by number of showtimes
filtered_movies.sort(
key=lambda m: (
m.rating if m.rating is not None else 0.0,
len(m.showtimes),
),
reverse=True,
)
return filtered_movies
