dotfiles/common/qutebrowser/userscripts/traggo-log-video

#!/usr/bin/env python3
"""Log the full length of the current page's video to Traggo as one finished span."""

import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import urllib.error
import urllib.request
from datetime import datetime, timedelta
from html import unescape

API = "https://time.rcjohnstone.com/graphql"
TOKEN_PATH = os.path.expanduser("~/.config/traggo/token")
TAGS = [
    {"key": "language", "value": "spanish"},
    {"key": "mode", "value": "listening"},
]
YTDLP_TIMEOUT = 45
HTTP_TIMEOUT = 20
UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
      "Chrome/124.0.0.0 Safari/537.36")

FIFO = os.environ.get("QUTE_FIFO")


class ProbeError(Exception):
    pass


def report(kind, text):
    text = " ".join(text.split())
    line = "message-{} {}\n".format(kind, shlex.quote(text))
    if FIFO:
        with open(FIFO, "a") as fh:
            fh.write(line)
    else:
        sys.stderr.write(line)


def die(text):
    report("error", "traggo: " + text)
    sys.exit(1)


def human(secs):
    hours, rem = divmod(secs, 3600)
    mins, s = divmod(rem, 60)
    if hours:
        return "{}h{:02d}m".format(hours, mins)
    if mins:
        return "{}m".format(mins)
    return "{}s".format(s)


def ytdlp_argv():
    local = shutil.which("yt-dlp")
    if local:
        return [local]
    # qutebrowser runs inside distrobox, so reach the host's copy if the container lacks one.
    host = shutil.which("distrobox-host-exec")
    if host:
        return [host, "yt-dlp"]
    raise ProbeError("yt-dlp not found")


def probe_ytdlp(url):
    argv = ytdlp_argv() + [
        "--skip-download", "--no-warnings",
        # We only ever want metadata, so a stale extractor failing on formats must not abort.
        "--ignore-no-formats-error",
        "--print", "%(duration)s|%(title)s", url,
    ]
    try:
        proc = subprocess.run(argv, capture_output=True, text=True,
                              timeout=YTDLP_TIMEOUT)
    except subprocess.TimeoutExpired:
        raise ProbeError("yt-dlp timed out after {}s".format(YTDLP_TIMEOUT))
    except OSError as exc:
        raise ProbeError("yt-dlp failed to start: {}".format(exc))

    line = next((ln for ln in reversed(proc.stdout.splitlines()) if "|" in ln), "")
    if not line:
        detail = " ".join(proc.stderr.split())[:160] or "no output"
        raise ProbeError("yt-dlp: " + detail)

    raw, _, title = line.partition("|")
    raw = raw.strip()
    if not raw.isdigit() or int(raw) <= 0:
        raise ProbeError("yt-dlp gave no duration (got {!r})".format(raw))
    return int(raw), title.strip()


def probe_html(url):
    req = urllib.request.Request(url, headers={
        "User-Agent": UA, "Accept-Language": "en-US,en;q=0.9"})
    try:
        page = urllib.request.urlopen(req, timeout=HTTP_TIMEOUT).read().decode(
            "utf-8", "replace")
    except Exception as exc:
        raise ProbeError("page fetch failed: {}".format(exc))

    match = re.search(r'"lengthSeconds":"(\d+)"', page)
    if not match:
        raise ProbeError("no lengthSeconds in page")
    secs = int(match.group(1))
    if secs <= 0:
        raise ProbeError("page reported zero length")
    title = re.search(r'<meta name="title" content="([^"]*)"', page)
    return secs, unescape(title.group(1)) if title else ""


def qute_title():
    return re.sub(r"\s+-\s+YouTube$", "", os.environ.get("QUTE_TITLE", "")).strip()


def create_span(token, start, end, note):
    query = ("mutation($s:Time!,$e:Time!,$t:[InputTimeSpanTag!],$n:String!){"
             "createTimeSpan(start:$s,end:$e,tags:$t,note:$n){id}}")
    body = json.dumps({"query": query, "variables": {
        "s": start, "e": end, "t": TAGS, "n": note}}).encode()
    req = urllib.request.Request(API, data=body, headers={
        "Content-Type": "application/json",
        "Authorization": "traggo " + token})
    try:
        raw = urllib.request.urlopen(req, timeout=HTTP_TIMEOUT).read().decode()
    except urllib.error.HTTPError as exc:
        die("server returned HTTP {}: {}".format(
            exc.code, exc.read().decode("utf-8", "replace")[:160]))
    except Exception as exc:
        die("cannot reach server: {}".format(exc))

    data = json.loads(raw)
    if data.get("errors"):
        die("api: {}".format(data["errors"][0].get("message", "unknown error")))
    return data["data"]["createTimeSpan"]["id"]


def main():
    url = os.environ.get("QUTE_URL", "").strip()
    if not url:
        die("no QUTE_URL in environment")

    try:
        token = open(TOKEN_PATH).read().strip()
    except OSError as exc:
        die("cannot read token: {}".format(exc))
    if not token:
        die("token file {} is empty".format(TOKEN_PATH))

    try:
        secs, title = probe_ytdlp(url)
    except ProbeError as primary:
        try:
            secs, title = probe_html(url)
        except ProbeError as fallback:
            die("no duration. {} / fallback: {}".format(primary, fallback))

    note = title or qute_title() or url
    end = datetime.now().astimezone().replace(microsecond=0)
    start = end - timedelta(seconds=secs)
    span = create_span(token, start.isoformat(), end.isoformat(), note)
    report("info", "traggo: logged {} [{}] - {}".format(human(secs), span, note))


if __name__ == "__main__":
    main()