path: root/tools/ci/issue_labeler.py

#!/usr/bin/env python3
#
# Copyright (c) 2026, RT-Thread Development Team
#
# SPDX-License-Identifier: Apache-2.0
#

"""Add existing RT-Thread labels from structured GitHub issue-form fields."""

import json
import os
import re
import sys
import unicodedata
from dataclasses import dataclass
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen


API_VERSION = "2022-11-28"
FORM_SECTION = re.compile(r"^###\s+(.+?)\s*$", re.MULTILINE)

AREA_LABELS = {
    "kernel": ("Kernel",),
    "board support package (bsp)": ("BSP",),
    "device drivers": ("Component", "component: drivers"),
    "file systems (dfs)": ("Component",),
    "networking": ("Component", "component: net"),
    "ipc": ("Component", "component: ipc"),
    "finsh": ("Component", "component: finsh"),
    "rt-smart": ("RT-Smart",),
    "architecture / libcpu": ("libcpu",),
    "documentation": ("Doc",),
    "build system and tools": ("tools",),
    "ci / github actions": ("action",),
    "tests and examples": ("testcase",),
}

VENDOR_LABELS = {
    "stm32": ("BSP", "BSP: STM32"),
    "nxp": ("BSP", "BSP: NXP"),
    "gd32": ("BSP", "BSP: GD32"),
    "at32": ("BSP", "BSP: AT32"),
    "infineon": ("BSP", "BSP: Infineon"),
    "renesas": ("BSP", "BSP: Renesas"),
    "phytium": ("BSP", "BSP: Phytium"),
    "cvitek": ("BSP", "BSP: Cvitek"),
    "wch": ("BSP", "BSP: WCH"),
    "raspberry pi": ("BSP", "BSP: Raspberry-Pi"),
    "rockchip": ("BSP", "BSP: Rockchip"),
    "loongson": ("BSP", "BSP: Loongson"),
    "esp32": ("BSP", "BSP: ESP32"),
    "hc32": ("BSP", "BSP: HC32"),
    "novosns": ("BSP", "BSP: Novosns"),
    "hpmicro": ("BSP", "BSP: HPMicro"),
    "bluetrum": ("BSP", "BSP: Bluetrum"),
    "k230": ("BSP", "BSP: K230"),
    "nuvoton": ("BSP", "BSP: Nuvoton"),
}

ARCHITECTURE_LABELS = {
    "arm / aarch64": ("Arch: ARM/AArch64",),
    "risc-v": ("Arch: RISC-V",),
}

FIELD_MAPPINGS = {
    "affected area": AREA_LABELS,
    "hardware/bsp vendor": VENDOR_LABELS,
    "architecture": ARCHITECTURE_LABELS,
}

MUTUALLY_EXCLUSIVE_FIELDS = {
    "hardware/bsp vendor": frozenset(
        label.casefold()
        for labels in VENDOR_LABELS.values()
        for label in labels
        if label.casefold().startswith("bsp:")
    ),
    "architecture": frozenset(
        label.casefold()
        for labels in ARCHITECTURE_LABELS.values()
        for label in labels
    ),
}
EMPTY_FIELD_VALUES = frozenset(("not applicable / other",))


@dataclass(frozen=True)
class EventContext:
    repository: str
    issue_number: int
    issue: dict


class GitHubAPIError(RuntimeError):
    def __init__(self, status, message):
        super().__init__(message)
        self.status = status

    def __str__(self):
        return "GitHub API request failed with status {}: {}".format(
            self.status, super().__str__()
        )


class GitHubClient:
    def __init__(self, token, api_url="https://api.github.com"):
        if not token:
            raise ValueError("GITHUB_TOKEN is required")
        self.token = token
        self.api_url = api_url.rstrip("/")

    def request(self, method, path, payload=None, expected=(200,)):
        headers = {
            "Accept": "application/vnd.github+json",
            "Authorization": "Bearer {}".format(self.token),
            "User-Agent": "rt-thread-issue-labeler",
            "X-GitHub-Api-Version": API_VERSION,
        }
        body = None
        if payload is not None:
            body = json.dumps(payload).encode("utf-8")
            headers["Content-Type"] = "application/json"

        request = Request(
            self.api_url + path, data=body, headers=headers, method=method
        )
        try:
            with urlopen(request, timeout=30) as response:
                status = response.getcode()
                response_body = response.read()
        except HTTPError as error:
            try:
                response_body = json.loads(error.read().decode("utf-8"))
                message = response_body.get("message", error.reason)
            except (ValueError, UnicodeDecodeError):
                message = error.reason
            raise GitHubAPIError(error.code, str(message)) from error
        except URLError as error:
            raise GitHubAPIError("network", str(error.reason)) from error

        if status not in expected:
            raise GitHubAPIError(status, "unexpected response status")
        return json.loads(response_body.decode("utf-8")) if response_body else None

    @staticmethod
    def repository_path(repository):
        owner, name = repository.split("/", 1)
        return "/repos/{}/{}".format(quote(owner, safe=""), quote(name, safe=""))

    def list_labels(self, repository):
        labels = []
        repository_path = self.repository_path(repository)
        for page in range(1, 11):
            query = urlencode({"per_page": 100, "page": page})
            batch = self.request("GET", "{}/labels?{}".format(repository_path, query))
            labels.extend(batch)
            if len(batch) < 100:
                break
        return labels

    def get_issue(self, repository, issue_number):
        return self.request(
            "GET",
            "{}/issues/{}".format(self.repository_path(repository), issue_number),
        )

    def add_labels(self, repository, issue_number, labels):
        return self.request(
            "POST",
            "{}/issues/{}/labels".format(
                self.repository_path(repository), issue_number
            ),
            {"labels": list(labels)},
        )

    def remove_label(self, repository, issue_number, label):
        try:
            return self.request(
                "DELETE",
                "{}/issues/{}/labels/{}".format(
                    self.repository_path(repository),
                    issue_number,
                    quote(label, safe=""),
                ),
            )
        except GitHubAPIError as error:
            if error.status == 404:
                return None
            raise


def normalize(value):
    return " ".join(
        unicodedata.normalize("NFKC", str(value or "")).strip().casefold().split()
    )


def parse_form_sections(body):
    text = str(body or "")
    matches = list(FORM_SECTION.finditer(text))
    sections = {}
    for index, match in enumerate(matches):
        start = match.end()
        end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
        value = text[start:end].strip()
        if value == "_No response_":
            value = ""
        sections[normalize(match.group(1))] = value
    return sections


def desired_labels(body):
    sections = parse_form_sections(body)
    result = []
    seen = set()
    for field, mapping in FIELD_MAPPINGS.items():
        value = normalize(sections.get(field, ""))
        for label in mapping.get(value, ()):
            key = label.casefold()
            if key not in seen:
                seen.add(key)
                result.append(label)
    return tuple(result)


def event_context(event):
    if not isinstance(event, dict):
        return None
    if event.get("action") not in ("opened", "edited", "reopened"):
        return None
    issue = event.get("issue") or {}
    if "pull_request" in issue:
        return None
    repository = (event.get("repository") or {}).get("full_name")
    try:
        issue_number = int(issue["number"])
    except (KeyError, TypeError, ValueError):
        return None
    if not repository or repository.count("/") != 1:
        return None
    return EventContext(repository, issue_number, issue)


def existing_label_names(issue):
    names = set()
    for label in issue.get("labels") or []:
        name = label.get("name") if isinstance(label, dict) else label
        if isinstance(name, str):
            names.add(name.casefold())
    return names


def existing_label_map(issue):
    labels = {}
    for label in issue.get("labels") or []:
        name = label.get("name") if isinstance(label, dict) else label
        if isinstance(name, str):
            labels[name.casefold()] = name
    return labels


def obsolete_labels(body, issue):
    sections = parse_form_sections(body)
    existing = existing_label_map(issue)
    obsolete = []
    for field, managed in MUTUALLY_EXCLUSIVE_FIELDS.items():
        value = normalize(sections.get(field, ""))
        mapping = FIELD_MAPPINGS[field]
        if value not in mapping and value not in EMPTY_FIELD_VALUES:
            continue
        selected = {label.casefold() for label in mapping.get(value, ())}
        obsolete.extend(
            existing[label]
            for label in sorted(managed)
            if label in existing and label not in selected
        )
    return tuple(obsolete)


def apply_labels(client, context, dry_run=False):
    issue = client.get_issue(context.repository, context.issue_number) or context.issue
    requested = desired_labels(issue.get("body", ""))
    available = {
        label["name"].casefold(): label["name"]
        for label in client.list_labels(context.repository)
        if isinstance(label, dict) and isinstance(label.get("name"), str)
    }
    existing = existing_label_names(issue)
    missing = [label for label in requested if label.casefold() not in available]
    to_add = [
        available[label.casefold()]
        for label in requested
        if label.casefold() in available and label.casefold() not in existing
    ]
    to_remove = list(obsolete_labels(issue.get("body", ""), issue))

    if missing:
        print("Skipping labels that do not exist: {}".format(", ".join(missing)))
    if to_add and not dry_run:
        client.add_labels(context.repository, context.issue_number, to_add)
    if to_remove and not dry_run:
        for label in to_remove:
            client.remove_label(context.repository, context.issue_number, label)

    return {
        "requested": list(requested),
        "added": to_add,
        "removed": to_remove,
        "missing": missing,
    }


def load_event(path):
    with open(path, "r", encoding="utf-8") as event_file:
        return json.load(event_file)


def main():
    event_path = os.getenv("GITHUB_EVENT_PATH")
    if not event_path:
        print("GITHUB_EVENT_PATH is required", file=sys.stderr)
        return 2
    try:
        event = load_event(event_path)
    except (OSError, ValueError) as error:
        print("Unable to read GitHub event: {}".format(error), file=sys.stderr)
        return 2

    context = event_context(event)
    if context is None:
        print("Ignoring event: no supported issue")
        return 0

    try:
        client = GitHubClient(
            os.getenv("GITHUB_TOKEN", ""),
            os.getenv("GITHUB_API_URL", "https://api.github.com"),
        )
        result = apply_labels(
            client,
            context,
            dry_run=normalize(os.getenv("DRY_RUN", "")) in ("1", "true", "yes"),
        )
    except (GitHubAPIError, ValueError) as error:
        print(str(error), file=sys.stderr)
        return 1

    print(json.dumps(result, ensure_ascii=True, sort_keys=True))
    return 0


if __name__ == "__main__":
    sys.exit(main())