marketplace/skills/sync/scripts/map.py

#!/usr/bin/env python3
"""
map.py — md <-> Gitea JSON. The whole translation, and only the translation.

Pure functions: no network, no filesystem, no argparse. Give it a payload and
it hands back a domain Issue; give it an Issue and it hands back a request
body. That purity is the point — it can be reasoned about and tested without a
Gitea anywhere, and it is the single file to open when the two representations
disagree.

Direction of knowledge: this module imports the domain (issue.py) and is
imported by the transport's callers. The domain never imports this.

What crosses the boundary, and what does not:

    domain            Gitea                 note
    ----------------------------------------------------------------------
    id (slug)         —                     local only; the tracker never sees it
    title             title                 verbatim, both ways
    body              body                  verbatim up, verbatim down except
                                            checkbox state — see
                                            merge_checkbox_state
    state             state                 open/closed, same vocabulary
    labels            labels[]              names both ways; ids only on write
    assignees         assignees[]           logins
    milestone         milestone.title       resolved to an id on write
    depends           —                     slugs; #N is translated at the edge
    —                 number, html_url      lands in extra as gitea:/url:
    —                 ref                   extra as branch:; push fills it from git

`depends:` is the authoritative graph and is always slugs. The body's
`## Depends on` section is human prose and is passed through UNCHANGED in both
directions: a pull seeds `depends:` from the `#N` it finds there, and a push
never rewrites what the author wrote. Deliberate — a translator that edits
prose churns the body on every round trip.
"""
import os
import sys

sys.path.insert(0, os.path.normpath(os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "..", "..", "issue", "scripts")))
import issue  # noqa: E402

# How the taxonomy is painted in Gitea's UI. A hex code says nothing about what
# an issue IS, which is exactly why it lives here and not in the domain.
LABEL_COLORS = {
    "type/bug": "#ee0701",
    "type/task": "#0e8a16",
    "type/refactor": "#1d76db",
    "type/test": "#fbca04",
    "type/feature": "#5319e7",
    "type/draft": "#cccccc",
    "severity/low": "#c2e0c6",
    "severity/medium": "#fbca04",
    "severity/high": "#eb6420",
    "severity/showstopper": "#ee0701",
    "severity/critical": "#b60205",
}
DEFAULT_COLOR = "#ededed"

# What this bridge writes into the domain's `origin:` field. The domain records
# that an issue exists somewhere else; only this module knows where.
ORIGIN = "gitea"

# Metadata key for Gitea's `ref` — the branch an issue is pinned to. A sync
# field: its value is a git branch name and means exactly `ref`, so the domain
# carries it in `extra` and never reads it.
BRANCH_KEY = "branch"


def label_specs(names):
    """{name: {color, description, exclusive}} for the transport to create.

    Exclusivity and meaning come from the domain taxonomy; only the color is
    decided here. `tea labels create` cannot set `exclusive` (as of 0.14.2),
    which is why these go through the API."""
    out = {}
    for name in names:
        desc = ""
        if name.startswith("type/"):
            desc = issue.TYPES.get(name.split("/", 1)[1], "")
        out[name] = {
            "color": LABEL_COLORS.get(name, DEFAULT_COLOR),
            "description": desc,
            "exclusive": name.startswith(issue.EXCLUSIVE_NS),
        }
    return out


def remote_key(repo, number):
    """Stable cross-repo handle: owner/repo#42."""
    return "%s#%d" % (repo, int(number))


def parse_remote_key(key):
    repo, _, num = (key or "").rpartition("#")
    return (repo, int(num)) if repo and num.isdigit() else (None, None)


# --------------------------------------------------------------------------
# Gitea -> domain
# --------------------------------------------------------------------------

def numbers_in_body(body):
    """`#N` referenced from the body's dependency sections, as ints. Used only
    to seed `depends:` on the first pull."""
    return [int(r[1:]) for r in issue.body_dep_refs(body) if r.startswith("#")]


def merge_checkbox_state(remote_body, local_body):
    """The remote body with every tick the local copy already had put back.

    The one exception to "a pull overwrites the body", and it is deliberately
    the narrowest one that works. A tick is **monotone** — an item only ever
    travels `[ ]` -> `[x]` — so the two sides are joined by a set union, not
    reconciled: no base version, no drift tracking, no conflict to resolve. The
    set is a set of item TEXTS, and an item comes out ticked when either side
    has it ticked. Everything else in the body is still the remote's word.

    Matching is on `Checkbox.text`, which the domain parser has already
    stripped and rejoined with single spaces, so rewrapping a long item does
    not cost it its tick. It is otherwise literal: reword an item and it is a
    different item — the tick stays with the wording it was put on.

    **The same text more than once** is read as the rule says, as a set: one
    ticked local item ticks every remote item with that text. The alternative —
    pairing duplicates up by order — is the reading that can still drop a tick
    (local `[ ]` then `[x]`, remote a single line: the ticked one pairs with
    nothing), and dropping a tick is the bug this exists to fix. Two items
    whose text is identical are the same item to whoever reads them.

    Pure: no store, no tracker, no I/O. A `local_body` of None or "" — a first
    pull, an empty store — returns the remote body untouched.

    The price, accepted explicitly: UNticking is not monotone, so a box
    unticked in the web UI comes back on the next pull. Untick locally, push.
    """
    ticked = {c.text for c in issue.checkboxes(local_body) if c.checked}
    if not ticked:
        return remote_body
    body = remote_body
    # set_checkbox trades one character for one character, so line numbers read
    # off `remote_body` stay valid against the partially rewritten `body`.
    for c in issue.checkboxes(remote_body):
        if not c.checked and c.text in ticked:
            body = issue.set_checkbox(body, c.line, True)
    return body


def from_api(payload, id, repo, id_for_number=None, extra_numbers=(), synced=None,
             local_body=None):
    """Build a domain Issue from a Gitea issue payload.

    id_for_number maps a Gitea number to a local slug — dependencies whose
    target has not been pulled yet are dropped from `depends:` (the body still
    names them, so nothing is lost) rather than invented.

    `local_body` is the body of the copy already in the store, when there is
    one. It contributes exactly one thing: its ticked checkboxes survive the
    overwrite (merge_checkbox_state). Pass None and the remote body is taken
    whole, which is what a first pull does."""
    body = merge_checkbox_state((payload.get("body") or "").strip(), local_body)
    id_for_number = id_for_number or {}

    numbers = list(numbers_in_body(body))
    for n in extra_numbers:
        if n not in numbers:
            numbers.append(n)
    depends, unresolved = [], []
    for n in numbers:
        slug = id_for_number.get(n)
        if slug and slug != id and slug not in depends:
            depends.append(slug)
        elif not slug:
            unresolved.append(n)

    extra = {
        "gitea": remote_key(repo, payload["number"]),
        "url": payload.get("html_url", ""),
        "synced": synced or "",
    }
    if payload.get("ref"):
        extra[BRANCH_KEY] = payload["ref"]
    if payload.get("updated_at"):
        extra["remote-updated"] = payload["updated_at"]
    if payload.get("comments"):
        extra["comments"] = payload["comments"]

    iss = issue.Issue(
        id=id,
        title=payload.get("title", ""),
        body=body,
        state=payload.get("state") or "open",
        labels=[l.get("name", "") for l in payload.get("labels") or []],
        assignees=[a.get("login", "") for a in payload.get("assignees") or []],
        milestone=(payload.get("milestone") or {}).get("title") or "",
        depends=depends,
        origin=ORIGIN,
        extra=extra)
    return iss, unresolved


def render_comments(comments):
    """Comment thread as flat markdown. Read-only: nothing writes it back."""
    out = []
    for c in comments:
        out.append("## comment %s — %s — %s" % (
            c.get("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)


# --------------------------------------------------------------------------
# domain -> Gitea
# --------------------------------------------------------------------------

def to_payload(iss, label_ids=None, milestone_id=None, include_state=False):
    """Request body for POST /issues or PATCH /issues/{n}.

    The body is sent verbatim — see the module docstring on why slugs in
    `## Depends on` are not rewritten to `#N`."""
    payload = {"title": iss.title, "body": iss.body.strip()}
    if label_ids is not None:
        payload["labels"] = [label_ids[l] for l in iss.labels if l in label_ids]
    if iss.assignees:
        payload["assignees"] = list(iss.assignees)
    if milestone_id is not None:
        payload["milestone"] = milestone_id
    if include_state:
        payload["state"] = iss.state
    # An empty `branch:` is "no opinion", not "no branch": sending ref="" would
    # clear whatever is set on the Gitea side, so the key is left out instead.
    branch = (iss.extra.get(BRANCH_KEY) or "").strip()
    if branch:
        payload["ref"] = branch
    return payload


def apply_remote(iss, payload, repo, synced):
    """Stamp the sync-owned fields onto an issue after a successful write.
    Mutates and returns it; `origin` is the one domain field this touches."""
    iss.origin = ORIGIN
    iss.extra["gitea"] = remote_key(repo, payload["number"])
    iss.extra["url"] = payload.get("html_url", "")
    iss.extra["synced"] = synced
    if payload.get("updated_at"):
        iss.extra["remote-updated"] = payload["updated_at"]
    return iss


def number_of(iss):
    """Gitea number for an already-synced issue, or None."""
    _repo, n = parse_remote_key(iss.extra.get("gitea", ""))
    return n