1import json
2import subprocess
3import sys
4import tomllib
5from pathlib import Path
6from typing import Any
7
8import click
9from plain.cli import register_cli
10from plain.cli.print import print_event
11from plain.cli.runtime import common_command, without_runtime_setup
12
13from .annotations import AnnotationResult, check_annotations
14from .oxc import OxcTool, install_oxc
15
16DEFAULT_RUFF_CONFIG = Path(__file__).parent / "ruff_defaults.toml"
17
18
19@without_runtime_setup
20@register_cli("code")
21@click.group()
22def cli() -> None:
23 """Code formatting and linting"""
24
25
26@without_runtime_setup
27@cli.command()
28@click.option("--force", is_flag=True, help="Reinstall even if up to date")
29@click.pass_context
30def install(ctx: click.Context, force: bool) -> None:
31 """Install or update oxlint and oxfmt binaries"""
32 config = get_code_config()
33
34 if not config.get("oxc", {}).get("enabled", True):
35 click.secho("Oxc is disabled in configuration", fg="yellow")
36 return
37
38 oxlint = OxcTool("oxlint")
39 oxfmt = OxcTool("oxfmt")
40
41 if force or not (oxlint.is_installed() and oxfmt.is_installed()):
42 version_to_install = config.get("oxc", {}).get("version", "")
43 if version_to_install:
44 click.secho(
45 f"Installing oxlint and oxfmt {version_to_install}...",
46 bold=True,
47 nl=False,
48 )
49 installed = install_oxc(version_to_install)
50 click.secho(f"oxlint and oxfmt {installed} installed", fg="green")
51 else:
52 ctx.invoke(update)
53 else:
54 click.secho("oxlint and oxfmt already installed", fg="green")
55
56
57@without_runtime_setup
58@cli.command()
59def update() -> None:
60 """Update oxlint and oxfmt to latest version"""
61 config = get_code_config()
62
63 if not config.get("oxc", {}).get("enabled", True):
64 click.secho("Oxc is disabled in configuration", fg="yellow")
65 return
66
67 click.secho("Updating oxlint and oxfmt...", bold=True)
68 version = install_oxc()
69 click.secho(f"oxlint and oxfmt {version} installed", fg="green")
70
71
72def _partition_paths(paths: tuple[str, ...]) -> tuple[tuple[str, ...], tuple[str, ...]]:
73 """Split paths into (python_paths, other_paths).
74
75 Directories go into both groups. Files are routed by extension.
76 """
77 python_paths: list[str] = []
78 other_paths: list[str] = []
79 for p in paths:
80 if Path(p).is_dir():
81 python_paths.append(p)
82 other_paths.append(p)
83 elif Path(p).suffix == ".py":
84 python_paths.append(p)
85 else:
86 other_paths.append(p)
87 return tuple(python_paths), tuple(other_paths)
88
89
90@without_runtime_setup
91@cli.command()
92@click.pass_context
93@click.argument("paths", nargs=-1)
94@click.option("--skip-ruff", is_flag=True, help="Skip Ruff checks")
95@click.option("--skip-ty", is_flag=True, help="Skip ty type checks")
96@click.option("--skip-oxc", is_flag=True, help="Skip oxlint and oxfmt checks")
97@click.option("--skip-annotations", is_flag=True, help="Skip type annotation checks")
98def check(
99 ctx: click.Context,
100 paths: tuple[str, ...],
101 skip_ruff: bool,
102 skip_ty: bool,
103 skip_oxc: bool,
104 skip_annotations: bool,
105) -> None:
106 """Check for formatting and linting issues"""
107 if not paths:
108 paths = (".",)
109
110 python_paths, other_paths = _partition_paths(paths)
111 ruff_args = ["--config", str(DEFAULT_RUFF_CONFIG)]
112 config = get_code_config()
113
114 for e in config.get("exclude", []):
115 ruff_args.extend(["--exclude", e])
116
117 def maybe_exit(return_code: int) -> None:
118 if return_code != 0:
119 click.secho(
120 "\nCode check failed. Run `plain fix` and/or fix issues manually.",
121 fg="red",
122 err=True,
123 )
124 sys.exit(return_code)
125
126 if not skip_ruff and python_paths:
127 print_event("ruff check...", newline=False)
128 result = subprocess.run(
129 ["ruff", "check", *python_paths, *ruff_args], check=False
130 )
131 maybe_exit(result.returncode)
132
133 print_event("ruff format --check...", newline=False)
134 result = subprocess.run(
135 ["ruff", "format", *python_paths, "--check", *ruff_args], check=False
136 )
137 maybe_exit(result.returncode)
138
139 if not skip_ty and python_paths and config.get("ty", {}).get("enabled", True):
140 print_event("ty check...", newline=False)
141 ty_args = ["ty", "check", *python_paths, "--no-progress"]
142 for e in config.get("exclude", []):
143 ty_args.extend(["--exclude", e])
144 result = subprocess.run(ty_args, check=False)
145 maybe_exit(result.returncode)
146
147 if not skip_oxc and other_paths and config.get("oxc", {}).get("enabled", True):
148 oxlint = OxcTool("oxlint")
149 oxfmt = OxcTool("oxfmt")
150
151 if not (oxlint.is_installed() and oxfmt.is_installed()):
152 ctx.invoke(install)
153
154 print_event("oxlint...", newline=False)
155 result = oxlint.invoke(*other_paths)
156 maybe_exit(result.returncode)
157
158 print_event("oxfmt --check...", newline=False)
159 result = oxfmt.invoke("--check", *other_paths)
160 maybe_exit(result.returncode)
161
162 if (
163 not skip_annotations
164 and python_paths
165 and config.get("annotations", {}).get("enabled", True)
166 ):
167 print_event("annotations...", newline=False)
168 # Combine top-level exclude with annotation-specific exclude
169 exclude_patterns = list(config.get("exclude", []))
170 exclude_patterns.extend(config.get("annotations", {}).get("exclude", []))
171 ann_result = check_annotations(
172 *python_paths, exclude_patterns=exclude_patterns or None
173 )
174 if ann_result.missing_count > 0:
175 click.secho(
176 f"{ann_result.missing_count} functions are untyped",
177 fg="red",
178 )
179 click.secho("Run 'plain code annotations --details' for details")
180 maybe_exit(1)
181 else:
182 click.secho("All functions typed!", fg="green")
183
184
185@without_runtime_setup
186@cli.command()
187@click.argument("paths", nargs=-1)
188@click.option("--details", is_flag=True, help="List untyped functions")
189@click.option("--json", "as_json", is_flag=True, help="Output as JSON")
190def annotations(paths: tuple[str, ...], details: bool, as_json: bool) -> None:
191 """Check type annotation status"""
192 if not paths:
193 paths = (".",)
194 config = get_code_config()
195 # Combine top-level exclude with annotation-specific exclude
196 exclude_patterns = list(config.get("exclude", []))
197 exclude_patterns.extend(config.get("annotations", {}).get("exclude", []))
198 result = check_annotations(*paths, exclude_patterns=exclude_patterns or None)
199 if as_json:
200 _print_annotations_json(result)
201 else:
202 _print_annotations_report(result, show_details=details)
203
204
205def _print_annotations_report(
206 result: AnnotationResult,
207 show_details: bool = False,
208) -> None:
209 """Print the annotation report with colors."""
210 if result.total_functions == 0:
211 click.echo("No functions found")
212 return
213
214 # Detailed output first (if enabled and there are untyped functions)
215 if show_details and result.missing_count > 0:
216 # Collect all untyped functions with full paths
217 untyped_items: list[tuple[str, str, int, list[str]]] = []
218
219 for stats in result.file_stats:
220 for func in stats.functions:
221 if not func.is_fully_typed:
222 issues = []
223 if not func.has_return_type:
224 issues.append("return type")
225 missing_params = func.total_params - func.typed_params
226 if missing_params > 0:
227 param_word = "param" if missing_params == 1 else "params"
228 issues.append(f"{missing_params} {param_word}")
229 untyped_items.append((stats.path, func.name, func.line, issues))
230
231 # Sort by file path, then line number
232 untyped_items.sort(key=lambda x: (x[0], x[2]))
233
234 # Print each untyped function
235 for file_path, func_name, line, issues in untyped_items:
236 location = click.style(f"{file_path}:{line}", fg="cyan")
237 issue_str = click.style(f"({', '.join(issues)})", dim=True)
238 click.echo(f"{location} {func_name} {issue_str}")
239
240 click.echo()
241
242 # Summary line
243 pct = result.coverage_percentage
244 color = "green" if result.missing_count == 0 else "red"
245 click.secho(
246 f"{pct:.1f}% typed ({result.fully_typed_functions}/{result.total_functions} functions)",
247 fg=color,
248 )
249
250 # Code smell indicators (only if present)
251 smells = []
252 if result.total_ignores > 0:
253 smells.append(f"{result.total_ignores} ignore")
254 if result.total_casts > 0:
255 smells.append(f"{result.total_casts} cast")
256 if result.total_asserts > 0:
257 smells.append(f"{result.total_asserts} assert")
258 if smells:
259 click.secho(f"{', '.join(smells)}", fg="yellow")
260
261
262def _print_annotations_json(result: AnnotationResult) -> None:
263 """Print the annotation report as JSON."""
264 output = {
265 "overall_coverage": result.coverage_percentage,
266 "total_functions": result.total_functions,
267 "fully_typed_functions": result.fully_typed_functions,
268 "total_ignores": result.total_ignores,
269 "total_casts": result.total_casts,
270 "total_asserts": result.total_asserts,
271 }
272 click.echo(json.dumps(output))
273
274
275@common_command
276@without_runtime_setup
277@register_cli("fix", shortcut_for="code fix")
278@cli.command()
279@click.pass_context
280@click.argument("paths", nargs=-1)
281@click.option("--unsafe-fixes", is_flag=True, help="Apply ruff unsafe fixes")
282@click.option("--add-noqa", is_flag=True, help="Add noqa comments to suppress errors")
283@click.option("--skip-oxc", is_flag=True, help="Skip oxlint and oxfmt")
284def fix(
285 ctx: click.Context,
286 paths: tuple[str, ...],
287 unsafe_fixes: bool,
288 add_noqa: bool,
289 skip_oxc: bool,
290) -> None:
291 """Fix formatting and linting issues"""
292 if not paths:
293 paths = (".",)
294
295 python_paths, other_paths = _partition_paths(paths)
296 ruff_args = ["--config", str(DEFAULT_RUFF_CONFIG)]
297 config = get_code_config()
298
299 for e in config.get("exclude", []):
300 ruff_args.extend(["--exclude", e])
301
302 if unsafe_fixes and add_noqa:
303 raise click.UsageError("Cannot use both --unsafe-fixes and --add-noqa")
304
305 if python_paths:
306 if unsafe_fixes:
307 print_event("ruff check --fix --unsafe-fixes...", newline=False)
308 result = subprocess.run(
309 ["ruff", "check", *python_paths, "--fix", "--unsafe-fixes", *ruff_args],
310 check=False,
311 )
312 elif add_noqa:
313 print_event("ruff check --add-noqa...", newline=False)
314 result = subprocess.run(
315 ["ruff", "check", *python_paths, "--add-noqa", *ruff_args], check=False
316 )
317 else:
318 print_event("ruff check --fix...", newline=False)
319 result = subprocess.run(
320 ["ruff", "check", *python_paths, "--fix", *ruff_args], check=False
321 )
322
323 if result.returncode != 0:
324 sys.exit(result.returncode)
325
326 print_event("ruff format...", newline=False)
327 result = subprocess.run(
328 ["ruff", "format", *python_paths, *ruff_args], check=False
329 )
330 if result.returncode != 0:
331 sys.exit(result.returncode)
332
333 if not skip_oxc and other_paths and config.get("oxc", {}).get("enabled", True):
334 oxlint = OxcTool("oxlint")
335 oxfmt = OxcTool("oxfmt")
336
337 if not (oxlint.is_installed() and oxfmt.is_installed()):
338 ctx.invoke(install)
339
340 if unsafe_fixes:
341 print_event("oxlint --fix-dangerously...", newline=False)
342 result = oxlint.invoke(*other_paths, "--fix-dangerously")
343 else:
344 print_event("oxlint --fix...", newline=False)
345 result = oxlint.invoke(*other_paths, "--fix")
346
347 if result.returncode != 0:
348 sys.exit(result.returncode)
349
350 print_event("oxfmt...", newline=False)
351 result = oxfmt.invoke(*other_paths)
352
353 if result.returncode != 0:
354 sys.exit(result.returncode)
355
356
357def get_code_config() -> dict[str, Any]:
358 pyproject = Path("pyproject.toml")
359 if not pyproject.exists():
360 return {}
361 with pyproject.open("rb") as f:
362 return tomllib.load(f).get("tool", {}).get("plain", {}).get("code", {})