devops-skills/skills/gitea-issue-devops-agent/scripts/change_scope.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path
from typing import Any

DOC_EXTS = {".md", ".rst", ".txt", ".adoc"}
TEST_MARKERS = ("/tests/", "/test/", "__tests__", ".spec.", ".test.")
CLIENT_PREFIXES = ("client/", "frontend/", "web/", "ui/")
SERVER_PREFIXES = ("server/", "backend/", "api/", "services/", "worker/")
INFRA_PREFIXES = (
    "deploy/",
    ".gitea/workflows/",
    ".github/workflows/",
    "infra/",
    "k8s/",
    "helm/",
    "nginx/",
)
SHARED_RUNTIME_FILES = {
    "package.json",
    "package-lock.json",
    "pnpm-lock.yaml",
    "yarn.lock",
    "requirements.txt",
    "pyproject.toml",
    "poetry.lock",
    "Dockerfile",
    "docker-compose.yml",
    "docker-compose.yaml",
}


def run_git(cwd: Path, *args: str) -> str:
    result = subprocess.run(
        ["git", *args],
        cwd=str(cwd),
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or f"git {' '.join(args)} failed")
    return result.stdout.strip()


def normalize_path(raw: str) -> str:
    return raw.replace("\\", "/").lstrip("./")


def classify_file(path: str) -> set[str]:
    categories: set[str] = set()
    normalized = normalize_path(path)
    lower = normalized.lower()
    suffix = Path(lower).suffix

    if lower.startswith("docs/") or suffix in DOC_EXTS:
        categories.add("docs")
    if any(marker in lower for marker in TEST_MARKERS):
        categories.add("tests")
    if any(lower.startswith(prefix) for prefix in CLIENT_PREFIXES):
        categories.add("client")
    if any(lower.startswith(prefix) for prefix in SERVER_PREFIXES):
        categories.add("server")
    if any(lower.startswith(prefix) for prefix in INFRA_PREFIXES):
        categories.add("infra")
    if Path(lower).name in SHARED_RUNTIME_FILES:
        categories.add("shared_runtime")

    if not categories:
        categories.add("unknown")
    return categories


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Detect branch change scope and recommend minimum deployment strategy."
    )
    parser.add_argument("--repo-path", default=".", help="Local git repository path")
    parser.add_argument("--base-ref", required=True, help="Base ref, e.g. origin/main")
    parser.add_argument("--head-ref", default="HEAD", help="Head ref, e.g. feature branch or SHA")
    parser.add_argument("--no-files", action="store_true", help="Do not include changed file list in output")
    args = parser.parse_args()

    repo_path = Path(args.repo_path).resolve()

    try:
        merge_base = run_git(repo_path, "merge-base", args.base_ref, args.head_ref)
        diff_output = run_git(repo_path, "diff", "--name-only", f"{merge_base}..{args.head_ref}")
    except Exception as error:  # noqa: BLE001
        print(json.dumps({"error": str(error)}, ensure_ascii=False, indent=2))
        sys.exit(2)

    changed_files = [normalize_path(line) for line in diff_output.splitlines() if line.strip()]
    file_classes: list[dict[str, Any]] = []
    category_union: set[str] = set()
    for path in changed_files:
        categories = classify_file(path)
        category_union.update(categories)
        file_classes.append({"path": path, "categories": sorted(categories)})

    has_docs = "docs" in category_union
    has_tests = "tests" in category_union
    has_client = "client" in category_union
    has_server = "server" in category_union
    has_infra = "infra" in category_union
    has_shared_runtime = "shared_runtime" in category_union
    has_unknown = "unknown" in category_union

    only_docs_tests = bool(changed_files) and category_union.issubset({"docs", "tests"})
    scope = "skip"
    reasons: list[str] = []

    if not changed_files:
        scope = "skip"
        reasons.append("no changed files")
    elif only_docs_tests:
        scope = "skip"
        reasons.append("docs/tests-only changes")
    else:
        require_server = has_server or has_shared_runtime
        require_client = has_client or has_shared_runtime

        if has_unknown:
            scope = "full_stack"
            reasons.append("unknown file changes detected -> conservative full-stack deploy")
            require_server = True
            require_client = True
        elif has_infra and not require_server and not require_client:
            scope = "infra_only"
            reasons.append("infra/workflow-only changes")
        elif require_client and not require_server:
            scope = "client_only"
            reasons.append("client-only changes")
        elif require_server and not require_client:
            scope = "server_only"
            reasons.append("server-only changes")
        else:
            scope = "full_stack"
            reasons.append("client/server/shared runtime changes")

    should_restart_server = scope in {"server_only", "full_stack"}
    should_restart_client = scope in {"client_only", "full_stack"}

    payload: dict[str, Any] = {
        "repo_path": str(repo_path.as_posix()),
        "base_ref": args.base_ref,
        "head_ref": args.head_ref,
        "merge_base": merge_base,
        "changed_files_count": len(changed_files),
        "scope": scope,
        "categories": {
            "docs": has_docs,
            "tests": has_tests,
            "client": has_client,
            "server": has_server,
            "infra": has_infra,
            "shared_runtime": has_shared_runtime,
            "unknown": has_unknown,
        },
        "only_docs_tests": only_docs_tests,
        "actions": {
            "skip_deploy": scope == "skip",
            "should_restart_client": should_restart_client,
            "should_restart_server": should_restart_server,
            "reuse_shared_server": not should_restart_server,
            "requires_dedicated_server": should_restart_server,
        },
        "reasons": reasons,
    }
    if not args.no_files:
        payload["changed_files"] = changed_files
        payload["file_classification"] = file_classes

    print(json.dumps(payload, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()