jf2yamtrack.py

jf2yamtrack.py

import http.client
import json
import argparse
import csv
import datetime
from urllib.parse import urlencode
from dateutil import parser as dateparser

parser = argparse.ArgumentParser()
parser.add_argument("--jellyfin", default="http://localhost:8096", help="Base URL for Jellyfin server, e.g. http://localhost:8096")
parser.add_argument("--user", required=True, help="Jellyfin username (case sensitive)")
parser.add_argument("--jfapi", required=True, help="Jellyfin API key")
parser.add_argument("--yamtrack", default="http://localhost:8000", help="Base URL for Yamtrack server, e.g. http://localhost:8000")
parser.add_argument("--yapi", help="Yamtrack API key") # each user has a unique API key
parser.add_argument("--api", action="store_true", help="Use the API") # use the API
parser.add_argument("--csv", action="store_true", help="Dump to CSV") # dump to CSV
parser.add_argument("--dry-run", action="store_true", help="Perform a dry run")

args = parser.parse_args()

class HTTPSession():
    """Basic requests session-like wrapper around http.client to support connection reuse"""
    def __init__(self, host):
        timeout = 5
        self.host = host
        if host.startswith("http://"):
            host = host[len("http://"):].rstrip("/")
            self.conn = http.client.HTTPConnection(host, timeout=timeout)
        elif host.startswith("https://"):
            host = host[len("https://"):].rstrip("/")
            self.conn = http.client.HTTPSConnection(host, timeout=timeout)
        else:
            raise ValueError("Host must start with http:// or https://")
        
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.conn.close()

    def request(self, method, path, param=None, body=None, headers=None):
        param = param or {}
        headers = headers or {}
        if param:
            path += '?' + urlencode(param)
        self.conn.request(method, path, body=body, headers=headers)
        response = self.conn.getresponse()
        if response.status >= 400: # raise for status
            raise Exception(f"HTTP {response.status} {response.reason} for {method} {path}: {response.read().decode('utf-8')}")
        return response.read().decode('utf-8')

def authenticated_jellyfin_request(session, apikey, method, path, params, body=None):
    headers = {"Authorization": f"MediaBrowser Token={apikey}"}
    return session.request(method, path, param=params, body=body, headers=headers)

def authenticated_yamtrack_request(session, apikey, method, path, params, body=None):
    headers = {"X-API-Key": apikey}
    headers["Content-Type"] = "application/json" # for POST and PATCH requests
    return session.request(method, path, param=params, body=body, headers=headers)

def get_jellyfin_user_id(session, username, apikey):
    response = authenticated_jellyfin_request(session, apikey, "GET", "/Users", {}, None)
    users = json.loads(response)
    for user in users:
        if user["Name"] == username:
            return user["Id"]
    raise ValueError(f"User {username} not found")

def get_user_watched_from_jellyfin(session, user_id, apikey):
    """
    {{baseUrl}}/Users/:userId/Items
    ?IncludeItemTypes=Movie&Recursive=true&Filters=IsPlayed&limit=1000000&Fields=ProviderIds,OriginalTitle
    - ProviderIds: { "Tmdb": "12345" }
    - OriginalTitle: get the English title for non-English libraries, only used for easy understanding of logs
    """
    params = {"IncludeItemTypes": "Movie", "Recursive": "true", 
            "Filters": "IsPlayed", "limit": 1000000, "Fields":"ProviderIds,OriginalTitle"}
    response = authenticated_jellyfin_request(session, apikey, "GET", f"/Users/{user_id}/Items", params)
    return response

series_tmdb_mapping = {} # global lookup for {series_id: [tmdb_id,status]}
season_tmdb_mapping = {} 

def lookup_item(item_id, session, user_id, apikey):
    """Lookup the item details, for seasons, get the TMDB ID, and get UserData.PlayedPercentage"""
    params = {"Fields": "ProviderIds"}
    response = authenticated_jellyfin_request(session, apikey, "GET", f"/Users/{user_id}/Items/{item_id}", params, None)
    data = json.loads(response)
    return data

def get_user_watched_episodes_from_jellyfin(session, user_id, apikey):
    """
    {{baseUrl}}/Users/:userId/Items
    ?IncludeItemTypes=Episode&Recursive=true&Filters=isPlayed&limit=1000000&Fields=Id,OriginalTitle,ProviderIds
    """
    params = {"IncludeItemTypes": "Episode", "Recursive": "true", "Filters": "IsPlayed", "limit": 1000000, "Fields": "Id,OriginalTitle,ProviderIds"}
    response = authenticated_jellyfin_request(session, apikey, "GET", f"/Users/{user_id}/Items", params)
    return response

def build_dict_helper(media_id, media_type, season_number="", episode_number="", status="", end_date=""):
    return {
        "media_id": media_id,
        "source": "tmdb",
        "media_type": media_type,
        "title": "",
        "image": "",
        "season_number": season_number,
        "episode_number": episode_number,
        "score": "",
        "status": status,
        "notes": "",
        "start_date": "",
        "end_date": end_date,
        "progress": "",
        "created_at": "",
        "progressed_at": ""
    }

def build_jellyfin_series_dict(jellyfin_episodes, session, user_id, apikey):
    """Build a dictionary for Yamtrack import from the watched episodes in Jellyfin, with TMDB lookup for series IDs
    For each episode
        Get series ID and season ID
        Get TMDB ID based on series ID and caches it
        Get PlayedPercentage for series and season, and season number for episode, and cache it
        Build the list for series, seasons, and episodes
    """
    if type(jellyfin_episodes) == str:
        try:
            jellyfin_episodes = json.loads(jellyfin_episodes)
        except json.JSONDecodeError:
            raise ValueError("Invalid JSON data for Jellyfin watched episodes")
    
    series = []
    seasons = []
    episodes = []
    episode_items = jellyfin_episodes.get("Items", [])
    for item in episode_items:
        series_id = item.get("SeriesId")
        season_id = item.get("SeasonId")
        if series_id not in series_tmdb_mapping:
            series_data = lookup_item(series_id, session, user_id, apikey)
            data = [series_data.get("ProviderIds", {}).get("Tmdb"), series_data.get("UserData", {}).get("PlayedPercentage")]
            series_tmdb_mapping[series_id] = data
        if season_id not in season_tmdb_mapping:
            season_data = lookup_item(season_id, session, user_id, apikey)
            season_tmdb_data = series_tmdb_mapping[series_id][0] # get the TMDB ID from the series mapping
            data = [season_tmdb_data, season_data.get("UserData", {}).get("PlayedPercentage"), season_data.get("IndexNumber")]
            season_tmdb_mapping[season_id] = data
        tmdb_id, _ = series_tmdb_mapping[series_id]
        season_number = item.get("ParentIndexNumber")
        episode_number = item.get("IndexNumber")
        end_date = item.get("UserData", {}).get("LastPlayedDate")
        if not end_date:
            print(f"SKIP   {item.get('Name', 'Unknown title')} - no LastPlayedDate")
            continue
        episodes.append(build_dict_helper(tmdb_id, "episode", season_number, episode_number, "", end_date))
    
    # YamTrack require "In progress" in lowercase 
    for _,(tmdb_id, played_percentage) in series_tmdb_mapping.items():
        status = "Completed" if played_percentage == 100 else "In progress"
        series.append(build_dict_helper(tmdb_id, "tv", "", "", status))
    for _,(tmdb_id, played_percentage, season_number) in season_tmdb_mapping.items():
        status = "Completed" if played_percentage == 100 else "In progress"
        seasons.append(build_dict_helper(tmdb_id, "season", season_number, "", status))
    sort_order = {"tv": 0, "season": 1, "episode": 2}
    _all = series + seasons + episodes
    _all.sort(key=lambda x: sort_order.get(x["media_type"], 3)) # tv comes before seasons which comes before episodes
    return _all


def get_yamtrack_movies_all(session, apikey) -> dict:
    """
    Yamtrack maximum limit is 200 per page
    Make initial request -> get .pagination.next until it's null
    """
    movies = {} # not a YamTrack API response, custom dict for lookup by TMDB ID
    """
    {"tmdb_id": {actual_yamtrack_results}, ...}
    """
    next_ = '/api/v1/media/movie/?limit=200&offset=0'
    while next_ is not None:
        response = authenticated_yamtrack_request(session, apikey, "GET", next_, {})
        data = json.loads(response)
        for result in data.get("results", []):
            movies[result['item']["media_id"]] = result
        next_ = data.get("pagination", {}).get("next")
        next_ = "/api/v1" + next_.split("/api/v1")[1] if next_ else None
    return movies

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

The minimum requirement for CSV import (for movies)
Note: the row must have all fields present even if only few are required
- `media_id`: the TMDB ID
	- this can be either quoted string or integer
- `source`: tmdb
- `media_type`: movie
- status: Completed (or Planned, In progress)
- `end_date`: when date when the movie was watched
	- `2019-12-10 23:10:00-08:00` also include timezone
"""

def dump_jellyfin_watched_to_csv(jellyfin_data, jellyfin_series_data=None):
    """Create the bare minimum CSV for Yamtrack import, Yamtrack handles the rest"""
    if type(jellyfin_data) == str:
        try:
            jellyfin_data = json.loads(jellyfin_data)
        except json.JSONDecodeError:
            raise ValueError("Invalid JSON data for Jellyfin watched items")

    with open("yamtrack_import.csv", "w", newline="", encoding="utf-8") as csvfile:
        fieldnames = ["media_id","source","media_type","title","image","season_number","episode_number","score","status","notes","start_date","end_date","progress","created_at","progressed_at"]
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        written_movies = 0
        written_series = 0
        written_seasons = 0
        written_episodes = 0
        skipped = 0
        for item in jellyfin_data.get("Items", []):
            media_id = item.get("ProviderIds", {}).get("Tmdb")
            if not media_id:
                skipped += 1
                print(f"SKIP   {item.get('Name', 'Unknown title')} - no TMDB ID")
                continue
            end_date = item.get("UserData", {}).get("LastPlayedDate")
            if not end_date:
                skipped += 1
                print(f"SKIP   {item.get('Name', 'Unknown title')} - no LastPlayedDate")
                continue
            writer.writerow(build_dict_helper(media_id, "movie", "", "", "Completed", end_date))
            written_movies += 1
        if jellyfin_series_data:
            for item in jellyfin_series_data:
                writer.writerow(item)
                if item["media_type"] == "tv": written_series += 1
                elif item["media_type"] == "season": written_seasons += 1
                elif item["media_type"] == "episode": written_episodes += 1
        
    print(f"CSV    yamtrack_import.csv - wrote {written_movies} movies, {written_series} series, {written_seasons} seasons, {written_episodes} episodes, skipped {skipped}")

"""
Manual Import via API
Using  `{{baseUrl}}/api/v1/media/:media_type/`
- `media_type`: movie
Minimum required payload
```json
{
  "media_id": "4108",
  "source": "tmdb",
  "media_type": "movie",
  "title": "TBD",
  "status": 3,
  "progress": 1,
  "end_date": "2022-08-27 17:32:33-08:00"
}
```
- this can be run repeatedly
- but it will create a new watched entry
"""


def post_to_yamtrack(session, apikey, media_id, end_date):
    """
    Modify server data, create new record, won't occur if --dry-run is specified
    """
    data = {
        "media_id": media_id,
        "source": "tmdb",
        "media_type": "movie",
        "title": "TBD",
        "status": 3, # Completed
        "progress": 1,
        "end_date": end_date
    }
    body = json.dumps(data)
    response = authenticated_yamtrack_request(session, apikey, "POST", "/api/v1/media/movie/", {}, body)
    return response

"""
Partial Update of a media item
```http
{{baseUrl}}/api/v1/media/:media_type/:source/:media_id/
```
- media_type: movie, source: tmdb, media_id: TMDB ID
Minimum required payload
```json
{
  "status": 3,
  "end_date": "2025-08-23 16:10:00-08:00"
}
```
"""

def patch_to_yamtrack(session, apikey, media_id, end_date):
    """
    Modify server data, update end_date only, won't occur if --dry-run is specified
    """
    data = {
        "status": 3, # Completed
        "end_date": end_date
    }
    body = json.dumps(data)
    response = authenticated_yamtrack_request(session, apikey, "PATCH", f"/api/v1/media/movie/tmdb/{media_id}/", {}, body)
    return response

def main():
    with HTTPSession(args.jellyfin) as jellyfin_session, HTTPSession(args.yamtrack) as yamtrack_session: 
        api_disabled = not args.api
        if args.api and not args.yapi:
            raise ValueError("--yapi is required when --api is specified")
        user_id = get_jellyfin_user_id(jellyfin_session, args.user, args.jfapi)
        jellyfin_watched = get_user_watched_from_jellyfin(jellyfin_session, user_id, args.jfapi)
        jellyfin_series_watched = get_user_watched_episodes_from_jellyfin(jellyfin_session, user_id, args.jfapi)
        jellyfin_series_data = build_jellyfin_series_dict(jellyfin_series_watched, jellyfin_session, user_id, args.jfapi)

        if args.csv:
            dump_jellyfin_watched_to_csv(jellyfin_watched, jellyfin_series_data)

        if not api_disabled:
            try:
                yamtrack_session.request("GET", "/api/docs/") # check if not 404
            except Exception as exc:
                print("ERROR  Yamtrack API check failed; use the :api Docker image")
                api_disabled = True
            if api_disabled:
                return
            yamtrack_movies = get_yamtrack_movies_all(yamtrack_session, args.yapi)
            print(len(yamtrack_movies.keys()), "movies found in Yamtrack for comparison")
            jellyfin_watched = json.loads(jellyfin_watched) if type(jellyfin_watched) == str else jellyfin_watched
            created = 0
            updated = 0
            skipped = 0
            for item in jellyfin_watched.get("Items", []):
                media_id = item.get("ProviderIds", {}).get("Tmdb")
                title = item.get("OriginalTitle") or item.get("Name", "Unknown title")
                if not media_id:
                    skipped += 1
                    print(f"SKIP   {title} - no TMDB ID")
                    continue
                end_date = item.get("UserData", {}).get("LastPlayedDate")
                if not end_date:
                    skipped += 1
                    print(f"SKIP   {title} ({media_id}) - no LastPlayedDate")
                    continue
                jellyfin_date = dateparser.isoparse(end_date) # isoparse needed for < Python 3.11
                if media_id in yamtrack_movies:
                    yamtrack_end_date = yamtrack_movies[media_id].get("end_date")
                    if not yamtrack_end_date:
                        yamtrack_date = None
                    else:
                        yamtrack_date = dateparser.isoparse(yamtrack_end_date)
                    if yamtrack_date and abs(jellyfin_date - yamtrack_date)\
                        < datetime.timedelta(hours=2): # if the dates are within 2 hours, consider them the same
                        skipped += 1
                        print(f"SKIP   {title} ({media_id}) - already watched near {jellyfin_date.astimezone()}")
                        continue
                    previous = yamtrack_date.astimezone() if yamtrack_date else "empty"
                    print(f"UPDATE {title} ({media_id}) - {previous} -> {jellyfin_date.astimezone()}")
                    if not args.dry_run:
                        patch_to_yamtrack(yamtrack_session, args.yapi, media_id, end_date)
                    updated += 1
                else:
                    print(f"CREATE {title} ({media_id}) - watched {jellyfin_date.astimezone()}")
                    if not args.dry_run:
                        post_to_yamtrack(yamtrack_session, args.yapi, media_id, end_date)
                    created += 1
            mode = "dry run" if args.dry_run else "applied"
            print(f"SUMMARY {mode}: created {created}, updated {updated}, skipped {skipped}")

if __name__ == "__main__":
    main()