Import MovieLens Ratings to Trakt.tv

README.md

Script for importing ratings and watched movies from MovieLens to Trakt.

Steps for running the script:

1. Download movielens-ratings.csv and movielens-logs.csv from https://movielens.org/profile/settings/import-export by clicking "export ratings" and "export activity logs"
2. Change the RATINGS_CSV and LOGS_CSV variables in secrets.py to point to the paths of the downloaded files
3. Change the ACCESS_TOKEN and CLIENT_ID variables to your Trakt API access token and client ID (see steps below)
4. Run the Python script

Steps to get Trakt API access token and client ID:

1. Create a new app trakt.tv/oauth/applications/new
2. Name it and set urn:ietf:wg:oauth:2.0:oob as the redirect uri.
3. Get your client_id, and go here, then either click on "Switch to Console" or "Try". Select body and replace the string here with your own client_id, send the query with "Call Resource"
4. Go to trakt.tv/activate and use the user_code from the 200 response from the previous query
5. Go to trakt.docs.apiary.io/#reference/authentication-devices/get-token/poll-for-the-access_token?console=1 and use the device_code from the previous query alongside the client_id and client_secret from your app to finally fetch the access_token

(Source: josh/imdb-trakt-sync#8 (comment))

License: MIT

import_movielens_to_trakt.py

"""Script for importing ratings and watched movies from MovieLens to Trakt."""

import csv
import datetime
import json
import re
import urllib.parse
import urllib.request

from secrets import ACCESS_TOKEN, CLIENT_ID, LOGS_CSV, RATINGS_CSV


def get_movies_to_rate():
    movies = {}

    with open(RATINGS_CSV, newline="") as f:
        reader = csv.reader(f)
        next(reader)
        for row in reader:
            movies[row[0]] = {
                "title": row[5],
                "ids": {"imdb": row[1], "tmdb": row[2]},
                "rating": int(float(row[3]) * 2),
            }

    with open(LOGS_CSV, newline="") as f:
        reader = csv.reader(f)
        next(reader)
        for row in reader:
            if row[2] == "rating":
                log_json = json.loads(row[3])
                if log_json["action"] == "ADD":
                    movie_id = str(log_json["movieId"])
                    if movie_id in movies:
                        rated_at = datetime.datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S.%f")
                        movies[movie_id]["rated_at"] = rated_at.isoformat() + ".000Z"

    return {
        "movies": list(movies.values()),
    }


def get_movies_to_mark_watched():
    body = get_movies_to_rate()
    for movie in body["movies"]:
        movie["watched_at"] = movie["rated_at"]
        del movie["rated_at"]
        del movie["rating"]
    return body


def get_headers():
    return {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "trakt-api-version": "2",
        "trakt-api-key": CLIENT_ID,
    }


def mark_movies_rated():
    print("Marking movies rated...")
    body = json.dumps(get_movies_to_rate())
    headers = get_headers()
    req = urllib.request.Request(
        "https://api.trakt.tv/sync/ratings", data=body.encode("utf-8"), headers=headers
    )
    response = urllib.request.urlopen(req)
    print(response.read())


def mark_movies_watched():
    print("Marking movies watched...")
    body = json.dumps(get_movies_to_mark_watched())
    headers = get_headers()
    req = urllib.request.Request(
        "https://api.trakt.tv/sync/history", data=body.encode("utf-8"), headers=headers
    )
    response = urllib.request.urlopen(req)
    print(response.read())


if __name__ == "__main__":
    mark_movies_rated()
    mark_movies_watched()

secrets.py

ACCESS_TOKEN = ""
CLIENT_ID = ""

RATINGS_CSV = "/path/to/movielens-ratings.csv"
LOGS_CSV = "/path/to/movielens-logs.csv"