Coverage for src/secchi/cli.py: 37%
248 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 23:28 +0000
1"""Terminal entry point: parse arguments, invoke workflows, render output."""
3from __future__ import annotations
5import argparse
6import asyncio
7import subprocess
8import sys
9from pathlib import Path
11from tomli_w import dumps as toml_dumps
13from secchi import __version__
14from secchi.config import find_config, list_projects
15from secchi.diagnostics import DiagnosticLog, DiagnosticStatus
16from secchi.errors import SecchiError
17from secchi.models import Registry
18from secchi.renderers.summary import render_summary
19from secchi.services.comparison import render_comparison
20from secchi.services.intelligence import PackageIntelligenceService
21from secchi.update import check_for_update
22from secchi.workflows import check, compare, dashboard, report, search, show
25def build_parser() -> argparse.ArgumentParser:
26 parser = argparse.ArgumentParser(
27 prog="secchi",
28 description="Open source package intelligence from your terminal.",
29 )
30 parser.add_argument("--version", action="version", version=f"secchi {__version__}")
31 parser.add_argument("--project", "-p", help="Project name from configuration")
32 parser.add_argument("--config", "-c", help="Path to secchi.toml or .secchi.toml")
33 parser.add_argument(
34 "--no-cache",
35 dest="refresh",
36 action="store_true",
37 help=(
38 "Bypass the local cache and re-fetch package, registry, GitHub, "
39 "and security-advisory signals"
40 ),
41 )
42 parser.add_argument(
43 "--security-no-cache",
44 dest="security_refresh",
45 action="store_true",
46 help="Bypass only the advisory cache and re-fetch OSV security data",
47 )
48 parser.add_argument(
49 "--list", "-l", action="store_true", help="List configured projects and exit"
50 )
51 parser.add_argument(
52 "--verbose",
53 action="store_true",
54 help="Show every SUCCESS, WARN, and FAILURE diagnostic event",
55 )
56 parser.add_argument(
57 "--log-file",
58 type=Path,
59 help="Write readable diagnostics for this run to a file",
60 )
62 sub = parser.add_subparsers(dest="command")
63 sub.add_parser("init", help="Interactively create secchi.toml")
65 dashboard_parser = sub.add_parser(
66 "dashboard", help="Launch the interactive dashboard"
67 )
68 dashboard_parser.add_argument(
69 "package", nargs="?", help="Package name or registry:name"
70 )
71 dashboard_parser.add_argument(
72 "--registry", choices=[item.value for item in Registry]
73 )
74 dashboard_parser.add_argument("--project", "-p", dest="dashboard_project")
75 dashboard_parser.add_argument("--config", "-c", dest="dashboard_config")
76 dashboard_parser.add_argument(
77 "--no-cache",
78 dest="dashboard_refresh",
79 action="store_true",
80 help="Re-fetch package data and security advisories instead of using cache",
81 )
82 dashboard_parser.add_argument(
83 "--security-no-cache",
84 dest="dashboard_security_refresh",
85 action="store_true",
86 help="Re-fetch only OSV security advisories instead of using their cache",
87 )
89 web_parser = sub.add_parser(
90 "web", help="Serve the interactive dashboard in a local browser"
91 )
92 web_parser.add_argument("package", nargs="?", help="Package name or registry:name")
93 web_parser.add_argument("--registry", choices=[item.value for item in Registry])
94 web_parser.add_argument("--project", "-p", dest="web_project")
95 web_parser.add_argument("--config", "-c", dest="web_config")
96 web_parser.add_argument(
97 "--port",
98 type=int,
99 default=8000,
100 help="Local browser server port (default: 8000)",
101 )
102 web_parser.add_argument(
103 "--no-cache",
104 dest="web_refresh",
105 action="store_true",
106 help="Re-fetch package data and security advisories instead of using cache",
107 )
108 web_parser.add_argument(
109 "--security-no-cache",
110 dest="web_security_refresh",
111 action="store_true",
112 help="Re-fetch only OSV security advisories instead of using their cache",
113 )
115 show_parser = sub.add_parser(
116 "show", help="Print a concise package intelligence summary"
117 )
118 show_parser.add_argument("package", help="Package name or registry:name")
119 show_parser.add_argument("--registry", choices=[item.value for item in Registry])
120 show_parser.add_argument(
121 "--no-cache",
122 dest="refresh",
123 action="store_true",
124 help="Re-fetch package data and security advisories instead of using cache",
125 )
126 show_parser.add_argument(
127 "--security-no-cache",
128 dest="show_security_refresh",
129 action="store_true",
130 help="Re-fetch only OSV security advisories instead of using their cache",
131 )
133 search_parser = sub.add_parser(
134 "search", help="Find packages across supported registries"
135 )
136 search_parser.add_argument("package", help="Exact package name")
137 search_parser.add_argument("--registry", choices=[item.value for item in Registry])
138 search_parser.add_argument("--no-cache", dest="refresh", action="store_true")
140 report_parser = sub.add_parser(
141 "report", help="Generate a package or project report"
142 )
143 report_parser.add_argument(
144 "package", nargs="?", help="Package name or registry:name"
145 )
146 report_parser.add_argument(
147 "--project", dest="report_project", help="Configured project name"
148 )
149 report_parser.add_argument(
150 "--config", dest="report_config", help="Workspace config path"
151 )
152 report_parser.add_argument("--registry", choices=[item.value for item in Registry])
153 report_parser.add_argument(
154 "--format", choices=["json", "html", "md", "markdown"], default="json"
155 )
156 report_parser.add_argument(
157 "--output", "-o", help="Target file path, or '-' for stdout"
158 )
159 report_parser.add_argument(
160 "--no-cache",
161 dest="refresh",
162 action="store_true",
163 help="Re-fetch package data and security advisories instead of using cache",
164 )
165 report_parser.add_argument(
166 "--security-no-cache",
167 dest="report_security_refresh",
168 action="store_true",
169 help="Re-fetch only OSV security advisories instead of using their cache",
170 )
172 check_parser = sub.add_parser(
173 "check", help="Evaluate simple package health policies"
174 )
175 check_parser.add_argument("package", help="Package name or registry:name")
176 check_parser.add_argument("--registry", choices=[item.value for item in Registry])
177 check_parser.add_argument("--min-health", type=int, default=70)
178 check_parser.add_argument("--require-ci", action="store_true")
179 check_parser.add_argument("--no-cache", dest="refresh", action="store_true")
180 check_parser.add_argument(
181 "--security-no-cache",
182 dest="check_security_refresh",
183 action="store_true",
184 help="Re-fetch only OSV security advisories instead of using their cache",
185 )
187 compare_parser = sub.add_parser(
188 "compare", help="Compare package choices with agent-readable evidence"
189 )
190 compare_parser.add_argument(
191 "packages",
192 nargs="+",
193 help="Two or more package names or registry:name references",
194 )
195 compare_parser.add_argument("--registry", choices=[item.value for item in Registry])
196 compare_parser.add_argument("--format", choices=["text", "json"], default="text")
197 compare_parser.add_argument("--no-cache", dest="refresh", action="store_true")
198 compare_parser.add_argument(
199 "--security-no-cache",
200 dest="compare_security_refresh",
201 action="store_true",
202 help="Re-fetch only OSV security advisories instead of using their cache",
203 )
205 monitor_parser = sub.add_parser("monitor", help="Alias for dashboard --project")
206 monitor_parser.add_argument("project_name", help="Project name to monitor")
207 monitor_parser.add_argument("--no-cache", dest="refresh", action="store_true")
208 monitor_parser.add_argument(
209 "--security-no-cache",
210 dest="security_refresh",
211 action="store_true",
212 help="Re-fetch only OSV security advisories instead of using their cache",
213 )
214 for command_parser in (
215 dashboard_parser,
216 web_parser,
217 show_parser,
218 search_parser,
219 report_parser,
220 check_parser,
221 compare_parser,
222 monitor_parser,
223 ):
224 command_parser.add_argument(
225 "--verbose",
226 action="store_true",
227 default=argparse.SUPPRESS,
228 help="Show every SUCCESS, WARN, and FAILURE diagnostic event",
229 )
230 command_parser.add_argument(
231 "--log-file",
232 type=Path,
233 default=argparse.SUPPRESS,
234 help="Write readable diagnostics for this run to a file",
235 )
236 sub.add_parser("mcp", help="Run the Model Context Protocol server over stdio")
237 return parser
240def cmd_init() -> None:
241 output_path = Path.cwd() / "secchi.toml"
242 print("🚀 secchi init — create a new config file\n")
243 if output_path.exists():
244 answer = input(f"'{output_path}' already exists. Overwrite? [y/N]: ")
245 if answer.lower() not in ("y", "yes"):
246 print("Aborted.")
247 return
248 projects: dict[str, dict] = {}
249 while True:
250 name = input("Project name (e.g., 'my-libs'): ").strip()
251 if not name:
252 print("Project name is required.")
253 continue
254 description = input(" Description (optional): ").strip()
255 favorite = input(" Favourite project? [y/N]: ").strip().lower() in ("y", "yes")
256 packages: list[dict[str, str]] = []
257 print(" Add packages. Leave name empty to finish.")
258 while True:
259 package_name = input(" Package name: ").strip()
260 if not package_name:
261 break
262 registry = input(
263 " Registry [pypi/crates.io/npm/homebrew/go/cran, default: pypi]: "
264 ).strip()
265 packages.append({"name": package_name, "registry": registry or "pypi"})
266 if packages:
267 projects[name] = {
268 "description": description,
269 "favorite": favorite,
270 "packages": packages,
271 }
272 if input("Add another project? [y/N]: ").strip().lower() not in ("y", "yes"):
273 break
274 if not projects:
275 print("No projects created. Aborting.")
276 return
277 output_path.write_text(toml_dumps({"projects": projects}))
278 print(f"\n✅ Config written to {output_path}")
281def _print_diagnostics(diagnostics: DiagnosticLog, *, verbose: bool) -> None:
282 events = diagnostics.snapshot()
283 visible = (
284 events
285 if verbose
286 else [
287 event
288 for event in events
289 if event.status in (DiagnosticStatus.WARN, DiagnosticStatus.FAILURE)
290 ]
291 )
292 if not visible:
293 return
294 print("\nDiagnostics:")
295 for event in visible:
296 print(f" {event.format()}")
299def _run_dashboard(
300 args: argparse.Namespace,
301 parser: argparse.ArgumentParser,
302 service: PackageIntelligenceService,
303 diagnostics: DiagnosticLog,
304) -> None:
305 try:
306 request = asyncio.run(
307 dashboard.run(
308 package=getattr(args, "package", None),
309 registry=getattr(args, "registry", None),
310 project_name=getattr(args, "dashboard_project", None) or args.project,
311 config=getattr(args, "dashboard_config", None) or args.config,
312 refresh=getattr(args, "dashboard_refresh", False) or args.refresh,
313 security_refresh=(
314 getattr(args, "dashboard_security_refresh", False)
315 or args.security_refresh
316 or args.refresh
317 ),
318 diagnostics=diagnostics,
319 )
320 )
321 except (SecchiError, ValueError) as exc:
322 parser.error(str(exc))
323 from secchi.ui.app import Secchi
325 Secchi(
326 project=request.project,
327 config_path=request.config_path,
328 force_refresh=request.refresh,
329 workspace=request.workspace,
330 force_security_refresh=request.security_refresh,
331 intelligence=service,
332 diagnostics=diagnostics,
333 ).run()
336def _run_web(
337 args: argparse.Namespace,
338 parser: argparse.ArgumentParser,
339 diagnostics: DiagnosticLog,
340) -> None:
341 refresh = getattr(args, "web_refresh", False) or args.refresh
342 security_refresh = (
343 getattr(args, "web_security_refresh", False) or args.security_refresh or refresh
344 )
345 project_name = getattr(args, "web_project", None) or args.project
346 config = getattr(args, "web_config", None) or args.config
347 try:
348 request = asyncio.run(
349 dashboard.run(
350 package=getattr(args, "package", None),
351 registry=getattr(args, "registry", None),
352 project_name=project_name,
353 config=config,
354 refresh=refresh,
355 security_refresh=security_refresh,
356 diagnostics=diagnostics,
357 )
358 )
359 except (SecchiError, ValueError) as exc:
360 parser.error(str(exc))
362 from secchi.web import TextualServeUnavailable, prepare_launch, run_textual_serve
364 launch = prepare_launch(
365 request,
366 package=getattr(args, "package", None),
367 registry=getattr(args, "registry", None),
368 project_name=project_name,
369 refresh=refresh,
370 security_refresh=security_refresh,
371 verbose=getattr(args, "verbose", False),
372 log_file=getattr(args, "log_file", None),
373 )
374 print(
375 "Serving Secchi through Textual's local web server. Open the URL shown "
376 "by the server to access this dashboard session."
377 )
378 try:
379 run_textual_serve(launch.command, port=args.port)
380 except TextualServeUnavailable as exc:
381 parser.error(str(exc))
382 except subprocess.CalledProcessError as exc:
383 parser.error(f"textual serve exited with status {exc.returncode}.")
386def _run_search(
387 args: argparse.Namespace, diagnostics: DiagnosticLog, *, verbose: bool
388) -> None:
389 results = asyncio.run(
390 search.run(
391 args.package,
392 registry=args.registry,
393 limit=10,
394 diagnostics=diagnostics,
395 )
396 )
397 if not results:
398 print(
399 f"No packages matching '{args.package}' found in the selected registries."
400 )
401 _print_diagnostics(diagnostics, verbose=verbose)
402 return
403 print(f"Matches for {args.package}:\n")
404 for result in results:
405 description = (result.description or "No description").splitlines()[0]
406 marker = "exact" if result.exact else "match"
407 print(
408 f"{result.registry.display_name:<10} {result.name:<24} {result.version or '—':<12} {marker:<6} {description}"
409 )
410 _print_diagnostics(diagnostics, verbose=verbose)
413def main() -> None:
414 parser = build_parser()
415 args = parser.parse_args()
416 if args.command == "init":
417 cmd_init()
418 return
419 if args.command == "mcp":
420 from secchi.mcp_server import main as mcp_main
422 mcp_main()
423 return
424 if args.command == "monitor":
425 args.package = None
426 args.dashboard_project = args.project_name
427 args.dashboard_config = None
428 args.dashboard_refresh = args.refresh
429 args.command = "dashboard"
430 diagnostics = DiagnosticLog(path=args.log_file)
431 service = PackageIntelligenceService(diagnostics=diagnostics)
432 if args.list:
433 config_path = find_config(args.config)
434 if not config_path:
435 parser.error("No config file found.")
436 for name in list_projects(config_path):
437 print(name)
438 return
439 notice = check_for_update()
440 if notice:
441 print(f"Notice: {notice.message}", file=sys.stderr)
442 if args.command == "dashboard" or args.command is None:
443 _run_dashboard(args, parser, service, diagnostics)
444 return
445 if args.command == "web":
446 _run_web(args, parser, diagnostics)
447 return
448 if args.command == "show":
449 try:
450 result = asyncio.run(
451 show.run(
452 args.package,
453 registry=args.registry,
454 refresh=args.refresh,
455 security_refresh=(
456 getattr(args, "show_security_refresh", False)
457 or args.security_refresh
458 or args.refresh
459 ),
460 service=service,
461 )
462 )
463 except (SecchiError, ValueError) as exc:
464 parser.error(str(exc))
465 print(render_summary(result.info, result.derived))
466 for warning in result.warnings:
467 print(f"Warning [{warning.source}]: {warning.message}")
468 _print_diagnostics(diagnostics, verbose=args.verbose)
469 return
470 if args.command == "search":
471 _run_search(args, diagnostics, verbose=args.verbose)
472 return
473 if args.command == "report":
474 try:
475 output = asyncio.run(
476 report.run(
477 package=args.package,
478 project_name=args.report_project,
479 config=args.report_config or args.config,
480 registry=args.registry,
481 format_name=args.format,
482 output=args.output,
483 refresh=args.refresh,
484 security_refresh=(
485 getattr(args, "report_security_refresh", False)
486 or args.security_refresh
487 or args.refresh
488 ),
489 service=service,
490 )
491 )
492 except (SecchiError, ValueError, FileNotFoundError) as exc:
493 parser.error(str(exc))
494 if args.output == "-":
495 print(output.content)
496 else:
497 output.target.parent.mkdir(parents=True, exist_ok=True)
498 output.target.write_text(output.content)
499 print(f"Wrote {output.format_name} report to {output.target}")
500 _print_diagnostics(diagnostics, verbose=args.verbose)
501 return
502 if args.command == "check":
503 try:
504 result = asyncio.run(
505 check.run(
506 args.package,
507 registry=args.registry,
508 min_health=args.min_health,
509 require_ci=args.require_ci,
510 refresh=args.refresh,
511 security_refresh=(
512 getattr(args, "check_security_refresh", False)
513 or args.security_refresh
514 or args.refresh
515 ),
516 service=service,
517 )
518 )
519 except (SecchiError, ValueError) as exc:
520 parser.error(str(exc))
521 for item in result.checks:
522 print(f"{'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}")
523 for warning in result.warnings:
524 print(f"Warning [{warning.source}]: {warning.message}")
525 _print_diagnostics(diagnostics, verbose=args.verbose)
526 if not result.passed:
527 raise SystemExit(1)
528 return
529 if args.command == "compare":
530 try:
531 comparison = asyncio.run(
532 compare.run(
533 args.packages,
534 registry=args.registry,
535 refresh=args.refresh,
536 security_refresh=(
537 getattr(args, "compare_security_refresh", False)
538 or args.security_refresh
539 or args.refresh
540 ),
541 service=service,
542 )
543 )
544 except (SecchiError, ValueError) as exc:
545 parser.error(str(exc))
546 if args.format == "json":
547 import json
549 print(json.dumps(comparison.as_dict(), indent=2))
550 else:
551 print(render_comparison(comparison))
552 _print_diagnostics(diagnostics, verbose=args.verbose)
553 return
554 parser.error("Unknown command")
557if __name__ == "__main__":
558 main()