marketplace/skills/use/scripts/_tea.py

#!/usr/bin/env python3
"""
_tea.py — shared helpers for the issue scripts (get / push / list / index).

Not a command. Holds the three things every script needs: the operator's
pinned login, a `tea api` wrapper, and the grep-friendly on-disk issue format.

On-disk format (tmp/issues/<n>.md) — every metadata field is ONE line so that
plain grep works without a parser:

    ---
    number: 42
    state: open
    labels: [type/task, tech/sql]
    assignees: [naudachu]
    milestone: v0.2
    depends: [#12, #15]
    comments: 3
    url: https://host/owner/repo/issues/42
    updated: 2026-08-05T11:20:00Z
    fetched: 2026-08-07T18:40:00Z
    ---
    # Title in English, imperative

    ## Summary
    ...

Login: resolved from .claude/settings.local.json (env.GITEA_LOGIN), walking up
from CWD — the same file /tea:auth writes and the tea-guard hook reads. The
scripts never accept a login argument: the operator's pin is the only identity
they will use. No pin -> exit with a pointer to /tea:auth.
"""
import datetime
import json
import os
import re
import subprocess
import sys
import urllib.parse

ISSUE_ROOT = os.path.join("tmp", "issues")
DRAFT_DIR = "drafts"
LABEL_CACHE = ".labels.json"
PAYLOAD_DIR = ".payload"

# Metadata keys in the order they are rendered. Keep them single-line.
META_ORDER = ["number", "state", "labels", "assignees", "milestone",
              "depends", "comments", "url", "updated", "fetched"]

# Canonical colors + descriptions from references/issue-format.md.
EXCLUSIVE_NS = ("type/", "severity/")
KNOWN_LABELS = {
    "type/bug": ("#ee0701", "Something behaves incorrectly in existing code"),
    "type/task": ("#0e8a16", "Implementation of new functionality"),
    "type/refactor": ("#1d76db", "Internal restructuring; behavior must not change"),
    "type/test": ("#fbca04", "Writing or fixing tests"),
    "type/feature": ("#5319e7", "Container: several issues delivering one unit of business value"),
    "type/draft": ("#cccccc", "Idea captured for later; not ready for work"),
    "severity/low": ("#c2e0c6", ""),
    "severity/medium": ("#fbca04", ""),
    "severity/high": ("#eb6420", ""),
    "severity/showstopper": ("#ee0701", ""),
    "severity/critical": ("#b60205", ""),
}
DEFAULT_COLOR = "#ededed"


def die(msg, code=1):
    sys.stderr.write("%s: %s\n" % (os.path.basename(sys.argv[0]), msg))
    sys.exit(code)


def warn(msg):
    sys.stderr.write("warning: %s\n" % msg)


# --------------------------------------------------------------------------
# login + api
# --------------------------------------------------------------------------

def find_pin(start_dir=None):
    """Walk up from start_dir; return login from the first
    .claude/settings.local.json carrying a non-empty env.GITEA_LOGIN."""
    d = os.path.abspath(start_dir or ".")
    while True:
        p = os.path.join(d, ".claude", "settings.local.json")
        if os.path.isfile(p):
            try:
                with open(p) as f:
                    v = (json.load(f).get("env") or {}).get("GITEA_LOGIN")
                if isinstance(v, str) and v.strip():
                    return v.strip()
            except Exception:
                pass
        parent = os.path.dirname(d)
        if parent == d:
            return None
        d = parent


def require_login():
    login = find_pin(os.getcwd())
    if not login:
        die("no login pinned (.claude/settings.local.json env.GITEA_LOGIN). Run /tea:auth.")
    return login


def tea_api(login, endpoint, method="GET", payload=None, payload_name=None,
            out_root=ISSUE_ROOT, allow_fail=False):
    """Call `tea api`; return parsed JSON (None on empty body).

    payload (a dict) is written to <out_root>/.payload/<payload_name>.json and
    passed as -d @file — the file survives the call for retries and debugging.
    allow_fail returns None instead of exiting when the call fails."""
    cmd = ["tea", "api", "--login", login]
    if method != "GET":
        cmd += ["-X", method]
    if payload is not None:
        pdir = os.path.join(out_root, PAYLOAD_DIR)
        os.makedirs(pdir, exist_ok=True)
        path = os.path.join(pdir, "%s.json" % (payload_name or "request"))
        with open(path, "w") as f:
            json.dump(payload, f, ensure_ascii=False, indent=2)
        cmd += ["-d", "@" + path]
    cmd.append(endpoint)

    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode != 0:
        if allow_fail:
            return None
        die("`tea api %s %s` failed:\n%s" % (method, endpoint, (r.stderr or r.stdout).strip()))
    body = r.stdout.strip()
    if not body:
        return None
    try:
        return json.loads(body)
    except json.JSONDecodeError:
        if allow_fail:
            return None
        die("`tea api %s` returned non-JSON:\n%s" % (endpoint, body[:500]))


def paginate(login, endpoint, limit=50, max_pages=40, **kw):
    """GET a list endpoint page by page; return the concatenated list."""
    sep = "&" if "?" in endpoint else "?"
    out = []
    for page in range(1, max_pages + 1):
        batch = tea_api(login, "%s%spage=%d&limit=%d" % (endpoint, sep, page, limit), **kw)
        if not isinstance(batch, list) or not batch:
            break
        out.extend(batch)
        if len(batch) < limit:
            break
    return out


def repo_base(repo=None):
    """API prefix. Without --repo, let tea fill {owner}/{repo} from CWD."""
    return "repos/%s" % repo if repo else "repos/{owner}/{repo}"


def parse_key(key):
    """Return (number, repo-or-None) from 42 / #42 / owner/repo#42 / URL."""
    key = key.strip()
    m = re.match(r'^https?://[^/]+/([^/]+)/([^/]+)/issues/(\d+)/?$', key)
    if m:
        return int(m.group(3)), "%s/%s" % (m.group(1), m.group(2))
    m = re.match(r'^([\w.-]+/[\w.-]+)#(\d+)$', key)
    if m:
        return int(m.group(2)), m.group(1)
    m = re.match(r'^#?(\d+)$', key)
    if m:
        return int(m.group(1)), None
    die("cannot parse issue key %r (want 42, #42, owner/repo#42, or an issue URL)" % key)


def now_iso():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


# --------------------------------------------------------------------------
# filters
# --------------------------------------------------------------------------

def resolve_milestone(login, base, value):
    """(id, title) for a milestone given by id or title. Exits if unknown.

    Gitea silently IGNORES an unresolvable `milestones=` filter and returns the
    whole backlog, so the milestone must be resolved before it is trusted."""
    got = paginate(login, "%s/milestones?state=all" % base, limit=100)
    for m in got or []:
        if str(m.get("id")) == str(value) or m.get("title") == str(value):
            return m["id"], m.get("title", "")
    have = ", ".join("%s (id %d)" % (m.get("title", ""), m["id"]) for m in got or [])
    die("no milestone %r in this repo — have: %s" % (value, have or "none"))


def issue_matches(iss, milestone_id=None, labels=()):
    """Client-side re-check of a server-side filter — see resolve_milestone."""
    if iss.get("pull_request"):
        return False
    if milestone_id is not None and (iss.get("milestone") or {}).get("id") != milestone_id:
        return False
    names = {l.get("name", "") for l in iss.get("labels") or []}
    return all(l in names for l in labels)


def list_issues(login, base, state="open", labels=(), query=None,
                milestone=None, limit=100):
    """Filtered issue list. Returns (issues, milestone_title).

    One request per page, and the payload already carries issue bodies — a
    whole milestone costs one call, not one per issue."""
    ms_id, ms_title = (None, None)
    if milestone is not None:
        ms_id, ms_title = resolve_milestone(login, base, milestone)

    params = {"state": state, "type": "issues"}
    if labels:
        params["labels"] = ",".join(labels)
    if query:
        params["q"] = query
    if ms_title:
        params["milestones"] = ms_title
    endpoint = "%s/issues?%s" % (base, urllib.parse.urlencode(params))

    got = paginate(login, endpoint, limit=min(limit, 50),
                   max_pages=max(1, -(-limit // min(limit, 50))))
    got = [i for i in got if issue_matches(i, ms_id, labels)]
    return got[:limit], ms_title


# --------------------------------------------------------------------------
# on-disk format
# --------------------------------------------------------------------------

def render_meta(meta):
    """Metadata block; lists inline on one line so grep sees them whole."""
    lines = ["---"]
    for k in META_ORDER:
        if k not in meta:
            continue
        v = meta[k]
        if isinstance(v, (list, tuple)):
            v = "[%s]" % ", ".join(str(x) for x in v)
        lines.append("%s: %s" % (k, v))
    lines.append("---")
    return "\n".join(lines)


def parse_meta(text):
    """Split a local issue/draft file into (meta, title, body).

    meta values are strings, or lists for the `[a, b]` inline form. title is
    the first `# ` heading below the block (stripped out of body)."""
    meta, rest = {}, text
    if text.startswith("---"):
        end = text.find("\n---", 3)
        if end != -1:
            for line in text[3:end].strip().splitlines():
                if ":" not in line:
                    continue
                k, v = line.split(":", 1)
                k, v = k.strip(), v.strip()
                if v.startswith("[") and v.endswith("]"):
                    v = [x.strip() for x in v[1:-1].split(",") if x.strip()]
                meta[k] = v
            rest = text[end + 4:]
    rest = rest.lstrip("\n")

    title = ""
    m = re.match(r'^#\s+(.+?)\s*\n', rest)
    if m:
        title = m.group(1).strip()
        rest = rest[m.end():].lstrip("\n")
    return meta, title, rest


def issue_meta(iss, comments=None):
    return {
        "number": iss["number"],
        "state": iss.get("state", ""),
        "labels": [l.get("name", "") for l in iss.get("labels") or []],
        "assignees": [a.get("login", "") for a in iss.get("assignees") or []],
        "milestone": (iss.get("milestone") or {}).get("title") or "none",
        "depends": ["#%d" % n for n in deps_of(iss)],
        "comments": iss.get("comments", 0) if comments is None else len(comments),
        "url": iss.get("html_url", ""),
        "updated": iss.get("updated_at", ""),
        "fetched": now_iso(),
    }


def render_issue(iss, extra_deps=()):
    meta = issue_meta(iss)
    for n in extra_deps:
        ref = "#%d" % n
        if ref not in meta["depends"]:
            meta["depends"].append(ref)
    body = (iss.get("body") or "").strip() or "(no body)"
    return "%s\n# %s\n\n%s\n" % (render_meta(meta), iss.get("title", ""), body)


def render_comments(number, comments):
    head = render_meta({"number": number, "comments": len(comments), "fetched": now_iso()})
    out = [head, ""]
    for c in comments:
        out.append("## comment %d — %s — %s" % (
            c["id"], (c.get("user") or {}).get("login", ""), (c.get("created_at") or "")[:10]))
        out.append("")
        out.append((c.get("body") or "(empty)").strip())
        out.append("")
    return "\n".join(out)


DEP_SECTIONS = ("## Depends on", "## Issues")


def deps_from_body(body):
    """Issue numbers referenced from the structured `## Depends on` /
    `## Issues` sections only — never from prose, or a --deps walk would drag
    in half the backlog."""
    out, active = [], False
    for line in (body or "").splitlines():
        if line.startswith("## "):
            active = line.strip() in DEP_SECTIONS
            continue
        if active:
            out.extend(int(n) for n in re.findall(r'#(\d+)', line))
    seen, uniq = set(), []
    for n in out:
        if n not in seen:
            seen.add(n)
            uniq.append(n)
    return uniq


def deps_of(iss):
    return deps_from_body(iss.get("body") or "")


# --------------------------------------------------------------------------
# paths
# --------------------------------------------------------------------------

def issue_path(root, n):
    return os.path.join(root, "%d.md" % n)


def comments_path(root, n):
    return os.path.join(root, "%d.comments.md" % n)


def tree_path(root, slug):
    return os.path.join(root, "tree-%s.md" % slug)


def write_file(path, text):
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w") as f:
        f.write(text)
    return path


def read_file(path):
    with open(path) as f:
        return f.read()


# --------------------------------------------------------------------------
# labels
# --------------------------------------------------------------------------

def load_label_ids(login, base, root, names):
    """Map label name -> id for every name in `names`, creating what the repo
    is missing. Cached in <root>/.labels.json; the cache is refreshed from the
    API before anything is created."""
    cache_path = os.path.join(root, LABEL_CACHE)
    cache = {}
    if os.path.isfile(cache_path):
        try:
            cache = json.load(open(cache_path))
        except Exception:
            cache = {}

    if any(n not in cache for n in names):
        cache = {l["name"]: l["id"]
                 for l in paginate(login, "%s/labels" % base, limit=100)}

    for name in names:
        if name in cache:
            continue
        color, desc = KNOWN_LABELS.get(name, (DEFAULT_COLOR, ""))
        payload = {"name": name, "color": color, "description": desc,
                   "exclusive": name.startswith(EXCLUSIVE_NS)}
        created = tea_api(login, "%s/labels" % base, "POST", payload,
                          payload_name="label-%s" % name.replace("/", "-"),
                          out_root=root)
        if not created or "id" not in created:
            die("could not create label %r" % name)
        cache[name] = created["id"]
        sys.stderr.write("created label %s%s\n" %
                         (name, " (exclusive)" if payload["exclusive"] else ""))

    write_file(cache_path, json.dumps(cache, indent=2, sort_keys=True))
    return {n: cache[n] for n in names}