519 lines
20 KiB
Python
519 lines
20 KiB
Python
from mcp_server import MCPToolset
|
|
from scrobbles.models import Scrobble
|
|
from scrobbles.constants import LONG_PLAY_MEDIA
|
|
|
|
|
|
class ScrobbleToolset(MCPToolset):
|
|
def list_recent_scrobbles(
|
|
self,
|
|
days: int = 7,
|
|
media_type: str | None = None,
|
|
limit: int = 50,
|
|
) -> list[dict]:
|
|
"""List scrobbles from the last N days, optionally filtered by media type.
|
|
Valid media_type values: Video, Track, PodcastEpisode, SportEvent, Book,
|
|
Paper, VideoGame, BoardGame, GeoLocation, Trail, Beer, Puzzle, Food, Task,
|
|
WebPage, LifeEvent, Mood, BrickSet, Channel, BirdingLocation, DiscGolfCourse
|
|
"""
|
|
qs = (
|
|
Scrobble.objects.filter(user=self.request.user)
|
|
.select_related(
|
|
"video", "track", "book", "video_game", "board_game",
|
|
"beer", "puzzle", "food", "trail", "task", "web_page",
|
|
"life_event", "mood", "brick_set", "podcast_episode",
|
|
"sport_event", "geo_location", "birding_location",
|
|
"disc_golf_course", "channel",
|
|
)
|
|
.order_by("-timestamp")
|
|
)
|
|
from django.utils import timezone
|
|
import datetime
|
|
qs = qs.filter(timestamp__gte=timezone.now() - datetime.timedelta(days=days))
|
|
if media_type:
|
|
qs = qs.filter(media_type=media_type)
|
|
qs = qs[:limit]
|
|
return [_scrobble_to_dict(s) for s in qs]
|
|
|
|
def get_scrobble(self, uuid: str) -> dict | None:
|
|
"""Get a single scrobble by its UUID."""
|
|
try:
|
|
s = Scrobble.objects.filter(user=self.request.user).get(uuid=uuid)
|
|
except Scrobble.DoesNotExist:
|
|
return None
|
|
return _scrobble_to_dict(s)
|
|
|
|
def search_scrobbles(
|
|
self, query: str, media_type: str | None = None, limit: int = 20
|
|
) -> list[dict]:
|
|
"""Search scrobbles by text in their log data or related media titles."""
|
|
from django.db.models import Q
|
|
qs = Scrobble.objects.filter(user=self.request.user).order_by("-timestamp")
|
|
if media_type:
|
|
qs = qs.filter(media_type=media_type)
|
|
qs = qs.filter(
|
|
Q(log__icontains=query)
|
|
| Q(video__title__icontains=query)
|
|
| Q(track__title__icontains=query)
|
|
| Q(book__title__icontains=query)
|
|
| Q(video_game__title__icontains=query)
|
|
| Q(board_game__title__icontains=query)
|
|
| Q(beer__title__icontains=query)
|
|
| Q(food__title__icontains=query)
|
|
| Q(trail__title__icontains=query)
|
|
| Q(task__title__icontains=query)
|
|
| Q(web_page__title__icontains=query)
|
|
| Q(life_event__title__icontains=query)
|
|
| Q(puzzle__title__icontains=query)
|
|
| Q(brick_set__title__icontains=query)
|
|
| Q(podcast_episode__title__icontains=query)
|
|
)[:limit]
|
|
return [_scrobble_to_dict(s) for s in qs]
|
|
|
|
def get_scrobbles_by_date(
|
|
self, date: str, media_type: str | None = None
|
|
) -> list[dict]:
|
|
"""Get scrobbles for a specific date (YYYY-MM-DD format)."""
|
|
import datetime
|
|
try:
|
|
dt = datetime.datetime.strptime(date, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
return []
|
|
qs = Scrobble.objects.filter(
|
|
user=self.request.user,
|
|
timestamp__date=dt,
|
|
).order_by("-timestamp")
|
|
if media_type:
|
|
qs = qs.filter(media_type=media_type)
|
|
return [_scrobble_to_dict(s) for s in qs]
|
|
|
|
def get_in_progress_scrobbles(
|
|
self, media_type: str | None = None
|
|
) -> list[dict]:
|
|
"""Get scrobbles currently in progress (started but not finished).
|
|
These are long-play items like books, video games, brick sets, or tasks."""
|
|
qs = Scrobble.objects.filter(
|
|
user=self.request.user,
|
|
in_progress=True,
|
|
).order_by("-timestamp")
|
|
if media_type:
|
|
qs = qs.filter(media_type=media_type)
|
|
return [_scrobble_to_dict(s) for s in qs]
|
|
|
|
def get_long_play_scrobbles(
|
|
self, status: str = "in_progress", media_type: str | None = None
|
|
) -> list[dict]:
|
|
"""Get long-play scrobbles (books, video games, brick sets, tasks).
|
|
Status can be 'in_progress' or 'completed'."""
|
|
types = list(LONG_PLAY_MEDIA.values())
|
|
qs = Scrobble.objects.filter(
|
|
user=self.request.user,
|
|
media_type__in=types,
|
|
).order_by("-timestamp")
|
|
if media_type:
|
|
qs = qs.filter(media_type=media_type)
|
|
if status == "in_progress":
|
|
qs = qs.filter(in_progress=True)
|
|
elif status == "completed":
|
|
qs = qs.filter(in_progress=False)
|
|
return [_scrobble_to_dict(s) for s in qs]
|
|
|
|
|
|
class MediaToolset(MCPToolset):
|
|
def get_book(self, uuid: str) -> dict | None:
|
|
"""Get a book by UUID."""
|
|
from books.models import Book
|
|
try:
|
|
b = Book.objects.get(uuid=uuid)
|
|
except Book.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(b, fields=["title", "pages", "language",
|
|
"first_publish_year", "isbn_13",
|
|
"publisher", "summary"])
|
|
|
|
def list_books(self, author: str | None = None, limit: int = 20) -> list[dict]:
|
|
"""List books, optionally filtered by author name."""
|
|
from books.models import Book
|
|
qs = Book.objects.all().order_by("title")
|
|
if author:
|
|
qs = qs.filter(authors__name__icontains=author)
|
|
return [_media_to_dict(b, fields=["title", "pages", "language",
|
|
"first_publish_year", "isbn_13",
|
|
"publisher"]) for b in qs[:limit]]
|
|
|
|
def get_track(self, uuid: str) -> dict | None:
|
|
"""Get a music track by UUID."""
|
|
from music.models import Track
|
|
try:
|
|
t = Track.objects.select_related("artist_fk").get(uuid=uuid)
|
|
except Track.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(t, fields=["title", "base_run_time_seconds",
|
|
"artist_fk__name", "genre"])
|
|
|
|
def list_tracks(self, artist: str | None = None, limit: int = 20) -> list[dict]:
|
|
"""List music tracks, optionally filtered by artist name."""
|
|
from music.models import Track
|
|
qs = Track.objects.select_related("artist_fk").all().order_by("title")
|
|
if artist:
|
|
qs = qs.filter(artist_fk__name__icontains=artist)
|
|
return [_media_to_dict(t, fields=["title", "base_run_time_seconds",
|
|
"artist_fk__name", "genre"])
|
|
for t in qs[:limit]]
|
|
|
|
def get_video(self, uuid: str) -> dict | None:
|
|
"""Get a video by UUID."""
|
|
from videos.models import Video
|
|
try:
|
|
v = Video.objects.select_related("tv_series", "channel").get(uuid=uuid)
|
|
except Video.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(v, fields=["title", "year", "overview",
|
|
"imdb_id", "imdb_rating",
|
|
"tv_series__name", "channel__title",
|
|
"season_number", "episode_number"])
|
|
|
|
def list_videos(self, series: str | None = None, limit: int = 20) -> list[dict]:
|
|
"""List videos, optionally filtered by series name."""
|
|
from videos.models import Video
|
|
qs = Video.objects.select_related("tv_series", "channel").all().order_by("title")
|
|
if series:
|
|
qs = qs.filter(tv_series__name__icontains=series)
|
|
return [_media_to_dict(v, fields=["title", "year", "overview",
|
|
"tv_series__name", "channel__title",
|
|
"season_number", "episode_number"])
|
|
for v in qs[:limit]]
|
|
|
|
def get_board_game(self, uuid: str) -> dict | None:
|
|
"""Get a board game by UUID."""
|
|
from boardgames.models import BoardGame
|
|
try:
|
|
bg = BoardGame.objects.get(uuid=uuid)
|
|
except BoardGame.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(bg, fields=["title", "genre"])
|
|
|
|
def list_board_games(self, limit: int = 20) -> list[dict]:
|
|
"""List board games."""
|
|
from boardgames.models import BoardGame
|
|
qs = BoardGame.objects.all().order_by("title")[:limit]
|
|
return [_media_to_dict(bg, fields=["title", "genre"]) for bg in qs]
|
|
|
|
def get_podcast_episode(self, uuid: str) -> dict | None:
|
|
"""Get a podcast episode by UUID."""
|
|
from podcasts.models import PodcastEpisode
|
|
try:
|
|
pe = PodcastEpisode.objects.select_related("podcast", "producer").get(
|
|
uuid=uuid
|
|
)
|
|
except PodcastEpisode.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(pe, fields=["title", "podcast__title",
|
|
"producer__name", "base_run_time_seconds"])
|
|
|
|
def get_beer(self, uuid: str) -> dict | None:
|
|
"""Get a beer by UUID."""
|
|
from beers.models import Beer
|
|
try:
|
|
b = Beer.objects.select_related("style", "producer").get(uuid=uuid)
|
|
except Beer.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(b, fields=["title", "style__name",
|
|
"producer__name", "abv"])
|
|
|
|
def get_brick_set(self, uuid: str) -> dict | None:
|
|
"""Get a brick set (LEGO) by UUID."""
|
|
from bricksets.models import BrickSet
|
|
try:
|
|
bs = BrickSet.objects.get(uuid=uuid)
|
|
except BrickSet.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(bs, fields=["title", "piece_count", "set_number"])
|
|
|
|
def get_video_game(self, uuid: str) -> dict | None:
|
|
"""Get a video game by UUID."""
|
|
from videogames.models import VideoGame
|
|
try:
|
|
vg = VideoGame.objects.get(uuid=uuid)
|
|
except VideoGame.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(vg, fields=["title", "genre",
|
|
"base_run_time_seconds"])
|
|
|
|
def get_puzzle(self, uuid: str) -> dict | None:
|
|
"""Get a puzzle by UUID."""
|
|
from puzzles.models import Puzzle
|
|
try:
|
|
p = Puzzle.objects.select_related("manufacturer").get(uuid=uuid)
|
|
except Puzzle.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(p, fields=["title", "piece_count",
|
|
"manufacturer__name"])
|
|
|
|
def get_web_page(self, uuid: str) -> dict | None:
|
|
"""Get a web page by UUID."""
|
|
from webpages.models import WebPage
|
|
try:
|
|
wp = WebPage.objects.select_related("domain").get(uuid=uuid)
|
|
except WebPage.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(wp, fields=["title", "url", "domain__name"])
|
|
|
|
def get_task(self, uuid: str) -> dict | None:
|
|
"""Get a task by UUID."""
|
|
from tasks.models import Task
|
|
try:
|
|
t = Task.objects.get(uuid=uuid)
|
|
except Task.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(t, fields=["title", "completed"])
|
|
|
|
def get_trail(self, uuid: str) -> dict | None:
|
|
"""Get a trail by UUID."""
|
|
from trails.models import Trail
|
|
try:
|
|
t = Trail.objects.get(uuid=uuid)
|
|
except Trail.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(t, fields=["title", "genre", "base_run_time_seconds"])
|
|
|
|
def get_geo_location(self, uuid: str) -> dict | None:
|
|
"""Get a geo location by UUID."""
|
|
from locations.models import GeoLocation
|
|
try:
|
|
gl = GeoLocation.objects.get(uuid=uuid)
|
|
except GeoLocation.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(gl, fields=["title", "latitude", "longitude"])
|
|
|
|
def get_life_event(self, uuid: str) -> dict | None:
|
|
"""Get a life event by UUID."""
|
|
from lifeevents.models import LifeEvent
|
|
try:
|
|
le = LifeEvent.objects.get(uuid=uuid)
|
|
except LifeEvent.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(le, fields=["title", "event_date", "genre"])
|
|
|
|
def get_mood(self, uuid: str) -> dict | None:
|
|
"""Get a mood entry by UUID."""
|
|
from moods.models import Mood
|
|
try:
|
|
m = Mood.objects.get(uuid=uuid)
|
|
except Mood.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(m, fields=["title", "mood_type", "mood_quality"])
|
|
|
|
def get_food(self, uuid: str) -> dict | None:
|
|
"""Get a food entry by UUID."""
|
|
from foods.models import Food
|
|
try:
|
|
f = Food.objects.select_related("category").get(uuid=uuid)
|
|
except Food.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(f, fields=["title", "category__name"])
|
|
|
|
def get_bird_sighting(self, uuid: str) -> dict | None:
|
|
"""Get a bird sighting by UUID."""
|
|
from birds.models import BirdSighting
|
|
try:
|
|
bs = BirdSighting.objects.select_related("bird").get(uuid=uuid)
|
|
except BirdSighting.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(bs, fields=["title", "bird__common_name",
|
|
"bird__scientific_name", "location"])
|
|
|
|
def get_disc_golf_course(self, uuid: str) -> dict | None:
|
|
"""Get a disc golf course by UUID."""
|
|
from discgolf.models import DiscGolfCourse
|
|
try:
|
|
dg = DiscGolfCourse.objects.get(uuid=uuid)
|
|
except DiscGolfCourse.DoesNotExist:
|
|
return None
|
|
return _media_to_dict(dg, fields=["title", "holes", "location"])
|
|
|
|
|
|
class StatsToolset(MCPToolset):
|
|
def get_scrobble_counts(self, days: int = 30) -> list[dict]:
|
|
"""Get scrobble counts grouped by media type for the last N days."""
|
|
from django.utils import timezone
|
|
import datetime
|
|
from django.db.models import Count
|
|
|
|
cutoff = timezone.now() - datetime.timedelta(days=days)
|
|
qs = (
|
|
Scrobble.objects.filter(user=self.request.user, timestamp__gte=cutoff)
|
|
.values("media_type")
|
|
.annotate(count=Count("id"))
|
|
.order_by("-count")
|
|
)
|
|
return list(qs)
|
|
|
|
def get_top_media(
|
|
self, media_type: str, days: int = 30, limit: int = 10
|
|
) -> list[dict]:
|
|
"""Get the most-scrobbled items of a given media type in the last N days.
|
|
Valid media_type values: Video, Track, Book, BoardGame, Beer, etc."""
|
|
from django.utils import timezone
|
|
import datetime
|
|
from django.db.models import Count
|
|
|
|
cutoff = timezone.now() - datetime.timedelta(days=days)
|
|
|
|
rel_field = _media_type_to_rel_field(media_type)
|
|
if not rel_field:
|
|
return []
|
|
|
|
qs = (
|
|
Scrobble.objects.filter(
|
|
user=self.request.user,
|
|
media_type=media_type,
|
|
timestamp__gte=cutoff,
|
|
)
|
|
.values(rel_field)
|
|
.annotate(count=Count("id"))
|
|
.order_by("-count")
|
|
)[:limit]
|
|
|
|
results = []
|
|
for row in qs:
|
|
obj_id = row[rel_field]
|
|
if obj_id is None:
|
|
continue
|
|
results.append({"id": obj_id, "count": row["count"]})
|
|
return results
|
|
|
|
|
|
def _media_type_to_rel_field(media_type: str) -> str | None:
|
|
mapping = {
|
|
"Video": "video",
|
|
"Track": "track",
|
|
"PodcastEpisode": "podcast_episode",
|
|
"SportEvent": "sport_event",
|
|
"Book": "book",
|
|
"Paper": "paper",
|
|
"VideoGame": "video_game",
|
|
"BoardGame": "board_game",
|
|
"GeoLocation": "geo_location",
|
|
"Trail": "trail",
|
|
"Beer": "beer",
|
|
"Puzzle": "puzzle",
|
|
"Food": "food",
|
|
"Task": "task",
|
|
"WebPage": "web_page",
|
|
"LifeEvent": "life_event",
|
|
"Mood": "mood",
|
|
"BrickSet": "brick_set",
|
|
"Channel": "channel",
|
|
"BirdingLocation": "birding_location",
|
|
"DiscGolfCourse": "disc_golf_course",
|
|
}
|
|
return mapping.get(media_type)
|
|
|
|
|
|
def _scrobble_to_dict(s: Scrobble) -> dict:
|
|
result = {
|
|
"uuid": str(s.uuid),
|
|
"media_type": s.media_type,
|
|
"timestamp": s.timestamp.isoformat() if s.timestamp else None,
|
|
"stop_timestamp": s.stop_timestamp.isoformat() if s.stop_timestamp else None,
|
|
"in_progress": s.in_progress,
|
|
"played_to_completion": s.played_to_completion,
|
|
"source": s.source,
|
|
"visibility": s.visibility,
|
|
"timezone": s.timezone,
|
|
}
|
|
if s.log:
|
|
result["log"] = s.log
|
|
|
|
rel = _scrobble_related_to_dict(s)
|
|
if rel:
|
|
result["media"] = rel
|
|
|
|
return result
|
|
|
|
|
|
def _scrobble_related_to_dict(s: Scrobble) -> dict | None:
|
|
if s.video:
|
|
return _media_to_dict(s.video, fields=["title", "year", "imdb_id",
|
|
"imdb_rating"])
|
|
if s.track:
|
|
return _media_to_dict(s.track, fields=["title",
|
|
"base_run_time_seconds"])
|
|
if s.book:
|
|
return _media_to_dict(s.book, fields=["title", "pages"])
|
|
if s.video_game:
|
|
return _media_to_dict(s.video_game, fields=["title",
|
|
"base_run_time_seconds"])
|
|
if s.board_game:
|
|
return _media_to_dict(s.board_game, fields=["title"])
|
|
if s.beer:
|
|
return _media_to_dict(s.beer, fields=["title"])
|
|
if s.puzzle:
|
|
return _media_to_dict(s.puzzle, fields=["title", "piece_count"])
|
|
if s.food:
|
|
return _media_to_dict(s.food, fields=["title"])
|
|
if s.trail:
|
|
return _media_to_dict(s.trail, fields=["title",
|
|
"base_run_time_seconds"])
|
|
if s.task:
|
|
return _media_to_dict(s.task, fields=["title"])
|
|
if s.web_page:
|
|
return _media_to_dict(s.web_page, fields=["title", "url"])
|
|
if s.life_event:
|
|
return _media_to_dict(s.life_event, fields=["title", "event_date"])
|
|
if s.mood:
|
|
return _media_to_dict(s.mood, fields=["title"])
|
|
if s.brick_set:
|
|
return _media_to_dict(s.brick_set, fields=["title", "set_number"])
|
|
if s.podcast_episode:
|
|
return _media_to_dict(s.podcast_episode, fields=["title"])
|
|
if s.sport_event:
|
|
return {"title": str(s.sport_event)}
|
|
if s.geo_location:
|
|
return _media_to_dict(s.geo_location, fields=["title", "latitude",
|
|
"longitude"])
|
|
if s.birding_location:
|
|
return _media_to_dict(s.birding_location, fields=["title"])
|
|
if s.disc_golf_course:
|
|
return _media_to_dict(s.disc_golf_course, fields=["title", "holes"])
|
|
if s.channel:
|
|
return _media_to_dict(s.channel, fields=["title"])
|
|
return None
|
|
|
|
|
|
def _media_to_dict(obj, fields: list[str] | None = None) -> dict:
|
|
if obj is None:
|
|
return {}
|
|
result = {}
|
|
if hasattr(obj, "uuid"):
|
|
result["uuid"] = str(obj.uuid)
|
|
if hasattr(obj, "title"):
|
|
result["title"] = obj.title
|
|
|
|
if fields is None:
|
|
return result
|
|
|
|
resolved = _resolve_fields(obj, fields)
|
|
for k, v in resolved.items():
|
|
if k not in result:
|
|
result[k] = v
|
|
|
|
return result
|
|
|
|
|
|
def _resolve_fields(obj, fields: list[str]) -> dict:
|
|
result = {}
|
|
for field in fields:
|
|
parts = field.split("__")
|
|
val = obj
|
|
try:
|
|
for part in parts:
|
|
val = getattr(val, part)
|
|
except AttributeError:
|
|
continue
|
|
if val is not None:
|
|
if hasattr(val, "all"):
|
|
val = [str(v) for v in val.all()]
|
|
result[field] = val
|
|
return result
|