#!/usr/bin/python3 """Import Dreaming Spanish progress into Traggo as tagged time spans. ds_to_traggo [--apply] [--from YYYY-MM-DD] [--include-initial] Dry-run unless --apply is given. Safe to re-run: spans this tool created are marked in their note and are not recreated. How the two DS endpoints relate (verified against live data): dayWatchedTime is the daily total of EVERYTHING -- platform videos plus every external entry, including 'talking'. externalTime is the itemised list of off-platform entries. So listening must be derived by subtraction, not by adding the two together: mode:speaking = external 'talking' mode:listening = dayWatchedTime - talking (platform + watching + listening) """ import argparse import json import subprocess import sys import urllib.request from collections import defaultdict from datetime import datetime, timedelta from zoneinfo import ZoneInfo DS_BASE = "https://app.dreaming.com/.netlify/functions" DS_TOKEN_FILE = "/tmp/dreaming_token" TRAGGO = "https://time.rcjohnstone.com/graphql" TZ = ZoneInfo("America/Louisville") MARKER = "[ds-import]" TAGS = {"language": "#4a90d9", "mode": "#7ab317"} def ds_get(endpoint, token): req = urllib.request.Request( "%s/%s" % (DS_BASE, endpoint), headers={"Authorization": "Bearer " + token}) with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read()) def gql(query, variables, token): body = json.dumps({"query": query, "variables": variables}).encode() req = urllib.request.Request( TRAGGO, data=body, headers={"Content-Type": "application/json", "Authorization": "traggo " + token}) with urllib.request.urlopen(req, timeout=60) as r: out = json.loads(r.read()) if out.get("errors"): raise RuntimeError(json.dumps(out["errors"])[:300]) return out["data"] def traggo_login(): pw = subprocess.run(["rbw", "get", "Traggo"], capture_output=True, text=True).stdout.strip() if not pw: sys.exit("Could not read the Traggo password from Bitwarden (rbw get Traggo).") q = ("mutation($u:String!,$p:String!){login(username:$u,pass:$p," "deviceName:\"ds-import\",type:NoExpiry,cookie:false){token}}") return gql(q, {"u": "connor", "p": pw}, "")["login"]["token"] def ensure_tags(token): have = {t["key"] for t in gql("{tags{key}}", {}, token)["tags"]} for key, color in TAGS.items(): if key not in have: gql("mutation($k:String!,$c:String!){createTag(key:$k,color:$c){key}}", {"k": key, "c": color}, token) print(" created tag key: %s" % key) def existing(token, first, last): """Dates already imported, as {(date, mode)}.""" q = ("query($f:Time!,$t:Time!,$c:InputCursor){timeSpans(fromInclusive:$f," "toInclusive:$t,cursor:$c){cursor{hasMore offset startId pageSize}" "timeSpans{start note tags{key value}}}}") seen, cursor = set(), {"offset": 0, "pageSize": 200} while True: page = gql(q, {"f": first, "t": last, "c": cursor}, token)["timeSpans"] for ts in page["timeSpans"]: if MARKER not in (ts.get("note") or ""): continue mode = next((t["value"] for t in ts["tags"] if t["key"] == "mode"), None) seen.add((ts["start"][:10], mode)) c = page["cursor"] if not c["hasMore"]: return seen cursor = {"offset": c["offset"], "startId": c["startId"], "pageSize": c["pageSize"]} def build(ds_token, start_from, include_initial): ext = ds_get("externalTime", ds_token)["externalTimes"] day = {d["date"]: d["timeSeconds"] for d in ds_get("dayWatchedTime", ds_token)} talk, notes = defaultdict(int), defaultdict(list) for e in ext: if e["type"] == "initial" and not include_initial: continue if e["type"] == "talking": talk[e["date"]] += e["timeSeconds"] if e.get("description"): notes[(e["date"], "speaking")].append(e["description"]) elif e.get("description"): notes[(e["date"], "listening")].append(e["description"]) plan = [] for date in sorted(day): if start_from and date < start_from: continue listening = day[date] - talk.get(date, 0) speaking = talk.get(date, 0) # Anchor at midnight local; speaking picks up where listening ends so # the two never overlap in the calendar view. DS records daily totals # only, so the clock times are synthetic either way. cursor = datetime(int(date[:4]), int(date[5:7]), int(date[8:10]), tzinfo=TZ) for mode, secs in (("listening", listening), ("speaking", speaking)): if secs <= 0: continue end = cursor + timedelta(seconds=secs) desc = ", ".join(dict.fromkeys(notes.get((date, mode), []))) plan.append({ "date": date, "mode": mode, "seconds": secs, "start": cursor.isoformat(), "end": end.isoformat(), "note": ("%s %s" % (MARKER, desc)).strip(), }) cursor = end return plan def main(): p = argparse.ArgumentParser() p.add_argument("--apply", action="store_true") p.add_argument("--from", dest="start_from", metavar="YYYY-MM-DD") p.add_argument("--include-initial", action="store_true", help="also import the 50h 'time prior to Dreaming Spanish' entry") a = p.parse_args() ds_token = open(DS_TOKEN_FILE).read().strip() plan = build(ds_token, a.start_from, a.include_initial) if not plan: print("Nothing to import.") return token = traggo_login() ensure_tags(token) already = existing(token, plan[0]["start"], plan[-1]["end"]) todo = [s for s in plan if (s["date"], s["mode"]) not in already] hrs = lambda rows, m: sum(r["seconds"] for r in rows if r["mode"] == m) / 3600 print(" planned : %d spans (%.1fh listening, %.1fh speaking) over %s..%s" % (len(plan), hrs(plan, "listening"), hrs(plan, "speaking"), plan[0]["date"], plan[-1]["date"])) print(" already : %d spans present from a previous run" % (len(plan) - len(todo))) print(" to create: %d spans (%.1fh listening, %.1fh speaking)" % (len(todo), hrs(todo, "listening"), hrs(todo, "speaking"))) if not a.apply: print("\n first 5:") for s in todo[:5]: print(" %s %-9s %5.0fm %s" % (s["date"], s["mode"], s["seconds"] / 60, s["note"][:48])) print("\n Dry run. Re-run with --apply to write these to Traggo.") return q = ("mutation($s:Time!,$e:Time!,$t:[InputTimeSpanTag!],$n:String!){" "createTimeSpan(start:$s,end:$e,tags:$t,note:$n){id}}") made = 0 for s in todo: tags = [{"key": "language", "value": "spanish"}, {"key": "mode", "value": s["mode"]}] try: gql(q, {"s": s["start"], "e": s["end"], "t": tags, "n": s["note"]}, token) made += 1 except Exception as err: print(" FAILED %s %s: %s" % (s["date"], s["mode"], err)) print(" created %d/%d spans" % (made, len(todo))) if __name__ == "__main__": main()