Playlist sharing with write permission

Code:
#!/usr/bin/env python3

"""

Script to manage Jellyfin playlist write permissions.



  • --list            : group by owner and print ASCII subtables

  • [user] [playlist] : grant or revoke edit permission (use --revoke)

  • --delay N        : seconds to wait after refresh before listing (default 5)

"""



import os

import sqlite3

import argparse

import getpass

import requests

import xml.etree.ElementTree as ET

import time



# --- 1) CONFIGURATION --------------------------------------------------



JELLYFIN_BASE = "http://localhost:8096"

PLAYLISTS_DIR = "/var/lib/jellyfin/data/playlists"

LIBRARY_DB    = "/var/lib/jellyfin/data/library.db"

USERS_DB      = "/var/lib/jellyfin/data/jellyfin.db"



CLIENT_HDR = (

    'Mediabrowser Client="script", '

    'Device="script", DeviceId="script", Version="1.0.0"'

)



# --- 2) AUTH UTILITIES -------------------------------------------------



def normalize_id(raw_id):

    return raw_id.replace("-", "").lower()



def authenticate(username):

    """

    Prompt for a password and authenticate to Jellyfin by username.

    Returns an access token.

    """

    pw = getpass.getpass(f"Password for '{username}': ")

    url = f"{JELLYFIN_BASE}/Users/AuthenticateByName"

    headers = {

        "accept": "application/json",

        "Content-Type": "application/json",

        "x-emby-authorization": CLIENT_HDR

    }

    resp = requests.post(url, json={"Username": username, "Pw": pw}, headers=headers)

    resp.raise_for_status()

    data = resp.json()

    return data.get("Data", data).get("AccessToken")



# --- 3) SINGLE-PLAYLIST REFRESH ----------------------------------------



def refresh_playlist(playlist_id, token):

    """

    Force reload of the specified playlist.

    Requires an Admin access token.

    """

    url = f"{JELLYFIN_BASE}/Items/{playlist_id}/Refresh"

    headers = {

        "accept": "application/json",

        "Content-Type": "application/json",

        "X-MediaBrowser-Token": token

    }

    resp = requests.post(url, headers=headers)

    try:

        resp.raise_for_status()

        print(f"✔ Playlist {playlist_id} successfully refreshed (Admin).")

    except Exception as e:

        print(f"✗ Error refreshing playlist {playlist_id}: {e}")



# --- 4) DATABASE AND FILESYSTEM LOADERS ---------------------------------



def load_users():

    conn = sqlite3.connect(USERS_DB)

    cur  = conn.cursor()

    cur.execute("SELECT Username, Id FROM Users")

    mapping = { normalize_id(uid): name for name, uid in cur.fetchall() }

    conn.close()

    return mapping



def load_playlists():

    conn = sqlite3.connect(LIBRARY_DB)

    cur  = conn.cursor()

    cur.execute("""

        SELECT Name, PresentationUniqueKey

        FROM TypedBaseItems

        WHERE Type='MediaBrowser.Controller.Playlists.Playlist'

    """)

    mapping = { name: pid.lower() for name, pid in cur.fetchall() if pid }

    conn.close()

    return mapping



def load_owners():

    owners = {}

    for d in os.listdir(PLAYLISTS_DIR):

        xml_path = os.path.join(PLAYLISTS_DIR, d, "playlist.xml")

        if not os.path.isfile(xml_path):

            continue

        tree = ET.parse(xml_path)

        owner_id = tree.findtext("OwnerUserId")

        if owner_id:

            owners[d] = normalize_id(owner_id)

    return owners



# --- 5) LIST PERMISSIONS GROUPED BY OWNER -----------------------------



def list_permissions():

    """

    Group playlists by owner and print ASCII subtables.

    No authentication needed since it reads from disk/DB only.

    """

    users    = load_users()

    playlists = load_playlists()

    owners    = load_owners()



    print("\n=== Current Permissions ===\n")

    by_owner = {}

    for pname in playlists:

        owner_norm = owners.get(pname)

        if not owner_norm:

            continue

        by_owner.setdefault(owner_norm, []).append(pname)



    for owner_norm, pnames in by_owner.items():

        owner_name = users.get(owner_norm, "<unknown>")

        print(f"Owner: {owner_name} ({owner_norm})")

        rows = []

        for pname in sorted(pnames):

            xml_path = os.path.join(PLAYLISTS_DIR, pname, "playlist.xml")

            editors  = []

            if os.path.isfile(xml_path):

                root = ET.parse(xml_path).getroot()

                for share in root.findall("Shares/Share"):

                    if share.findtext("CanEdit") == "true":

                        uid_norm = normalize_id(share.findtext("UserId") or "")

                        if uid_norm != owner_norm:

                            editors.append(users.get(uid_norm, uid_norm))

            rows.append([pname, ", ".join(editors) or "-"])



        headers    = ["Playlist", "CanEdit by"]

        col_widths = [max(len(str(c)) for c in col) for col in zip(headers, *rows)]

        sep = "+" + "+".join("-"*(w+2) for w in col_widths) + "+"



        print(sep)

        print("| " + " | ".join(h.ljust(w) for h, w in zip(headers, col_widths)) + " |")

        print(sep)

        for row in rows:

            print("| " + " | ".join(str(c).ljust(w) for c, w in zip(row, col_widths)) + " |")

        print(sep + "\n")



# --- 6) MODIFY PERMISSION + AUTO-LIST ---------------------------------



def modify_permission(target_user, playlist, revoke=False, delay=5):

    users    = load_users()

    playlists = load_playlists()

    owners    = load_owners()



    if playlist not in playlists:

        raise SystemExit(f"Playlist '{playlist}' not found.")

    if target_user not in users.values():

        raise SystemExit(f"User '{target_user}' not found.")



    pid  = playlists[playlist]

    tuid  = next(u for u, n in users.items() if n == target_user)

    onorm = owners.get(playlist)

    if not onorm:

        raise SystemExit("Cannot determine playlist owner.")



    # 1) Owner authentication and grant/revoke

    owner_name  = users[onorm]

    print(f"Playlist '{playlist}' ({pid}) is owned by {owner_name}.")

    token_owner = authenticate(owner_name)



    headers = {

        "accept":              "application/json",

        "Content-Type":        "application/json",

        "X-MediaBrowser-Token": token_owner

    }

    url  = f"{JELLYFIN_BASE}/Playlists/{pid}/Users/{tuid}"

    body = {"UserId": tuid, "CanEdit": not revoke}

    resp = requests.post(url, headers=headers, json=body)

    action = "revoked" if revoke else "granted"

    if resp.status_code in (200, 204):

        print(f"Edit permission {action} for '{target_user}'.")

    else:

        print(f"Error {resp.status_code}: {resp.text}")

        return



    # 2) Admin authentication and refresh single playlist

    print("Admin authentication to refresh playlist…")

    admin_user  = input("Admin username: ")

    admin_token = authenticate(admin_user)

    print(f"Triggering refresh for playlist '{playlist}' ({pid})…")

    refresh_playlist(pid, admin_token)



    # 2.b) Delay to allow DB to synchronize

    if delay > 0:

        print(f"Waiting {delay} seconds to allow DB update…")

        time.sleep(delay)



    # 3) Immediately show updated permissions

    list_permissions()



# --- 7) ENTRYPOINT AND ARG PARSING -------------------------------------



def main():

    parser = argparse.ArgumentParser(

        description="Manage Jellyfin playlist write permissions"

    )

    parser.add_argument("user", nargs="?", help="Target username")

    parser.add_argument("playlist", nargs="?", help="Playlist name")

    parser.add_argument("--revoke", action="store_true", help="Revoke edit permission")

    parser.add_argument("--list", action="store_true",

                        help="List permissions grouped by owner")

    parser.add_argument("--delay", type=int, default=5,

                        help="Seconds to wait after refresh (default: 5)")

    args = parser.parse_args()



    if args.list:

        list_permissions()

    elif args.user and args.playlist:

        modify_permission(args.user, args.playlist,

                          revoke=args.revoke,

                          delay=args.delay)

    else:

        parser.print_help()



if __name__ == "__main__":

    main()