Gitea-Tools/reviewer_mergeability_skip.py

"""Non-mergeable PR skip conflict-proof verifier for reviewer reports (#322)."""

from __future__ import annotations

import re
from typing import Any

_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE)
_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
_MERGEABILITY_FALSE_RE = re.compile(
    r"(?:mergeable\s*:\s*false|not mergeable|non[- ]mergeable|mergeability\s*:\s*false|"
    r"mergeability result\s*:\s*false)",
    re.IGNORECASE,
)
_CONFLICT_PROOF_RE = re.compile(
    r"(?:merge-tree|git merge-tree|gitea_view_pr|gitea_check_pr_eligibility|"
    r"conflict proof|mergeability tool|merge simulation|conflicting files?)",
    re.IGNORECASE,
)
_CONFLICTING_FILES_RE = re.compile(
    r"(?:conflicting files?|conflict files?)\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_HEAD_CHANGED_RE = re.compile(
    r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|"
    r"head changed since|head unchanged since)",
    re.IGNORECASE,
)
_MERGEABILITY_UNVERIFIED_RE = re.compile(
    r"\bMERGEABILITY_UNVERIFIED\b",
)


def _pr_section(text: str, pr_number: int) -> str:
    """Return report lines likely describing one skipped PR."""
    lines = (text or "").splitlines()
    chunks: list[str] = []
    capture = False
    token = f"#{pr_number}"
    for line in lines:
        lower = line.lower()
        if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""):
            capture = True
            chunks.append(line)
            continue
        if capture:
            if _PR_NUMBER_RE.search(line) and token not in line.lower():
                break
            if line.strip() == "" and len(chunks) > 3:
                break
            chunks.append(line)
    if chunks:
        return "\n".join(chunks)
    return text or ""


def _has_head_sha(text: str, head_sha: str | None) -> bool:
    if not head_sha:
        return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text))
    head = head_sha.strip().lower()
    if head in text.lower():
        return True
    if len(head) >= 7 and head[:7] in text.lower():
        return True
    return bool(_FULL_SHA.search(text))


def assess_non_mergeable_skip_proof(
    report_text: str,
    *,
    skipped_prs: list[dict] | None = None,
) -> dict[str, Any]:
    """Validate skipped non-mergeable PR documentation in reviewer reports (#322)."""
    text = report_text or ""
    reasons: list[str] = []
    assessments: list[dict[str, Any]] = []

    for entry in skipped_prs or []:
        pr_number = entry.get("pr_number")
        if pr_number is None:
            reasons.append("skipped PR entry missing pr_number")
            continue
        pr_number = int(pr_number)
        section = _pr_section(text, pr_number)
        mergeable = entry.get("mergeable")
        verified = entry.get("mergeability_verified")
        classification = (entry.get("classification") or "").strip().upper()
        head_sha = (entry.get("head_sha") or "").strip() or None

        item_reasons: list[str] = []

        if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower():
            item_reasons.append(f"skipped PR #{pr_number} not documented in final report")

        if mergeable is False or verified is False:
            if not _MERGEABILITY_UNVERIFIED_RE.search(section) and verified is False:
                if "MERGEABILITY_UNVERIFIED" not in classification:
                    item_reasons.append(
                        f"PR #{pr_number} conflict proof unavailable; report must classify "
                        "MERGEABILITY_UNVERIFIED"
                    )
            elif verified is not False:
                if not _MERGEABILITY_FALSE_RE.search(section):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing mergeability:false proof"
                    )
                if not _has_head_sha(section, head_sha):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing current head SHA"
                    )
                if not _CONFLICT_PROOF_RE.search(section):
                    item_reasons.append(
                        f"PR #{pr_number} skip missing conflict proof command/tool"
                    )
                if entry.get("head_changed_since_prior_blocker") is not None:
                    if not _HEAD_CHANGED_RE.search(section):
                        item_reasons.append(
                            f"PR #{pr_number} skip missing head-changed-since-blocker proof"
                        )

        if (
            entry.get("conflicting_files")
            and not _CONFLICTING_FILES_RE.search(section)
            and not any(
                path.lower() in section.lower()
                for path in (entry.get("conflicting_files") or [])
            )
        ):
            item_reasons.append(
                f"PR #{pr_number} has conflicting files in session proof but "
                "report omits file-level conflict proof"
            )

        proven = not item_reasons
        assessments.append({
            "pr_number": pr_number,
            "proven": proven,
            "classification": classification or (
                "MERGEABILITY_UNVERIFIED" if verified is False else "NON_MERGEABLE_SKIPPED"
            ),
            "reasons": item_reasons,
        })
        reasons.extend(item_reasons)

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "downgraded": False,
        "assessments": assessments,
        "reasons": reasons,
        "safe_next_action": (
            "document each skipped non-mergeable PR with head SHA, mergeability result, "
            "conflict proof command, conflicting files, and head-change status"
            if reasons
            else "proceed"
        ),
    }