paperclip-harness/bin/lib/project.py

"""The project convention: one project, one folder, one machine.

Paperclip owns the project record, its tasks, blockers, comments, runs, cost and
worktrees. What it cannot know is which folder on which machine holds the work,
which TUIOS session shows it, and which task the viewer is looking at. That is
this file's whole job:

    <project>/.harness/harness.toml

Nothing else is stored. The task list, the blocker edges and the worktrees are
read from Paperclip every time.
"""

import os
import socket
import sys
import tomllib

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import harness as h  # noqa: E402

REL = os.path.join(".harness", "harness.toml")


def path(root):
    return os.path.join(root, REL)


def exists(root):
    return os.path.isfile(path(root))


def load(root):
    p = path(root)
    if not os.path.isfile(p):
        raise h.Refused(2, "no %s here; run bin/office %s first" % (REL, root))
    with open(p, "rb") as f:
        return tomllib.load(f)


def save(root, cfg):
    os.makedirs(os.path.dirname(path(root)), exist_ok=True)
    lines = [
        "# Written by bin/office. One project, one folder, one machine.",
        "# Paperclip owns the project, its tasks and its worktrees; this file only",
        "# remembers where the work is and what to show.",
        "",
        'session = "%s"' % cfg["session"],
        'watching = "%s"' % (cfg.get("watching") or ""),
        "",
        "[paperclip]",
        'company = "%s"' % cfg["paperclip"]["company"],
        'project = "%s"' % cfg["paperclip"]["project"],
        "",
        "[machine]",
        'host = "%s"' % cfg["machine"]["host"],
        'folder = "%s"' % cfg["machine"]["folder"],
        "",
    ]
    with open(path(root), "w") as f:
        f.write("\n".join(lines))


def this_machine():
    return socket.gethostname()


def check_machine(cfg):
    """Refuse to act on a project that belongs to another machine."""
    want = (cfg.get("machine") or {}).get("host") or ""
    here = this_machine()
    if want and want not in (here, here.split(".")[0]):
        raise h.Refused(
            6,
            "this project lives on %r but this machine is %r; run the command there, "
            "or pass --host %s" % (want, here, want),
        )


def resolve_project(ref):
    """Accept a project id, url key or name, and return the Paperclip record."""
    projects = h.api("GET", "/api/companies/%s/projects" % h.COMPANY)
    for p in projects:
        if ref in (p.get("id"), p.get("urlKey"), p.get("name")):
            return p
    raise h.Refused(2, "no Paperclip project matches %r" % ref)


def active_tasks(project_id):
    issues = h.api("GET", "/api/companies/%s/issues" % h.COMPANY)
    mine = [i for i in issues if i.get("projectId") == project_id]
    order = {"in_progress": 0, "blocked": 1, "in_review": 2, "todo": 3, "backlog": 4, "done": 5, "cancelled": 6}
    mine.sort(key=lambda i: (order.get(i.get("status"), 9), i.get("updatedAt") or ""), reverse=False)
    return mine


def watch_task(cfg):
    """Which task the reply and log panes should show.

    A person's last choice wins if it still exists; otherwise the most active
    task in the project, so the panes are never empty for no reason.
    """
    tasks = active_tasks(cfg["paperclip"]["project"])
    wanted = cfg.get("watching") or ""
    for t in tasks:
        if t.get("identifier") == wanted or t.get("id") == wanted:
            return t["identifier"]
    for t in tasks:
        if t.get("status") not in ("done", "cancelled"):
            return t["identifier"]
    return tasks[0]["identifier"] if tasks else "none"


def set_watch(root, cfg, identifier):
    cfg["watching"] = identifier
    save(root, cfg)


# --- the registry of adopted projects -------------------------------------
#
# One project is one folder on one machine. Paperclip owns the project record;
# what it cannot know is where the folder is and which session shows it, so that
# is kept here, beside the person's other configuration.

REGISTRY = os.path.join(os.path.expanduser("~"), ".config", "paperclip-harness", "projects.toml")


def register(cfg, name):
    """Record an adopted project so it can be listed and switched to later."""
    os.makedirs(os.path.dirname(REGISTRY), exist_ok=True)
    rows = {}
    if os.path.isfile(REGISTRY):
        with open(REGISTRY, "rb") as f:
            try:
                rows = tomllib.load(f).get("projects", {})
            except Exception:
                rows = {}
    key = cfg["paperclip"]["project"]
    rows[key] = {
        "name": name,
        "folder": cfg["machine"]["folder"],
        "host": cfg["machine"]["host"],
        "session": cfg["session"],
    }
    lines = ["# Adopted projects: one folder, one machine, one session each.",
             "# Written by bin/office. Paperclip owns the project itself; this is only where it lives.", ""]
    for k, v in rows.items():
        lines.append("[projects.%r]" % k)
        lines.append('name = "%s"' % v["name"])
        lines.append('folder = "%s"' % v["folder"])
        lines.append('host = "%s"' % v["host"])
        lines.append('session = "%s"' % v["session"])
        lines.append("")
    with open(REGISTRY, "w") as f:
        f.write("\n".join(lines))


def registered():
    """Every adopted project, as a list of dicts."""
    if not os.path.isfile(REGISTRY):
        return []
    with open(REGISTRY, "rb") as f:
        try:
            data = tomllib.load(f)
        except Exception:
            return []
    return [dict(v, id=k) for k, v in data.get("projects", {}).items()]