#!/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" UP_HOST = "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 # How the emulator addresses this proxy. The app decides a video/channel is # "local" by matching its host against the instance host, so we relabel the # upstream host to ours in the JSON (only the `host` fields, never URLs, so # thumbnails/media keep pointing at peertube2). MOCK_HOST = f"10.0.2.2:{PORT}" def relabel_hosts(obj): if isinstance(obj, dict): return {k: (MOCK_HOST if k == "host" and v == UP_HOST else relabel_hosts(v)) for k, v in obj.items()} if isinstance(obj, list): return [relabel_hosts(x) for x in obj] return obj 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(relabel_hosts(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. # A handle the app built with our host (name@10.0.2.2:PORT) is mapped back # to the upstream host before we ask peertube2. upstream = self.path.replace(MOCK_HOST, UP_HOST).replace( MOCK_HOST.replace(":", "%3A"), UP_HOST) try: code, ctype, body = fetch(upstream) if "json" in ctype: self._json(json.loads(body), code) # relabel hosts in proxied JSON too else: 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()