vikunja-agents/vk

#!/usr/bin/env python3
"""
vk -- Vikunja from the command line, for agent harnesses that can shell out.

Every subcommand takes --json for machine-readable output. The default is a
compact line format that is still trivially parseable (leading `#<id>`), so a
harness can grep it without asking for JSON.

Auth resolution order: flags, then VIKUNJA_TOKEN / VIKUNJA_URL / VIKUNJA_PROFILE,
then ~/.config/vikunja-agents/config.json. See `vk bootstrap` to create a token.
"""

import argparse
import getpass
import json
import os
import subprocess
import sys
import time
from datetime import datetime, timedelta, timezone

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from vkclient import CONFIG_PATH, Vikunja, VikunjaError, load_config, save_config  # noqa: E402

# What an agent bot needs and nothing more: it works tasks, talks in comments,
# and reads the project list to know where to file things. It cannot touch
# users, sharing, or its own account (Vikunja excludes `user_*` and `tokens`
# from token-grantable routes anyway).
DEFAULT_PERMISSIONS = {
    "tasks": ["read_all", "read_one", "create", "update"],
    "tasks_comments": ["read_all", "create"],
    "tasks_assignees": ["read_all", "create"],
    "tasks_labels": ["read_all", "create"],
    # `projectusers` resolves a username to an id when assigning; the
    # instance-wide /users search is not token-grantable.
    "projects": ["read_all", "read_one", "projectusers"],
    "labels": ["read_all"],
}


def out(args, data, lines):
    if args.json:
        json.dump(data, sys.stdout, indent=2, default=str)
        sys.stdout.write("\n")
    else:
        for line in lines:
            print(line)


def task_line(t):
    mark = "x" if t.get("done") else " "
    who = ",".join(a.get("username", "?") for a in (t.get("assignees") or []))
    labels = ",".join(l.get("title", "") for l in (t.get("labels") or []))
    bits = [f"[{mark}] #{t.get('id')}", t.get("title", "")]
    if who:
        bits.append(f"@{who}")
    if labels:
        bits.append(f"+{labels}")
    return "  ".join(bits)


# -- commands -------------------------------------------------------------


def cmd_bootstrap(vk, args):
    """Password login -> mint an API token -> store it. Run once per bot."""
    password = args.password or os.environ.get("VIKUNJA_PASSWORD") or getpass.getpass(
        f"Password for {args.username}: "
    )
    jwt = vk.login(args.username, password, args.totp)

    wanted = json.loads(args.permissions) if args.permissions else DEFAULT_PERMISSIONS
    granted, missing = vk.token_permissions(jwt, wanted)
    if missing:
        print(f"note: server does not offer {', '.join(missing)} -- skipped", file=sys.stderr)
    if not granted:
        raise VikunjaError("None of the requested permissions exist on this server.")

    expires = (datetime.now(timezone.utc) + timedelta(days=args.days)).replace(microsecond=0)
    created = vk.create_token(
        jwt, args.title or f"{args.username} ({os.uname().nodename})", granted,
        expires.isoformat().replace("+00:00", "Z"),
    )
    token = created.get("token")
    if not token:
        raise VikunjaError(f"Token created but not returned: {created}")

    profile = args.profile or args.username
    config = load_config()
    config.setdefault("url", vk.url)
    config.setdefault("profiles", {})[profile] = {
        "url": vk.url,
        "token": token,
        # Recorded because /user is unreachable with an API token; `--mine`
        # and `whoami` read the username from here.
        "username": args.username,
    }
    config.setdefault("default_profile", profile)
    save_config(config)

    print(f"Token stored as profile '{profile}' in {CONFIG_PATH}")
    print(f"Permissions: {json.dumps(granted)}")
    print(f"Expires:     {expires.isoformat()}")
    print(f"Token:       {token}")


def cmd_profile(vk, args):
    """Store a token minted elsewhere, reading it from stdin.

    Bot users have no password, so `bootstrap` cannot be used for them: their
    owner mints the token instead (`owner_id` on PUT /tokens) and it arrives
    out-of-band. Reading from stdin keeps it out of argv and shell history.
    """
    token = sys.stdin.read().strip()
    if not token:
        raise VikunjaError("No token on stdin. Try: pbpaste | vk profile <name> --username <user>")

    url = args.url or vk.url
    probe = Vikunja(url=url, token=token)
    probe.username = args.username
    # Fail before writing: a token that cannot read projects is not worth saving.
    projects = probe.projects()

    config = load_config()
    config.setdefault("url", url)
    config.setdefault("profiles", {})[args.name] = {
        "url": url,
        "token": token,
        "username": args.username,
    }
    config.setdefault("default_profile", args.name)
    save_config(config)
    out(args, {"profile": args.name, "username": args.username, "projects": len(projects)},
        [f"stored profile '{args.name}' for {args.username} in {CONFIG_PATH}",
         f"token verified: {len(projects)} project(s) visible"])


def cmd_whoami(vk, args):
    who = {"username": vk.whoami(), "url": vk.url}
    out(args, who, [f"{who['username'] or '(unknown)'} @ {who['url']}"])


def cmd_projects(vk, args):
    p = vk.projects()
    out(args, p, [f"#{x['id']} {x.get('title')}" for x in p])


def cmd_tasks(vk, args):
    filters = list(args.filter or [])
    if args.mine:
        filters.append(f"assignees in '{vk.whoami()}'")
    if not args.all:
        filters.append("done = false")
    tasks = vk.tasks(
        project_id=args.project,
        filter=" && ".join(filters) or None,
        sort_by=args.sort,
        limit=args.limit,
    )
    out(args, tasks, [task_line(t) for t in tasks] or ["(no tasks)"])


def cmd_show(vk, args):
    t = vk.task(args.task_id)
    comments = vk.comments(args.task_id) if args.comments else []
    lines = [task_line(t), f"project: {t.get('project_id')}"]
    if t.get("description"):
        lines += ["", t["description"]]
    for c in comments:
        author = (c.get("author") or {}).get("username", "?")
        lines += ["", f"--- {author} @ {c.get('created')}", c.get("comment", "")]
    out(args, {"task": t, "comments": comments}, lines)


def cmd_create(vk, args):
    t = vk.create_task(args.project, args.title, args.description, args.priority, args.due)
    if args.assign:
        user = vk.find_user(args.assign, args.project)
        if not user:
            raise VikunjaError(f"No such user: {args.assign}")
        vk.assign(t["id"], user["id"])
        t = vk.task(t["id"])
    out(args, t, [task_line(t)])


def cmd_comment(vk, args):
    text = args.text if args.text != "-" else sys.stdin.read()
    c = vk.comment(args.task_id, text)
    out(args, c, [f"commented on #{args.task_id}"])


def cmd_update(vk, args):
    fields = {}
    if args.title:
        fields["title"] = args.title
    if args.description:
        fields["description"] = args.description
    if args.priority is not None:
        fields["priority"] = args.priority
    if args.done:
        fields["done"] = True
    if args.undone:
        fields["done"] = False
    if not fields:
        raise VikunjaError("Nothing to update -- pass at least one field.")
    t = vk.update_task(args.task_id, **fields)
    out(args, t, [task_line(t)])


def cmd_done(vk, args):
    t = vk.update_task(args.task_id, done=True)
    out(args, t, [task_line(t)])


def cmd_assign(vk, args):
    task = vk.task(args.task_id)
    user = vk.find_user(args.username, task["project_id"])
    if not user:
        raise VikunjaError(f"No user '{args.username}' with access to project {task['project_id']}")
    vk.assign(args.task_id, user["id"])
    t = vk.task(args.task_id)
    out(args, t, [task_line(t)])


STATE_DIR = os.path.expanduser(
    os.environ.get("VIKUNJA_STATE_DIR", "~/.local/state/vikunja-agents")
)
SEEN_TTL = 30 * 86400  # forget handled tasks after a month so state stays small


def _state_path(profile):
    return os.path.join(STATE_DIR, f"watch-{profile or 'default'}.json")


def _load_seen(path):
    """State for both triggers.

    Shape: {"tasks": {task_id: ts}, "comments": {task_id: last_comment_id}}.
    Older files were a bare {task_id: ts} map; those are migrated in place so an
    upgrade does not re-fire every task already handled.
    """
    try:
        with open(path) as f:
            state = json.load(f)
    except (FileNotFoundError, ValueError):
        return {"tasks": {}, "comments": {}}

    if "tasks" not in state:
        state = {"tasks": state, "comments": {}}
    state.setdefault("comments", {})

    cutoff = time.time() - SEEN_TTL
    state["tasks"] = {k: v for k, v in state["tasks"].items() if v > cutoff}
    return state


def _save_seen(path, seen):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        json.dump(seen, f)
    os.replace(tmp, path)  # atomic; a crash mid-write can't corrupt the state


def _substitute(template, task, vk):
    values = {
        "task_id": task.get("id"),
        "title": task.get("title"),
        "project_id": task.get("project_id"),
        "priority": task.get("priority"),
        "url": f"{vk.url}/tasks/{task.get('id')}",
        "labels": ",".join(l.get("title", "") for l in (task.get("labels") or [])),
    }
    out_str = template
    for key, value in values.items():
        out_str = out_str.replace("{" + key + "}", "" if value is None else str(value))
    return out_str


def _run_command(args, command, env_extra, log):
    """Run the configured command once. Never raises."""
    env = dict(os.environ)
    env.update({k: str(v) for k, v in env_extra.items() if v is not None})
    if args.profile:
        env["VIKUNJA_PROFILE"] = args.profile
    try:
        result = subprocess.run(command, cwd=args.cwd, env=env,
                                stdin=subprocess.DEVNULL, timeout=args.timeout)
        log(f"finished rc={result.returncode}")
    except subprocess.TimeoutExpired:
        log(f"timed out after {args.timeout}s")
    except FileNotFoundError:
        log(f"FAILED: no such command: {command[0]}")
    except Exception as e:
        log(f"FAILED: {e!r}")


def _poll_comments(vk, args, state, state_file, me, log):
    """Fire for comments written by someone other than this account.

    Scans the account's assigned tasks *including done ones* -- a follow-up
    question almost always lands on a task that was just closed. On first sight
    of a task the newest comment id is recorded without firing, so enabling this
    on an existing board does not replay months of history.
    """
    filters = [f"assignees in '{me}'"] + list(args.filter or [])
    tasks = vk.tasks(project_id=args.project, filter=" && ".join(filters), sort_by="id")

    for task in tasks:
        key = str(task["id"])
        try:
            comments = vk.comments(task["id"])
        except VikunjaError as e:
            log(f"comment poll failed for #{task['id']}: {e}")
            continue
        if not comments:
            continue

        newest = max(int(c["id"]) for c in comments)
        if key not in state["comments"]:
            # First time we have looked at this task: adopt, do not replay.
            state["comments"][key] = newest
            _save_seen(state_file, state)
            continue

        last = int(state["comments"][key])
        fresh = [c for c in comments
                 if int(c["id"]) > last
                 and (c.get("author") or {}).get("username") != me]
        if not fresh:
            state["comments"][key] = max(last, newest)
            continue

        state["comments"][key] = newest
        _save_seen(state_file, state)  # at-most-once, same as task triggers

        for c in fresh:
            author = (c.get("author") or {}).get("username", "?")
            body = c.get("comment", "")
            log(f"comment on #{task['id']} by {author}: {body[:60]!r}")
            if args.dry_run:
                continue
            command = [_substitute(part, task, vk) for part in args.command]
            _run_command(args, command, {
                "VIKUNJA_TASK_ID": task["id"],
                "VIKUNJA_TASK_TITLE": task.get("title") or "",
                "VIKUNJA_TASK_URL": f"{vk.url}/tasks/{task['id']}",
                "VIKUNJA_TRIGGER": "comment",
                "VIKUNJA_COMMENT_ID": c.get("id"),
                "VIKUNJA_COMMENT_AUTHOR": author,
                "VIKUNJA_COMMENT": body,
            }, lambda m: log(f"task #{task['id']} {m}"))


def cmd_watch(vk, args):
    """Poll this bot's assigned queue and run a command for each new task.

    Pull rather than push, deliberately. Vikunja's WebSocket only accepts a JWT
    (not an API token) and its subscribable events are notifications and timers
    -- no task events -- so there is nothing to subscribe to. Polling also means
    a harness on another machine needs only outbound HTTPS: no inbound port, no
    SSH key, and nothing to reconfigure when a device moves or sits behind NAT.
    """
    me = vk.whoami()
    if not me:
        raise VikunjaError("Cannot determine my own username; re-run `vk bootstrap`.")

    # argparse.REMAINDER hands back the `--` separator itself; drop it so the
    # command is what the user actually typed after it.
    if args.command and args.command[0] == "--":
        args.command = args.command[1:]
    if not args.command:
        raise VikunjaError("No command given. Put it after `--`, e.g. "
                           "`vk watch -- nullclaw --task {task_id}`")

    state_file = args.state or _state_path(args.profile or me)
    state = _load_seen(state_file)
    seen = state["tasks"]
    log = lambda msg: print(f"{datetime.now().isoformat(timespec='seconds')} {msg}", flush=True)

    triggers = "new tasks" + (" + comments" if args.comments else "")
    log(f"watching {triggers} as {me} every {args.interval}s "
        f"({'dry run' if args.dry_run else ' '.join(args.command)})")

    backoff = args.interval
    while True:
        try:
            filters = [f"assignees in '{me}'", "done = false"] + list(args.filter or [])
            tasks = vk.tasks(project_id=args.project, filter=" && ".join(filters),
                             sort_by="id")
            backoff = args.interval
        except VikunjaError as e:
            # Never exit on a transient API failure -- this runs unattended.
            log(f"poll failed: {e}")
            time.sleep(min(backoff, 600))
            backoff = min(backoff * 2, 600)
            if args.once:
                return
            continue

        for task in tasks:
            key = str(task["id"])
            if key in seen:
                continue

            # Record before running, not after: at-most-once. A harness that
            # crashes the machine should not be relaunched on every poll.
            seen[key] = time.time()
            _save_seen(state_file, state)

            command = [_substitute(part, task, vk) for part in args.command]
            log(f"task #{task['id']} {task.get('title')!r} -> {' '.join(command)}")
            if args.dry_run:
                continue

            _run_command(args, command, {
                "VIKUNJA_TASK_ID": task["id"],
                "VIKUNJA_TASK_TITLE": task.get("title") or "",
                "VIKUNJA_TASK_URL": f"{vk.url}/tasks/{task['id']}",
                "VIKUNJA_TRIGGER": "task",
            }, lambda m: log(f"task #{task['id']} {m}"))

        if args.comments:
            try:
                _poll_comments(vk, args, state, state_file, me, log)
            except VikunjaError as e:
                log(f"comment poll failed: {e}")

        if args.once:
            return
        time.sleep(args.interval)


def build_parser():
    p = argparse.ArgumentParser(prog="vk", description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--url")
    p.add_argument("--token")
    p.add_argument("--profile", help="Named profile from the config file.")
    p.add_argument("--json", action="store_true", help="Emit raw JSON.")

    # The same global flags again, accepted *after* the subcommand -- `vk tasks
    # --json` is what anyone actually types. SUPPRESS keeps an unspecified flag
    # here from clobbering one given before the subcommand.
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--url", default=argparse.SUPPRESS)
    common.add_argument("--token", default=argparse.SUPPRESS)
    common.add_argument("--profile", default=argparse.SUPPRESS)
    common.add_argument("--json", action="store_true", default=argparse.SUPPRESS,
                        help="Emit raw JSON.")

    sub = p.add_subparsers(dest="command", required=True)
    _add = sub.add_parser

    def add_parser(name, **kw):
        kw.setdefault("parents", [common])
        return _add(name, **kw)

    sub.add_parser = add_parser

    b = sub.add_parser("bootstrap", help="Mint and store an API token for a bot account.")
    b.add_argument("--username", required=True)
    b.add_argument("--password", help="Prompted for, or $VIKUNJA_PASSWORD, if omitted.")
    b.add_argument("--totp")
    b.add_argument("--title", help="Token title shown in Vikunja's UI.")
    b.add_argument("--days", type=int, default=365)
    b.add_argument("--permissions", help="JSON permission map; defaults to a task-work set.")
    b.set_defaults(func=cmd_bootstrap)

    pr = sub.add_parser(
        "profile",
        help="Store a token minted elsewhere (reads it from stdin).",
        description="For bot users, whose owner mints the token for them.\n"
                    "Example: vk profile bot-nullclaw --username bot-nullclaw < token.txt",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    pr.add_argument("name", help="Profile name to store it under.")
    pr.add_argument("--username", required=True, help="The Vikunja account the token belongs to.")
    pr.set_defaults(func=cmd_profile)

    sub.add_parser("whoami", help="Show the account this token belongs to.").set_defaults(func=cmd_whoami)
    sub.add_parser("projects", help="List projects.").set_defaults(func=cmd_projects)

    t = sub.add_parser("tasks", help="List tasks (undone by default).")
    t.add_argument("--project", type=int)
    t.add_argument("--mine", action="store_true", help="Only tasks assigned to me.")
    t.add_argument("--all", action="store_true", help="Include done tasks.")
    t.add_argument("--filter", action="append", help="Raw Vikunja filter, repeatable (ANDed).")
    t.add_argument("--sort", default="due_date")
    t.add_argument("--limit", type=int)
    t.set_defaults(func=cmd_tasks)

    s = sub.add_parser("show", help="Show one task.")
    s.add_argument("task_id", type=int)
    s.add_argument("--comments", action="store_true")
    s.set_defaults(func=cmd_show)

    c = sub.add_parser("create", help="Create a task.")
    c.add_argument("--project", type=int, required=True)
    c.add_argument("--title", required=True)
    c.add_argument("--description")
    c.add_argument("--priority", type=int)
    c.add_argument("--due", help="RFC3339 timestamp.")
    c.add_argument("--assign", help="Username to assign it to.")
    c.set_defaults(func=cmd_create)

    m = sub.add_parser("comment", help="Comment on a task ('-' reads stdin).")
    m.add_argument("task_id", type=int)
    m.add_argument("text")
    m.set_defaults(func=cmd_comment)

    u = sub.add_parser("update", help="Update task fields.")
    u.add_argument("task_id", type=int)
    u.add_argument("--title")
    u.add_argument("--description")
    u.add_argument("--priority", type=int)
    u.add_argument("--done", action="store_true")
    u.add_argument("--undone", action="store_true")
    u.set_defaults(func=cmd_update)

    d = sub.add_parser("done", help="Mark a task done.")
    d.add_argument("task_id", type=int)
    d.set_defaults(func=cmd_done)

    w = sub.add_parser(
        "watch",
        help="Poll my assigned queue and run a command per new task.",
        description="Run on each device that hosts a harness. Outbound HTTPS only.\n"
                    "Example: vk watch --interval 30 -- nullclaw --task {task_id}",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    w.add_argument("--interval", type=int, default=60, help="Seconds between polls.")
    w.add_argument("--project", type=int, help="Only watch one project.")
    w.add_argument("--filter", action="append", help="Extra Vikunja filter, repeatable.")
    w.add_argument("--cwd", help="Working directory for the command.")
    w.add_argument("--timeout", type=int, default=3600, help="Per-task timeout in seconds.")
    w.add_argument("--state", help="Override the handled-task state file.")
    w.add_argument("--once", action="store_true", help="One pass then exit (for cron).")
    w.add_argument("--dry-run", action="store_true", help="Log what would run; run nothing.")
    w.add_argument("--comments", action="store_true",
                   help="Also fire when someone else comments on one of my tasks "
                        "(includes done tasks -- follow-ups usually land there).")
    w.add_argument("command", nargs=argparse.REMAINDER,
                   help="Command to run, after `--`. Placeholders: {task_id} {title} "
                        "{project_id} {priority} {labels} {url}")
    w.set_defaults(func=cmd_watch)

    a = sub.add_parser("assign", help="Assign a task to a user.")
    a.add_argument("task_id", type=int)
    a.add_argument("username")
    a.set_defaults(func=cmd_assign)
    return p


def main():
    args = build_parser().parse_args()
    try:
        args.func(Vikunja(args.url, args.token, args.profile), args)
    except VikunjaError as e:
        print(f"error: {e}", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        sys.exit(130)


if __name__ == "__main__":
    main()