Initial commit — Movie Night media discovery app
AI-powered web app that recommends unwatched movies from a Jellyfin library based on natural language mood input. Jellyfin auth, modular LLM backend (Claude/OpenAI/Ollama), two-tier pre-filter + AI ranking, mobile-responsive dark theme UI with poster cards and deep links. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,189 @@
|
||||
import httpx
|
||||
from app.config import settings
|
||||
|
||||
_headers = {
|
||||
"X-MediaBrowser-Token": settings.jellyfin_api_key,
|
||||
}
|
||||
|
||||
_client: httpx.AsyncClient | None = None
|
||||
|
||||
|
||||
def _get_client() -> httpx.AsyncClient:
|
||||
global _client
|
||||
if _client is None or _client.is_closed:
|
||||
_client = httpx.AsyncClient(
|
||||
base_url=settings.jellyfin_url,
|
||||
headers=_headers,
|
||||
timeout=30.0,
|
||||
)
|
||||
return _client
|
||||
|
||||
|
||||
async def authenticate_user(username: str, password: str) -> dict | None:
|
||||
"""Authenticate a user via Jellyfin. Returns user info + token or None."""
|
||||
client = _get_client()
|
||||
auth_header = (
|
||||
'MediaBrowser Client="Movie Night", Device="Web", DeviceId="movie-night-app", Version="1.0.0"'
|
||||
)
|
||||
try:
|
||||
res = await client.post(
|
||||
"/Users/AuthenticateByName",
|
||||
json={"Username": username, "Pw": password},
|
||||
headers={"X-Emby-Authorization": auth_header},
|
||||
)
|
||||
if res.status_code == 200:
|
||||
data = res.json()
|
||||
return {
|
||||
"user_id": data["User"]["Id"],
|
||||
"username": data["User"]["Name"],
|
||||
"token": data["AccessToken"],
|
||||
}
|
||||
except httpx.HTTPError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
async def get_users() -> list[dict]:
|
||||
"""Get all Jellyfin users."""
|
||||
client = _get_client()
|
||||
try:
|
||||
res = await client.get("/Users")
|
||||
if res.status_code == 200:
|
||||
return [{"id": u["Id"], "name": u["Name"]} for u in res.json()]
|
||||
except httpx.HTTPError:
|
||||
pass
|
||||
return []
|
||||
|
||||
|
||||
async def get_all_movies(user_id: str) -> list[dict]:
|
||||
"""Fetch all movies from Jellyfin with full metadata, paginated."""
|
||||
client = _get_client()
|
||||
movies = []
|
||||
start_index = 0
|
||||
page_size = 200
|
||||
fields = "Genres,Overview,CommunityRating,CriticRating,OfficialRating,Studios,People,Tags,RunTimeTicks"
|
||||
|
||||
while True:
|
||||
try:
|
||||
res = await client.get(
|
||||
f"/Users/{user_id}/Items",
|
||||
params={
|
||||
"IncludeItemTypes": "Movie",
|
||||
"Recursive": "true",
|
||||
"Fields": fields,
|
||||
"Limit": page_size,
|
||||
"StartIndex": start_index,
|
||||
"SortBy": "SortName",
|
||||
"SortOrder": "Ascending",
|
||||
},
|
||||
)
|
||||
if res.status_code != 200:
|
||||
break
|
||||
|
||||
data = res.json()
|
||||
items = data.get("Items", [])
|
||||
if not items:
|
||||
break
|
||||
|
||||
for item in items:
|
||||
runtime_ticks = item.get("RunTimeTicks")
|
||||
runtime_minutes = int(runtime_ticks / 600_000_000) if runtime_ticks else None
|
||||
|
||||
people = []
|
||||
for person in (item.get("People") or [])[:5]:
|
||||
people.append({
|
||||
"name": person.get("Name", ""),
|
||||
"role": person.get("Role", person.get("Type", "")),
|
||||
})
|
||||
|
||||
movies.append({
|
||||
"jellyfin_id": item["Id"],
|
||||
"title": item.get("Name", "Unknown"),
|
||||
"sort_title": item.get("SortName"),
|
||||
"year": item.get("ProductionYear"),
|
||||
"genres": item.get("Genres", []),
|
||||
"overview": item.get("Overview"),
|
||||
"community_rating": item.get("CommunityRating"),
|
||||
"critic_rating": item.get("CriticRating"),
|
||||
"runtime_minutes": runtime_minutes,
|
||||
"content_rating": item.get("OfficialRating"),
|
||||
"studios": [s.get("Name", "") for s in (item.get("Studios") or [])],
|
||||
"people": people,
|
||||
"tags": item.get("Tags", []),
|
||||
"is_played": item.get("UserData", {}).get("Played", False),
|
||||
})
|
||||
|
||||
total = data.get("TotalRecordCount", 0)
|
||||
start_index += page_size
|
||||
if start_index >= total:
|
||||
break
|
||||
except httpx.HTTPError:
|
||||
break
|
||||
|
||||
return movies
|
||||
|
||||
|
||||
async def get_played_movie_ids(user_id: str) -> set[str]:
|
||||
"""Get set of movie IDs that have been played by a user."""
|
||||
client = _get_client()
|
||||
played_ids = set()
|
||||
start_index = 0
|
||||
page_size = 200
|
||||
|
||||
while True:
|
||||
try:
|
||||
res = await client.get(
|
||||
f"/Users/{user_id}/Items",
|
||||
params={
|
||||
"IncludeItemTypes": "Movie",
|
||||
"Recursive": "true",
|
||||
"IsPlayed": "true",
|
||||
"Limit": page_size,
|
||||
"StartIndex": start_index,
|
||||
"Fields": "",
|
||||
},
|
||||
)
|
||||
if res.status_code != 200:
|
||||
break
|
||||
|
||||
data = res.json()
|
||||
items = data.get("Items", [])
|
||||
if not items:
|
||||
break
|
||||
|
||||
for item in items:
|
||||
played_ids.add(item["Id"])
|
||||
|
||||
total = data.get("TotalRecordCount", 0)
|
||||
start_index += page_size
|
||||
if start_index >= total:
|
||||
break
|
||||
except httpx.HTTPError:
|
||||
break
|
||||
|
||||
return played_ids
|
||||
|
||||
|
||||
async def get_poster(item_id: str) -> bytes | None:
|
||||
"""Fetch poster image bytes for a movie."""
|
||||
client = _get_client()
|
||||
try:
|
||||
res = await client.get(
|
||||
f"/Items/{item_id}/Images/Primary",
|
||||
params={"maxWidth": 400, "quality": 80},
|
||||
)
|
||||
if res.status_code == 200:
|
||||
return res.content
|
||||
except httpx.HTTPError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def get_poster_url(item_id: str) -> str:
|
||||
"""Get the proxied poster URL."""
|
||||
return f"/api/poster/{item_id}"
|
||||
|
||||
|
||||
def get_deep_link(item_id: str) -> str:
|
||||
"""Get a direct link to the movie in Jellyfin."""
|
||||
return f"{settings.jellyfin_external_url}/web/index.html#!/details?id={item_id}"
|
||||
@@ -0,0 +1,157 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.config import settings
|
||||
from app.database import get_db
|
||||
from app.services.jellyfin import get_all_movies, get_played_movie_ids, get_users
|
||||
|
||||
logger = logging.getLogger("movie-night.sync")
|
||||
|
||||
|
||||
async def sync_movie_metadata():
|
||||
"""Full sync of movie metadata from Jellyfin."""
|
||||
logger.info("Starting movie metadata sync...")
|
||||
|
||||
users = await get_users()
|
||||
if not users:
|
||||
logger.warning("No Jellyfin users found, skipping sync")
|
||||
return
|
||||
|
||||
# Use first user to fetch library (all users see the same movies)
|
||||
user_id = users[0]["id"]
|
||||
movies = await get_all_movies(user_id)
|
||||
logger.info(f"Fetched {len(movies)} movies from Jellyfin")
|
||||
|
||||
db = await get_db()
|
||||
try:
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
for movie in movies:
|
||||
await db.execute(
|
||||
"""INSERT OR REPLACE INTO movies
|
||||
(jellyfin_id, title, sort_title, year, genres, overview,
|
||||
community_rating, critic_rating, runtime_minutes, content_rating,
|
||||
studios, people, tags, synced_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(
|
||||
movie["jellyfin_id"],
|
||||
movie["title"],
|
||||
movie["sort_title"],
|
||||
movie["year"],
|
||||
json.dumps(movie["genres"]),
|
||||
movie["overview"],
|
||||
movie["community_rating"],
|
||||
movie["critic_rating"],
|
||||
movie["runtime_minutes"],
|
||||
movie["content_rating"],
|
||||
json.dumps(movie["studios"]),
|
||||
json.dumps(movie["people"]),
|
||||
json.dumps(movie["tags"]),
|
||||
now,
|
||||
),
|
||||
)
|
||||
|
||||
# Update sync timestamp
|
||||
await db.execute(
|
||||
"INSERT OR REPLACE INTO sync_status (key, value) VALUES ('last_metadata_sync', ?)",
|
||||
(now,),
|
||||
)
|
||||
await db.commit()
|
||||
logger.info(f"Movie metadata sync complete: {len(movies)} movies")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def sync_watch_state():
|
||||
"""Sync watch state for all users."""
|
||||
logger.info("Starting watch state sync...")
|
||||
|
||||
users = await get_users()
|
||||
db = await get_db()
|
||||
try:
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
|
||||
for user in users:
|
||||
played_ids = await get_played_movie_ids(user["id"])
|
||||
logger.info(f"User {user['name']}: {len(played_ids)} played movies")
|
||||
|
||||
# Get all movie IDs
|
||||
cursor = await db.execute("SELECT jellyfin_id FROM movies")
|
||||
all_movie_ids = {row["jellyfin_id"] for row in await cursor.fetchall()}
|
||||
|
||||
for movie_id in all_movie_ids:
|
||||
is_played = 1 if movie_id in played_ids else 0
|
||||
await db.execute(
|
||||
"""INSERT OR REPLACE INTO watch_state
|
||||
(jellyfin_id, user_id, is_played, synced_at)
|
||||
VALUES (?, ?, ?, ?)""",
|
||||
(movie_id, user["id"], is_played, now),
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"INSERT OR REPLACE INTO sync_status (key, value) VALUES ('last_watch_sync', ?)",
|
||||
(now,),
|
||||
)
|
||||
await db.commit()
|
||||
logger.info("Watch state sync complete")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def needs_metadata_sync() -> bool:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT value FROM sync_status WHERE key = 'last_metadata_sync'"
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
return True
|
||||
last_sync = datetime.fromisoformat(row["value"])
|
||||
age_hours = (datetime.now(timezone.utc) - last_sync).total_seconds() / 3600
|
||||
return age_hours > settings.sync_interval_hours
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def needs_watch_sync() -> bool:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT value FROM sync_status WHERE key = 'last_watch_sync'"
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
return True
|
||||
last_sync = datetime.fromisoformat(row["value"])
|
||||
age_hours = (datetime.now(timezone.utc) - last_sync).total_seconds() / 3600
|
||||
return age_hours > settings.watch_state_sync_hours
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def run_periodic_sync():
|
||||
"""Background task that periodically syncs library data."""
|
||||
# Initial sync on startup
|
||||
try:
|
||||
if await needs_metadata_sync():
|
||||
await sync_movie_metadata()
|
||||
if await needs_watch_sync():
|
||||
await sync_watch_state()
|
||||
except Exception as e:
|
||||
logger.error(f"Initial sync failed: {e}")
|
||||
|
||||
# Periodic sync loop
|
||||
while True:
|
||||
try:
|
||||
await asyncio.sleep(3600) # Check every hour
|
||||
|
||||
if await needs_metadata_sync():
|
||||
await sync_movie_metadata()
|
||||
if await needs_watch_sync():
|
||||
await sync_watch_state()
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"Periodic sync failed: {e}")
|
||||
@@ -0,0 +1,17 @@
|
||||
from app.config import settings
|
||||
from app.services.llm.base import LLMProvider
|
||||
|
||||
|
||||
def get_llm_provider() -> LLMProvider:
|
||||
provider = settings.llm_provider.lower()
|
||||
if provider == "anthropic":
|
||||
from app.services.llm.anthropic import AnthropicProvider
|
||||
return AnthropicProvider()
|
||||
elif provider == "openai":
|
||||
from app.services.llm.openai_provider import OpenAIProvider
|
||||
return OpenAIProvider()
|
||||
elif provider == "ollama":
|
||||
from app.services.llm.ollama import OllamaProvider
|
||||
return OllamaProvider()
|
||||
else:
|
||||
raise ValueError(f"Unknown LLM provider: {provider}. Must be 'anthropic', 'openai', or 'ollama'.")
|
||||
@@ -0,0 +1,48 @@
|
||||
import json
|
||||
import logging
|
||||
|
||||
import anthropic
|
||||
|
||||
from app.config import settings
|
||||
from app.models import Movie
|
||||
from app.services.llm.base import SYSTEM_PROMPT, LLMProvider, build_user_message
|
||||
|
||||
logger = logging.getLogger("movie-night.llm.anthropic")
|
||||
|
||||
|
||||
class AnthropicProvider(LLMProvider):
|
||||
def __init__(self):
|
||||
self.client = anthropic.AsyncAnthropic(api_key=settings.llm_api_key)
|
||||
self.model = settings.llm_model or "claude-sonnet-4-6"
|
||||
|
||||
async def get_recommendations(self, mood: str, candidates: list[Movie], max_results: int = 6) -> list[dict]:
|
||||
system = SYSTEM_PROMPT.format(max_results=max_results)
|
||||
user_msg = build_user_message(mood, candidates)
|
||||
|
||||
logger.info(f"Calling Anthropic ({self.model}) with {len(candidates)} candidates")
|
||||
|
||||
response = await self.client.messages.create(
|
||||
model=self.model,
|
||||
max_tokens=2048,
|
||||
system=system,
|
||||
messages=[{"role": "user", "content": user_msg}],
|
||||
)
|
||||
|
||||
text = response.content[0].text.strip()
|
||||
|
||||
# Parse JSON response
|
||||
try:
|
||||
data = json.loads(text)
|
||||
return data.get("recommendations", [])
|
||||
except json.JSONDecodeError:
|
||||
# Try to extract JSON from the response
|
||||
start = text.find("{")
|
||||
end = text.rfind("}") + 1
|
||||
if start >= 0 and end > start:
|
||||
try:
|
||||
data = json.loads(text[start:end])
|
||||
return data.get("recommendations", [])
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
logger.error(f"Failed to parse LLM response: {text[:200]}")
|
||||
return []
|
||||
@@ -0,0 +1,54 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from app.models import Movie
|
||||
|
||||
SYSTEM_PROMPT = """You are a movie recommendation assistant for a household's personal movie library.
|
||||
The user will describe their mood or what kind of movie night they want. You will receive a list of
|
||||
unwatched movies from their library and recommend the best matches.
|
||||
|
||||
Rules:
|
||||
- ONLY recommend movies from the provided list — these are movies they already own but haven't watched
|
||||
- Consider genre, themes, tone, cast, era, and the movie's overview when matching to the mood
|
||||
- Provide a brief, enthusiastic 1-2 sentence explanation for each pick that connects it to the mood
|
||||
- Rank by how well they match the described mood, not by rating alone
|
||||
- If the mood mentions kids, children, or family, only recommend age-appropriate content (G, PG, or PG-13)
|
||||
- Return exactly {max_results} recommendations, or fewer only if the library has very few matches
|
||||
|
||||
Respond with ONLY valid JSON in this exact format, no other text:
|
||||
{{
|
||||
"recommendations": [
|
||||
{{
|
||||
"jellyfin_id": "the-exact-id-from-the-list",
|
||||
"title": "Movie Title",
|
||||
"reasoning": "Why this fits the mood",
|
||||
"match_score": 0.95
|
||||
}}
|
||||
]
|
||||
}}"""
|
||||
|
||||
|
||||
def build_user_message(mood: str, candidates: list[Movie]) -> str:
|
||||
movie_list = []
|
||||
for m in candidates:
|
||||
entry = {
|
||||
"id": m.jellyfin_id,
|
||||
"title": m.title,
|
||||
"year": m.year,
|
||||
"genres": m.genres,
|
||||
"rating": m.community_rating,
|
||||
"runtime_min": m.runtime_minutes,
|
||||
"content_rating": m.content_rating,
|
||||
"overview": (m.overview or "")[:200],
|
||||
}
|
||||
movie_list.append(entry)
|
||||
|
||||
import json
|
||||
movies_json = json.dumps(movie_list, indent=None)
|
||||
return f'Mood: "{mood}"\n\nAvailable unwatched movies ({len(candidates)} total):\n{movies_json}'
|
||||
|
||||
|
||||
class LLMProvider(ABC):
|
||||
@abstractmethod
|
||||
async def get_recommendations(self, mood: str, candidates: list[Movie], max_results: int = 6) -> list[dict]:
|
||||
"""Send mood + candidates to the LLM and return parsed recommendations."""
|
||||
...
|
||||
@@ -0,0 +1,50 @@
|
||||
import json
|
||||
import logging
|
||||
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
from app.config import settings
|
||||
from app.models import Movie
|
||||
from app.services.llm.base import SYSTEM_PROMPT, LLMProvider, build_user_message
|
||||
|
||||
logger = logging.getLogger("movie-night.llm.ollama")
|
||||
|
||||
|
||||
class OllamaProvider(LLMProvider):
|
||||
def __init__(self):
|
||||
base_url = settings.llm_base_url or "http://localhost:11434/v1"
|
||||
self.client = AsyncOpenAI(api_key="ollama", base_url=base_url)
|
||||
self.model = settings.llm_model or "llama3"
|
||||
|
||||
async def get_recommendations(self, mood: str, candidates: list[Movie], max_results: int = 6) -> list[dict]:
|
||||
system = SYSTEM_PROMPT.format(max_results=max_results)
|
||||
user_msg = build_user_message(mood, candidates)
|
||||
|
||||
logger.info(f"Calling Ollama ({self.model}) with {len(candidates)} candidates")
|
||||
|
||||
response = await self.client.chat.completions.create(
|
||||
model=self.model,
|
||||
max_tokens=2048,
|
||||
messages=[
|
||||
{"role": "system", "content": system},
|
||||
{"role": "user", "content": user_msg},
|
||||
],
|
||||
)
|
||||
|
||||
text = response.choices[0].message.content.strip()
|
||||
|
||||
try:
|
||||
data = json.loads(text)
|
||||
return data.get("recommendations", [])
|
||||
except json.JSONDecodeError:
|
||||
# Try to extract JSON from response
|
||||
start = text.find("{")
|
||||
end = text.rfind("}") + 1
|
||||
if start >= 0 and end > start:
|
||||
try:
|
||||
data = json.loads(text[start:end])
|
||||
return data.get("recommendations", [])
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
logger.error(f"Failed to parse Ollama response: {text[:200]}")
|
||||
return []
|
||||
@@ -0,0 +1,44 @@
|
||||
import json
|
||||
import logging
|
||||
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
from app.config import settings
|
||||
from app.models import Movie
|
||||
from app.services.llm.base import SYSTEM_PROMPT, LLMProvider, build_user_message
|
||||
|
||||
logger = logging.getLogger("movie-night.llm.openai")
|
||||
|
||||
|
||||
class OpenAIProvider(LLMProvider):
|
||||
def __init__(self):
|
||||
kwargs = {"api_key": settings.llm_api_key}
|
||||
if settings.llm_base_url:
|
||||
kwargs["base_url"] = settings.llm_base_url
|
||||
self.client = AsyncOpenAI(**kwargs)
|
||||
self.model = settings.llm_model or "gpt-4o"
|
||||
|
||||
async def get_recommendations(self, mood: str, candidates: list[Movie], max_results: int = 6) -> list[dict]:
|
||||
system = SYSTEM_PROMPT.format(max_results=max_results)
|
||||
user_msg = build_user_message(mood, candidates)
|
||||
|
||||
logger.info(f"Calling OpenAI ({self.model}) with {len(candidates)} candidates")
|
||||
|
||||
response = await self.client.chat.completions.create(
|
||||
model=self.model,
|
||||
max_tokens=2048,
|
||||
messages=[
|
||||
{"role": "system", "content": system},
|
||||
{"role": "user", "content": user_msg},
|
||||
],
|
||||
response_format={"type": "json_object"},
|
||||
)
|
||||
|
||||
text = response.choices[0].message.content.strip()
|
||||
|
||||
try:
|
||||
data = json.loads(text)
|
||||
return data.get("recommendations", [])
|
||||
except json.JSONDecodeError:
|
||||
logger.error(f"Failed to parse LLM response: {text[:200]}")
|
||||
return []
|
||||
@@ -0,0 +1,168 @@
|
||||
import json
|
||||
import re
|
||||
|
||||
from app.models import Movie
|
||||
|
||||
# Mood signal → genre boosts and filters
|
||||
MOOD_SIGNALS = {
|
||||
"kids": {"boost": ["Family", "Animation", "Comedy", "Adventure"], "penalize": ["Horror", "Thriller"], "max_rating": "PG-13"},
|
||||
"children": {"boost": ["Family", "Animation", "Comedy", "Adventure"], "penalize": ["Horror", "Thriller"], "max_rating": "PG"},
|
||||
"family": {"boost": ["Family", "Animation", "Comedy", "Adventure"], "penalize": ["Horror", "Thriller"], "max_rating": "PG-13"},
|
||||
"scary": {"boost": ["Horror", "Thriller", "Mystery"], "penalize": [], "max_rating": None},
|
||||
"horror": {"boost": ["Horror", "Thriller"], "penalize": [], "max_rating": None},
|
||||
"spooky": {"boost": ["Horror", "Thriller", "Mystery", "Fantasy"], "penalize": [], "max_rating": None},
|
||||
"creepy": {"boost": ["Horror", "Thriller", "Mystery"], "penalize": [], "max_rating": None},
|
||||
"funny": {"boost": ["Comedy"], "penalize": ["Horror", "War"], "max_rating": None},
|
||||
"comedy": {"boost": ["Comedy"], "penalize": [], "max_rating": None},
|
||||
"laugh": {"boost": ["Comedy"], "penalize": [], "max_rating": None},
|
||||
"light": {"boost": ["Comedy", "Romance", "Animation", "Family"], "penalize": ["Horror", "Thriller", "War"], "max_rating": None},
|
||||
"fun": {"boost": ["Comedy", "Adventure", "Animation", "Action"], "penalize": ["Horror", "War"], "max_rating": None},
|
||||
"feel-good": {"boost": ["Comedy", "Romance", "Family", "Animation"], "penalize": ["Horror", "Thriller", "War"], "max_rating": None},
|
||||
"relax": {"boost": ["Comedy", "Romance", "Drama"], "penalize": ["Horror", "Thriller", "Action"], "max_rating": None},
|
||||
"action": {"boost": ["Action", "Adventure", "Sci-Fi", "Thriller"], "penalize": [], "max_rating": None},
|
||||
"exciting": {"boost": ["Action", "Adventure", "Thriller"], "penalize": [], "max_rating": None},
|
||||
"adventure": {"boost": ["Adventure", "Action", "Fantasy", "Sci-Fi"], "penalize": [], "max_rating": None},
|
||||
"intense": {"boost": ["Action", "Thriller", "Drama", "War"], "penalize": [], "max_rating": None},
|
||||
"romantic": {"boost": ["Romance", "Comedy", "Drama"], "penalize": ["Horror", "War"], "max_rating": None},
|
||||
"romance": {"boost": ["Romance", "Comedy", "Drama"], "penalize": [], "max_rating": None},
|
||||
"date night": {"boost": ["Romance", "Comedy", "Drama", "Thriller"], "penalize": [], "max_rating": None},
|
||||
"date": {"boost": ["Romance", "Comedy", "Drama"], "penalize": [], "max_rating": None},
|
||||
"sad": {"boost": ["Drama", "Romance"], "penalize": ["Comedy", "Animation"], "max_rating": None},
|
||||
"cry": {"boost": ["Drama", "Romance", "War"], "penalize": [], "max_rating": None},
|
||||
"drama": {"boost": ["Drama"], "penalize": [], "max_rating": None},
|
||||
"sci-fi": {"boost": ["Science Fiction", "Sci-Fi", "Fantasy"], "penalize": [], "max_rating": None},
|
||||
"space": {"boost": ["Science Fiction", "Sci-Fi"], "penalize": [], "max_rating": None},
|
||||
"fantasy": {"boost": ["Fantasy", "Adventure"], "penalize": [], "max_rating": None},
|
||||
"mystery": {"boost": ["Mystery", "Thriller", "Crime"], "penalize": [], "max_rating": None},
|
||||
"crime": {"boost": ["Crime", "Thriller", "Mystery"], "penalize": [], "max_rating": None},
|
||||
"documentary": {"boost": ["Documentary"], "penalize": [], "max_rating": None},
|
||||
"war": {"boost": ["War", "History", "Drama"], "penalize": [], "max_rating": None},
|
||||
"classic": {"boost": [], "penalize": [], "max_rating": None},
|
||||
"animated": {"boost": ["Animation"], "penalize": [], "max_rating": None},
|
||||
"anime": {"boost": ["Animation"], "penalize": [], "max_rating": None},
|
||||
"music": {"boost": ["Music", "Musical"], "penalize": [], "max_rating": None},
|
||||
"musical": {"boost": ["Music", "Musical"], "penalize": [], "max_rating": None},
|
||||
"western": {"boost": ["Western"], "penalize": [], "max_rating": None},
|
||||
"superhero": {"boost": ["Action", "Adventure", "Science Fiction"], "penalize": [], "max_rating": None},
|
||||
}
|
||||
|
||||
# Content rating hierarchy for family filtering
|
||||
RATING_ORDER = ["G", "PG", "PG-13", "R", "NC-17", "NR", "Not Rated", None]
|
||||
|
||||
|
||||
def _parse_decade(mood: str) -> tuple[int, int] | None:
|
||||
"""Extract decade filter from mood text."""
|
||||
match = re.search(r"\b(19|20)(\d)0s\b", mood.lower())
|
||||
if match:
|
||||
decade_start = int(match.group(1) + match.group(2) + "0")
|
||||
return (decade_start, decade_start + 9)
|
||||
|
||||
match = re.search(r"\b(old|classic|vintage|retro)\b", mood.lower())
|
||||
if match:
|
||||
return (1920, 1989)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _is_rating_appropriate(content_rating: str | None, max_rating: str | None) -> bool:
|
||||
"""Check if a movie's content rating is at or below the max allowed."""
|
||||
if max_rating is None:
|
||||
return True
|
||||
if content_rating is None:
|
||||
return True # Unknown rating, let it through
|
||||
|
||||
try:
|
||||
movie_idx = RATING_ORDER.index(content_rating)
|
||||
max_idx = RATING_ORDER.index(max_rating)
|
||||
return movie_idx <= max_idx
|
||||
except ValueError:
|
||||
return True # Unknown rating format, let it through
|
||||
|
||||
|
||||
def _parse_movie(raw: dict) -> Movie:
|
||||
"""Convert a raw DB row dict into a Movie model."""
|
||||
return Movie(
|
||||
jellyfin_id=raw["jellyfin_id"],
|
||||
title=raw["title"],
|
||||
sort_title=raw.get("sort_title"),
|
||||
year=raw.get("year"),
|
||||
genres=json.loads(raw.get("genres") or "[]"),
|
||||
overview=raw.get("overview"),
|
||||
community_rating=raw.get("community_rating"),
|
||||
critic_rating=raw.get("critic_rating"),
|
||||
runtime_minutes=raw.get("runtime_minutes"),
|
||||
content_rating=raw.get("content_rating"),
|
||||
studios=json.loads(raw.get("studios") or "[]"),
|
||||
people=json.loads(raw.get("people") or "[]"),
|
||||
tags=json.loads(raw.get("tags") or "[]"),
|
||||
)
|
||||
|
||||
|
||||
def prefilter_candidates(movies_raw: list[dict], mood: str, max_candidates: int = 200) -> list[Movie]:
|
||||
"""Score and filter movies based on mood signals. Returns top candidates as Movie models."""
|
||||
mood_lower = mood.lower()
|
||||
|
||||
# Collect all active signals
|
||||
boost_genres: set[str] = set()
|
||||
penalize_genres: set[str] = set()
|
||||
max_rating: str | None = None
|
||||
decade = _parse_decade(mood)
|
||||
|
||||
for keyword, signals in MOOD_SIGNALS.items():
|
||||
if keyword in mood_lower:
|
||||
boost_genres.update(signals["boost"])
|
||||
penalize_genres.update(signals["penalize"])
|
||||
if signals["max_rating"] and (max_rating is None or RATING_ORDER.index(signals["max_rating"]) < RATING_ORDER.index(max_rating)):
|
||||
max_rating = signals["max_rating"]
|
||||
|
||||
# Remove any genres that appear in both boost and penalize
|
||||
penalize_genres -= boost_genres
|
||||
|
||||
scored: list[tuple[float, dict]] = []
|
||||
|
||||
for raw in movies_raw:
|
||||
movie_genres = set(json.loads(raw.get("genres") or "[]"))
|
||||
content_rating = raw.get("content_rating")
|
||||
|
||||
# Filter by content rating
|
||||
if not _is_rating_appropriate(content_rating, max_rating):
|
||||
continue
|
||||
|
||||
# Filter by decade
|
||||
year = raw.get("year")
|
||||
if decade and year:
|
||||
if year < decade[0] or year > decade[1]:
|
||||
continue
|
||||
|
||||
# Score the movie
|
||||
score = 0.0
|
||||
|
||||
# Genre match bonus
|
||||
if boost_genres:
|
||||
genre_overlap = len(movie_genres & boost_genres)
|
||||
score += genre_overlap * 3.0
|
||||
|
||||
# Genre penalty
|
||||
if penalize_genres:
|
||||
penalty_overlap = len(movie_genres & penalize_genres)
|
||||
score -= penalty_overlap * 2.0
|
||||
|
||||
# Rating bonus (higher rated movies get a small boost)
|
||||
rating = raw.get("community_rating")
|
||||
if rating:
|
||||
score += rating * 0.3
|
||||
|
||||
# Keyword match in overview
|
||||
overview = (raw.get("overview") or "").lower()
|
||||
mood_words = [w for w in mood_lower.split() if len(w) > 3]
|
||||
for word in mood_words:
|
||||
if word in overview:
|
||||
score += 1.0
|
||||
|
||||
scored.append((score, raw))
|
||||
|
||||
# Sort by score descending
|
||||
scored.sort(key=lambda x: x[0], reverse=True)
|
||||
|
||||
# Return top candidates as Movie models
|
||||
return [_parse_movie(raw) for _, raw in scored[:max_candidates]]
|
||||
Reference in New Issue
Block a user