Nextcue/source/resi_probe.py

#!/usr/bin/env python3
"""
resi_probe.py — does the Resi cloud hand us live cue + playhead data for OUR account?

One read-only pass over the endpoints a Companion "cue countdown" module would need:

    login            POST /api/v3/login?newToken=true
    who am i          GET /api_v2.svc/users/me
    venues            GET /api/v3/customers/{customerId}/venues
    decoders          GET /api_v2.svc/users/{venueUuid}/players      <- playhead lives here
    event             GET /api/v3/customers/{customerId}/events/{eventId}
    cues              GET /api_v2.svc/streamprofiles/{eventProfileId}/events/{eventId}/cues

Nothing is created, changed or deleted. Your password is read with getpass (never
echoed), sent only to https://central.resi.io, and never written to disk.

This talks to Resi's INTERNAL API — the one studio.resi.io itself uses. It is
undocumented and unsupported; don't open a Resi ticket about it.

Usage:
    python3 resi_probe.py --user you@yourchurch.org
    python3 resi_probe.py --user you@yourchurch.org --watch 10   # 10s playhead watch
    python3 resi_probe.py --user you@yourchurch.org --save ./resi-capture
"""

import argparse
import getpass
import http.cookiejar
import json
import os
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

BASE = "https://central.resi.io"

_jar = http.cookiejar.CookieJar()
_opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(_jar))
_token = None
_save_dir = None


def _cookie(name):
    for c in _jar:
        if c.name == name:
            return c.value
    return None


def _headers():
    h = {"Accept": "application/json", "User-Agent": "resi-probe/1.0"}
    if _token:
        h["Authorization"] = "X-Bearer " + _token
    return h


def _request(method, path, body=None):
    url = path if path.startswith("http") else BASE + path
    data = None
    h = _headers()
    if body is not None:
        data = json.dumps(body).encode()
        h["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=h, method=method)
    try:
        with _opener.open(req, timeout=20) as resp:
            raw = resp.read()
    except urllib.error.HTTPError as e:
        raise RuntimeError("HTTP %s on %s %s — %s" % (e.code, method, path, e.read()[:300].decode("utf-8", "replace")))
    except ssl.SSLError as e:
        raise RuntimeError("TLS failure (%s). If this Mac's Python has no CA bundle, run the "
                           "'Install Certificates.command' that ships with your Python install." % e)
    except urllib.error.URLError as e:
        raise RuntimeError("Could not reach %s — %s" % (url, e.reason))
    if not raw:
        return None
    try:
        return json.loads(raw)
    except ValueError:
        return raw.decode("utf-8", "replace")


def get(path):
    out = _request("GET", path)
    if _save_dir is not None:
        name = urllib.parse.quote(path.split("?")[0].strip("/"), safe="") + ".json"
        with open(os.path.join(_save_dir, name), "w") as fh:
            json.dump(out, fh, indent=2, sort_keys=True)
    return out


def login(username, password):
    global _token
    _request("POST", "/api/v3/login?newToken=true", {"userName": username, "password": password})
    _token = _cookie("livingas1")
    if not _token:
        raise RuntimeError("Login returned 2xx but set no 'livingas1' cookie — the auth shape has changed. "
                           "Cookies seen: %s" % ", ".join(c.name for c in _jar))


def ms(position):
    """'01:02:03.500' -> milliseconds. Returns None on anything unparseable."""
    if not position:
        return None
    try:
        hh, mm_, rest = position.split(":")
        return int((int(hh) * 3600 + int(mm_) * 60 + float(rest)) * 1000)
    except (ValueError, AttributeError):
        return None


def hms(milliseconds):
    s, sign = int(abs(milliseconds) // 1000), "-" if milliseconds < 0 else ""
    return "%s%02d:%02d:%02d" % (sign, s // 3600, (s % 3600) // 60, s % 60)


def main():
    global _save_dir
    ap = argparse.ArgumentParser(description="Read-only probe of the Resi internal API.")
    ap.add_argument("--user", help="Resi Studio login (email)")
    ap.add_argument("--watch", type=int, default=0, metavar="SEC",
                    help="after the report, poll the playhead for SEC seconds to prove it advances")
    ap.add_argument("--save", metavar="DIR",
                    help="write each raw JSON response to DIR (contains venue/decoder names and uuids)")
    args = ap.parse_args()

    if args.save:
        os.makedirs(args.save, exist_ok=True)
        _save_dir = args.save

    username = args.user or input("Resi login (email): ").strip()
    password = getpass.getpass("Password (not echoed, not stored): ")

    print("\n== auth ==")
    login(username, password)
    print("  login OK, bearer token acquired")

    me = get("/api_v2.svc/users/me")
    customer_id = me.get("customerId")
    print("  user      : %s" % me.get("userName"))
    print("  org       : %s (customerId %s)" % (me.get("customerName"), customer_id))
    if not customer_id:
        sys.exit("  no customerId on this account — stop here, nothing else will resolve")

    print("\n== venues ==")
    venues = get("/api/v3/customers/%s/venues" % customer_id) or []
    if not venues:
        print("  none returned. Either this login has no venue access, or the plan has no Multisite.")
    for v in venues:
        print("  %-40s %s" % (v.get("userName") or v.get("name"), v.get("uuid")))

    print("\n== decoders / players ==")
    live = []
    for v in venues:
        players = []
        try:
            players = get("/api_v2.svc/users/%s/players" % v.get("uuid")) or []
        except RuntimeError as e:
            print("  [%s] players call failed: %s" % (v.get("userName"), e))
            continue
        if not players:
            print("  [%s] no players" % v.get("userName"))
        for p in players:
            print("  [%s] %-28s pos=%-14s eventId=%s"
                  % (v.get("userName"), p.get("userName"), p.get("position"), p.get("eventId")))
            extra = sorted(set(p) - {"playerHardwareId", "userName", "position", "eventId"})
            if extra:
                print("        other fields on this player: %s" % ", ".join(extra))
            if p.get("eventId"):
                live.append((v, p))

    if not live:
        print("\nNo decoder is on an event right now. Re-run this DURING a service —"
              "\nposition and eventId are what the whole countdown hangs on.")
        return

    for v, p in live:
        print("\n== event + cues for %s / %s ==" % (v.get("userName"), p.get("userName")))
        event = get("/api/v3/customers/%s/events/%s" % (customer_id, p.get("eventId")))
        profile_id = event.get("eventProfileId")
        print("  event     : %s" % event.get("name"))
        print("  startTime : %s" % event.get("startTime"))
        print("  profileId : %s" % profile_id)
        if not profile_id:
            print("  no eventProfileId — cannot fetch cues for this event")
            continue

        cues = get("/api_v2.svc/streamprofiles/%s/events/%s/cues?canISetCues=1"
                   % (profile_id, p.get("eventId"))) or []
        cues = sorted(cues, key=lambda c: ms(c.get("position")) or 0)
        print("  %d cue(s):" % len(cues))
        here = ms(p.get("position"))
        for c in cues:
            at = ms(c.get("position"))
            marker = "  "
            if here is not None and at is not None:
                marker = "->" if at > here else "  "
            print("   %s %-14s %s" % (marker, c.get("position"), c.get("name")))

        if here is None:
            print("\n  playhead unparseable (%r) — countdown not possible from this field"
                  % p.get("position"))
            continue
        nxt = next((c for c in cues if (ms(c.get("position")) or 0) > here), None)
        print("\n  THE BUTTON WOULD READ:")
        print("    Playback position : %s" % hms(here))
        if nxt:
            print("    Next cue          : %s" % nxt.get("name"))
            print("    Next cue in       : %s" % hms((ms(nxt.get("position")) or 0) - here))
        else:
            print("    Next cue          : (none left on this event)")

    if args.watch:
        v, p = live[0]
        print("\n== watching %s for %ds (does the playhead advance?) ==" % (p.get("userName"), args.watch))
        first = last = None
        deadline = time.time() + args.watch
        while time.time() < deadline:
            players = get("/api_v2.svc/users/%s/players" % v.get("uuid")) or []
            now = next((x.get("position") for x in players
                        if x.get("playerHardwareId") == p.get("playerHardwareId")), None)
            print("  %s  %s" % (time.strftime("%H:%M:%S"), now))
            if first is None:
                first = ms(now)
            last = ms(now)
            time.sleep(2)
        if first is not None and last is not None:
            moved = last - first
            print("  moved %s over ~%ds -> %s"
                  % (hms(moved), args.watch,
                     "PLAYING (a play/pause variable is derivable)" if moved > 0 else "frozen / paused"))


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as exc:
        sys.exit("\nFAILED: %s" % exc)
    except KeyboardInterrupt:
        sys.exit("\ninterrupted")