jtube-tv/tools/mock/mock_peertube.py

#!/usr/bin/env python3
"""
Mock PeerTube backend for README screenshots.

It is a thin HTTP proxy to the public PeerTube demo instance
(peertube2.cpy.re, real CC/Blender content) with a few overrides so the
authenticated screens are populated without a real account:

  - login always succeeds (oauth-clients/local + users/token are canned),
  - /users/me is a "Demo User" with an avatar,
  - Continue watching / My videos / Subscriptions each proxy a DIFFERENT
    window of the public video list, so the rows don't repeat thumbnails,
  - My videos gets Private/Internal/Public badges,
  - a channel page shows the full video grid (channel .../videos is widened),
  - a video opened from Continue watching gets a fake resume position.

Everything else (video lists, video detail with its absolute HLS URLs, channel
banners/avatars, config, thumbnails) is proxied verbatim, so media and images
load straight from peertube2 over HTTPS; only the JSON + thumbnails pass through
here. The app runs UNMODIFIED against it (a debug-only cleartext flag in
app/src/debug lets it talk HTTP to this proxy).

Usage:
  python3 tools/mock/mock_peertube.py           # listens on 0.0.0.0:8080
Then in the app's setup screen use  http://10.0.2.2:8080  (emulator) and sign
in with any username/password.
"""
import json
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

UPSTREAM = "https://peertube2.cpy.re"
AVATAR = UPSTREAM + "/lazy-static/avatars/73e78544-cbae-439e-8c84-98b0756fbde7.jpg"
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8080


def fetch(path_with_query):
    req = urllib.request.Request(UPSTREAM + path_with_query, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=30) as r:
        return r.status, r.headers.get_content_type(), r.read()


def fetch_videos(start, count, sort="-publishedAt"):
    _, _, body = fetch(f"/api/v1/videos?isLocal=true&start={start}&count={count}&sort={sort}")
    return json.loads(body)


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *a):  # quiet
        pass

    def _json(self, obj, code=200):
        data = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def _raw(self, code, ctype, body):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _q(self, query, name, default):
        return urllib.parse.parse_qs(query).get(name, [default])[0]

    def do_POST(self):
        path = urllib.parse.urlparse(self.path).path
        self.rfile.read(int(self.headers.get("Content-Length", 0) or 0))
        if path == "/api/v1/users/token":
            return self._json({
                "token_type": "Bearer", "access_token": "demo",
                "refresh_token": "demo", "expires_in": 86400,
            })
        if re.match(r"^/api/v1/videos/[^/]+/watching$", path):
            self.send_response(204)
            self.end_headers()
            return
        self._json({}, 204)

    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        path, query = parsed.path, parsed.query
        start = lambda off: int(self._q(query, "start", "0")) + off  # noqa: E731
        count = int(self._q(query, "count", "25"))

        if path == "/api/v1/oauth-clients/local":
            return self._json({"client_id": "demo", "client_secret": "demo"})
        if path == "/api/v1/users/me":
            return self._json({
                "username": "demo",
                "account": {"displayName": "Demo User",
                            "avatars": [{"width": 120, "fileUrl": AVATAR}]},
            })
        if path == "/api/v1/config":
            return self._json({
                "instance": {"name": "PeerTube Demo",
                             "shortDescription": "Public demo content"},
                "search": {"searchIndex": {"enabled": False}},
            })
        if path == "/api/v1/users/me/videos":
            data = fetch_videos(start(4), count)
            for i, v in enumerate(data.get("data", [])):
                v["privacy"] = {0: {"id": 3, "label": "Private"},
                                1: {"id": 4, "label": "Internal"}}.get(i % 3, {"id": 1, "label": "Public"})
            return self._json(data)
        if path == "/api/v1/users/me/history/videos":
            return self._json(fetch_videos(start(8), count))
        if path == "/api/v1/users/me/subscriptions/videos":
            return self._json(fetch_videos(start(16), count))
        if path.startswith("/api/v1/users/me/subscriptions/exist"):
            return self._json({})
        if re.match(r"^/api/v1/video-channels/[^/]+/videos$", path):
            return self._json(fetch_videos(start(0), count))  # full grid on the channel page
        m = re.match(r"^/api/v1/videos/([^/]+)$", path)
        if m and self._q(query, "userHistory", "") == "true":
            _, _, body = fetch(path)
            detail = json.loads(body)
            detail["userHistory"] = {"currentTime": int(detail.get("duration", 0) * 0.35)}
            return self._json(detail)

        # Everything else (video lists, detail, channels, thumbnails) is proxied.
        try:
            code, ctype, body = fetch(self.path)
            self._raw(code, ctype, body)
        except urllib.error.HTTPError as e:
            self._raw(e.code, "application/json", e.read())
        except Exception as e:  # noqa: BLE001
            self._raw(502, "text/plain", str(e).encode())


if __name__ == "__main__":
    print(f"Mock PeerTube proxy on http://0.0.0.0:{PORT}  ->  {UPSTREAM}")
    print(f"Emulator: set instance URL to http://10.0.2.2:{PORT} and sign in with anything.")
    ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()