114 lines
4.9 KiB
Python
114 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Watch data of the test account on the server, before and after a test run (docs/desktop.md).
|
|
|
|
watchdata.py snapshot [FILE] saves the user's watch data (positions, played, counts, dates, favourites)
|
|
watchdata.py diff [FILE] what the run changed, compared with the snapshot
|
|
watchdata.py restore [FILE] puts the changed items back as they were in the snapshot
|
|
|
|
FILE defaults to $STAND/watchdata.json. The user is JF_USER: its token is taken from the test stand's saved accounts
|
|
($STAND/config/sleepyowl/jellyfin_session.preferences_pb, STAND=~/.cache/sleepyowl-test) or from JF_TOKEN.
|
|
Server JF_URL and user JF_USER come from local/dev.env (template tools/dev.env.example) or the environment.
|
|
|
|
Restore writes the saved UserData back (POST /UserItems/{id}/UserData). An item the run started from nothing (not in
|
|
the snapshot) is reset: played and its date removed (DELETE /UserPlayedItems), position and count back to 0.
|
|
The resume list follows LastPlayedDate, so its order comes back with the dates.
|
|
"""
|
|
import json, os, re, sys, urllib.request
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
import devenv
|
|
|
|
STAND = os.environ.get("STAND") or os.path.expanduser("~/.cache/sleepyowl-test")
|
|
URL = devenv.need("JF_URL")
|
|
USER = devenv.need("JF_USER")
|
|
FIELDS = ("PlaybackPositionTicks", "PlayCount", "Played", "LastPlayedDate", "IsFavorite")
|
|
|
|
|
|
def account():
|
|
token = os.environ.get("JF_TOKEN")
|
|
path = os.path.join(STAND, "config", "sleepyowl", "jellyfin_session.preferences_pb")
|
|
raw = open(path, "rb").read().decode("utf-8", "replace") if os.path.exists(path) else ""
|
|
m = re.search(r"\[\{.*\}\]", raw)
|
|
for a in json.loads(m.group(0)) if m else []:
|
|
if a["userName"] == USER:
|
|
return token or a["token"], a["userId"]
|
|
sys.exit(f"no saved account {USER!r} in {path}: log in on the stand first (or set JF_TOKEN and JF_USER_ID)")
|
|
|
|
|
|
def api(method, path, body=None):
|
|
req = urllib.request.Request(URL + path, method=method, data=None if body is None else json.dumps(body).encode())
|
|
req.add_header("Authorization", f'MediaBrowser Token="{TOKEN}"')
|
|
req.add_header("Content-Type", "application/json")
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
data = r.read()
|
|
return json.loads(data) if data else None
|
|
|
|
|
|
def current():
|
|
"""Every item with any watch data: resumable, played or favourite (episodes, movies, series)."""
|
|
items = {}
|
|
for flt in ("IsResumable", "IsPlayed", "IsFavorite"):
|
|
res = api("GET", f"/Items?userId={USER_ID}&Recursive=true&Filters={flt}&Fields=UserData&EnableImages=false")
|
|
for it in res["Items"]:
|
|
ud = it.get("UserData", {})
|
|
items[it["Id"]] = {"Name": it.get("SeriesName", "") + (" · " if it.get("SeriesName") else "") + it.get("Name", ""),
|
|
**{k: ud.get(k) for k in FIELDS}}
|
|
return items
|
|
|
|
|
|
def user_data(item_id):
|
|
ud = api("GET", f"/UserItems/{item_id}/UserData?userId={USER_ID}")
|
|
return {k: ud.get(k) for k in FIELDS}
|
|
|
|
|
|
def changes(before, after):
|
|
out = []
|
|
for item_id in sorted(set(before) | set(after)):
|
|
b, a = before.get(item_id), after.get(item_id)
|
|
if b is None:
|
|
a = {**a, **user_data(item_id)}
|
|
if a is None:
|
|
a = {"Name": b["Name"], **user_data(item_id)}
|
|
if b is None or any(b[k] != a[k] for k in FIELDS):
|
|
out.append((item_id, b, a))
|
|
return out
|
|
|
|
|
|
def show(item_id, b, a):
|
|
name = (b or a)["Name"]
|
|
diff = ", ".join(f"{k}: {None if b is None else b[k]} → {a[k]}" for k in FIELDS if b is None or b[k] != a[k])
|
|
print(f" {name} ({item_id}): {diff}")
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2 or sys.argv[1] not in ("snapshot", "diff", "restore"):
|
|
sys.exit(__doc__)
|
|
path = sys.argv[2] if len(sys.argv) > 2 else os.path.join(STAND, "watchdata.json")
|
|
if sys.argv[1] == "snapshot":
|
|
items = current()
|
|
json.dump(items, open(path, "w"), ensure_ascii=False, indent=1)
|
|
print(f"{len(items)} items with watch data saved to {path}")
|
|
return
|
|
before = json.load(open(path))
|
|
diff = changes(before, current())
|
|
if not diff:
|
|
print("no changes")
|
|
return
|
|
for c in diff: show(*c)
|
|
if sys.argv[1] == "diff":
|
|
return
|
|
for item_id, b, _ in diff:
|
|
if b is None:
|
|
api("DELETE", f"/UserPlayedItems/{item_id}?userId={USER_ID}")
|
|
api("POST", f"/UserItems/{item_id}/UserData?userId={USER_ID}", {"PlaybackPositionTicks": 0, "PlayCount": 0, "Played": False})
|
|
else:
|
|
api("POST", f"/UserItems/{item_id}/UserData?userId={USER_ID}", {k: b[k] for k in FIELDS})
|
|
left = changes(before, current())
|
|
print(f"restored {len(diff)} items" + ("" if not left else f"; still different: {len(left)}"))
|
|
for c in left: show(*c)
|
|
|
|
|
|
TOKEN, USER_ID = (os.environ["JF_TOKEN"], os.environ["JF_USER_ID"]) if os.environ.get("JF_USER_ID") else account()
|
|
if __name__ == "__main__":
|
|
main()
|