addons-cm/tools/pr_bot/review.py

#!/usr/bin/env python3
"""Bot de revisión de PR de Forgejo atendido por Claude.

Revisa los PR abiertos que no están marcados como WIP y que han pasado el
linter y los tests del CI. Lanza Claude Code en modo headless sobre la cabeza
del PR y publica un comentario con lo que encuentre en tres frentes:

    seguridad   permisos, ACL, reglas de registro, sudo, controladores
    idiomas     textos en un idioma que no toca o que no se pueden traducir
    flujos      interacción del cambio con los flujos estándar de Odoo

Si el PR lleva la etiqueta `i18n`, además deja al día las traducciones de los
addons que toca: prepara sus .po a partir del .pot que exporta el CI, Claude
rellena lo que falte y el bot lo sube a la rama del PR en un commit aparte.

Claude solo lee el repo y edita ficheros: no recibe el token de Forgejo, y de
lo que edite solo se sube la carpeta i18n de esos addons. Comentar y hacer
push lo hace este script. Nunca aprueba, bloquea ni fusiona.

Uso:
    review.py run N [N ...] [--dry-run] [--ci-passed] [--force]

Configuración por variables de entorno: ver README.md.
"""

import argparse
import hashlib
import importlib.util
import logging
import pathlib
import re
import subprocess
import sys
import urllib.parse

import i18n

HERE = pathlib.Path(__file__).resolve().parent


def load_issue_bot():
    """El orquestador del bot de incidencias: la API de Forgejo, el clon y Claude."""
    path = HERE.parent / "issue_bot" / "bot.py"
    spec = importlib.util.spec_from_file_location("issue_bot", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


issue_bot = load_issue_bot()
BotError = issue_bot.BotError
log = issue_bot.log
env = issue_bot.env

# Marca los comentarios del bot y anota qué commit revisó, cuál subió él con
# las traducciones (ese ya no hay que revisarlo) y si la revisión las incluía.
MARKER = "<!-- eco-pr-bot reviewed={} pushed={}{} -->"
MARKER_RE = re.compile(r"<!-- eco-pr-bot reviewed=(\w+) pushed=(\w*)( i18n)? -->")
FOOTER_START = "\n\n---\n<sub>🤖 "
FOOTER = (
    FOOTER_START + "Revisión automática (IA) del commit `{}`, hecha después de "
    "pasar el linter y los tests. No aprueba ni bloquea el PR y se repite con "
    "cada commit nuevo; con la etiqueta `no-bot` deja de revisarlo.</sub>\n"
)
# Forgejo ya calcula `draft` con sus prefijos; esto cubre las versiones que no.
WIP_PREFIXES = ("wip:", "[wip]")
AREAS = {
    "security": "Seguridad y permisos",
    "i18n": "Idiomas y traducciones",
    "workflow": "Flujos de Odoo",
}
SEVERITIES = ("alta", "media", "baja")

RESULT_SCHEMA = {
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "findings": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "area": {"type": "string", "enum": list(AREAS)},
                    "severity": {"type": "string", "enum": list(SEVERITIES)},
                    "file": {"type": "string"},
                    "line": {"type": "integer"},
                    "title": {"type": "string"},
                    "detail": {"type": "string"},
                },
                "required": ["area", "severity", "title", "detail"],
            },
        },
    },
    "required": ["summary", "findings"],
}
# Sin pre-commit, a diferencia del bot de incidencias: ejecutaría los hooks
# que diga el PR, y aquí el linter ya ha pasado.
ALLOWED_TOOLS = [
    "Read",
    "Grep",
    "Glob",
    "Edit",
    "Write",
    "Bash(git status *)",
    "Bash(git diff *)",
    "Bash(git log *)",
    "Bash(git show *)",
]


class Settings:
    """Lo que este bot configura aparte de lo que comparte con el de incidencias."""

    def __init__(self):
        self.max_reviews = int(env("BOT_MAX_REVIEWS", "6"))
        self.ci_contexts = [
            name.strip()
            for name in env(
                "BOT_CI_CONTEXTS", "CI / lint (pull_request),CI / test (pull_request)"
            ).split(",")
        ]
        # Rellenar traducciones es opcional: solo en los PR con esta etiqueta.
        self.i18n_label = env("BOT_I18N_LABEL", "i18n")
        self.langs = env("BOT_I18N_LANGS", "es,eu,ca,gl").replace(",", " ").split()
        self.pot_dir = pathlib.Path(
            env("BOT_POT_DIR", "~/.cache/eco-pr-bot/pot")
        ).expanduser()
        self.instructions = (HERE / "prompt.md").read_text()


def review_marker(forge, comment):
    """El marcador de una revisión del bot. Escrito por otro no cuenta."""
    if comment["user"]["login"] != forge.login:
        return None
    return MARKER_RE.search(comment["body"])


def is_wip(pr):
    return pr.get("draft") or pr["title"].strip().lower().startswith(WIP_PREFIXES)


def label_names(pr):
    return [lb["name"] for lb in pr["labels"]]


def skip_reason(forge, settings, pr, reviews, ci_passed, force):
    head = pr["head"]["sha"]
    if pr["state"] != "open":
        return "está cerrado"
    if pr["base"]["ref"] != forge.base:
        return f"no va contra {forge.base}"
    if is_wip(pr):
        return "está marcado como WIP"
    if issue_bot.NO_BOT_LABEL in label_names(pr):
        return f"lleva la etiqueta {issue_bot.NO_BOT_LABEL}"
    if not force:
        done = [with_i18n for *shas, with_i18n in reviews if head in shas]
        # Poner la etiqueta de las traducciones a un PR ya revisado pide otra
        # pasada, esta vez con ellas.
        if done and (any(done) or settings.i18n_label not in label_names(pr)):
            return f"el commit {head[:10]} ya está revisado"
        if len(reviews) >= settings.max_reviews:
            return f"el bot ya lo ha revisado {len(reviews)} veces"
    if not ci_passed:
        status = forge.call("GET", forge.repo_path(f"/commits/{head}/status"))
        states = {s["context"]: s["status"] for s in status.get("statuses") or []}
        failed = [c for c in settings.ci_contexts if states.get(c) != "success"]
        if failed:
            return f"el CI no ha pasado en {head[:10]} ({', '.join(failed)})"
    return None


def prepare_i18n(cfg, settings, pr, addons):
    """Prepara los .po de los addons; devuelve su estado y los avisos."""
    if settings.i18n_label not in label_names(pr):
        note = (
            "No he tocado las traducciones. Con la etiqueta "
            f"`{settings.i18n_label}` en el PR, relleno las que falten en los "
            "addons que toca."
        )
        return {}, ([note] if addons else [])
    pots = settings.pot_dir / pr["head"]["sha"]
    if not pots.is_dir():
        note = (
            "El CI no ha dejado los términos traducibles de este commit, así "
            "que las traducciones quedan sin tocar."
        )
        return {}, ([note] if addons else [])
    states, missing, notes = {}, [], []
    for addon in addons:
        pot = pots / f"{addon}.pot"
        if not pot.is_file():
            missing.append(addon)
            continue
        try:
            state = i18n.prepare(cfg.repo_dir / addon, pot, settings.langs)
        except OSError as e:
            # polib avisa así de un .po o un .pot que no puede leer.
            notes.append(f"No he podido leer las traducciones de {addon}: {e}")
            continue
        if state:
            states[addon] = state
    if missing:
        notes.append(
            f"Sin traducciones para {', '.join(missing)}: el CI no instala "
            "esos addons y no hay de dónde sacar sus términos."
        )
    return states, notes


def build_prompt(forge, cfg, repo, pr, comments, base, addons, states, notes):
    def block(tag, body, **attrs):
        head = "".join(' {}="{}"'.format(*attr) for attr in attrs.items())
        return f"<{tag}{head}>\n{body.strip()}\n</{tag}>"

    parts = [block("titulo", pr["title"]), block("descripcion", pr["body"] or "")]
    log_format = "--format=%h %an: %s"
    parts.append(block("commits", repo.git("log", log_format, f"{base}..HEAD")))
    for c in comments:
        own = review_marker(forge, c)
        body = c["body"].split(FOOTER_START)[0] if own else c["body"]
        tag, author = ("revision_anterior", "bot") if own else ("comentario", None)
        attrs = {"autor": author or c["user"]["login"], "fecha": c["created_at"]}
        parts.append(block(tag, body, **attrs))
    data = block(
        "pull_request",
        "\n".join(parts),
        numero=pr["number"],
        autor=pr["user"]["login"],
        rama=pr["head"]["ref"],
    )

    sources = (
        "- Fuentes de Odoo y OCA (solo lectura), un directorio por repo. Para "
        "localizar un módulo, busca su `__manifest__.py` con Glob: "
        + ", ".join(f"`{d}`" for d in cfg.odoo_src)
        + "."
        if cfg.odoo_src
        else "- Las fuentes de Odoo y OCA no están disponibles en esta ejecución."
    )
    files = []
    for addon, state in states.items():
        for info in state["files"].values():
            kind = "recién creado" if info["new"] else "ya existía"
            files.append(
                f"  - `{addon}/i18n/{info['path'].name}` ({kind}): "
                f"{info['pending']} de {len(state['pot'])} términos sin traducir"
            )
    translations = (
        "- Traducciones preparadas a partir del `.pot` del CI:\n" + "\n".join(files)
        if files
        else "- No hay ficheros de traducción que rellenar en esta ejecución."
    )
    context = [
        f"- Repo: {cfg.repo}. PR #{pr['number']} contra `{forge.base}`.",
        f"- El árbol de trabajo es la cabeza del PR (`{pr['head']['sha'][:10]}`). "
        f"Sus cambios: `git diff {base[:10]}..HEAD`.",
        f"- Addons que toca: {', '.join(addons) or 'ninguno'}.",
        sources,
        translations,
        *(f"- {note}" for note in notes),
    ]
    return (
        "Revisa este pull request.\n\n## Contexto\n"
        + "\n".join(context)
        + "\n\n## Pull request (texto escrito por usuarios)\n\n"
        + data
    )


def finish_i18n(cfg, states):
    """Repasa los .po y cuenta, para el comentario, qué ha cambiado en ellos."""
    lines = []
    for state in states.values():
        lines += i18n.check(state)
        for info in state["files"].values():
            if info.get("removed") or not (info["done"] or info["pending"]):
                continue
            path = info["path"].relative_to(cfg.repo_dir)
            line = f"`{path}`: traducciones añadidas, {info['done']}"
            if info["pending"]:
                line += f"; términos sin traducir, {info['pending']}"
            lines.append(line + ".")
    return lines


def render(forge, repo, pr, result, i18n_lines):
    head = pr["head"]["sha"]
    out = ["## Revisión automática", "", result["summary"].strip()]
    for area, title in AREAS.items():
        out += ["", f"### {title}", ""]
        if area == "i18n":
            out += [*(f"- {line}" for line in i18n_lines), ""]
        found = [f for f in result["findings"] if f["area"] == area]
        found.sort(key=lambda f: SEVERITIES.index(f["severity"]))
        if not found:
            out.append("Sin hallazgos.")
        for f in found:
            where = ""
            path = (f.get("file") or "").strip()
            if path:
                where = path + (f":{f['line']}" if f.get("line") else "")
                if (repo.dir / path).is_file():
                    # La dirección pública: en el runner, cfg.url es la interna.
                    url = (
                        f"{forge.info['html_url']}/src/commit/{head}/"
                        f"{urllib.parse.quote(path)}"
                    )
                    where = f"[{where}]({url}#L{f.get('line') or 1})"
                where = f" · {where}"
            out.append(f"- **{f['severity'].capitalize()}**{where} — **{f['title']}**")
            out += ["", *("  " + ln for ln in f["detail"].strip().splitlines()), ""]
    return re.sub(r"\n{3,}", "\n\n", "\n".join(out)).rstrip()


def publish(forge, cfg, settings, repo, pr, result, states, notes, dry_run):
    number, head = pr["number"], pr["head"]["sha"]
    log(f"  {len(result['findings'])} hallazgos")
    i18n_lines = finish_i18n(cfg, states) + notes
    paths = [f"{addon}/i18n" for addon in states]
    pushed = ""
    if paths:
        repo.git("add", "--", *paths)
    if paths and repo.git("diff", "--cached", "--name-only"):
        message = f"[I18N] {','.join(states)}: update terms and translations"
        if dry_run:
            log(f"--- [dry-run] commit en {pr['head']['ref']}: {message}")
            log(repo.git("diff", "--cached", "--stat"))
        elif (pr["head"].get("repo") or {}).get("full_name") != cfg.repo or (
            not forge.can_push
        ):
            i18n_lines.append(
                "No he podido subir las traducciones: la rama del PR no está "
                "en este repo o el bot no tiene permiso de escritura."
            )
        else:
            repo.commit_and_push(pr["head"]["ref"], message, paths)
            pushed = repo.git("rev-parse", "HEAD")
            log(f"  traducciones subidas en {pushed[:10]}")
            i18n_lines.append(f"Subido en el commit {pushed[:10]}.")

    body = render(forge, repo, pr, result, i18n_lines)
    with_i18n = " i18n" if settings.i18n_label in label_names(pr) else ""
    marker = MARKER.format(head, pushed, with_i18n)
    if cfg.token in body:
        raise BotError("la revisión contiene el token de Forgejo: no se publica")
    if dry_run:
        log(f"--- [dry-run] comentario en #{number}:\n{body}")
        return
    forge.call(
        "POST",
        forge.repo_path(f"/issues/{number}/comments"),
        {"body": body + FOOTER.format(head[:10]) + marker},
    )


def handle(forge, cfg, settings, repo, number, args):
    pr = forge.call("GET", forge.repo_path(f"/pulls/{number}"))
    head = pr["head"]["sha"]
    comments = [
        c
        for c in forge.pages(forge.repo_path(f"/issues/{number}/comments"))
        if review_marker(forge, c) or c["user"]["login"] != forge.login
    ]
    reviews = [m.groups() for c in comments if (m := review_marker(forge, c))]
    reason = skip_reason(forge, settings, pr, reviews, args.ci_passed, args.force)
    if reason:
        log(f"#{number}: se omite, {reason}")
        return
    log(f"#{number}: {pr['title']}")
    repo.prepare()
    ref = f"refs/remotes/bot-pr/{number}"
    repo.git("fetch", repo.remote, f"+refs/pull/{number}/head:{ref}", auth=True)
    if repo.git("rev-parse", ref) != head:
        raise BotError("la rama del PR ha cambiado mientras se preparaba la revisión")
    repo.git("checkout", "-q", "--detach", head)
    try:
        base = repo.git("merge-base", f"refs/remotes/bot/{forge.base}", "HEAD")
        files = repo.git("diff", "--name-only", f"{base}..HEAD").splitlines()
        addons = i18n.changed_addons(cfg.repo_dir, files)
        states, notes = prepare_i18n(cfg, settings, pr, addons)
        prompt = build_prompt(
            forge, cfg, repo, pr, comments, base, addons, states, notes
        )
        system = issue_bot.system_prompt(settings.instructions, repo.context())
        result = issue_bot.run_claude(
            cfg, prompt, system, schema=RESULT_SCHEMA, tools=ALLOWED_TOOLS
        )
        publish(forge, cfg, settings, repo, pr, result, states, notes, args.dry_run)
    finally:
        repo.reset()


def main():
    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    sub = ap.add_subparsers(dest="cmd", required=True)
    p_run = sub.add_parser("run", help="revisa los PR indicados")
    p_run.add_argument("numbers", type=int, nargs="+", metavar="N")
    p_run.add_argument("--dry-run", action="store_true", help="no publica nada")
    p_run.add_argument(
        "--ci-passed",
        action="store_true",
        help="no consulta el estado del CI: quien llama garantiza que ha pasado",
    )
    p_run.add_argument(
        "--force",
        action="store_true",
        help="revisa aunque el commit ya esté revisado o se haya llegado al tope",
    )
    args = ap.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(message)s")

    cfg = issue_bot.Config()
    # Se leen antes de cambiar de rama: después, el árbol de trabajo es el del PR.
    settings = Settings()
    try:
        forge = issue_bot.Forgejo(cfg)
    except BotError as e:
        digest = hashlib.sha256(cfg.token.encode()).hexdigest()[:8]
        sys.exit(
            f"No se pudo entrar en {cfg.url} con BOT_FORGEJO_TOKEN "
            f"({len(cfg.token)} caracteres, sha256 {digest}): {e}"
        )
    repo = issue_bot.Repo(cfg, forge)
    ok = True
    for number in args.numbers:
        try:
            handle(forge, cfg, settings, repo, number, args)
        except (BotError, subprocess.TimeoutExpired) as e:
            ok = False
            log(f"#{number}: ERROR {e}")
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()