jellyfin-daily-mix/main.py

from jellyfin_api_client.models import *
from jellyfin_api_client.api.items import get_items
from jellyfin_api_client.api.user import get_users
from jellyfin_api_client.api.playlists import create_playlist, update_playlist
from jellyfin_api_client.types import Response
from jellyfin_api_client import AuthenticatedClient

import os
from dotenv import load_dotenv
from pprint import pp

load_dotenv()

def log_response(response):
    request = response.request
    print(f"Response event hook: {request.method} {request.url} - Status {response.status_code}")
    if response.status_code >= 400:
        raise Exception(response.read())

server_url = os.getenv('SERVER_URL')
api_token = os.getenv('API_TOKEN')
device = os.getenv('HOSTNAME')
user_names = os.getenv('USER_NAMES').split()
playlist_name = os.getenv('PLAYLIST')

client = AuthenticatedClient(
    base_url=server_url,
    prefix="",
    token=f"MediaBrowser Client=\"jellyfin-client\", Device=\"{device}\", DeviceId=\"1234\", Version=\"1.0.0\", Token=\"{api_token}\"",
    httpx_args={"event_hooks": {"response": [log_response]}})


with client as client:
    users = get_users.sync(client=client)
    user_ids = [user.id for user in users if user.name in user_names]

    playlists = get_items.sync(
        client=client,
        recursive=True,
        include_item_types=[BaseItemKind.PLAYLIST],
        search_term=playlist_name
    )

    songs: list[BaseItemDto] = []
    for user_id in user_ids:
        result = get_items.sync(
            client=client,
            user_id=user_ids,
            recursive=True,
            include_item_types=[BaseItemKind.AUDIO],
            limit=420,
            sort_by=[ItemSortBy.PLAYCOUNT, ItemSortBy.RANDOM])
        songs.extend([item for item in result.items if not any(s.id == item.id for s in songs)])

    try:
        playlist_id = next(p.id for p in playlists.items if p.name == playlist_name)
        body = UpdatePlaylistDto(
            name=playlist_name,
            users=[PlaylistUserPermissions(user_id=id, can_edit=True) for id in user_ids],
            ids=[item.id for item in songs],
            is_public=True
        )
        update_playlist.sync(
            playlist_id=playlist_id,
            client=client,
            body=body
        )
    except StopIteration:
        create_playlist.sync(
            client=client,
            body=CreatePlaylistDto(
                name=playlist_name,
                user_id=user_ids[0],
                users=[PlaylistUserPermissions(user_id=id, can_edit=True) for id in user_ids],
                ids=[item.id for item in songs],
                is_public=True,
                media_type=CreatePlaylistDtoMediaType.AUDIO
            )
        )
    print()
    print(f"Successfully updated playlist '{playlist_name}' with {len(songs)} songs")