Flaro/Tools/JellyfinStub/server.py

"""Ein Jellyfin-Attrappenserver — gerade so viel, dass die App anmelden und
stöbern kann. Nur zum Ansehen der Oberfläche; kein echter Server wird berührt."""
import json, os, struct, zlib, hashlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

USER_ID = "user-1"
TOKEN = "stub-token"
SAMPLE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sample.mp4")
# Was an Wiedergabemeldungen hereinkam — damit sich prüfen lässt, ob die App
# den Fortschritt tatsächlich meldet.
REPORTS = []
# Mit `--unanalysed` melden die Quellen keine Spuren. Dann entscheidet die App
# auf Umrechnen, und der Weg über PlaybackInfo lässt sich prüfen.
UNANALYSED = "--unanalysed" in os.sys.argv

TITLES = [
    ("m1", "Der weite Himmel", "Movie", "Ein Mann, ein Berg, zwei Stunden Schweigen."),
    ("m2", "Nachtzug nach Genf", "Movie", "Krimi in engen Abteilen."),
    ("s1", "Küstenwache", "Series", "Zwölf Folgen Nieselregen und Verdacht."),
    ("m3", "Die Kartografin", "Movie", "Sie zeichnet Länder, die es nicht gibt."),
    ("s2", "Hinter dem Deich", "Series", "Eine Familie, ein Hof, drei Staffeln."),
    ("m4", "Letzter Sommer", "Movie", "Zwei Wochen, die alles ändern."),
    ("m5", "Kalte Küste", "Movie", "Ein Leuchtturm und eine Frage zu viel."),
]

def item(key, name, kind, overview, *, resume=None, series=None):
    d = {
        "Id": key, "Name": name, "Type": kind, "Overview": overview,
        "ProductionYear": 2021 + (len(name) % 4),
        "RunTimeTicks": (90 + len(name)) * 60 * 10_000_000,
        "OfficialRating": "12",
        "ImageTags": {"Primary": "p-" + key},
        "BackdropImageTags": ["b-" + key],
        "Genres": ["Drama"],
        "UserData": {"PlayCount": 0, "Played": False, "PlaybackPositionTicks": 0},
        "MediaSources": [{
            "Id": "src-" + key,
            "Container": "mp4",
            "Size": 28135,
            "Bitrate": 1_200_000,
            "RunTimeTicks": 10 * 10_000_000,
            "SupportsDirectPlay": True,
            "SupportsDirectStream": True,
            "SupportsTranscoding": True,
            "MediaStreams": [] if UNANALYSED else [
                {"Index": 0, "Type": "Video", "Codec": "h264",
                 "Width": 640, "Height": 360},
                {"Index": 1, "Type": "Audio", "Codec": "aac", "Channels": 2,
                 "Language": "ger", "DisplayTitle": "Deutsch", "IsDefault": True},
            ],
        }],
    }
    if resume:
        d["UserData"]["PlaybackPositionTicks"] = int(d["RunTimeTicks"] * resume)
    if series:
        d.update({"SeriesId": series[0], "SeriesName": series[1],
                  "ParentIndexNumber": 1, "IndexNumber": series[2],
                  "SeriesPrimaryImageTag": "p-" + series[0]})
    return d

def page(items):
    return {"Items": items, "TotalRecordCount": len(items), "StartIndex": 0}

def png(colour, width=960, height=540):
    """Eine einfarbige Fläche. Reicht, um zu sehen, dass das Bild wechselt."""
    raw = b"".join(b"\x00" + bytes(colour) * width for _ in range(height))
    def chunk(tag, data):
        return struct.pack(">I", len(data)) + tag + data + struct.pack(
            ">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
    return (b"\x89PNG\r\n\x1a\n"
            + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0))
            + chunk(b"IDAT", zlib.compress(raw, 6))
            + chunk(b"IEND", b""))

def colour_for(key):
    h = hashlib.sha1(key.encode()).digest()
    # Gedeckte Töne, damit weiße Schrift darauf lesbar bleibt.
    return (40 + h[0] % 90, 40 + h[1] % 90, 50 + h[2] % 90)

POLLS = {"count": 0}

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, *args):
        pass

    def reply(self, payload, content_type="application/json"):
        body = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        url = urlparse(self.path)
        p, q = url.path, parse_qs(url.query)

        if p == "/System/Info/Public":
            return self.reply({"ServerName": "Attrappe", "Version": "10.10.3",
                               "Id": "stub", "StartupWizardCompleted": True})
        if p == "/QuickConnect/Enabled":
            return self.reply(b"true", "text/plain")
        if p == "/QuickConnect/Connect":
            POLLS["count"] += 1
            # Nach drei Abfragen gilt der Code als bestätigt — lange genug,
            # um den Bildschirm mit dem Code zu sehen.
            return self.reply({"Authenticated": POLLS["count"] >= 3,
                               "Secret": "s3cr3t", "Code": "428193"})
        if p == f"/Users/{USER_ID}/Views":
            return self.reply(page([
                {"Id": "lib-movies", "Name": "Filme", "Type": "CollectionFolder",
                 "CollectionType": "movies"},
                {"Id": "lib-shows", "Name": "Serien", "Type": "CollectionFolder",
                 "CollectionType": "tvshows"},
            ]))
        if p == f"/Users/{USER_ID}/Items/Latest":
            return self.reply([item(*t) for t in TITLES[:6]])
        if p == f"/Users/{USER_ID}/Items/Resume":
            return self.reply(page([
                item("e1", "Der Nebel kommt", "Episode", "Erste Folge, letzte Ruhe.",
                     resume=0.4, series=("s1", "Küstenwache", 3)),
                item("m2", TITLES[1][1], "Movie", TITLES[1][3], resume=0.2),
            ]))
        if p == "/Shows/NextUp":
            return self.reply(page([
                item("e2", "Das Wrack", "Episode", "Was der Sturm freilegt.",
                     series=("s1", "Küstenwache", 4)),
            ]))
        if p.startswith("/Videos/") and p.endswith("/stream"):
            return self.serve_file(SAMPLE, "video/mp4")
        if p.startswith(f"/Users/{USER_ID}/Items/") and p.count("/") == 4:
            key = p.rsplit("/", 1)[-1]
            match = next((t for t in TITLES if t[0] == key), None)
            if match:
                return self.reply(item(*match))
            return self.reply(item(key, "Unbekannt", "Movie", "Aus dem Nichts."))
        if p.startswith("/Items/") and "/Images/" in p:
            key = p.split("/")[2]
            return self.reply(png(colour_for(key)), "image/png")
        if p == f"/Users/{USER_ID}/Items":
            parent = q.get("ParentId", [""])[0]
            kind = "Series" if parent == "lib-shows" else "Movie"
            chosen = [t for t in TITLES if t[2] == kind] or TITLES
            return self.reply(page([item(*t) for t in chosen]))
        if p == "/__reports":
            return self.reply(REPORTS)
        if p == "/Genres":
            return self.reply(page([{"Id": "g1", "Name": "Drama", "Type": "Genre"}]))
        return self.reply(page([]))

    def serve_file(self, path, content_type):
        """Mit Bereichsanfragen — AVPlayer holt eine Datei nie am Stück."""
        with open(path, "rb") as handle:
            data = handle.read()
        rng = self.headers.get("Range")
        start, end = 0, len(data) - 1
        status = 200
        if rng and rng.startswith("bytes="):
            piece = rng[6:].split("-")
            start = int(piece[0] or 0)
            if len(piece) > 1 and piece[1]:
                end = min(int(piece[1]), len(data) - 1)
            status = 206
        body = data[start:end + 1]
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Accept-Ranges", "bytes")
        self.send_header("Content-Length", str(len(body)))
        if status == 206:
            self.send_header("Content-Range", f"bytes {start}-{end}/{len(data)}")
        self.end_headers()
        self.wfile.write(body)

    def do_DELETE(self):
        REPORTS.append(("encoding-stopped", self.path))
        self.reply({})

    def do_POST(self):
        url = urlparse(self.path)
        length = int(self.headers.get("Content-Length") or 0)
        body = self.rfile.read(length)
        if url.path in ("/Users/AuthenticateWithQuickConnect", "/Users/AuthenticateByName"):
            return self.reply({
                "AccessToken": TOKEN,
                "User": {"Id": USER_ID, "Name": "Testkonto",
                         "Policy": {"IsAdministrator": True}},
            })
        if url.path.startswith("/Sessions/Playing"):
            REPORTS.append((url.path, body.decode() if body else ""))
            return self.reply({})
        if url.path.startswith("/Items/") and url.path.endswith("/PlaybackInfo"):
            key = url.path.split("/")[2]
            # Keine echte Umrechnung — die Adresse zeigt auf dieselbe Datei.
            # Geprüft wird hier der Weg, nicht der Transcoder.
            return self.reply({
                "MediaSources": [{
                    "Id": "src-" + key,
                    "Container": "mp4",
                    "SupportsTranscoding": True,
                    "TranscodingUrl": f"/Videos/{key}/stream?static=true&ApiKey={TOKEN}",
                }],
                "PlaySessionId": "ps-" + key,
            })
        if url.path == "/QuickConnect/Initiate":
            POLLS["count"] = 0
            return self.reply({"Secret": "s3cr3t", "Code": "428193"})
        return self.reply({})

ThreadingHTTPServer(("127.0.0.1", 8096), Handler).serve_forever()