Podcast research briefing agent with Spotify

import getpass
import json
import os

from IPython.display import display, Markdown
from mistralai.client import Mistral
from mistralai.client.models import (
    FunctionCallEvent,
    FunctionResultEntry,
    MessageOutputEvent,
)

if not os.environ.get("MISTRAL_API_KEY"):
    os.environ["MISTRAL_API_KEY"] = getpass.getpass("Mistral API key: ")

if not os.environ.get("SPOTIFY_CLIENT_ID"):
    os.environ["SPOTIFY_CLIENT_ID"] = getpass.getpass("Spotify Client ID: ")

if not os.environ.get("SPOTIFY_CLIENT_SECRET"):
    os.environ["SPOTIFY_CLIENT_SECRET"] = getpass.getpass("Spotify Client Secret: ")

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

                        ┌─────────────────────┐
                        │   Mistral Agent      │
                        │   (zai-glm-5-2)     │
                        └──────┬──────┬────────┘
                               │      │
              ┌────────────────┘      └────────────────┐
              │                │                       │
    ┌─────────▼────────┐  ┌───▼──────────────┐  ┌─────▼──────────┐
    │ Spotify functions │  │ Briefing function│  │ Web Search     │
    │ (spotipy client   │  │ (mistral LLM     │  │ (built-in)     │
    │  credentials)     │  │  chat completion) │  │                │
    └──────────────────┘  └──────────────────┘  └────────────────┘

import spotipy
from spotipy.oauth2 import SpotifyClientCredentials

sp = spotipy.Spotify(auth_manager=SpotifyClientCredentials(
    client_id=os.environ["SPOTIFY_CLIENT_ID"],
    client_secret=os.environ["SPOTIFY_CLIENT_SECRET"],
))

MODEL = "zai-glm-5-2"

BRIEFING_SYSTEM_PROMPT = """You are a research analyst. Given a topic, podcast data from
Spotify, and web research, produce a concise markdown briefing with:
- Executive summary (2-3 sentences)
- Ranked episode recommendations with relevance score, episode/show name, Spotify link, duration, release date, and a one-sentence summary
- Key themes across episodes
- Notable experts and guests
- Gaps and limitations

Use ONLY exact URLs from the input data. Never fabricate Spotify links."""


def _format_duration(ms: int) -> str:
    """Convert milliseconds to a human-readable duration string."""
    minutes = ms // 60000
    if minutes >= 60:
        hours = minutes // 60
        remaining = minutes % 60
        return f"{hours}h {remaining}m"
    return f"{minutes}m"


def search_podcasts(query: str, limit: int = 10) -> str:
    """Search for podcast shows on Spotify."""
    try:
        results = sp.search(q=query, type="show", limit=limit)
        shows = []
        for item in results.get("shows", {}).get("items", []):
            if item is None:
                continue
            shows.append({
                "id": item["id"],
                "name": item["name"],
                "publisher": item.get("publisher", "Unknown"),
                "description": (item.get("description") or "")[:500],
                "total_episodes": item.get("total_episodes", 0),
                "url": item.get("external_urls", {}).get("spotify", ""),
            })
        return json.dumps(shows, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})


def search_episodes(query: str, limit: int = 10) -> str:
    """Search for specific podcast episodes on Spotify."""
    try:
        results = sp.search(q=query, type="episode", limit=limit)
        episodes = []
        for item in results.get("episodes", {}).get("items", []):
            if item is None:
                continue
            episodes.append({
                "id": item["id"],
                "name": item["name"],
                "show_name": item.get("show", {}).get("name", "Unknown"),
                "description": (item.get("description") or "")[:500],
                "duration": _format_duration(item.get("duration_ms", 0)),
                "release_date": item.get("release_date", "Unknown"),
                "url": item.get("external_urls", {}).get("spotify", ""),
            })
        return json.dumps(episodes, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})


def get_podcast_details(show_id: str) -> str:
    """Get full details for a specific podcast show."""
    try:
        show = sp.show(show_id)
        return json.dumps({
            "id": show["id"],
            "name": show["name"],
            "publisher": show.get("publisher", "Unknown"),
            "description": (show.get("description") or "")[:1000],
            "total_episodes": show.get("total_episodes", 0),
            "languages": show.get("languages", []),
            "url": show.get("external_urls", {}).get("spotify", ""),
        }, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})


def get_podcast_episodes(show_id: str, limit: int = 10) -> str:
    """Get episodes from a specific podcast show."""
    try:
        results = sp.show_episodes(show_id, limit=limit)
        episodes = []
        for item in results.get("items", []):
            if item is None:
                continue
            episodes.append({
                "id": item["id"],
                "name": item["name"],
                "description": (item.get("description") or "")[:500],
                "duration": _format_duration(item.get("duration_ms", 0)),
                "release_date": item.get("release_date", "Unknown"),
                "url": item.get("external_urls", {}).get("spotify", ""),
            })
        return json.dumps(episodes, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})


def get_episode_details(episode_id: str) -> str:
    """Get full details for a specific podcast episode."""
    try:
        episode = sp.episode(episode_id)
        return json.dumps({
            "id": episode["id"],
            "name": episode["name"],
            "show_name": episode.get("show", {}).get("name", "Unknown"),
            "description": (episode.get("description") or "")[:2000],
            "duration": _format_duration(episode.get("duration_ms", 0)),
            "release_date": episode.get("release_date", "Unknown"),
            "language": episode.get("language", "Unknown"),
            "url": episode.get("external_urls", {}).get("spotify", ""),
        }, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})


def generate_research_briefing(topic: str, podcast_data: str, web_research: str) -> str:
    """Generate a structured research briefing from podcast data and web research."""
    try:
        response = client.chat.complete(
            model=MODEL,
            messages=[
                {"role": "system", "content": BRIEFING_SYSTEM_PROMPT},
                {"role": "user", "content": f"""Topic: {topic}

Podcast data:
{podcast_data}

Web research:
{web_research}"""},
            ],
        )
        return response.choices[0].message.content
    except Exception as e:
        return json.dumps({"error": str(e)})