A simple web UI for downloading media from Plex without the Plex Pass.
from datetime import timedelta

import asyncio
import humanize
from fastapi import Request
from fastapi.responses import RedirectResponse, FileResponse
from nicegui import ui, app
from plexapi.collection import *
from plexapi.video import *
from plexapi.media import MediaPart, Media

from config import config
from login import check_login, logout
from plex import get_server, get_self
from locales import _

from common import io_bound


@app.middleware("http")
async def check_auth(request: Request, call_next):
    # https://github.com/zauberzeug/nicegui/blob/main/examples/authentication/main.py
    if not await check_login():
        if not request.url.path.startswith("/_nicegui") and request.url.path != "/login":
            app.storage.user['referrer_path'] = request.url.path
            return RedirectResponse('/login')
    return await call_next(request)


# todo: find a way
# class CustomGZipMiddleware(GZipMiddleware):
#     def __init__(
#         self, *args, **kwargs
#     ) -> None:
#         super().__init__(*args, **kwargs)

#     async def __call__(self, scope, receive, send) -> None:
#         if scope["type"] == "http":
#             headers = Headers(scope=scope)
#             if "gzip" in headers.get("Accept-Encoding", ""):
#                 responder = GZipResponder(
#                     self.app, self.minimum_size, compresslevel=self.compresslevel
#                 )
#                 await responder(scope, receive, send)
#                 return
#         await self.app(scope, receive, send)

del app.user_middleware[-1]  # gzip removes content-length


async def header():
    """
    Displays thecommon page header
    """
    user = await get_self()

    def logout_handler():
        logout()
        ui.open("/login")

    with ui.row():
        ui.button(_("home"), on_click=lambda: ui.open("/"))
        ui.button(_("logout"), on_click=logout_handler)
        ui.label(_("user", user=user.email))


@app.get("/download/{media}/{index}")
async def download(media: int, index: int):
    """
    Downloads the specified media part from Plex
    """
    part = (await get_server()).fetchItem(media).media[index].parts[0]  # is there ever more than one part per media?
    filename = os.path.basename(part.file)
    return FileResponse(part.file, filename=filename, stat_result=os.stat(part.file))


@ui.page("/", title="PlexDLWeb")
async def index():
    ui.add_head_html("""
        <link rel="manifest" href="/plexdlweb.webmanifest">
        <style>
            /* for search results */
            .cursor-pointer-rec, .cursor-pointer-rec * {
                cursor: pointer;
            }
        </style>
    """)

    ui.add_body_html("""
        <script>
            if ('serviceWorker' in navigator) {
                navigator.serviceWorker.register('/sw.js');
            }
        </script>
    """)

    def fake_button_group():
        return ui.element("div").classes(add="q-btn-group row flex-nowrap")

    def fake_button(text):
        with ui.element("div").classes(add="q-btn q-btn-item no-outline") as e:
            with ui.element("span").classes(add="text-left items-center justify-center col row"):
                ui.html(text)
            return e

    def fake_button_label(text):
        return ui.label(text).classes(add="m-3 q-btn q-btn--flat p-0").style("min-height: 0")
    
    def duration_to_string(ms: int):
        dur = timedelta(milliseconds=ms)
        if dur > timedelta(hours=1):
            return f"{dur.seconds // 3600}h{(dur.seconds // 60) % 60}m"
        elif dur > timedelta(minutes=5):
            return f"{dur.seconds // 60}m"
        else:
            return f"{dur.seconds // 60}m{dur.seconds % 60}s"

    def merge(query, new):
        """
        Merge the previous query history with the new one
        Example: merge([A,B], [C,D]) == [A,B,C,D]
        Handles overlaps: merge([A,B,C], [B,D,E]) == [A,B,D,E]
        """
        try:
            idx = query.index(new[0])
            return query[:idx] + new
        except ValueError:
            return [*query, *new]

    @ui.refreshable
    def result_list(query, results):
        def refresh(query2, results2):
            # don't refresh if the parameters haven't changed
            if (query2, results2) == (query, results):
                return
            result_list.refresh(query2, results2)

        if not query:
            return
        if not results:
            ui.label(_("no_results"))
            return
        kinds = {
            Movie: (
                _("movie"),
                "bg-green-300",
                lambda m: (f"<span style='font-size: 70%'>{m.editionTitle}</span><br>" if m.editionTitle else "") + m.title,
                lambda m: print(list(m.iterParts()))
            ),
            Show: (
                _("show"),
                "bg-yellow-300",
                lambda s: s.title,
                lambda s: refresh([*query, s], s.seasons())
            ),
            Season: (
                _("season"),
                "bg-red-300",
                lambda s: f"{s.parentTitle} - {s.title}",
                lambda s: refresh([*query, s], s.episodes())
            ),
            Episode: (
                _("episode"),
                "bg-teal-300",
                lambda
                    e: f"<span style='font-size: 70%'>{e.grandparentTitle} - {e.parentTitle} - {_('episode')} {e.index}</span><br>{e.title}",
                lambda e: refresh(merge(query, [e.show(), sea := e.season()]), sea.episodes())
            ),
            Collection: (
                _("collection"),
                "bg-violet-300",
                lambda c: c.title,
                lambda c: refresh([*query, c], c.items())
            ),
            str: (
                _("search_noun"),
                "bg-blue-300",
                lambda s: s,
                lambda s: do_search(s, True)
            )
        }
        result_as_list = app.storage.user.get("result_as_list", False)
        with ui.row():
            def format_change(e):
                app.storage.user['result_as_list'] = e.value
                result_list.refresh(query, results)

            ui.label(_("display"))
            ui.toggle({False: _("grid"), True: _("list")}, value=result_as_list, on_change=format_change)

        with ui.row():
            ui.label(_("history"))

            def display_crumb(i, part):
                kind, color, namer, clicked = kinds[type(part)]

                async def handler():
                    del query[i:]
                    await clicked(part)

                with fake_button_group().on("click", handler).classes(add="cursor-pointer-rec"):
                    fake_button(kind).classes(add=color)
                    fake_button(namer(part))

            for i, part in enumerate(query):
                display_crumb(i, part)
                
        with ui.column() if result_as_list else ui.grid(columns=3):
            def display_result(r):
                opts = kinds.get(type(r), None)
                if opts is None:
                    return
                kind, color, namer, clicked = opts

                def dl_button():
                    if isinstance(r, Playable):
                        if len(r.media) > 1:
                            def handler():
                                def part_line(i, media: Media):
                                    part = media.parts[0]
                                    fake_button_label(duration_to_string(part.duration)).style("text-transform: none").classes(add="text-right")
                                    fake_button_label(f"{media.width}x{media.height}").style("text-transform: none")
                                    fake_button_label(humanize.naturalsize(part.size)).classes(add="text-right")
                                    ui.button(icon="download").props("flat").on("click.stop", lambda: ui.download(f"/download/{r.ratingKey}/{i}")).classes(add="px-3")
                                with ui.dialog() as dialog, ui.card():
                                    with ui.grid(columns=4):
                                        for i, media in enumerate(r.media):
                                            part_line(i, media)
                                dialog.open()
                        else:
                            def handler():
                                ui.download(f"/download/{r.ratingKey}/0")
                        part = r.media[0].parts[0]
                        fake_button_label(duration_to_string(part.duration)).classes(add="ml-auto self-center").style(
                            "text-transform: none; font-size: 90%")
                        fake_button_label(humanize.naturalsize(part.size)).classes(add="mx-0 self-center").style(
                            "font-size: 90%")
                        ui.button(icon="download").props("flat").on("click.stop", handler).classes(add="px-3")

                if result_as_list:
                    with fake_button_group().on("click", lambda: clicked(r)).classes(add="w-full cursor-pointer-rec"):
                        fake_button(kind).classes(add=color)
                        fake_button(namer(r))
                        dl_button()
                else:
                    with ui.card().tight().on("click", lambda: clicked(r)).classes(add="cursor-pointer-rec"):
                        with ui.card_section().classes(add=color).classes(add="p-0 row"):
                            fake_button_label(kind)
                            dl_button()
                        ui.image("/plex" + r.thumb)
                        with ui.card_section():
                            ui.html("<span style='font-size: 120%'>" + namer(r) + "</span>")

            for r in results:
                display_result(r)

    last_search = None

    server = await get_server()

    async def do_search(query, force=False):
        nonlocal last_search
        previous, last_search = last_search, query
        if not force and query == previous or len(query) < 3:
            return

        loading.set_visibility(True)
        results = await io_bound(server.search, query)
        async def all_editions(x):
            if isinstance(x, Movie):
                return [x, *(await io_bound(x.editions))]
            return [x]
        result_list.refresh([query], [item for editions in (await asyncio.gather(*[all_editions(res) for res in results])) for item in editions])
        loading.set_visibility(False)

    debounce = None

    def name_change(e):
        nonlocal debounce
        if debounce is not None:
            debounce.deactivate()
        debounce = ui.timer(0.3, lambda: do_search(e.value), once=True)

    await header()

    with ui.column():
        ui.input(label=_("search_verb"), placeholder=_("search_placeholder"), on_change=name_change)

    loading = ui.spinner(size="lg")
    loading.set_visibility(False)

    result_list([], [])


@app.get("/plexdlweb.webmanifest")
def manifest():
    return {
        "name": "PlexDLWeb",
        "short_name": "Volume",
        "start_url": "/",
        "display": "standalone",
        "background_color": "#282a2d",
        "theme_color": "#000",
        "icons": [
            {
                "src": "https://app.plex.tv/desktop/static/icon-ipad@2x.png",
                "sizes": "512x512",
                "type": "image/png"
            }
        ]
    }


@app.get("/sw.js")
def sw():
    return """
    self.addEventListener('install', function(event) {});
    self.addEventListener('fetch', function(event) {});
    self.addEventListener('activate', function(event) {});
    """


ui.run(host=config.host, port=config.port, show=False, storage_secret=config.secret)

__main__.py
zdimension/plexdlweb
import asyncio
from concurrent.futures import ThreadPoolExecutor
from functools import partial

executor = ThreadPoolExecutor()
# todo: use nicegui.run.io_bound when it's released
async def io_bound(func, *args, **kwargs):
    return await asyncio.get_running_loop().run_in_executor(executor, partial(func, *args, **kwargs))
common.py
zdimension/plexdlweb
from dataclasses import dataclass, field
from dataclasses_json import dataclass_json
import os
import uuid


@dataclass_json
@dataclass
class Config:
    server_url: str = "http://SERVER_HOST:32400"
    server_id: str = "SERVER_ID_HERE"
    lang: str = "en"
    host: str = "127.0.0.1"
    port: int = 8766
    uuid: str = field(default_factory=lambda: str(uuid.uuid4()))
    secret: str = field(default_factory=lambda: str(uuid.uuid4()))

def save_config():
    with open("config.json", "w") as config_file:
        config_file.write(config.to_json())

if os.path.exists("config.json"):
    try:
        with open("config.json", "r") as config_file:
            config = Config.from_json(config_file.read())
    except Exception as e:
        print("Error while reading config.json")
        raise
else:
    config = Config()
    print("No config.json found, creating it -- please fill it with your server details")
    save_config()
    exit()

save_config()

config.py
zdimension/plexdlweb
from config import config
import json


def get_data(lang: str) -> dict[str, str]:
    with open(f"locales/{lang}.json", "r", encoding="utf-8") as f:
        return json.load(f)


default_lang = get_data("en")
user_lang = get_data(config.lang)


def _(key: str, **kwargs) -> str:
    return user_lang.get(key, default_lang[key]).format(**kwargs)

locales.py
zdimension/plexdlweb
"""
This module uses code from https://github.com/Tautulli/Tautulli for the Plex OAuth login, licensed under GNU GPLv3.
"""
from typing import Optional

import asyncio
from nicegui import ui, app
from config import config
import json
from fastapi.responses import RedirectResponse
from plex import check_user_token, check_server_token, get_server_token
from locales import _


async def check_login() -> Optional[tuple[str, str]]:
    if (user_token := app.storage.user.get("user_token")) and (server_token := app.storage.user.get("server_token")) \
            and all(await asyncio.gather(check_user_token(user_token), check_server_token(server_token))):
        return user_token, server_token
    else:
        return None


@ui.page("/logout", title=_("logout"))
def logout():
    app.storage.user.pop("user_token")
    app.storage.user.pop("server_token")
    return RedirectResponse("/login")


@ui.page("/login", title=_("login"))
async def page_login():
    if await check_login():
        # already logged in, redirect to home
        return RedirectResponse("/")

    HEADERS = {
        "Accept": "application/json",
        "X-Plex-Product": "PlexDLWeb",
        "X-Plex-Version": "Plex OAuth",
        "X-Plex-Client-Identifier": str(config.uuid),
        "X-Plex-Platform": "PlexDLWeb",
        "X-Plex-Platform-Version": "0.1",
        "X-Plex-Model": "Plex OAuth",
        "X-Plex-Device": "PlexDLWeb",
        "X-Plex-Device-Name": "PlexDLWeb",
        "X-Plex-Device-Screen-Resolution": "1920x1080",
        "X-Plex-Language": "fr"
    }
    ui.add_body_html(f"""
        <script>
            const HEADERS = {json.dumps(HEADERS)};
        </script>
        <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.7.1/jquery.min.js" integrity="sha512-v2CJ7UaYy4JwqLDIrZUI/4hqeoQieOmAZNXBeQyjo21dadnwR+8ZaIJVT8EE2iyI61OV8e6M8PP2/4hpQINQ/g==" crossorigin="anonymous" referrerpolicy="no-referrer"></script>
    """)
    ui.add_body_html("""
        <script>
            getPlexOAuthPin = function () {
                var deferred = $.Deferred();

                $.ajax({
                    url: 'https://plex.tv/api/v2/pins?strong=true',
                    type: 'POST',
                    headers: HEADERS,
                    success: function(data) {
                        deferred.resolve({pin: data.id, code: data.code});
                    },
                    error: function() {
                        closePlexOAuthWindow();
                        deferred.reject();
                    }
                });
                return deferred;
            };

            let plex_oauth_window = null;

            function closePlexOAuthWindow() {
                if (plex_oauth_window) {
                    plex_oauth_window.close();
                }
            }

            function PopupCenter(url, title, w, h) {
                // Fixes dual-screen position                         Most browsers      Firefox
                var dualScreenLeft = window.screenLeft != undefined ? window.screenLeft : window.screenX;
                var dualScreenTop = window.screenTop != undefined ? window.screenTop : window.screenY;

                var width = window.innerWidth ? window.innerWidth : document.documentElement.clientWidth ? document.documentElement.clientWidth : screen.width;
                var height = window.innerHeight ? window.innerHeight : document.documentElement.clientHeight ? document.documentElement.clientHeight : screen.height;

                var left = ((width / 2) - (w / 2)) + dualScreenLeft;
                var top = ((height / 2) - (h / 2)) + dualScreenTop;
                var newWindow = window.open(url, title, 'scrollbars=yes, width=' + w + ', height=' + h + ', top=' + top + ', left=' + left);

                // Puts focus on the newWindow
                if (window.focus) {
                    newWindow.focus();
                }

                return newWindow;
            }

            const plex_oauth_loader = '<style>' +
                    '.login-loader-container {' +
                        'font-family: "Open Sans", Arial, sans-serif;' +
                        'position: absolute;' +
                        'top: 0;' +
                        'right: 0;' +
                        'bottom: 0;' +
                        'left: 0;' +
                    '}' +
                    '.login-loader-message {' +
                        'color: #282A2D;' +
                        'text-align: center;' +
                        'position: absolute;' +
                        'left: 50%;' +
                        'top: 25%;' +
                        'transform: translate(-50%, -50%);' +
                    '}' +
                    '.login-loader {' +
                        'border: 5px solid #ccc;' +
                        '-webkit-animation: spin 1s linear infinite;' +
                        'animation: spin 1s linear infinite;' +
                        'border-top: 5px solid #282A2D;' +
                        'border-radius: 50%;' +
                        'width: 50px;' +
                        'height: 50px;' +
                        'position: relative;' +
                        'left: calc(50% - 25px);' +
                    '}' +
                    '@keyframes spin {' +
                        '0% { transform: rotate(0deg); }' +
                        '100% { transform: rotate(360deg); }' +
                    '}' +
                '</style>' +
                '<div class="login-loader-container">' +
                    '<div class="login-loader-message">' +
                        '<div class="login-loader"></div>' +
                        '<br>' +
                        '""" + _("redirecting_to_plex") + """...' +
                    '</div>' +
                '</div>';

            function encodeData(data) {
                return Object.keys(data).map(function(key) {
                    return [key, data[key]].map(encodeURIComponent).join("=");
                }).join("&");
            }

            function PlexOAuth(success, error) {
                closePlexOAuthWindow();
                plex_oauth_window = PopupCenter('', 'Plex-OAuth', 600, 700);
                $(plex_oauth_window.document.body).html(plex_oauth_loader);

                getPlexOAuthPin().then(function (data) {
                    const pin = data.pin;
                    const code = data.code;

                    var oauth_params = {
                        'clientID': HEADERS['X-Plex-Client-Identifier'],
                        'context[device][product]': HEADERS['X-Plex-Product'],
                        'context[device][version]': HEADERS['X-Plex-Version'],
                        'context[device][platform]': HEADERS['X-Plex-Platform'],
                        'context[device][platformVersion]': HEADERS['X-Plex-Platform-Version'],
                        'context[device][device]': HEADERS['X-Plex-Device'],
                        'context[device][deviceName]': HEADERS['X-Plex-Device-Name'],
                        'context[device][model]': HEADERS['X-Plex-Model'],
                        'context[device][screenResolution]': HEADERS['X-Plex-Device-Screen-Resolution'],
                        'context[device][layout]': 'desktop',
                        'code': code
                    }

                    plex_oauth_window.location = 'https://app.plex.tv/auth/#!?' + encodeData(oauth_params);
                    polling = pin;

                    (function poll() {
                        $.ajax({
                            url: 'https://plex.tv/api/v2/pins/' + pin,
                            type: 'GET',
                            headers: HEADERS,
                            success: function (data) {
                                if (data.authToken){
                                    closePlexOAuthWindow();
                                    if (typeof success === "function") {
                                        success(data.authToken)
                                    }
                                }
                            },
                            error: function (jqXHR, textStatus, errorThrown) {
                                if (textStatus !== "timeout") {
                                    closePlexOAuthWindow();
                                    if (typeof error === "function") {
                                        error()
                                    }
                                }
                            },
                            complete: function () {
                                if (!plex_oauth_window.closed && polling === pin){
                                    setTimeout(function() {poll()}, 1000);
                                }
                            },
                            timeout: 10000
                        });
                    })();
                }, function () {
                    closePlexOAuthWindow();
                    if (typeof error === "function") {
                        error()
                    }
                });
            }

            function getToken() {
                return new Promise((resolve, reject) => {
                    PlexOAuth(function (token) {
                        resolve(token);
                    }, function () {
                        reject();
                    });
                });
            }
        </script>
    """)

    async def connect_plex():
        login_btn.text = _("logging_in")
        app.storage.user["user_token"] = await ui.run_javascript("return await getToken();", timeout=120)
        app.storage.user["server_token"] = await get_server_token(app.storage.user["user_token"])
        ui.open("/")

    with ui.column():
        login_btn = ui.button(_("login_with_plex"), on_click=connect_plex)

login.py
zdimension/plexdlweb
from nicegui import app
from plexapi.server import PlexServer
from plexapi.myplex import MyPlexAccount
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask
import httpx
from config import config
from common import io_bound

SERVER_ID = "1af7c05328fb1a2bc68ca4eb9ee9c4ac9dd90bca"


@app.on_event("startup")
async def startup():
    app.state.httpx_client = httpx.AsyncClient()


@app.on_event("shutdown")
async def shutdown():
    await app.state.httpx_client.aclose()


@app.get("/plex/{path:path}")
async def streaming(path: str) -> StreamingResponse:
    """
    Forwards a request to the Plex server
    """
    # https://stackoverflow.com/a/74556972/2196124
    client = app.state.httpx_client
    url = httpx.URL(config.server_url + "/" + path)
    req = httpx.Request("GET", url, headers={"X-Plex-Token": app.storage.user.get("server_token")})
    resp = await client.send(req, stream=True)
    return StreamingResponse(
        resp.aiter_raw(),
        status_code=resp.status_code,
        headers=resp.headers,
        background=BackgroundTask(resp.aclose)
    )


async def get_server():
    return await io_bound(PlexServer, config.server_url, app.storage.user.get("server_token"))


async def get_self():
    return await io_bound(MyPlexAccount, token=app.storage.user.get("user_token"))


async def get_server_token(user_token: str) -> str:
    acc = await io_bound(MyPlexAccount, token=user_token)
    srv = [r for r in await io_bound(acc.resources) if r.clientIdentifier == config.server_id][0]
    return srv.accessToken


async def check_user_token(token: str) -> bool:
    try:
        await io_bound(MyPlexAccount, token=token)
        return True
    except Exception as e:
        print(e)
        return False


async def check_server_token(token: str) -> bool:
    try:
        await io_bound(PlexServer, config.server_url, token=token)
        return True
    except Exception as e:
        print(e)
        return False

plex.py
zdimension/plexdlweb