Coverage for src/secchi/cli.py: 0%

184 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-08-04 22:15 +0000

1"""Terminal entry point: parse arguments, invoke workflows, render output.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import asyncio 

7from pathlib import Path 

8 

9from tomli_w import dumps as toml_dumps 

10 

11from secchi import __version__ 

12from secchi.config import find_config, list_projects 

13from secchi.errors import SecchiError 

14from secchi.models import Registry 

15from secchi.renderers.summary import render_summary 

16from secchi.services.comparison import render_comparison 

17from secchi.services.intelligence import PackageIntelligenceService 

18from secchi.workflows import check, compare, dashboard, report, search, show 

19 

20 

21def build_parser() -> argparse.ArgumentParser: 

22 parser = argparse.ArgumentParser( 

23 prog="secchi", 

24 description="Open source package intelligence from your terminal.", 

25 ) 

26 parser.add_argument("--version", action="version", version=f"secchi {__version__}") 

27 parser.add_argument("--project", "-p", help="Project name from configuration") 

28 parser.add_argument("--config", "-c", help="Path to secchi.toml or .secchi.toml") 

29 parser.add_argument( 

30 "--refresh", "-r", action="store_true", help="Bypass local cache" 

31 ) 

32 parser.add_argument( 

33 "--list", "-l", action="store_true", help="List configured projects and exit" 

34 ) 

35 

36 sub = parser.add_subparsers(dest="command") 

37 sub.add_parser("init", help="Interactively create secchi.toml") 

38 

39 dashboard_parser = sub.add_parser( 

40 "dashboard", help="Launch the interactive dashboard" 

41 ) 

42 dashboard_parser.add_argument( 

43 "package", nargs="?", help="Package name or registry:name" 

44 ) 

45 dashboard_parser.add_argument( 

46 "--registry", choices=[item.value for item in Registry] 

47 ) 

48 dashboard_parser.add_argument("--project", "-p", dest="dashboard_project") 

49 dashboard_parser.add_argument("--config", "-c", dest="dashboard_config") 

50 dashboard_parser.add_argument( 

51 "--refresh", "-r", dest="dashboard_refresh", action="store_true" 

52 ) 

53 

54 show_parser = sub.add_parser( 

55 "show", help="Print a concise package intelligence summary" 

56 ) 

57 show_parser.add_argument("package", help="Package name or registry:name") 

58 show_parser.add_argument("--registry", choices=[item.value for item in Registry]) 

59 show_parser.add_argument("--refresh", "-r", action="store_true") 

60 

61 search_parser = sub.add_parser( 

62 "search", help="Find packages across supported registries" 

63 ) 

64 search_parser.add_argument("package", help="Exact package name") 

65 search_parser.add_argument("--registry", choices=[item.value for item in Registry]) 

66 search_parser.add_argument("--refresh", "-r", action="store_true") 

67 

68 report_parser = sub.add_parser( 

69 "report", help="Generate a package or project report" 

70 ) 

71 report_parser.add_argument( 

72 "package", nargs="?", help="Package name or registry:name" 

73 ) 

74 report_parser.add_argument( 

75 "--project", dest="report_project", help="Configured project name" 

76 ) 

77 report_parser.add_argument( 

78 "--config", dest="report_config", help="Workspace config path" 

79 ) 

80 report_parser.add_argument("--registry", choices=[item.value for item in Registry]) 

81 report_parser.add_argument( 

82 "--format", choices=["json", "html", "md", "markdown"], default="json" 

83 ) 

84 report_parser.add_argument( 

85 "--output", "-o", help="Target file path, or '-' for stdout" 

86 ) 

87 report_parser.add_argument("--refresh", "-r", action="store_true") 

88 

89 check_parser = sub.add_parser( 

90 "check", help="Evaluate simple package health policies" 

91 ) 

92 check_parser.add_argument("package", help="Package name or registry:name") 

93 check_parser.add_argument("--registry", choices=[item.value for item in Registry]) 

94 check_parser.add_argument("--min-health", type=int, default=70) 

95 check_parser.add_argument("--require-ci", action="store_true") 

96 check_parser.add_argument("--refresh", "-r", action="store_true") 

97 

98 compare_parser = sub.add_parser( 

99 "compare", help="Compare package choices with agent-readable evidence" 

100 ) 

101 compare_parser.add_argument( 

102 "packages", 

103 nargs="+", 

104 help="Two or more package names or registry:name references", 

105 ) 

106 compare_parser.add_argument("--registry", choices=[item.value for item in Registry]) 

107 compare_parser.add_argument("--format", choices=["text", "json"], default="text") 

108 compare_parser.add_argument("--refresh", "-r", action="store_true") 

109 

110 monitor_parser = sub.add_parser("monitor", help="Alias for dashboard --project") 

111 monitor_parser.add_argument("project_name", help="Project name to monitor") 

112 sub.add_parser("mcp", help="Run the Model Context Protocol server over stdio") 

113 return parser 

114 

115 

116def cmd_init() -> None: 

117 output_path = Path.cwd() / "secchi.toml" 

118 print("🚀 secchi init — create a new config file\n") 

119 if output_path.exists(): 

120 answer = input(f"'{output_path}' already exists. Overwrite? [y/N]: ") 

121 if answer.lower() not in ("y", "yes"): 

122 print("Aborted.") 

123 return 

124 projects: dict[str, dict] = {} 

125 while True: 

126 name = input("Project name (e.g., 'my-libs'): ").strip() 

127 if not name: 

128 print("Project name is required.") 

129 continue 

130 description = input(" Description (optional): ").strip() 

131 favorite = input(" Favourite project? [y/N]: ").strip().lower() in ("y", "yes") 

132 packages: list[dict[str, str]] = [] 

133 print(" Add packages. Leave name empty to finish.") 

134 while True: 

135 package_name = input(" Package name: ").strip() 

136 if not package_name: 

137 break 

138 registry = input( 

139 " Registry [pypi/crates.io/npm/homebrew/go/cran, default: pypi]: " 

140 ).strip() 

141 packages.append({"name": package_name, "registry": registry or "pypi"}) 

142 if packages: 

143 projects[name] = { 

144 "description": description, 

145 "favorite": favorite, 

146 "packages": packages, 

147 } 

148 if input("Add another project? [y/N]: ").strip().lower() not in ("y", "yes"): 

149 break 

150 if not projects: 

151 print("No projects created. Aborting.") 

152 return 

153 output_path.write_text(toml_dumps({"projects": projects})) 

154 print(f"\n✅ Config written to {output_path}") 

155 

156 

157def _run_dashboard( 

158 args: argparse.Namespace, 

159 parser: argparse.ArgumentParser, 

160 service: PackageIntelligenceService, 

161) -> None: 

162 try: 

163 request = asyncio.run( 

164 dashboard.run( 

165 package=getattr(args, "package", None), 

166 registry=getattr(args, "registry", None), 

167 project_name=getattr(args, "dashboard_project", None) or args.project, 

168 config=getattr(args, "dashboard_config", None) or args.config, 

169 refresh=getattr(args, "dashboard_refresh", False) or args.refresh, 

170 ) 

171 ) 

172 except (SecchiError, ValueError) as exc: 

173 parser.error(str(exc)) 

174 from secchi.ui.app import Secchi 

175 

176 Secchi( 

177 project=request.project, 

178 config_path=request.config_path, 

179 force_refresh=request.refresh, 

180 workspace=request.workspace, 

181 intelligence=service, 

182 ).run() 

183 

184 

185def _run_search(args: argparse.Namespace) -> None: 

186 results = asyncio.run(search.run(args.package, registry=args.registry, limit=10)) 

187 if not results: 

188 print( 

189 f"No packages matching '{args.package}' found in the selected registries." 

190 ) 

191 return 

192 print(f"Matches for {args.package}:\n") 

193 for result in results: 

194 description = (result.description or "No description").splitlines()[0] 

195 marker = "exact" if result.exact else "match" 

196 print( 

197 f"{result.registry.display_name:<10} {result.name:<24} {result.version or '—':<12} {marker:<6} {description}" 

198 ) 

199 

200 

201def main() -> None: 

202 parser = build_parser() 

203 args = parser.parse_args() 

204 if args.command == "init": 

205 cmd_init() 

206 return 

207 if args.command == "mcp": 

208 from secchi.mcp_server import main as mcp_main 

209 

210 mcp_main() 

211 return 

212 if args.command == "monitor": 

213 args.package = None 

214 args.dashboard_project = args.project_name 

215 args.dashboard_config = None 

216 args.dashboard_refresh = args.refresh 

217 args.command = "dashboard" 

218 service = PackageIntelligenceService() 

219 if args.list: 

220 config_path = find_config(args.config) 

221 if not config_path: 

222 parser.error("No config file found.") 

223 for name in list_projects(config_path): 

224 print(name) 

225 return 

226 if args.command == "dashboard" or args.command is None: 

227 _run_dashboard(args, parser, service) 

228 return 

229 if args.command == "show": 

230 try: 

231 result = asyncio.run( 

232 show.run( 

233 args.package, 

234 registry=args.registry, 

235 refresh=args.refresh, 

236 service=service, 

237 ) 

238 ) 

239 except (SecchiError, ValueError) as exc: 

240 parser.error(str(exc)) 

241 print(render_summary(result.info, result.derived)) 

242 for warning in result.warnings: 

243 print(f"Warning [{warning.source}]: {warning.message}") 

244 return 

245 if args.command == "search": 

246 _run_search(args) 

247 return 

248 if args.command == "report": 

249 try: 

250 output = asyncio.run( 

251 report.run( 

252 package=args.package, 

253 project_name=args.report_project, 

254 config=args.report_config or args.config, 

255 registry=args.registry, 

256 format_name=args.format, 

257 output=args.output, 

258 refresh=args.refresh, 

259 service=service, 

260 ) 

261 ) 

262 except (SecchiError, ValueError, FileNotFoundError) as exc: 

263 parser.error(str(exc)) 

264 if args.output == "-": 

265 print(output.content) 

266 else: 

267 output.target.parent.mkdir(parents=True, exist_ok=True) 

268 output.target.write_text(output.content) 

269 print(f"Wrote {output.format_name} report to {output.target}") 

270 return 

271 if args.command == "check": 

272 try: 

273 result = asyncio.run( 

274 check.run( 

275 args.package, 

276 registry=args.registry, 

277 min_health=args.min_health, 

278 require_ci=args.require_ci, 

279 refresh=args.refresh, 

280 service=service, 

281 ) 

282 ) 

283 except (SecchiError, ValueError) as exc: 

284 parser.error(str(exc)) 

285 for item in result.checks: 

286 print(f"{'PASS' if item.passed else 'FAIL'} {item.name}: {item.detail}") 

287 for warning in result.warnings: 

288 print(f"Warning [{warning.source}]: {warning.message}") 

289 if not result.passed: 

290 raise SystemExit(1) 

291 return 

292 if args.command == "compare": 

293 try: 

294 comparison = asyncio.run( 

295 compare.run( 

296 args.packages, 

297 registry=args.registry, 

298 refresh=args.refresh, 

299 service=service, 

300 ) 

301 ) 

302 except (SecchiError, ValueError) as exc: 

303 parser.error(str(exc)) 

304 if args.format == "json": 

305 import json 

306 

307 print(json.dumps(comparison.as_dict(), indent=2)) 

308 else: 

309 print(render_comparison(comparison)) 

310 return 

311 parser.error("Unknown command") 

312 

313 

314if __name__ == "__main__": 

315 main()