diff --git a/pyproject.toml b/pyproject.toml index ce4a3ca..ffdfa04 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ python = ">=3.11,<3.15" Django = "^4.0.3" django-extensions = "^3.1.5" python-dateutil = "^2.8.2" -python-dotenv = "^0.20.0" +python-dotenv = ">=0.20.0,<2" python-json-logger = "^2.0.2" colorlog = "^6.6.0" httpx = "<=0.27.2" @@ -43,6 +43,7 @@ ipython = "^8.14.0" pendulum = "^3" trafilatura = "^1.6.3" django-imagekit = "^5.0.0" +django-mcp-server = "^0.5.7" thefuzz = "^0.22.1" dataclass-wizard = "^0.35.0" webdavclient3 = "^3.14.6" diff --git a/vrobbler/apps/scrobbles/mcp.py b/vrobbler/apps/scrobbles/mcp.py new file mode 100644 index 0000000..ca78a47 --- /dev/null +++ b/vrobbler/apps/scrobbles/mcp.py @@ -0,0 +1,518 @@ +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 diff --git a/vrobbler/settings.py b/vrobbler/settings.py index 28280e8..cfc9a3a 100644 --- a/vrobbler/settings.py +++ b/vrobbler/settings.py @@ -236,6 +236,7 @@ INSTALLED_APPS = [ "allauth.account", "allauth.socialaccount", "django_celery_results", + "mcp_server", ] SITE_ID = 1 @@ -329,6 +330,13 @@ REST_FRAMEWORK = { "PAGE_SIZE": 200, } +DJANGO_MCP_AUTHENTICATION_CLASSES = [ + "oauth2_provider.contrib.rest_framework.OAuth2Authentication", + "rest_framework.authentication.TokenAuthentication", + "rest_framework.authentication.SessionAuthentication", +] +DJANGO_MCP_GET_SERVER_INSTRUCTIONS_TOOL = True + LOGIN_REDIRECT_URL = "/" AUTH_PASSWORD_VALIDATORS = [ diff --git a/vrobbler/urls.py b/vrobbler/urls.py index baf15c0..0c90d6f 100644 --- a/vrobbler/urls.py +++ b/vrobbler/urls.py @@ -188,6 +188,7 @@ urlpatterns = [ path("", include(people_urls, namespace="people")), path("", include(charts_urls, namespace="charts")), path("", include(trends_urls, namespace="trends")), + path("", include("mcp_server.urls")), path("", scrobbles_views.RecentScrobbleList.as_view(), name="vrobbler-home"), ] if settings.DEBUG: