A script to export all the watching data from Jellyfin to a Yamtrack-compatible CSV format

jellyfin2yamtrack.py

"""
Jellyfin to Yamtrack CSV Exporter

This script fetches a user's library and watch progress from a Jellyfin server, maps the 
media to precise TMDB IDs and images using the TMDB API, and outputs a Yamtrack-compatible CSV file.

Features:
- Fetches Movies, TV Shows, Seasons, and Episodes.
- Resolves Jellyfin Provider IDs (Imdb, Tvdb) to TMDB IDs via the TMDB API.
- Properly maps Season and Episode rows to their Parent TV Show's TMDB ID (Required by Yamtrack).
- Gathers TMDB image links (posters for movies/shows/seasons, stills for episodes).
- Converts Jellyfin 'LastPlayedDate' into the strictly required ISO-8601 Yamtrack format.

Usage: jellyfin2yamtrack.py [-h] --jellyfin-url JELLYFIN_URL --jellyfin-api-key JELLYFIN_API_KEY
                            --jellyfin-username JELLYFIN_USERNAME --tmdb-api-key TMDB_API_KEY [--movies-only | --episodes-only]
                            [--output OUTPUT]
                            
The script requires the presence of the requests library in the current Python environment. 
I recommend running it with uv, like this:
uv run --with requests jellyfin2yamtrack.py --jellyfin-url https://jellyfin.org --jellyfin-api-key jellyfin_api_key --jellyfin-username username --tmdb-api-key tmdb_key --episodes-only
"""

import argparse
import csv
import sys
import time
from datetime import datetime

import requests

# Yamtrack CSV Specification Columns
COLUMNS = [
    "media_id", "source", "media_type", "title", "image",
    "season_number", "episode_number", "score", "status",
    "notes", "start_date", "end_date", "progress"
]

TMDB_IMAGE_BASE_URL = "https://image.tmdb.org/t/p/w500"

class TMDBClient:
    """Handles interactions with the TMDB API."""
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.themoviedb.org/3"
        self.session = requests.Session()
        self.session.params = {"api_key": self.api_key}
        self.cache = {}

    def _get(self, endpoint, params=None):
        cache_key = (endpoint, frozenset(params.items()) if params else None)
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        url = f"{self.base_url}{endpoint}"
        retries = 3
        
        for attempt in range(retries):
            try:
                response = self.session.get(url, params=params)
                if response.status_code == 429:
                    time.sleep(1)
                    continue
                response.raise_for_status()
                data = response.json()
                self.cache[cache_key] = data
                return data
            except requests.exceptions.RequestException as e:
                if attempt == retries - 1:
                    print(f"TMDB API Error on {endpoint}: {e}", file=sys.stderr)
                    return None
                time.sleep(0.5)
        return None

    def find_by_external_id(self, external_id, external_source):
        return self._get(f"/find/{external_id}", {"external_source": external_source})

    def search_movie(self, title, year=None):
        params = {"query": title}
        if year:
            params["primary_release_year"] = year
        res = self._get("/search/movie", params)
        return res["results"][0] if res and res.get("results") else None

    def search_show(self, title, year=None):
        params = {"query": title}
        if year:
            params["first_air_date_year"] = year
        res = self._get("/search/tv", params)
        return res["results"][0] if res and res.get("results") else None

    def get_movie(self, tmdb_id):
        return self._get(f"/movie/{tmdb_id}")

    def get_show(self, tmdb_id):
        return self._get(f"/tv/{tmdb_id}")

    def get_season(self, show_tmdb_id, season_number):
        return self._get(f"/tv/{show_tmdb_id}/season/{season_number}")

    def get_episode(self, show_tmdb_id, season_number, episode_number):
        return self._get(f"/tv/{show_tmdb_id}/season/{season_number}/episode/{episode_number}")


class JellyfinClient:
    """Handles interactions with the Jellyfin API."""
    def __init__(self, url, api_key, username):
        self.url = url.rstrip('/')
        self.api_key = api_key
        self.username = username
        self.headers = {
            "Authorization": f"MediaBrowser Token={self.api_key}",
            "Content-Type": "application/json"
        }
        self.user_id = self._get_user_id()

    def _get_user_id(self):
        try:
            response = requests.get(f"{self.url}/Users", headers=self.headers)
            response.raise_for_status()
            for user in response.json():
                if user.get("Name").lower() == self.username.lower():
                    return user.get("Id")
            raise ValueError(f"User '{self.username}' not found on Jellyfin server.")
        except requests.exceptions.RequestException as e:
            print(f"Jellyfin API Error (Users): {e}", file=sys.stderr)
            sys.exit(1)

    def get_items(self, item_types):
        params = {
            "Recursive": "true",
            "IncludeItemTypes": item_types,
            "Fields": "ProviderIds,DateLastPlayed,ParentId,UserData",
        }
        try:
            response = requests.get(
                f"{self.url}/Users/{self.user_id}/Items",
                headers=self.headers,
                params=params
            )
            response.raise_for_status()
            return response.json().get("Items", [])
        except requests.exceptions.RequestException as e:
            print(f"Jellyfin API Error (Items): {e}", file=sys.stderr)
            sys.exit(1)


# --- Helper Functions ---

def format_datetime(jellyfin_date_str):
    if not jellyfin_date_str:
        return ""
    try:
        if jellyfin_date_str.endswith('Z'):
            jellyfin_date_str = jellyfin_date_str[:-1]
        if '.' in jellyfin_date_str:
            jellyfin_date_str = jellyfin_date_str.split('.')[0] 
            
        dt = datetime.strptime(jellyfin_date_str, "%Y-%m-%dT%H:%M:%S")
        return dt.strftime("%Y-%m-%d %H:%M:%S+00:00")
    except Exception:
        return ""

def get_status(user_data):
    if not user_data:
        return "Planning"
    if user_data.get("Played"):
        return "Completed"
    if user_data.get("PlaybackPositionTicks", 0) > 0:
        return "In progress"
    return "Planning"

def get_tmdb_image_url(path):
    if not path:
        return ""
    return f"{TMDB_IMAGE_BASE_URL}{path}"

def resolve_tmdb_id(item, tmdb_client, tmdb_type):
    provider_ids = item.get("ProviderIds", {})
    
    if "Tmdb" in provider_ids:
        return provider_ids["Tmdb"]
        
    for ext_id, source_name in [("Imdb", "imdb_id"), ("Tvdb", "tvdb_id")]:
        if ext_id in provider_ids:
            res = tmdb_client.find_by_external_id(provider_ids[ext_id], source_name)
            if res:
                results = res.get(f"{tmdb_type}_results", [])
                if results:
                    return results[0].get("id")

    title = item.get("Name")
    year = item.get("ProductionYear")
    data = tmdb_client.search_movie(title, year) if tmdb_type == "movie" else tmdb_client.search_show(title, year)
    return data.get("id") if data else None


# --- Processing Functions ---

def process_movie(item, tmdb_client):
    tmdb_id = resolve_tmdb_id(item, tmdb_client, "movie")
    poster_path = ""
    title = item.get("Name")
    
    if tmdb_id:
        tmdb_data = tmdb_client.get_movie(tmdb_id)
        if tmdb_data:
            poster_path = tmdb_data.get("poster_path")
    else:
        print(f"Warning: Could not resolve TMDB ID for Movie '{title}'")
        
    user_data = item.get("UserData", {})
    last_played = format_datetime(user_data.get("LastPlayedDate"))
    
    return {
        "media_id": tmdb_id or "",
        "source": "tmdb",
        "media_type": "movie",
        "title": title,
        "image": get_tmdb_image_url(poster_path),
        "season_number": "",
        "episode_number": "",
        "score": "",
        "status": get_status(user_data),
        "notes": "",
        "start_date": last_played,
        "end_date": last_played,
        "progress": ""
    }

def process_series(item, tmdb_client, series_cache):
    jellyfin_id = item.get("Id")
    if jellyfin_id in series_cache:
        return None 
        
    tmdb_id = resolve_tmdb_id(item, tmdb_client, "tv")
    poster_path = ""
    title = item.get("Name")
    
    if tmdb_id:
        tmdb_data = tmdb_client.get_show(tmdb_id)
        if tmdb_data:
            poster_path = tmdb_data.get("poster_path")
        series_cache[jellyfin_id] = tmdb_id
    else:
        print(f"Warning: Could not resolve TMDB ID for TV Show '{title}'")
        
    user_data = item.get("UserData", {})
    last_played = format_datetime(user_data.get("LastPlayedDate"))
    
    return {
        "media_id": tmdb_id or "",
        "source": "tmdb",
        "media_type": "tv",
        "title": title,
        "image": get_tmdb_image_url(poster_path),
        "season_number": "",
        "episode_number": "",
        "score": "",
        "status": get_status(user_data),
        "notes": "",
        "start_date": last_played,
        "end_date": last_played,
        "progress": ""
    }

def process_season(item, tmdb_client, series_cache):
    season_number = item.get("IndexNumber")
    if season_number is None:
        return None 
        
    show_tmdb_id = series_cache.get(item.get("SeriesId"))
    if not show_tmdb_id:
        return None

    poster_path = ""
    title = item.get("Name")
    
    tmdb_data = tmdb_client.get_season(show_tmdb_id, season_number)
    if tmdb_data:
        poster_path = tmdb_data.get("poster_path")
        title = title or tmdb_data.get("name")
            
    user_data = item.get("UserData", {})
    last_played = format_datetime(user_data.get("LastPlayedDate"))
    
    return {
        "media_id": show_tmdb_id, # FIX: Must be the TV Show's TMDB ID
        "source": "tmdb",
        "media_type": "season",
        "title": title or f"Season {season_number}",
        "image": get_tmdb_image_url(poster_path),
        "season_number": season_number,
        "episode_number": "",
        "score": "",
        "status": get_status(user_data),
        "notes": "",
        "start_date": last_played,
        "end_date": last_played,
        "progress": ""
    }

def process_episode(item, tmdb_client, series_cache):
    season_number = item.get("ParentIndexNumber", 0) 
    episode_number = item.get("IndexNumber")
    
    if episode_number is None:
        return None 
        
    show_tmdb_id = series_cache.get(item.get("SeriesId"))
    if not show_tmdb_id:
        return None

    still_path = ""
    title = item.get("Name")
    
    tmdb_data = tmdb_client.get_episode(show_tmdb_id, season_number, episode_number)
    if tmdb_data:
        still_path = tmdb_data.get("still_path")
        title = title or tmdb_data.get("name")
            
    user_data: dict = item.get("UserData", {})
    last_played = format_datetime(user_data.get("LastPlayedDate"))
    
    status = get_status(user_data)
    percentage = user_data.get("PlayedPercentage", 0) * 100
    progress = 100 if status == "Completed" else (percentage if status == "Planning" else 0)
    return {
        "media_id": show_tmdb_id, # FIX: Must be the TV Show's TMDB ID
        "source": "tmdb",
        "media_type": "episode",
        "title": title or f"Episode {episode_number}",
        "image": get_tmdb_image_url(still_path),
        "season_number": season_number,
        "episode_number": episode_number,
        "score": "",
        "status": status,
        "notes": "",
        "start_date": last_played,
        "end_date": last_played,
        "progress": progress
    }


def main():
    parser = argparse.ArgumentParser(description="Export Jellyfin data to Yamtrack-compatible CSV")
    parser.add_argument("--jellyfin-url", required=True, help="Jellyfin server URL (e.g. http://192.168.1.100:8096)")
    parser.add_argument("--jellyfin-api-key", required=True, help="Jellyfin API Key")
    parser.add_argument("--jellyfin-username", required=True, help="Jellyfin Username to export")
    parser.add_argument("--tmdb-api-key", required=True, help="TMDB API Key")
    
    group = parser.add_mutually_exclusive_group()
    group.add_argument("--movies-only", action="store_true", help="Only export movies")
    group.add_argument("--episodes-only", action="store_true", help="Only export TV Shows, Seasons, and Episodes")
    
    parser.add_argument("--output", default="yamtrack_export.csv", help="Output CSV file path")
    args = parser.parse_args()

    tmdb = TMDBClient(args.tmdb_api_key)
    jellyfin = JellyfinClient(args.jellyfin_url, args.jellyfin_api_key, args.jellyfin_username)
    
    if args.movies_only:
        item_types = ["Movie"]
    elif args.episodes_only:
        item_types = ["Series", "Season", "Episode"]
    else:
        item_types = ["Movie", "Series", "Season", "Episode"]
        
    print(f"Fetching {', '.join(item_types)} from Jellyfin (Username: {args.jellyfin_username})...")
    items = jellyfin.get_items(",".join(item_types))
    
    movies, series, seasons, episodes = [], {}, {}, []
    
    for item in items:
        itype = item.get("Type")
        if itype == "Movie":
            movies.append(item)
        elif itype == "Series":
            series[item.get("Id")] = item
        elif itype == "Season":
            seasons[item.get("Id")] = item
        elif itype == "Episode":
            episodes.append(item)
            
    rows = []
    series_cache = {}

    print(f"Processing {len(movies)} Movies...")
    for m in movies:
        row = process_movie(m, tmdb)
        if row: rows.append(row)
            
    print(f"Processing {len(series)} TV Shows...")
    for _, s in series.items():
        row = process_series(s, tmdb, series_cache)
        if row: rows.append(row)
            
    print(f"Processing {len(seasons)} Seasons...")
    for _, s in seasons.items():
        row = process_season(s, tmdb, series_cache)
        if row: rows.append(row)
            
    print(f"Processing {len(episodes)} Episodes...")
    for e in episodes:
        row = process_episode(e, tmdb, series_cache)
        if row: rows.append(row)
            
    with open(args.output, "w", newline='', encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=COLUMNS)
        writer.writeheader()
        writer.writerows(rows)
        
    print(f"\nSuccess! Exported {len(rows)} total records to {args.output}")

if __name__ == "__main__":
    main()