marketplace/skills/sync/scripts/labels.py

#!/usr/bin/env python3
"""
labels.py — put the canonical label set into a repository, in one run.

Every `type/*` and every `severity/*` the domain taxonomy defines, created up
front instead of trickling in as a side effect of whichever push first happens
to use one. Until a name exists in the repository nobody can filter by it in
the web UI, so somebody makes their own with a foreign color and without
`exclusive`, and the set arrives in pieces over months.

    labels.py --dry-run            print the plan, write nothing
    labels.py                      create whatever is missing
    labels.py --fix                also patch color / `exclusive` drift
    labels.py --repo owner/repo    outside the repository's own checkout

No label name is spelled out in this file. The names are assembled from the
domain — issue.TYPES, issue.SEVERITIES, issue.EXCLUSIVE_NS — and painted by
map.label_specs; add a type over in skills/issue and the next run creates it.
`tea labels create` cannot set `exclusive` (tea 0.14.2), so creation goes
through `tea api`.

The repository's own labels are read before anything is written. A name that
matches exactly is left alone — never re-created, never patched; a color or
`exclusive` that disagrees with the spec is reported, and corrected only under
--fix. A name that merely RESEMBLES a canonical one (the same tail, up to
case, separator and whatever namespace is in front: `X`, `x`, `kind/x`,
`type: x` against `type/x`) is reported with its id and never touched —
renaming somebody else's label is a decision, not a step.

Out of scope by design: `tech/*` and `comp/*`, which are open-ended and get
created by push as they come up, and deleting or renaming anything at all.
Only repository labels are read; an organization's own labels sit behind a
different endpoint and are neither read nor written.

Login: the operator's pin from .claude/settings.local.json (see /tea:auth).
"""
import argparse
import os
import re
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path[:0] = [_HERE, os.path.normpath(os.path.join(_HERE, "..", "..", "issue", "scripts"))]

import _gitea  # noqa: E402
import issue  # noqa: E402
import map as gmap  # noqa: E402


# --------------------------------------------------------------------------
# the canonical set
# --------------------------------------------------------------------------

# Which taxonomy collection fills which exclusive namespace. Both sides are the
# domain's — this dict is only the join between them, and it is the whole
# reason no name has to be repeated here.
MEMBERS = {"type/": issue.TYPES, "severity/": issue.SEVERITIES}


def canonical_names():
    """Every name in the canonical set, in taxonomy order.

    Which namespaces are exclusive is issue.EXCLUSIVE_NS; what lives in each
    is MEMBERS, i.e. the domain again. A namespace the domain declares but
    MEMBERS does not know about is handed back separately — better reported
    than quietly missing from the set."""
    names, orphan = [], []
    for ns in issue.EXCLUSIVE_NS:
        if ns in MEMBERS:
            names += [ns + m for m in MEMBERS[ns]]
        else:
            orphan.append(ns)
    return names, orphan


# --------------------------------------------------------------------------
# lookalikes
# --------------------------------------------------------------------------

WORDS = re.compile(r'[^a-z0-9]+')


def akin(name):
    """Comparison keys for a label name: its tail, and the whole name squashed.

    Case, separators and the namespace in front are noise — what a person
    meant is the tail. `x`, `X`, `kind/x` all reduce to the same tail as
    `type/x`, and `severity: x y` to the same squashed form as `severity/xy`.
    Two names resemble each other when these sets intersect."""
    parts = [p for p in WORDS.split(name.lower()) if p]
    return {parts[-1], "".join(parts)} if parts else set()


# --------------------------------------------------------------------------
# plan
# --------------------------------------------------------------------------

def color_of(value):
    """Gitea reports colors bare, map.py writes them with a `#`. Same color."""
    return (value or "").lstrip("#").lower()


def drift_of(spec, got):
    """Where an existing label disagrees with the spec, as (field, is, want).

    Only color and `exclusive` — a description somebody rewrote is theirs, and
    the name matched exactly or we would not be here."""
    out = []
    if color_of(got.get("color")) != color_of(spec.get("color")):
        out.append(("color", color_of(got.get("color")), color_of(spec.get("color"))))
    if bool(got.get("exclusive")) != bool(spec.get("exclusive")):
        out.append(("exclusive", str(bool(got.get("exclusive"))).lower(),
                    str(bool(spec.get("exclusive"))).lower()))
    return out


def plan(specs, existing):
    """(rows, similar) for one repository, decided before anything is written.

    A row is (name, spec, got, drift), one per canonical label in taxonomy
    order: `got` is the repository's own payload when that exact name is
    already there (None when it is not), `drift` what disagrees with the spec.

    `similar` is (name, id, [canonical it resembles]) for the repository's
    other labels. They are reported and left alone: this script owns the
    canonical names, not everything that looks like one."""
    by_name = dict((l.get("name", ""), l) for l in existing or [])

    rows = []
    for name in specs:
        got = by_name.get(name)
        rows.append((name, specs[name], got, drift_of(specs[name], got) if got else []))

    keys = dict((name, akin(name)) for name in specs)
    similar = []
    for l in existing or []:
        name = l.get("name", "")
        if name in specs:
            continue
        mine = akin(name)
        hits = [n for n in specs if keys[n] & mine]
        if hits:
            similar.append((name, l.get("id"), hits))
    return rows, similar


# --------------------------------------------------------------------------
# run
# --------------------------------------------------------------------------

def main():
    ap = argparse.ArgumentParser(
        description="Create the canonical type/* and severity/* labels in a repository")
    ap.add_argument("--dry-run", action="store_true",
                    help="print the plan; not one writing request")
    ap.add_argument("--fix", action="store_true",
                    help="also patch color/exclusive on labels that already exist")
    ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)")
    args = ap.parse_args()

    names, orphan = canonical_names()
    for ns in orphan:
        _gitea.warn("namespace %r is exclusive in the domain but has no members here "
                    "— nothing created for it" % ns)
    specs = gmap.label_specs(names)

    login = _gitea.require_login()
    base = _gitea.repo_base(args.repo)

    # Read first, always: the plan is decided against the repository itself,
    # never against tmp/issues/.labels.json. That cache is what makes
    # _gitea.ensure_labels cheap for push.py and wrong for a bootstrap — it
    # answers "what did we create last time", and the answer here has to be
    # "what does the repository have right now".
    existing = _gitea.paginate(login, "%s/labels" % base, limit=100)
    rows, similar = plan(specs, existing)

    fixed, drifted = 0, 0
    for name, spec, got, drift in rows:
        mark = "  exclusive" if spec.get("exclusive") else ""

        if got is None:
            if args.dry_run:
                print("create   %-20s %s%s" % (name, spec["color"], mark))
                continue
            payload = dict(spec, name=name)
            new = _gitea.api(login, "%s/labels" % base, "POST", payload,
                             payload_name="label-%s" % name.replace("/", "-"),
                             out_root=issue.ISSUE_ROOT)
            if not new or "id" not in new:
                _gitea.die("could not create label %r" % name)
            print("created  %-20s id %-5s %s%s" % (name, new["id"], spec["color"], mark))
            continue

        if not drift:
            print("present  %-20s id %s" % (name, got.get("id")))
            continue

        drifted += 1
        shown = ", ".join("%s %s -> %s" % d for d in drift)
        if not args.fix:
            print("present  %-20s id %-5s drift: %s" % (name, got.get("id"), shown))
            continue
        if args.dry_run:
            print("fix      %-20s id %-5s %s" % (name, got.get("id"), shown))
            continue
        # Gitea 1.26 patches only the fields it is given, but the unchanged
        # name and description ride along anyway: they cost nothing and an
        # older server that reads an absent field as empty would blank them.
        patch = {"name": name, "description": got.get("description") or ""}
        for field, _is, _want in drift:
            patch[field] = spec[field]
        _gitea.api(login, "%s/labels/%s" % (base, got.get("id")), "PATCH", patch,
                   payload_name="label-%s" % name.replace("/", "-"),
                   out_root=issue.ISSUE_ROOT)
        fixed += 1
        print("fixed    %-20s id %-5s %s" % (name, got.get("id"), shown))

    for name, id, hits in similar:
        _gitea.warn("%r (id %s) resembles %s — left alone; rename it by hand or ignore it"
                    % (name, id, ", ".join(hits)))

    missing = sum(1 for r in rows if r[2] is None)
    print("%d canonical label(s): %d %s, %d present%s%s"
          % (len(rows), missing, "to create" if args.dry_run else "created",
             len(rows) - missing,
             " (%d drifted, %d fixed)" % (drifted, fixed) if drifted else "",
             ", %d similar" % len(similar) if similar else ""))
    if drifted and not args.fix:
        print("drift is shown, not applied — re-run with --fix to patch color/exclusive")
    if args.dry_run:
        print("dry-run — nothing was written")


if __name__ == "__main__":
    main()