From b56a9efac29bf18a4b9db7277fc628629706e288 Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Thu, 21 May 2026 18:12:17 +0200 Subject: [PATCH] feat: add Quizzical Beats MCP automation interface --- docs/developer-guide/mcp.md | 63 ++++ mkdocs.yml | 3 +- musicround/mcp_server.py | 225 ++++++++++++ musicround/services/__init__.py | 1 + musicround/services/automation.py | 579 ++++++++++++++++++++++++++++++ requirements.txt | 1 + tests/test_automation_service.py | 129 +++++++ tests/test_import_queue.py | 48 ++- 8 files changed, 1047 insertions(+), 2 deletions(-) create mode 100644 docs/developer-guide/mcp.md create mode 100644 musicround/mcp_server.py create mode 100644 musicround/services/__init__.py create mode 100644 musicround/services/automation.py create mode 100644 tests/test_automation_service.py diff --git a/docs/developer-guide/mcp.md b/docs/developer-guide/mcp.md new file mode 100644 index 0000000..5400473 --- /dev/null +++ b/docs/developer-guide/mcp.md @@ -0,0 +1,63 @@ +# MCP Interface + +Quizzical Beats includes an MCP server for agentic round production workflows. +It exposes the same catalog, round, export, email, and custom-audio capabilities +used by the Flask application. + +## Run Locally + +Install dependencies and start the MCP server from the repository root: + +```bash +pip install -r requirements.txt +python -m musicround.mcp_server +``` + +The server uses the normal Quizzical Beats Flask configuration. Set the same +environment variables you use for the web app, including `SECRET_KEY`, +`AUTOMATION_TOKEN`, database configuration, mail settings, and any Spotify, +Deezer, OpenAI, AWS Polly, or ElevenLabs credentials needed by the tools you +plan to call. + +## Tools + +The MCP server exposes these tools: + +| Tool | Purpose | +| --- | --- | +| `find_songs` | Search the existing Quizzical Beats catalog before adding duplicates. | +| `add_song` | Add or update a catalog song, including platform IDs and tags. | +| `import_catalog_item` | Import a Spotify or Deezer track, album, or playlist. | +| `compile_round` | Create a named round from explicit song IDs or selection criteria. | +| `rename_round` | Set or clear a round name. | +| `create_round_from_playlist` | Import a playlist and turn the imported songs into a round. | +| `generate_round_assets` | Generate the round PDF and/or MP3. | +| `inspect_round_mp3` | Check round MP3 duration, loudness, silence, and clipping indicators. | +| `inspect_round_pdf` | Check round PDF existence and basic structural validity. | +| `send_round_email` | Generate assets and email the finished round bundle. | +| `generate_tts_snippet` | Generate and assign custom intro, replay, or outro TTS MP3s. | + +## Intended Workflow + +1. Search with `find_songs` to avoid duplicates. +2. Add missing tracks with `add_song` or import platform content with + `import_catalog_item`. +3. Create the round with `compile_round` or `create_round_from_playlist`. +4. Generate PDF and MP3 files with `generate_round_assets`. +5. Inspect the generated files with `inspect_round_pdf` and `inspect_round_mp3`. +6. Send the completed bundle with `send_round_email`. + +For Spotify imports, pass a `user_id` for a user with connected Spotify tokens. +For email, either pass an explicit recipient or use a selected user that has an +email address. + +## Custom Audio + +Use `generate_tts_snippet` to update the reusable audio segments: + +- `intro`: lead-in before the first song. +- `replay`: announcement before the repeat section. +- `outro`: lead-out after the round. + +Supported TTS services follow the existing application helper: `openai`, `polly`, +and `elevenlabs`. diff --git a/mkdocs.yml b/mkdocs.yml index 0a25657..2d6fed2 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -59,9 +59,10 @@ nav: - Developer Guide: - Architecture: developer-guide/architecture.md - API Reference: developer-guide/api-reference.md + - MCP Interface: developer-guide/mcp.md - Database Schema: developer-guide/database-schema.md - OAuth Integration: developer-guide/oauth-integration.md - Contributing: developer-guide/contributing.md - FAQ: faq.md - Changelog: changelog.md - - Brand Identity: brand-identity.md \ No newline at end of file + - Brand Identity: brand-identity.md diff --git a/musicround/mcp_server.py b/musicround/mcp_server.py new file mode 100644 index 0000000..6c89034 --- /dev/null +++ b/musicround/mcp_server.py @@ -0,0 +1,225 @@ +"""MCP server for agentic Quizzical Beats workflows. + +Run with: + python -m musicround.mcp_server +""" + +from __future__ import annotations + +from functools import lru_cache +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from musicround import create_app +from musicround.services import automation + + +mcp = FastMCP("Quizzical Beats") + + +@lru_cache(maxsize=1) +def _app(): + """Create the Flask app once for the MCP server process.""" + return create_app() + + +def _with_app_context(func, *args, **kwargs) -> dict[str, Any]: + """Run a service function inside the Quizzical Beats app context.""" + app = _app() + with app.app_context(): + return func(*args, **kwargs) + + +@mcp.tool() +def find_songs( + query: str | None = None, + title: str | None = None, + artist: str | None = None, + spotify_id: str | None = None, + deezer_id: str | None = None, + isrc: str | None = None, + limit: int = 20, +) -> dict[str, Any]: + """Search the local Quizzical Beats catalog before adding a song.""" + return _with_app_context( + automation.find_songs, + query=query, + title=title, + artist=artist, + spotify_id=spotify_id, + deezer_id=deezer_id, + isrc=isrc, + limit=limit, + ) + + +@mcp.tool() +def add_song( + title: str, + artist: str, + album_name: str | None = None, + genre: str | None = None, + year: int | None = None, + preview_url: str | None = None, + cover_url: str | None = None, + spotify_id: str | None = None, + deezer_id: str | None = None, + isrc: str | None = None, + tags: list[str] | None = None, + source: str = "manual", +) -> dict[str, Any]: + """Add a song to Quizzical Beats if it is not already present.""" + return _with_app_context( + automation.add_song, + title=title, + artist=artist, + album_name=album_name, + genre=genre, + year=year, + preview_url=preview_url, + cover_url=cover_url, + spotify_id=spotify_id, + deezer_id=deezer_id, + isrc=isrc, + tags=tags, + source=source, + ) + + +@mcp.tool() +def import_catalog_item( + service_name: str, + item_type: str, + item_id_or_url: str, + user_id: int | None = None, +) -> dict[str, Any]: + """Import a Spotify or Deezer track, album, or playlist into the catalog.""" + return _with_app_context( + automation.import_catalog_item, + service_name=service_name, + item_type=item_type, + item_id_or_url=item_id_or_url, + user_id=user_id, + ) + + +@mcp.tool() +def compile_round( + name: str | None = None, + round_type: str = "random", + count: int = 8, + criteria: str | None = None, + song_ids: list[int] | None = None, +) -> dict[str, Any]: + """Compile songs into a named quiz round.""" + return _with_app_context( + automation.create_round, + name=name, + round_type=round_type, + count=count, + criteria=criteria, + song_ids=song_ids, + ) + + +@mcp.tool() +def rename_round(round_id: int, name: str | None) -> dict[str, Any]: + """Set or clear the display name for a round.""" + return _with_app_context(automation.rename_round, round_id=round_id, name=name) + + +@mcp.tool() +def create_round_from_playlist( + service_name: str, + playlist_id_or_url: str, + name: str | None = None, + count: int = 8, + user_id: int | None = None, +) -> dict[str, Any]: + """Import a Spotify or Deezer playlist and turn it into a quiz round.""" + return _with_app_context( + automation.create_round_from_playlist, + service_name=service_name, + playlist_id_or_url=playlist_id_or_url, + name=name, + count=count, + user_id=user_id, + ) + + +@mcp.tool() +def generate_round_assets( + round_id: int, + user_id: int | None = None, + include_pdf: bool = True, + include_mp3: bool = True, +) -> dict[str, Any]: + """Generate the PDF and/or MP3 files for a round.""" + return _with_app_context( + automation.generate_round_assets, + round_id=round_id, + user_id=user_id, + include_pdf=include_pdf, + include_mp3=include_mp3, + ) + + +@mcp.tool() +def inspect_round_mp3(path: str | None = None, round_id: int | None = None) -> dict[str, Any]: + """Check a round MP3 for duration, loudness, clipping, and silence issues.""" + return _with_app_context(automation.inspect_mp3_quality, path=path, round_id=round_id) + + +@mcp.tool() +def inspect_round_pdf(path: str | None = None, round_id: int | None = None) -> dict[str, Any]: + """Check that a round PDF exists and has a valid basic PDF structure.""" + return _with_app_context(automation.inspect_pdf_quality, path=path, round_id=round_id) + + +@mcp.tool() +def send_round_email( + round_id: int, + recipient: str | None = None, + user_id: int | None = None, + subject: str | None = None, + body_text: str | None = None, +) -> dict[str, Any]: + """Generate assets and email the completed round bundle.""" + return _with_app_context( + automation.email_round, + round_id=round_id, + recipient=recipient, + user_id=user_id, + subject=subject, + body_text=body_text, + ) + + +@mcp.tool() +def generate_tts_snippet( + user_id: int, + mp3_type: str, + text: str, + service: str = "openai", + voice: str | None = None, + model: str | None = None, + stability: float | None = None, + similarity: float | None = None, +) -> dict[str, Any]: + """Generate and assign a custom intro, replay, or outro TTS MP3.""" + return _with_app_context( + automation.generate_tts_snippet, + user_id=user_id, + mp3_type=mp3_type, + text=text, + service=service, + voice=voice, + model=model, + stability=stability, + similarity=similarity, + ) + + +if __name__ == "__main__": + mcp.run() diff --git a/musicround/services/__init__.py b/musicround/services/__init__.py new file mode 100644 index 0000000..9448332 --- /dev/null +++ b/musicround/services/__init__.py @@ -0,0 +1 @@ +"""Service-layer helpers for Quizzical Beats.""" diff --git a/musicround/services/automation.py b/musicround/services/automation.py new file mode 100644 index 0000000..d00b936 --- /dev/null +++ b/musicround/services/automation.py @@ -0,0 +1,579 @@ +"""Automation services used by the MCP server and agent workflows.""" + +from __future__ import annotations + +import os +import re +from datetime import datetime +from typing import Any, Iterable + +from flask import current_app +from flask_login import login_user, logout_user +from pydub import AudioSegment +from sqlalchemy import or_ + +from musicround import db +from musicround.helpers.email_helper import send_email +from musicround.helpers.import_helper import ImportHelper +from musicround.helpers.utils import generate_tts_mp3 +from musicround.models import Round, RoundExport, Song, Tag, User + + +class AutomationError(ValueError): + """Raised when an automation request cannot be completed.""" + + +def _song_summary(song: Song) -> dict[str, Any]: + data = song.to_dict() + return { + "id": data["id"], + "title": data["title"], + "artist": data["artist"], + "genre": data["genre"], + "year": data["year"], + "source": song.source, + "preview_url": data["preview_url"], + "spotify_id": data["spotify_id"], + "deezer_id": data["deezer_id"], + "isrc": data["isrc"], + "tags": data["tags"], + } + + +def _round_summary(round_obj: Round) -> dict[str, Any]: + ids = [int(song_id) for song_id in round_obj.songs.split(",") if song_id] + songs = Song.query.filter(Song.id.in_(ids)).all() + songs_by_id = {song.id: song for song in songs} + ordered = [songs_by_id[song_id] for song_id in ids if song_id in songs_by_id] + return { + "id": round_obj.id, + "name": round_obj.name, + "round_type": round_obj.round_type, + "criteria": round_obj.round_criteria_used, + "song_ids": ids, + "songs": [_song_summary(song) for song in ordered], + "mp3_generated": round_obj.mp3_generated, + "pdf_generated": round_obj.pdf_generated, + "last_generated_at": ( + round_obj.last_generated_at.isoformat() if round_obj.last_generated_at else None + ), + } + + +def _find_user(user_id: int | None = None) -> User: + if user_id is not None: + user = db.session.get(User, user_id) + if not user: + raise AutomationError(f"User {user_id} was not found.") + return user + + users = User.query.order_by(User.id).limit(2).all() + if len(users) == 1: + return users[0] + if not users: + raise AutomationError( + "No users exist yet. Create a user before generating user-owned assets." + ) + raise AutomationError( + "Multiple users exist. Pass user_id so the action uses the right account." + ) + + +def _parse_external_id(service_name: str, item_type: str, value: str) -> str: + service = service_name.lower() + item = item_type.lower() + stripped = value.strip() + if service == "spotify": + match = re.search(rf"spotify\.com/{item}/([A-Za-z0-9]+)", stripped) + if match: + return match.group(1) + match = re.search(rf"spotify:{item}:([A-Za-z0-9]+)", stripped) + if match: + return match.group(1) + if service == "deezer": + match = re.search(r"deezer\.page\.link/([A-Za-z0-9]+)", stripped) + if match: + return match.group(1) + match = re.search(rf"deezer\.com/(?:[a-z]{{2}}/)?{item}/(\d+)", stripped) + if match: + return match.group(1) + return stripped.split("?")[0].rstrip("/") + + +def _attach_tags(song: Song, tag_names: Iterable[str] | None) -> None: + for raw_name in tag_names or []: + tag_name = raw_name.strip() + if not tag_name: + continue + tag = Tag.query.filter(Tag.name.ilike(tag_name)).first() + if not tag: + tag = Tag(name=tag_name) + db.session.add(tag) + db.session.flush() + if tag not in song.tags: + song.tags.append(tag) + + +def add_song( + title: str, + artist: str, + album_name: str | None = None, + genre: str | None = None, + year: int | None = None, + preview_url: str | None = None, + cover_url: str | None = None, + spotify_id: str | None = None, + deezer_id: str | None = None, + isrc: str | None = None, + tags: list[str] | None = None, + source: str = "manual", +) -> dict[str, Any]: + """Add or update a song in the local catalog.""" + if not title or not artist: + raise AutomationError("Both title and artist are required.") + + existing = None + if isrc: + existing = Song.query.filter_by(isrc=isrc).first() + if not existing and spotify_id: + existing = Song.query.filter_by(spotify_id=spotify_id).first() + if not existing and deezer_id: + existing = Song.query.filter_by(deezer_id=str(deezer_id)).first() + + song = existing or Song(title=title.strip(), artist=artist.strip()) + song.title = title.strip() + song.artist = artist.strip() + song.album_name = album_name or song.album_name + song.genre = genre or song.genre + song.year = year or song.year + song.preview_url = preview_url or song.preview_url + song.cover_url = cover_url or song.cover_url + song.spotify_id = spotify_id or song.spotify_id + song.deezer_id = str(deezer_id) if deezer_id else song.deezer_id + song.isrc = isrc or song.isrc + song.source = source or song.source or "manual" + _attach_tags(song, tags) + + if not existing: + db.session.add(song) + db.session.commit() + return {"created": existing is None, "song": _song_summary(song)} + + +def find_songs( + query: str | None = None, + title: str | None = None, + artist: str | None = None, + spotify_id: str | None = None, + deezer_id: str | None = None, + isrc: str | None = None, + limit: int = 20, +) -> dict[str, Any]: + """Search the local catalog before adding or importing tracks.""" + if limit < 1 or limit > 100: + raise AutomationError("limit must be between 1 and 100.") + + filters = [] + if query: + pattern = f"%{query.strip()}%" + filters.append(or_(Song.title.ilike(pattern), Song.artist.ilike(pattern))) + if title: + filters.append(Song.title.ilike(f"%{title.strip()}%")) + if artist: + filters.append(Song.artist.ilike(f"%{artist.strip()}%")) + if spotify_id: + filters.append(Song.spotify_id == spotify_id) + if deezer_id: + filters.append(Song.deezer_id == str(deezer_id)) + if isrc: + filters.append(Song.isrc == isrc) + + song_query = Song.query + for condition in filters: + song_query = song_query.filter(condition) + songs = song_query.order_by(Song.artist, Song.title).limit(limit).all() + return {"count": len(songs), "songs": [_song_summary(song) for song in songs]} + + +def import_catalog_item( + service_name: str, + item_type: str, + item_id_or_url: str, + user_id: int | None = None, +) -> dict[str, Any]: + """Import a track, album, or playlist from Spotify or Deezer.""" + service = service_name.lower() + item = item_type.lower() + external_id = _parse_external_id(service, item, item_id_or_url) + if service == "spotify": + user = _find_user(user_id) + with current_app.test_request_context(): + login_user(user) + try: + result = ImportHelper.import_item(service, item, external_id) + finally: + logout_user() + else: + result = ImportHelper.import_item(service, item, external_id) + + if result.get("error_count", 0) > 0: + current_app.logger.warning("Import completed with errors: %s", result.get("errors", [])) + return {"service_name": service, "item_type": item, "item_id": external_id, "result": result} + + +def _songs_for_round( + round_type: str, + count: int, + criteria: str | None = None, + song_ids: list[int] | None = None, +) -> tuple[str, str, list[Song]]: + from musicround.routes.generate import ( + get_random_songs, + get_random_songs_from_decade, + get_random_songs_from_genre, + get_random_songs_from_least_used_decade, + get_random_songs_from_least_used_genre, + get_songs_by_tag, + ) + + normalized = round_type.lower().strip() + if song_ids: + songs_by_id = {song.id: song for song in Song.query.filter(Song.id.in_(song_ids)).all()} + songs = [songs_by_id[song_id] for song_id in song_ids if song_id in songs_by_id] + if len(songs) != len(song_ids): + missing = sorted(set(song_ids) - set(songs_by_id)) + raise AutomationError(f"Unknown song IDs: {missing}") + return "Manual", "Explicit song selection", songs[:count] + + if normalized == "random": + return "Random", "Random Selection", get_random_songs(count) + if normalized == "genre": + if criteria: + return "Genre", criteria, get_random_songs_from_genre(criteria, x=count) + songs, chosen = get_random_songs_from_least_used_genre(count) + return "Genre", chosen or "Least Used Genre", songs + if normalized == "decade": + if criteria: + return "Decade", criteria, get_random_songs_from_decade(criteria, x=count) + songs, chosen = get_random_songs_from_least_used_decade(count) + return "Decade", chosen or "Least Used Decade", songs + if normalized == "tag": + if not criteria: + raise AutomationError("Tag rounds require criteria with the tag name.") + return "Tag", criteria, get_songs_by_tag(criteria, count) + raise AutomationError("round_type must be one of random, genre, decade, tag, or manual.") + + +def create_round( + name: str | None = None, + round_type: str = "random", + count: int = 8, + criteria: str | None = None, + song_ids: list[int] | None = None, +) -> dict[str, Any]: + """Create and persist a quiz round.""" + if count < 1: + raise AutomationError("count must be at least 1.") + + resolved_type, resolved_criteria, songs = _songs_for_round( + round_type, count, criteria, song_ids + ) + if not songs: + raise AutomationError("No songs matched the requested round criteria.") + + round_obj = Round( + name=name, + round_type=resolved_type, + round_criteria_used=resolved_criteria, + songs=",".join(str(song.id) for song in songs), + created_at=datetime.utcnow(), + ) + db.session.add(round_obj) + for song in songs: + song.used_count = (song.used_count or 0) + 1 + song.last_used = datetime.utcnow() + db.session.commit() + return {"round": _round_summary(round_obj)} + + +def rename_round(round_id: int, name: str | None) -> dict[str, Any]: + """Rename a persisted round.""" + round_obj = db.session.get(Round, round_id) + if not round_obj: + raise AutomationError(f"Round {round_id} was not found.") + round_obj.name = name.strip() if name and name.strip() else None + db.session.commit() + return {"round": _round_summary(round_obj)} + + +def _spotify_playlist_song_ids(playlist_id: str, limit: int, user_id: int | None) -> list[int]: + from musicround.routes.generate import get_songs_from_spotify_playlist + + user = _find_user(user_id) + with current_app.test_request_context(): + login_user(user) + try: + songs = get_songs_from_spotify_playlist(playlist_id) + finally: + logout_user() + return [song.id for song in songs[:limit]] + + +def _deezer_playlist_song_ids(playlist_id: str, limit: int) -> list[int]: + deezer_client = current_app.config.get("deezer") + if not deezer_client: + raise AutomationError("Deezer client is not configured.") + + tracks = deezer_client.get_playlist_tracks(playlist_id) + song_ids = [] + lastfm_key = current_app.config.get("LASTFM_API_KEY") + for track in tracks[:limit]: + track_id = track.get("id") + if not track_id: + continue + song, _ = deezer_client.import_track(track_id, lastfm_api_key=lastfm_key) + if song: + song_ids.append(song.id) + db.session.commit() + return song_ids + + +def create_round_from_playlist( + service_name: str, + playlist_id_or_url: str, + name: str | None = None, + count: int = 8, + user_id: int | None = None, +) -> dict[str, Any]: + """Import a playlist and create a manual round from the imported songs.""" + imported = import_catalog_item(service_name, "playlist", playlist_id_or_url, user_id=user_id) + playlist_id = imported["item_id"] + if service_name.lower() == "spotify": + song_ids = _spotify_playlist_song_ids(playlist_id, count, user_id) + else: + song_ids = imported.get("result", {}).get( + "imported_song_ids" + ) or _deezer_playlist_song_ids(playlist_id, count) + if not song_ids: + raise AutomationError("Playlist import did not return song IDs to build a round.") + round_result = create_round( + name=name, round_type="manual", count=count, song_ids=song_ids[:count] + ) + return {"import": imported, "round": round_result["round"]} + + +def generate_round_pdf(round_id: int) -> dict[str, Any]: + from musicround.routes.rounds import generate_pdf + + round_obj = db.session.get(Round, round_id) + if not round_obj: + raise AutomationError(f"Round {round_id} was not found.") + pdf_data = generate_pdf(round_id) + if isinstance(pdf_data, str): + raise AutomationError(pdf_data) + round_obj.pdf_generated = True + round_obj.last_generated_at = datetime.utcnow() + db.session.commit() + path = os.path.join("/data/pdfs", f"round_{round_id}.pdf") + return {"round_id": round_id, "path": path, "bytes": len(pdf_data)} + + +def generate_round_mp3(round_id: int, user_id: int | None = None) -> dict[str, Any]: + from musicround.routes.rounds import round_mp3 + + round_obj = db.session.get(Round, round_id) + if not round_obj: + raise AutomationError(f"Round {round_id} was not found.") + user = _find_user(user_id) + with current_app.test_request_context(headers={"X-Requested-With": "XMLHttpRequest"}): + login_user(user) + try: + response = round_mp3(round_id) + finally: + logout_user() + + if hasattr(response, "get_json"): + payload = response.get_json(silent=True) or {} + if payload.get("success") is False or payload.get("error"): + raise AutomationError(payload.get("error", "MP3 generation failed.")) + + path = os.path.join("/data/rounds", f"round_{round_id}.mp3") + if not os.path.exists(path): + raise AutomationError(f"MP3 generation did not create {path}.") + return {"round_id": round_id, "path": path, "bytes": os.path.getsize(path)} + + +def generate_round_assets( + round_id: int, + user_id: int | None = None, + include_pdf: bool = True, + include_mp3: bool = True, +) -> dict[str, Any]: + """Generate requested round assets.""" + assets: dict[str, Any] = {"round_id": round_id} + if include_pdf: + assets["pdf"] = generate_round_pdf(round_id) + if include_mp3: + assets["mp3"] = generate_round_mp3(round_id, user_id=user_id) + return assets + + +def email_round( + round_id: int, + recipient: str | None = None, + user_id: int | None = None, + subject: str | None = None, + body_text: str | None = None, +) -> dict[str, Any]: + """Generate assets and send a round as an email attachment bundle.""" + user = _find_user(user_id) + target = recipient or user.email + if not target: + raise AutomationError("No recipient was provided and the selected user has no email.") + + assets = generate_round_assets(round_id, user_id=user.id) + round_obj = db.session.get(Round, round_id) + title = round_obj.name if round_obj and round_obj.name else f"Quizzical Beats Round {round_id}" + email_subject = subject or title + email_body = body_text or "Attached are the MP3 and PDF files for your quiz round." + + attachments = [] + with open(assets["pdf"]["path"], "rb") as pdf_file: + attachments.append( + { + "data": pdf_file.read(), + "filename": f"round_{round_id}.pdf", + "mimetype": "application/pdf", + } + ) + with open(assets["mp3"]["path"], "rb") as mp3_file: + attachments.append( + { + "data": mp3_file.read(), + "filename": f"round_{round_id}.mp3", + "mimetype": "audio/mpeg", + } + ) + + success, message = send_email(target, email_subject, email_body, attachments) + export = RoundExport( + round_id=round_id, + user_id=user.id, + export_type="email", + destination=target, + include_mp3s=True, + status="success" if success else "failed", + error_message=None if success else message, + ) + db.session.add(export) + db.session.commit() + if not success: + raise AutomationError(message) + return {"success": True, "message": message, "recipient": target, "assets": assets} + + +def inspect_mp3_quality(path: str | None = None, round_id: int | None = None) -> dict[str, Any]: + """Inspect basic MP3 quality and flag common generation issues.""" + if not path: + if round_id is None: + raise AutomationError("Pass either path or round_id.") + path = os.path.join("/data/rounds", f"round_{round_id}.mp3") + if not os.path.exists(path): + raise AutomationError(f"MP3 file not found: {path}") + + audio = AudioSegment.from_mp3(path) + warnings = [] + if len(audio) < 1000: + warnings.append("Audio is shorter than one second.") + if audio.dBFS == float("-inf"): + warnings.append("Audio appears to be silent.") + elif audio.dBFS < -35: + warnings.append("Average loudness is very low.") + elif audio.dBFS > -8: + warnings.append("Average loudness is high; check for limiting or clipping.") + + samples = audio.get_array_of_samples() + max_possible = float(1 << (8 * audio.sample_width - 1)) + clipped = sum(1 for sample in samples if abs(sample) >= max_possible * 0.99) + clipping_ratio = clipped / len(samples) if samples else 0 + if clipping_ratio > 0.001: + warnings.append("Potential clipping detected.") + + return { + "path": path, + "duration_seconds": round(len(audio) / 1000, 3), + "channels": audio.channels, + "frame_rate": audio.frame_rate, + "sample_width_bytes": audio.sample_width, + "average_dbfs": None if audio.dBFS == float("-inf") else round(audio.dBFS, 2), + "peak_dbfs": round(audio.max_dBFS, 2), + "clipping_ratio": round(clipping_ratio, 6), + "warnings": warnings, + "ok": not warnings, + } + + +def inspect_pdf_quality(path: str | None = None, round_id: int | None = None) -> dict[str, Any]: + """Inspect basic PDF integrity for generated round sheets.""" + if not path: + if round_id is None: + raise AutomationError("Pass either path or round_id.") + path = os.path.join("/data/pdfs", f"round_{round_id}.pdf") + if not os.path.exists(path): + raise AutomationError(f"PDF file not found: {path}") + + with open(path, "rb") as pdf_file: + data = pdf_file.read() + warnings = [] + if not data.startswith(b"%PDF-"): + warnings.append("File does not start with a PDF header.") + if b"%%EOF" not in data[-2048:]: + warnings.append("PDF EOF marker was not found near the end of the file.") + if len(data) < 1024: + warnings.append("PDF file is unusually small.") + page_count = data.count(b"/Type /Page") + if page_count == 0: + warnings.append("No PDF pages were detected.") + + return { + "path": path, + "bytes": len(data), + "page_count_estimate": page_count, + "warnings": warnings, + "ok": not warnings, + } + + +def generate_tts_snippet( + user_id: int, + mp3_type: str, + text: str, + service: str = "openai", + voice: str | None = None, + model: str | None = None, + stability: float | None = None, + similarity: float | None = None, +) -> dict[str, Any]: + """Generate and assign a custom intro, replay, or outro MP3 for a user.""" + if mp3_type not in {"intro", "replay", "outro"}: + raise AutomationError("mp3_type must be intro, replay, or outro.") + if not text: + raise AutomationError("text is required for TTS generation.") + + user = _find_user(user_id) + path = generate_tts_mp3( + text=text, + username=user.username, + mp3_type=mp3_type, + service=service, + voice=voice, + model=model, + stability=stability, + similarity=similarity, + ) + if not path: + raise AutomationError("TTS generation failed.") + + setattr(user, f"{mp3_type}_mp3", path) + db.session.commit() + return {"user_id": user.id, "mp3_type": mp3_type, "path": path} diff --git a/requirements.txt b/requirements.txt index b574beb..b16db2a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -19,6 +19,7 @@ authlib>=1.6.5 Flask-Caching gunicorn psutil +mcp[cli] # Testing dependencies pytest>=7.4.0 diff --git a/tests/test_automation_service.py b/tests/test_automation_service.py new file mode 100644 index 0000000..5359d0b --- /dev/null +++ b/tests/test_automation_service.py @@ -0,0 +1,129 @@ +"""Tests for agent automation services.""" + +import os +import shutil +import tempfile +from unittest.mock import patch + +import pytest + +from musicround.models import Song, User, db +from musicround.services import automation + + +def _create_user(username="agentuser", email="agent@example.com"): + user = User(username=username, email=email) + user.password = "AgentPass123!" + db.session.add(user) + db.session.commit() + return user + + +def _create_song(title="Song", artist="Artist", **kwargs): + song = Song(title=title, artist=artist, **kwargs) + db.session.add(song) + db.session.commit() + return song + + +class TestSongAutomation: + """Tests for catalog lookup and mutation.""" + + def test_find_songs_by_query(self, app): + with app.app_context(): + _create_song(title="Blue Monday", artist="New Order", genre="Synthpop") + + result = automation.find_songs(query="blue") + + assert result["count"] == 1 + assert result["songs"][0]["title"] == "Blue Monday" + + def test_add_song_reuses_existing_by_isrc_and_adds_tags(self, app): + with app.app_context(): + _create_song(title="Old Title", artist="Old Artist", isrc="ABC123") + + result = automation.add_song( + title="New Title", + artist="New Artist", + isrc="ABC123", + tags=["warmup", "classic"], + ) + + assert result["created"] is False + assert Song.query.count() == 1 + song = Song.query.first() + assert song.title == "New Title" + assert sorted(tag.name for tag in song.tags) == ["classic", "warmup"] + + +class TestRoundAutomation: + """Tests for round creation and naming.""" + + def test_create_and_rename_manual_round(self, app): + with app.app_context(): + song_one = _create_song(title="One", artist="A") + song_two = _create_song(title="Two", artist="B") + + created = automation.create_round( + name="Initial Name", + round_type="manual", + song_ids=[song_one.id, song_two.id], + ) + renamed = automation.rename_round(created["round"]["id"], "Final Name") + + assert created["round"]["song_ids"] == [song_one.id, song_two.id] + assert renamed["round"]["name"] == "Final Name" + assert Song.query.get(song_one.id).used_count == 1 + + +class TestAssetInspection: + """Tests for generated asset quality checks.""" + + def test_inspect_pdf_quality(self, app): + with app.app_context(), tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp: + tmp.write(b"%PDF-1.4\n") + tmp.write(b"1 0 obj << /Type /Page >> endobj\n" * 40) + tmp.write(b"%%EOF") + tmp_path = tmp.name + + try: + with app.app_context(): + result = automation.inspect_pdf_quality(path=tmp_path) + + assert result["ok"] is True + assert result["page_count_estimate"] > 0 + finally: + os.remove(tmp_path) + + def test_inspect_mp3_quality_existing_fixture(self, app): + if not shutil.which("ffprobe"): + pytest.skip("ffprobe is required for MP3 inspection") + + fixture_path = os.path.abspath("musicround/mp3/intro.mp3") + + with app.app_context(): + result = automation.inspect_mp3_quality(path=fixture_path) + + assert result["duration_seconds"] > 0 + assert result["channels"] >= 1 + assert "ok" in result + + +class TestTTSAutomation: + """Tests for TTS snippet assignment.""" + + def test_generate_tts_snippet_updates_user_audio_path(self, app): + with app.app_context(): + user = _create_user() + + with patch("musicround.services.automation.generate_tts_mp3") as mock_tts: + mock_tts.return_value = "custommp3/agentuser/intro.mp3" + result = automation.generate_tts_snippet( + user_id=user.id, + mp3_type="intro", + text="Welcome to the quiz", + service="openai", + ) + + assert result["path"] == "custommp3/agentuser/intro.mp3" + assert User.query.get(user.id).intro_mp3 == "custommp3/agentuser/intro.mp3" diff --git a/tests/test_import_queue.py b/tests/test_import_queue.py index d6d2905..d9ca7a5 100644 --- a/tests/test_import_queue.py +++ b/tests/test_import_queue.py @@ -2,7 +2,10 @@ import pytest import threading import time -from musicround.helpers.import_queue import ImportJob, ImportQueue +from unittest.mock import patch + +from musicround.helpers.import_queue import ImportJob, ImportQueue, ImportWorker +from musicround.models import User, db class TestImportJob: @@ -154,3 +157,46 @@ class TestImportQueue: assert not errors assert len(results) == 5 + + +class TestImportWorker: + """Tests for ImportWorker job processing.""" + + def test_process_job_imports_as_user(self, app): + """Test that _process_job logs in the target user and imports the item.""" + with app.app_context(): + user = User(username='workeruser', email='worker@example.com') + user.password = 'WorkerPass123!' + db.session.add(user) + db.session.commit() + + worker = ImportWorker(app, ImportQueue()) + job = ImportJob( + priority=1, + service_name='deezer', + item_type='track', + item_id='123', + user_id=user.id, + ) + + with patch('musicround.helpers.import_queue.ImportHelper.import_item') as mock_import: + worker._process_job(job) + + mock_import.assert_called_once_with('deezer', 'track', '123') + + def test_process_job_unknown_user_does_not_import(self, app): + """Test that jobs for missing users are ignored.""" + with app.app_context(): + worker = ImportWorker(app, ImportQueue()) + job = ImportJob( + priority=1, + service_name='deezer', + item_type='track', + item_id='123', + user_id=999, + ) + + with patch('musicround.helpers.import_queue.ImportHelper.import_item') as mock_import: + worker._process_job(job) + + mock_import.assert_not_called()