Convert a manual Trakt export zip into a Yamtrack (v0.25.3) import CSV — full history, rewatches, ratings, watchlist, TMDB metadata prefill

trakt_to_yamtrack.py

#!/usr/bin/env python3
"""Convert a manual Trakt export (.zip contents) into a Yamtrack import CSV.

Targets Yamtrack v0.25.3's CSV importer (integrations/imports/yamtrack.py).
Key importer facts this script relies on:
- Row order determines bulk-create order: movie, tv, season, episode.
- TV is unique per (user, item); Season per (related_tv, item); Episode and
  Movie rows may repeat (that is how rewatches are stored).
- Episodes are linked to seasons via (media_id, season_number) after seasons
  are created (update_episode_references), so season rows must precede
  episode rows in the file.
- If title AND image are both non-empty, the importer makes no provider API
  calls for that row. We prefill both from TMDB here.
- `progressed_at` (when present) becomes the history date -> activity heatmap.
- EpisodeForm only reads end_date; TvForm/SeasonForm read score/status/notes;
  MovieForm reads score/status/start_date/end_date/notes.

Also merges the current Yamtrack CSV export (tmdb movie/tv/season/episode rows
only; books etc. are excluded so overwrite mode won't touch them).
"""

import csv
import json
import re
import sys
import time
import urllib.request
import urllib.error
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path

BASE = Path(__file__).parent
TRAKT_DIR = BASE / "trakt"
YAMTRACK_EXPORT = Path(sys.argv[1]) if len(sys.argv) > 1 else None
OUT_CSV = BASE / "yamtrack-import.csv"
CACHE_FILE = BASE / "tmdb_cache.json"

TMDB_KEY = "61572be02f0a068658828f6396aacf60"  # yamtrack's public default key
IMG_NONE = (
    "https://www.themoviedb.org/assets/2/v4/glyphicons/basic/"
    "glyphicons-basic-38-picture-grey-c2ebdbb057f2a7614185931650f8cee23fa137b93812ccb132b9df511df1cfac.svg"
)
IMG_BASE = "https://image.tmdb.org/t/p/w500"

COLUMNS = [
    "media_id", "source", "media_type", "title", "image",
    "season_number", "episode_number", "score", "status", "notes",
    "start_date", "end_date", "progress", "created_at", "progressed_at",
]

warnings = []


def ts(trakt_date):
    """Trakt '2026-08-01T18:48:00.000Z' -> '2026-08-01 18:48:00+00:00'."""
    if not trakt_date:
        return ""
    dt = datetime.fromisoformat(trakt_date.replace("Z", "+00:00"))
    dt = dt.replace(second=0, microsecond=0)
    return dt.strftime("%Y-%m-%d %H:%M:%S+00:00")


def load_pages(prefix):
    """Concatenate trakt/<prefix>.json and trakt/<prefix>-<n>.json in order."""
    pattern = re.compile(rf"^{re.escape(prefix)}(-(\d+))?\.json$")
    matches = []
    for path in TRAKT_DIR.iterdir():
        m = pattern.match(path.name)
        if m:
            matches.append((int(m.group(2) or 0), path))
    entries = []
    for _, path in sorted(matches):
        entries.extend(json.loads(path.read_text()))
    return entries


# ---------------------------------------------------------------- TMDB fetch

cache = json.loads(CACHE_FILE.read_text()) if CACHE_FILE.exists() else {}


def tmdb_get(path):
    if path in cache:
        return cache[path]
    url = f"https://api.themoviedb.org/3{path}?api_key={TMDB_KEY}"
    for attempt in range(5):
        try:
            with urllib.request.urlopen(url, timeout=30) as resp:
                data = json.loads(resp.read())
                cache[path] = data
                return data
        except urllib.error.HTTPError as e:
            if e.code == 404:
                cache[path] = None
                return None
            if e.code == 429:
                time.sleep(2 * (attempt + 1))
                continue
            raise
        except OSError:
            time.sleep(1 + attempt)
    raise RuntimeError(f"TMDB fetch kept failing for {path}")


def fetch_all(paths):
    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(tmdb_get, paths))


# ------------------------------------------------------------- collect trakt

history = load_pages("watched-history")
history.sort(key=lambda e: e["watched_at"])

movie_watches = defaultdict(list)   # tmdb_id -> [watched_at, ...]
movie_titles = {}
ep_watches = {}                     # (show_id, s, e, end_ts) -> watched_at iso
show_titles = {}

for entry in history:
    kind = entry["type"]
    data = entry.get(kind == "movie" and "movie" or "show")
    tmdb_id = data["ids"].get("tmdb")
    if not tmdb_id:
        warnings.append(f"no tmdb id: {data.get('title')}")
        continue
    tmdb_id = str(tmdb_id)
    if kind == "movie":
        movie_watches[tmdb_id].append(entry["watched_at"])
        movie_titles[tmdb_id] = data["title"]
    elif kind == "episode":
        show_titles[tmdb_id] = data["title"]
        key = (tmdb_id, entry["episode"]["season"], entry["episode"]["number"],
               ts(entry["watched_at"]))
        # identical show/season/episode/minute duplicates collapse to one
        ep_watches[key] = entry["watched_at"]

movie_ratings, show_ratings, season_ratings = {}, {}, {}
for r in load_pages("ratings-movies"):
    mid = r["movie"]["ids"].get("tmdb")
    if mid:
        movie_ratings[str(mid)] = (r["rating"], r["rated_at"], r["movie"]["title"])
for r in load_pages("ratings-shows"):
    sid = r["show"]["ids"].get("tmdb")
    if sid:
        show_ratings[str(sid)] = (r["rating"], r["rated_at"], r["show"]["title"])
for r in load_pages("ratings-seasons"):
    sid = r["show"]["ids"].get("tmdb")
    if sid:
        season_ratings[(str(sid), r["season"]["number"])] = r["rating"]

skipped_episode_ratings = len(load_pages("ratings-episodes"))

watchlist_movies, watchlist_shows = {}, {}
for w in load_pages("lists-watchlist"):
    data = w.get(w["type"])
    if w["type"] not in ("movie", "show") or not data:
        warnings.append(f"watchlist entry skipped (type {w['type']})")
        continue
    tmdb_id = data["ids"].get("tmdb")
    if not tmdb_id:
        continue
    target = watchlist_movies if w["type"] == "movie" else watchlist_shows
    target[str(tmdb_id)] = (w["listed_at"], data["title"])

# ------------------------------------------------- merge current yamtrack csv

yt_tv_rows, yt_season_rows = {}, {}
yt_movie_rows = []
if YAMTRACK_EXPORT and YAMTRACK_EXPORT.exists():
    for row in csv.DictReader(YAMTRACK_EXPORT.open()):
        if row["source"] != "tmdb":
            continue
        mt = row["media_type"]
        if mt == "movie":
            yt_movie_rows.append(row)
        elif mt == "tv":
            yt_tv_rows[row["media_id"]] = row
        elif mt == "season":
            yt_season_rows[(row["media_id"], int(row["season_number"]))] = row
        elif mt == "episode":
            key = (row["media_id"], int(row["season_number"]),
                   int(row["episode_number"]), row["end_date"])
            if key not in ep_watches:
                ep_watches[key] = None  # keep original end_date string
                show_titles.setdefault(row["media_id"], row["title"])

# ------------------------------------------------------------------ fetch tmdb

all_show_ids = (set(show_titles) | set(show_ratings) | set(watchlist_shows))
all_movie_ids = (set(movie_watches) | set(movie_ratings) | set(watchlist_movies)
                 | {r["media_id"] for r in yt_movie_rows})

season_numbers = defaultdict(set)
for (show_id, season, _ep, _ts) in ep_watches:
    season_numbers[show_id].add(season)
for (show_id, season) in season_ratings:
    season_numbers[show_id].add(season)
for (show_id, season) in yt_season_rows:
    season_numbers[show_id].add(season)

print(f"TMDB: {len(all_movie_ids)} movies, {len(all_show_ids)} shows, "
      f"{sum(len(v) for v in season_numbers.values())} seasons")
fetch_all([f"/movie/{mid}" for mid in all_movie_ids])
fetch_all([f"/tv/{sid}" for sid in all_show_ids])
fetch_all([f"/tv/{sid}/season/{n}"
           for sid, nums in season_numbers.items() for n in nums])
CACHE_FILE.write_text(json.dumps(cache))

# --------------------------------------------------------------- build rows


def img(poster_path, fallback=IMG_NONE):
    return f"{IMG_BASE}{poster_path}" if poster_path else fallback


def base_row(**kw):
    row = {c: "" for c in COLUMNS}
    row.update(kw)
    return row


movie_rows, tv_rows, season_rows, episode_rows = [], [], [], []

# movies: one row per watch; rating applied to the latest watch
for mid in sorted(all_movie_ids, key=int):
    meta = tmdb_get(f"/movie/{mid}")
    rating = movie_ratings.get(mid)
    title = (meta and meta.get("title")) or movie_titles.get(mid) \
        or (rating and rating[2]) or (watchlist_movies.get(mid) or ("", ""))[1]
    image = img(meta and meta.get("poster_path"))
    watches = sorted(movie_watches.get(mid, []))

    yt_dates = set()
    for row in yt_movie_rows:
        if row["media_id"] == mid:
            yt_dates.add(row["end_date"])

    if watches:
        for i, watched_at in enumerate(watches):
            if ts(watched_at) in yt_dates:
                continue
            last = i == len(watches) - 1
            movie_rows.append(base_row(
                media_id=mid, source="tmdb", media_type="movie",
                title=title, image=image,
                score=rating[0] if (rating and last) else "",
                status="Completed",
                start_date=ts(watched_at), end_date=ts(watched_at),
                progressed_at=ts(watched_at),
            ))
    elif rating:
        movie_rows.append(base_row(
            media_id=mid, source="tmdb", media_type="movie",
            title=title, image=image, score=rating[0], status="Completed",
            progressed_at=ts(rating[1]),
        ))
    elif mid in watchlist_movies:
        listed_at, _ = watchlist_movies[mid]
        movie_rows.append(base_row(
            media_id=mid, source="tmdb", media_type="movie",
            title=title, image=image, status="Planning",
            progressed_at=ts(listed_at),
        ))
    for row in yt_movie_rows:
        if row["media_id"] == mid:
            movie_rows.append(row)

# shows/seasons/episodes
eps_by_show = defaultdict(lambda: defaultdict(dict))
for (show_id, season, ep, end), watched_at in ep_watches.items():
    eps_by_show[show_id][season][(ep, end)] = watched_at

for sid in sorted(all_show_ids, key=int):
    meta = tmdb_get(f"/tv/{sid}")
    rating = show_ratings.get(sid)
    title = (meta and meta.get("name")) or show_titles.get(sid) \
        or (rating and rating[2]) or (watchlist_shows.get(sid) or ("", ""))[1]
    show_image = img(meta and meta.get("poster_path"))
    seasons = eps_by_show.get(sid, {})

    # A season only counts as completed against its FULL episode count
    # (yamtrack keeps airing seasons "In progress" while unreleased
    # episodes remain, even when the user is caught up).
    episode_counts = {
        season_meta["season_number"]: season_meta.get("episode_count") or 0
        for season_meta in (meta or {}).get("seasons", [])
    }
    last_aired = (meta or {}).get("last_episode_to_air") or {}
    has_upcoming = bool((meta or {}).get("next_episode_to_air"))

    all_watched = []
    season_status = {}
    for season, watches in seasons.items():
        watched_eps = {ep for (ep, _end) in watches}
        count = episode_counts.get(season)
        done = count is not None and count > 0 and len(watched_eps) >= count
        if done and has_upcoming and season == last_aired.get("season_number"):
            done = False
        season_status[season] = "Completed" if done else "In progress"
        all_watched.extend(
            watched_at or end
            for (ep, end), watched_at in watches.items()
        )

    if seasons:
        last_season = last_aired.get("season_number")
        regular = [s for s in seasons if s != 0]
        show_done = (
            not has_upcoming
            and last_season is not None
            and season_status.get(last_season) == "Completed"
            and all(season_status[s] == "Completed" for s in regular)
        )
        status = "Completed" if show_done else "In progress"
    elif rating:
        status = "Completed"
    else:
        status = "Planning"

    watch_ts = sorted(ts(w) if w.endswith("Z") else w for w in all_watched)
    progressed = watch_ts[-1] if watch_ts else ts(
        (rating and rating[1]) or (watchlist_shows.get(sid) or ("",))[0])

    yt_row = yt_tv_rows.get(sid)
    tv_rows.append(base_row(
        media_id=sid, source="tmdb", media_type="tv",
        title=title, image=show_image,
        score=(rating and rating[0]) or (yt_row or {}).get("score", ""),
        status=status,
        notes=(yt_row or {}).get("notes", ""),
        progressed_at=progressed,
    ))

    for season in sorted(seasons):
        smeta = tmdb_get(f"/tv/{sid}/season/{season}")
        season_image = img(smeta and smeta.get("poster_path"), show_image)
        season_watch_ts = sorted(
            (watched_at and ts(watched_at)) or end
            for (ep, end), watched_at in seasons[season].items()
        )
        yt_srow = yt_season_rows.get((sid, season))
        season_rows.append(base_row(
            media_id=sid, source="tmdb", media_type="season",
            title=title, image=season_image, season_number=season,
            score=season_ratings.get((sid, season), "")
            or (yt_srow or {}).get("score", ""),
            status=season_status[season],
            notes=(yt_srow or {}).get("notes", ""),
            progressed_at=season_watch_ts[-1],
        ))

        stills = {}
        for ep_meta in (smeta or {}).get("episodes", []):
            stills[ep_meta["episode_number"]] = img(ep_meta.get("still_path"))

        for (ep, end), watched_at in sorted(seasons[season].items(),
                                            key=lambda kv: (kv[0][1], kv[0][0])):
            end_ts = (watched_at and ts(watched_at)) or end
            episode_rows.append(base_row(
                media_id=sid, source="tmdb", media_type="episode",
                title=title, image=stills.get(ep, IMG_NONE),
                season_number=season, episode_number=ep,
                end_date=end_ts, progressed_at=end_ts,
            ))

# rated/watchlist seasons without watches for shows we emit: add season rows
for (sid, season), rating in season_ratings.items():
    if sid in eps_by_show and season in eps_by_show[sid]:
        continue
    smeta = tmdb_get(f"/tv/{sid}/season/{season}")
    show_meta = tmdb_get(f"/tv/{sid}")
    season_rows.append(base_row(
        media_id=sid, source="tmdb", media_type="season",
        title=(show_meta and show_meta.get("name")) or show_titles.get(sid, ""),
        image=img(smeta and smeta.get("poster_path"),
                  img(show_meta and show_meta.get("poster_path"))),
        season_number=season, score=rating, status="Completed",
    ))

# yamtrack-only seasons not covered above (e.g. tracked but 0 episodes watched)
emitted_seasons = {(r["media_id"], r["season_number"]) for r in season_rows}
for (sid, season), row in yt_season_rows.items():
    if (sid, season) not in emitted_seasons:
        season_rows.append(row)

with OUT_CSV.open("w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=COLUMNS, quoting=csv.QUOTE_ALL)
    writer.writeheader()
    for row in movie_rows + tv_rows + season_rows + episode_rows:
        writer.writerow({c: row.get(c, "") for c in COLUMNS})

print(f"\nwrote {OUT_CSV}")
print(f"movies: {len(movie_rows)}  tv: {len(tv_rows)}  "
      f"seasons: {len(season_rows)}  episodes: {len(episode_rows)}")
print(f"episode ratings skipped (no score field on Episode): "
      f"{skipped_episode_ratings}")
if warnings:
    print(f"\nwarnings ({len(warnings)}):")
    for w in dict.fromkeys(warnings):
        print(" -", w)