xref: /aosp_15_r20/external/pytorch/tools/linter/adapters/actionlint_linter.py (revision da0073e96a02ea20f0ac840b70461e3646d07c45) 
1from __future__ import annotations
2
3import argparse
4import concurrent.futures
5import json
6import logging
7import os
8import re
9import subprocess
10import sys
11import time
12from enum import Enum
13from typing import NamedTuple
14
15
16LINTER_CODE = "ACTIONLINT"
17
18
19class LintSeverity(str, Enum):
20    ERROR = "error"
21    WARNING = "warning"
22    ADVICE = "advice"
23    DISABLED = "disabled"
24
25
26class LintMessage(NamedTuple):
27    path: str | None
28    line: int | None
29    char: int | None
30    code: str
31    severity: LintSeverity
32    name: str
33    original: str | None
34    replacement: str | None
35    description: str | None
36
37
38RESULTS_RE: re.Pattern[str] = re.compile(
39    r"""(?mx)
40    ^
41    (?P<file>.*?):
42    (?P<line>\d+):
43    (?P<char>\d+):
44    \s(?P<message>.*)
45    \s(?P<code>\[.*\])
46    $
47    """
48)
49
50
51def run_command(
52    args: list[str],
53) -> subprocess.CompletedProcess[bytes]:
54    logging.debug("$ %s", " ".join(args))
55    start_time = time.monotonic()
56    try:
57        return subprocess.run(
58            args,
59            capture_output=True,
60        )
61    finally:
62        end_time = time.monotonic()
63        logging.debug("took %dms", (end_time - start_time) * 1000)
64
65
66def check_file(
67    binary: str,
68    file: str,
69) -> list[LintMessage]:
70    try:
71        proc = run_command(
72            [
73                binary,
74                "-ignore",
75                '"runs-on" section must be sequence node but got mapping node with "!!map" tag',
76                file,
77            ]
78        )
79    except OSError as err:
80        return [
81            LintMessage(
82                path=None,
83                line=None,
84                char=None,
85                code=LINTER_CODE,
86                severity=LintSeverity.ERROR,
87                name="command-failed",
88                original=None,
89                replacement=None,
90                description=(f"Failed due to {err.__class__.__name__}:\n{err}"),
91            )
92        ]
93    stdout = str(proc.stdout, "utf-8").strip()
94    return [
95        LintMessage(
96            path=match["file"],
97            name=match["code"],
98            description=match["message"],
99            line=int(match["line"]),
100            char=int(match["char"]),
101            code=LINTER_CODE,
102            severity=LintSeverity.ERROR,
103            original=None,
104            replacement=None,
105        )
106        for match in RESULTS_RE.finditer(stdout)
107    ]
108
109
110if __name__ == "__main__":
111    parser = argparse.ArgumentParser(
112        description="actionlint runner",
113        fromfile_prefix_chars="@",
114    )
115    parser.add_argument(
116        "--binary",
117        required=True,
118        help="actionlint binary path",
119    )
120    parser.add_argument(
121        "filenames",
122        nargs="+",
123        help="paths to lint",
124    )
125
126    args = parser.parse_args()
127
128    if not os.path.exists(args.binary):
129        err_msg = LintMessage(
130            path="<none>",
131            line=None,
132            char=None,
133            code=LINTER_CODE,
134            severity=LintSeverity.ERROR,
135            name="command-failed",
136            original=None,
137            replacement=None,
138            description=(
139                f"Could not find actionlint binary at {args.binary},"
140                " you may need to run `lintrunner init`."
141            ),
142        )
143        print(json.dumps(err_msg._asdict()), flush=True)
144        sys.exit(0)
145
146    with concurrent.futures.ThreadPoolExecutor(
147        max_workers=os.cpu_count(),
148        thread_name_prefix="Thread",
149    ) as executor:
150        futures = {
151            executor.submit(
152                check_file,
153                args.binary,
154                filename,
155            ): filename
156            for filename in args.filenames
157        }
158        for future in concurrent.futures.as_completed(futures):
159            try:
160                for lint_message in future.result():
161                    print(json.dumps(lint_message._asdict()), flush=True)
162            except Exception:
163                logging.critical('Failed at "%s".', futures[future])
164                raise
165