Coverage for sources/librovore/cli.py: 46%
164 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-03 21:59 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
4#============================================================================#
5# #
6# Licensed under the Apache License, Version 2.0 (the "License"); #
7# you may not use this file except in compliance with the License. #
8# You may obtain a copy of the License at #
9# #
10# http://www.apache.org/licenses/LICENSE-2.0 #
11# #
12# Unless required by applicable law or agreed to in writing, software #
13# distributed under the License is distributed on an "AS IS" BASIS, #
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
15# See the License for the specific language governing permissions and #
16# limitations under the License. #
17# #
18#============================================================================#
21''' Command-line interface. '''
24from . import __
25from . import cacheproxy as _cacheproxy
26from . import exceptions as _exceptions
27from . import functions as _functions
28from . import interfaces as _interfaces
29from . import results as _results
30from . import server as _server
31from . import state as _state
33_scribe = __.acquire_scribe( __name__ )
36def intercept_errors( ) -> __.cabc.Callable[
37 [ __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ] ],
38 __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]
39]:
40 ''' Decorator for CLI handlers to intercept exceptions.
42 Catches Omnierror exceptions and renders them appropriately.
43 Other exceptions are logged and formatted simply.
44 '''
45 def decorator(
46 func: __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]
47 ) -> __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]:
48 @__.funct.wraps( func )
49 async def wrapper(
50 self: __.typx.Any,
51 auxdata: _state.Globals,
52 display: __.DisplayTarget,
53 display_format: _interfaces.DisplayFormat,
54 color: bool,
55 *posargs: __.typx.Any,
56 **nomargs: __.typx.Any,
57 ) -> None:
58 stream = await display.provide_stream( )
59 try:
60 return await func(
61 self, auxdata, display, display_format, color,
62 *posargs, **nomargs )
63 except _exceptions.Omnierror as exc:
64 match display_format:
65 case _interfaces.DisplayFormat.JSON:
66 serialized = dict( exc.render_as_json( ) )
67 error_message = __.json.dumps( serialized, indent = 2 )
68 case _interfaces.DisplayFormat.Markdown:
69 lines = exc.render_as_markdown( )
70 error_message = '\n'.join( lines )
71 print( error_message, file = stream )
72 raise SystemExit( 1 ) from None
73 except Exception as exc:
74 _scribe.error( f"{func.__name__} failed: %s", exc )
75 match display_format:
76 case _interfaces.DisplayFormat.JSON:
77 error_data = {
78 "type": "unexpected_error",
79 "title": "Unexpected Error",
80 "message": str( exc ),
81 "suggestion": (
82 "Please report this issue if it persists." ),
83 }
84 error_message = __.json.dumps( error_data, indent = 2 )
85 case _interfaces.DisplayFormat.Markdown:
86 error_message = f"❌ Unexpected error: {exc}"
87 print( error_message, file = stream )
88 raise SystemExit( 1 ) from None
90 return wrapper
91 return decorator
94GroupByArgument: __.typx.TypeAlias = __.typx.Annotated[
95 __.typx.Optional[ str ],
96 __.tyro.conf.arg( help = __.access_doctab( 'group by argument' ) ),
97]
98PortArgument: __.typx.TypeAlias = __.typx.Annotated[
99 __.typx.Optional[ int ],
100 __.tyro.conf.arg( help = __.access_doctab( 'server port argument' ) ),
101]
102TermArgument: __.typx.TypeAlias = __.typx.Annotated[
103 __.tyro.conf.Positional[ str ],
104 __.tyro.conf.arg( help = __.access_doctab( 'term argument' ) ),
105]
106ResultsMax: __.typx.TypeAlias = __.typx.Annotated[
107 int,
108 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),
109]
110LocationArgument: __.typx.TypeAlias = __.typx.Annotated[
111 __.tyro.conf.Positional[ str ],
112 __.tyro.conf.arg( help = __.access_doctab( 'location argument' ) ),
113]
114TransportArgument: __.typx.TypeAlias = __.typx.Annotated[
115 __.typx.Optional[ str ],
116 __.tyro.conf.arg( help = __.access_doctab( 'transport argument' ) ),
117]
120_search_behaviors_default = _interfaces.SearchBehaviors( )
121_filters_default = __.immut.Dictionary[ str, __.typx.Any ]( )
123_MARKDOWN_OBJECT_LIMIT = 10
124_MARKDOWN_CONTENT_LIMIT = 200
127def decide_rich_markdown(
128 stream: __.typx.TextIO, colorize: bool
129) -> bool:
130 ''' Determines whether to use Rich markdown rendering. '''
131 return (
132 colorize
133 and stream.isatty( )
134 and not __.os.environ.get( 'NO_COLOR' )
135 )
138async def _render_and_print_result(
139 result: _results.ResultBase,
140 display_format: _interfaces.DisplayFormat,
141 stream: __.typx.TextIO,
142 color: bool,
143 **nomargs: __.typx.Any
144) -> None:
145 ''' Centralizes result rendering logic with Rich formatting support. '''
146 match display_format:
147 case _interfaces.DisplayFormat.JSON:
148 nomargs_filtered = {
149 key: value for key, value in nomargs.items()
150 if key in [ 'lines_max' ] # Only pass relevant nomargs
151 }
152 serialized = dict( result.render_as_json( **nomargs_filtered ) )
153 output = __.json.dumps( serialized, indent = 2 )
154 print( output, file = stream )
155 case _interfaces.DisplayFormat.Markdown:
156 lines = result.render_as_markdown( **nomargs )
157 if decide_rich_markdown( stream, color ):
158 from rich.console import (
159 Console,
160 )
161 from rich.markdown import (
162 Markdown,
163 )
164 console = Console( file = stream, force_terminal = True )
165 markdown_obj = Markdown( '\n'.join( lines ) )
166 console.print( markdown_obj )
167 else:
168 output = '\n'.join( lines )
169 print( output, file = stream )
172class _CliCommand(
173 __.immut.DataclassProtocol, __.typx.Protocol,
174 decorators = ( __.typx.runtime_checkable, ),
175):
176 ''' CLI command. '''
178 @__.abc.abstractmethod
179 def __call__(
180 self,
181 auxdata: _state.Globals,
182 display: __.DisplayTarget,
183 display_format: _interfaces.DisplayFormat,
184 color: bool,
185 ) -> __.cabc.Awaitable[ None ]:
186 ''' Executes command with global state. '''
187 raise NotImplementedError
190class DetectCommand(
191 _CliCommand, decorators = ( __.standard_tyro_class, ),
192):
193 ''' Detect which processors can handle a documentation source. '''
195 location: LocationArgument
196 genus: __.typx.Annotated[
197 _interfaces.ProcessorGenera,
198 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),
199 ]
200 processor_name: __.typx.Annotated[
201 __.typx.Optional[ str ],
202 __.tyro.conf.arg( help = "Specific processor to use." ),
203 ] = None
205 @intercept_errors( )
206 async def __call__(
207 self,
208 auxdata: _state.Globals,
209 display: __.DisplayTarget,
210 display_format: _interfaces.DisplayFormat,
211 color: bool,
212 ) -> None:
213 stream = await display.provide_stream( )
214 processor_name = (
215 self.processor_name if self.processor_name is not None
216 else __.absent )
217 result = await _functions.detect(
218 auxdata, self.location, self.genus,
219 processor_name = processor_name )
220 await _render_and_print_result(
221 result, display_format, stream, color, reveal_internals = True )
224class QueryInventoryCommand(
225 _CliCommand, decorators = ( __.standard_tyro_class, ),
226):
227 ''' Explores documentation structure and object inventory.
229 Use before content searches to:
231 - Discover available topics and object types
232 - Identify relevant search terms and filters
233 - Understand documentation scope and organization
234 '''
236 location: LocationArgument
237 term: TermArgument
238 details: __.typx.Annotated[
239 _interfaces.InventoryQueryDetails,
240 __.tyro.conf.arg(
241 help = __.access_doctab( 'query details argument' ) ),
242 ] = _interfaces.InventoryQueryDetails.Name
243 filters: __.typx.Annotated[
244 __.cabc.Mapping[ str, __.typx.Any ],
245 __.tyro.conf.arg( prefix_name = False ),
246 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )
247 search_behaviors: __.typx.Annotated[
248 _interfaces.SearchBehaviors,
249 __.tyro.conf.arg( prefix_name = False ),
250 ] = __.dcls.field(
251 default_factory = lambda: _interfaces.SearchBehaviors( ) )
252 results_max: __.typx.Annotated[
253 int,
254 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),
255 ] = 5
257 @intercept_errors( )
258 async def __call__(
259 self,
260 auxdata: _state.Globals,
261 display: __.DisplayTarget,
262 display_format: _interfaces.DisplayFormat,
263 color: bool,
264 ) -> None:
265 stream = await display.provide_stream( )
266 result = await _functions.query_inventory(
267 auxdata,
268 self.location,
269 self.term,
270 search_behaviors = self.search_behaviors,
271 filters = self.filters,
272 results_max = self.results_max,
273 details = self.details )
274 await _render_and_print_result(
275 result, display_format, stream, color,
276 reveal_internals = True )
279class QueryContentCommand(
280 _CliCommand, decorators = ( __.standard_tyro_class, ),
281):
282 ''' Searches documentation with flexible preview/extraction modes.
284 Workflows:
286 - Sample: Use --lines-max 5-10 to preview results and identify relevant
287 content
288 - Extract: Use --content-id from sample results to retrieve full
289 content
290 - Direct: Search with higher --lines-max for immediate full results
291 '''
293 location: LocationArgument
294 term: TermArgument
295 search_behaviors: __.typx.Annotated[
296 _interfaces.SearchBehaviors,
297 __.tyro.conf.arg( prefix_name = False ),
298 ] = __.dcls.field(
299 default_factory = lambda: _interfaces.SearchBehaviors( ) )
300 filters: __.typx.Annotated[
301 __.cabc.Mapping[ str, __.typx.Any ],
302 __.tyro.conf.arg( prefix_name = False ),
303 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )
304 results_max: ResultsMax = 10
305 lines_max: __.typx.Annotated[
306 int,
307 __.tyro.conf.arg(
308 help = (
309 "Lines per result for preview/sampling. Use 5-10 for "
310 "discovery, omit for full content extraction via "
311 "content-id." ) ),
312 ] = 40
313 content_id: __.typx.Annotated[
314 __.typx.Optional[ str ],
315 __.tyro.conf.arg(
316 help = (
317 "Extract full content for specific result. Obtain IDs from "
318 "previous query-content calls with limited lines-max." ) ),
319 ] = None
321 @intercept_errors( )
322 async def __call__(
323 self,
324 auxdata: _state.Globals,
325 display: __.DisplayTarget,
326 display_format: _interfaces.DisplayFormat,
327 color: bool,
328 ) -> None:
329 stream = await display.provide_stream( )
330 content_id_ = (
331 __.absent if self.content_id is None else self.content_id )
332 result = await _functions.query_content(
333 auxdata, self.location, self.term,
334 search_behaviors = self.search_behaviors,
335 filters = self.filters,
336 content_id = content_id_,
337 results_max = self.results_max,
338 lines_max = self.lines_max )
339 await _render_and_print_result(
340 result, display_format, stream, color,
341 reveal_internals = True, lines_max = self.lines_max )
346class SurveyProcessorsCommand(
347 _CliCommand, decorators = ( __.standard_tyro_class, ),
348):
349 ''' List processors for specified genus and their capabilities. '''
351 genus: __.typx.Annotated[
352 _interfaces.ProcessorGenera,
353 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),
354 ]
355 name: __.typx.Annotated[
356 __.typx.Optional[ str ],
357 __.tyro.conf.arg( help = "Name of processor to describe" ),
358 ] = None
360 @intercept_errors( )
361 async def __call__(
362 self,
363 auxdata: _state.Globals,
364 display: __.DisplayTarget,
365 display_format: _interfaces.DisplayFormat,
366 color: bool,
367 ) -> None:
368 stream = await display.provide_stream( )
369 nomargs: __.NominativeArguments = { 'genus': self.genus }
370 if self.name is not None: nomargs[ 'name' ] = self.name
371 result = await _functions.survey_processors( auxdata, **nomargs )
372 await _render_and_print_result(
373 result, display_format, stream, color,
374 reveal_internals = False )
378class ServeCommand(
379 _CliCommand, decorators = ( __.standard_tyro_class, ),
380):
381 ''' Starts MCP server. '''
383 port: PortArgument = None
384 transport: TransportArgument = None
385 extra_functions: __.typx.Annotated[
386 bool,
387 __.tyro.conf.arg(
388 help = "Enable extra functions (detect and survey-processors)." ),
389 ] = False
390 serve_function: __.typx.Callable[
391 [ _state.Globals ], __.cabc.Awaitable[ None ]
392 ] = _server.serve
393 async def __call__(
394 self,
395 auxdata: _state.Globals,
396 display: __.DisplayTarget,
397 display_format: _interfaces.DisplayFormat,
398 color: bool,
399 ) -> None:
400 nomargs: __.NominativeArguments = { }
401 if self.port is not None: nomargs[ 'port' ] = self.port
402 if self.transport is not None: nomargs[ 'transport' ] = self.transport
403 nomargs[ 'extra_functions' ] = self.extra_functions
404 await self.serve_function( auxdata, **nomargs )
407class Cli( __.immut.DataclassObject, decorators = ( __.simple_tyro_class, ) ):
408 ''' MCP server CLI. '''
410 display: __.DisplayTarget
411 display_format: __.typx.Annotated[
412 _interfaces.DisplayFormat,
413 __.tyro.conf.arg( help = "Output format for command results." ),
414 ] = _interfaces.DisplayFormat.Markdown
415 color: __.typx.Annotated[
416 bool,
417 __.tyro.conf.arg(
418 aliases = ( "--ansi-sgr", ),
419 help = "Enable colored output and terminal formatting"
420 ),
421 ] = True
422 command: __.typx.Union[
423 __.typx.Annotated[
424 DetectCommand,
425 __.tyro.conf.subcommand( 'detect', prefix_name = False ),
426 ],
427 __.typx.Annotated[
428 QueryInventoryCommand,
429 __.tyro.conf.subcommand( 'query-inventory', prefix_name = False ),
430 ],
431 __.typx.Annotated[
432 QueryContentCommand,
433 __.tyro.conf.subcommand( 'query-content', prefix_name = False ),
434 ],
435 __.typx.Annotated[
436 SurveyProcessorsCommand,
437 __.tyro.conf.subcommand(
438 'survey-processors', prefix_name = False ),
439 ],
440 __.typx.Annotated[
441 ServeCommand,
442 __.tyro.conf.subcommand( 'serve', prefix_name = False ),
443 ],
444 ]
445 logfile: __.typx.Annotated[
446 __.typx.Optional[ str ],
447 __.ddoc.Doc( ''' Path to log capture file. ''' ),
448 ] = None
450 async def __call__( self ):
451 ''' Invokes command after library preparation. '''
452 nomargs = self.prepare_invocation_args( )
453 async with __.ctxl.AsyncExitStack( ) as exits:
454 auxdata = await _prepare( exits = exits, **nomargs )
455 from . import xtnsmgr
456 await xtnsmgr.register_processors( auxdata )
457 await self.command(
458 auxdata = auxdata,
459 display = self.display,
460 display_format = self.display_format,
461 color = self.color )
463 def prepare_invocation_args(
464 self,
465 ) -> __.cabc.Mapping[ str, __.typx.Any ]:
466 ''' Prepares arguments for initial configuration. '''
467 args: dict[ str, __.typx.Any ] = dict(
468 environment = True,
469 logfile = self.logfile,
470 )
471 return args
474def execute( ) -> None:
475 ''' Entrypoint for CLI execution. '''
476 config = (
477 __.tyro.conf.HelptextFromCommentsOff,
478 )
479 with __.warnings.catch_warnings( ):
480 __.warnings.filterwarnings(
481 'ignore',
482 message = r'Mutable type .* is used as a default value.*',
483 category = UserWarning,
484 module = 'tyro.constructors._struct_spec_dataclass' )
485 try: __.asyncio.run( __.tyro.cli( Cli, config = config )( ) )
486 except SystemExit: raise
487 except BaseException as exc:
488 __.report_exceptions( exc, _scribe )
489 raise SystemExit( 1 ) from None
502async def _prepare(
503 environment: __.typx.Annotated[
504 bool,
505 __.ddoc.Doc( ''' Whether to configure environment. ''' )
506 ],
507 exits: __.typx.Annotated[
508 __.ctxl.AsyncExitStack,
509 __.ddoc.Doc( ''' Exit stack for resource management. ''' )
510 ],
511 logfile: __.typx.Annotated[
512 __.typx.Optional[ str ],
513 __.ddoc.Doc( ''' Path to log capture file. ''' )
514 ],
515) -> __.typx.Annotated[
516 _state.Globals,
517 __.ddoc.Doc( ''' Configured global state. ''' )
518]:
519 ''' Configures application based on arguments. '''
520 nomargs: __.NominativeArguments = {
521 'environment': environment,
522 'exits': exits,
523 }
524 if logfile:
525 logfile_p = __.Path( logfile ).resolve( )
526 ( logfile_p.parent ).mkdir( parents = True, exist_ok = True )
527 logstream = exits.enter_context( logfile_p.open( 'w' ) )
528 inscription = __.appcore.inscription.Control(
529 level = 'debug', target = logstream )
530 nomargs[ 'inscription' ] = inscription
531 auxdata = await __.appcore.prepare( **nomargs )
532 content_cache, probe_cache, robots_cache = _cacheproxy.prepare( auxdata )
533 return _state.Globals(
534 application = auxdata.application,
535 configuration = auxdata.configuration,
536 directories = auxdata.directories,
537 distribution = auxdata.distribution,
538 exits = auxdata.exits,
539 content_cache = content_cache,
540 probe_cache = probe_cache,
541 robots_cache = robots_cache )