Coverage for sources/librovore/cli.py: 46%
181 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-06 02:25 +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__ )
38class TargetStream( __.enum.Enum ):
39 ''' Output stream selection. '''
40 Stdout = 'stdout'
41 Stderr = 'stderr'
44TargetMutex = __.tyro.conf.create_mutex_group( required = False )
47class DisplayOptions( __.immut.DataclassObject ):
48 ''' Consolidated display configuration for CLI output. '''
49 format: _interfaces.DisplayFormat = _interfaces.DisplayFormat.Markdown
50 target_stream: __.typx.Annotated[
51 __.typx.Optional[ TargetStream ],
52 TargetMutex,
53 __.tyro.conf.arg( help = "Output to stdout or stderr." )
54 ] = TargetStream.Stderr
55 target_file: __.typx.Annotated[
56 __.typx.Optional[ __.Path ],
57 TargetMutex,
58 __.tyro.conf.arg( help = "Output to specified file." )
59 ] = None
60 color: __.typx.Annotated[
61 bool,
62 __.tyro.conf.arg(
63 aliases = ( "--ansi-sgr", ),
64 help = "Enable colored output and terminal formatting."
65 ),
66 ] = True
68 async def provide_stream(
69 self, exits: __.ctxl.AsyncExitStack
70 ) -> __.typx.TextIO:
71 ''' Provides the target output stream. '''
72 if self.target_file is not None:
73 target_path = self.target_file.resolve( )
74 target_path.parent.mkdir( parents = True, exist_ok = True )
75 return exits.enter_context( target_path.open( 'w' ) )
76 target_stream = self.target_stream or TargetStream.Stderr
77 match target_stream:
78 case TargetStream.Stdout: return __.sys.stdout
79 case TargetStream.Stderr: return __.sys.stderr
80 case _: return __.sys.stderr
82 def decide_rich_markdown( self, stream: __.typx.TextIO ) -> bool:
83 ''' Determines whether to use Rich markdown rendering. '''
84 return decide_rich_markdown( stream, self.color )
87def intercept_errors( ) -> __.cabc.Callable[
88 [ __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ] ],
89 __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]
90]:
91 ''' Decorator for CLI handlers to intercept exceptions.
93 Catches Omnierror exceptions and renders them appropriately.
94 Other exceptions are logged and formatted simply.
95 '''
96 def decorator(
97 func: __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]
98 ) -> __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]:
99 @__.funct.wraps( func )
100 async def wrapper(
101 self: __.typx.Any,
102 auxdata: _state.Globals,
103 display: DisplayOptions,
104 *posargs: __.typx.Any,
105 **nomargs: __.typx.Any,
106 ) -> None:
107 stream = await display.provide_stream( auxdata.exits )
108 try:
109 return await func(
110 self, auxdata, display, *posargs, **nomargs )
111 except _exceptions.Omnierror as exc:
112 match display.format:
113 case _interfaces.DisplayFormat.JSON:
114 serialized = dict( exc.render_as_json( ) )
115 error_message = __.json.dumps( serialized, indent = 2 )
116 case _interfaces.DisplayFormat.Markdown:
117 lines = exc.render_as_markdown( )
118 error_message = '\n'.join( lines )
119 print( error_message, file = stream )
120 raise SystemExit( 1 ) from None
121 except Exception as exc:
122 _scribe.error( f"{func.__name__} failed: %s", exc )
123 match display.format:
124 case _interfaces.DisplayFormat.JSON:
125 error_data = {
126 "type": "unexpected_error",
127 "title": "Unexpected Error",
128 "message": str( exc ),
129 "suggestion": (
130 "Please report this issue if it persists." ),
131 }
132 error_message = __.json.dumps( error_data, indent = 2 )
133 case _interfaces.DisplayFormat.Markdown:
134 error_message = f"❌ Unexpected error: {exc}"
135 print( error_message, file = stream )
136 raise SystemExit( 1 ) from None
138 return wrapper
139 return decorator
142GroupByArgument: __.typx.TypeAlias = __.typx.Annotated[
143 __.typx.Optional[ str ],
144 __.tyro.conf.arg( help = __.access_doctab( 'group by argument' ) ),
145]
146PortArgument: __.typx.TypeAlias = __.typx.Annotated[
147 __.typx.Optional[ int ],
148 __.tyro.conf.arg( help = __.access_doctab( 'server port argument' ) ),
149]
150TermArgument: __.typx.TypeAlias = __.typx.Annotated[
151 __.tyro.conf.Positional[ str ],
152 __.tyro.conf.arg( help = __.access_doctab( 'term argument' ) ),
153]
154ResultsMax: __.typx.TypeAlias = __.typx.Annotated[
155 int,
156 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),
157]
158LocationArgument: __.typx.TypeAlias = __.typx.Annotated[
159 __.tyro.conf.Positional[ str ],
160 __.tyro.conf.arg( help = __.access_doctab( 'location argument' ) ),
161]
162TransportArgument: __.typx.TypeAlias = __.typx.Annotated[
163 __.typx.Optional[ str ],
164 __.tyro.conf.arg( help = __.access_doctab( 'transport argument' ) ),
165]
168_search_behaviors_default = _interfaces.SearchBehaviors( )
169_filters_default = __.immut.Dictionary[ str, __.typx.Any ]( )
171_MARKDOWN_OBJECT_LIMIT = 10
172_MARKDOWN_CONTENT_LIMIT = 200
175def decide_rich_markdown(
176 stream: __.typx.TextIO, colorize: bool
177) -> bool:
178 ''' Determines whether to use Rich markdown rendering. '''
179 return (
180 colorize
181 and stream.isatty( )
182 and not __.os.environ.get( 'NO_COLOR' )
183 )
186async def _render_and_print_result(
187 result: _results.ResultBase,
188 display: DisplayOptions,
189 exits: __.ctxl.AsyncExitStack,
190 **nomargs: __.typx.Any
191) -> None:
192 ''' Centralizes result rendering logic with Rich formatting support. '''
193 stream = await display.provide_stream( exits )
194 match display.format:
195 case _interfaces.DisplayFormat.JSON:
196 nomargs_filtered = {
197 key: value for key, value in nomargs.items()
198 if key in [ 'lines_max', 'reveal_internals' ]
199 }
200 serialized = dict( result.render_as_json( **nomargs_filtered ) )
201 output = __.json.dumps( serialized, indent = 2 )
202 print( output, file = stream )
203 case _interfaces.DisplayFormat.Markdown:
204 lines = result.render_as_markdown( **nomargs )
205 if display.decide_rich_markdown( stream ):
206 from rich.console import Console
207 from rich.markdown import Markdown
208 console = Console( file = stream, force_terminal = True )
209 markdown_obj = Markdown( '\n'.join( lines ) )
210 console.print( markdown_obj )
211 else:
212 output = '\n'.join( lines )
213 print( output, file = stream )
216class _CliCommand(
217 __.immut.DataclassProtocol, __.typx.Protocol,
218 decorators = ( __.typx.runtime_checkable, ),
219):
220 ''' CLI command. '''
222 @__.abc.abstractmethod
223 def __call__(
224 self,
225 auxdata: _state.Globals,
226 display: DisplayOptions,
227 ) -> __.cabc.Awaitable[ None ]:
228 ''' Executes command with global state. '''
229 raise NotImplementedError
232class DetectCommand(
233 _CliCommand, decorators = ( __.standard_tyro_class, ),
234):
235 ''' Detect which processors can handle a documentation source. '''
237 location: LocationArgument
238 genus: __.typx.Annotated[
239 _interfaces.ProcessorGenera,
240 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),
241 ]
242 processor_name: __.typx.Annotated[
243 __.typx.Optional[ str ],
244 __.tyro.conf.arg( help = "Specific processor to use." ),
245 ] = None
247 @intercept_errors( )
248 async def __call__(
249 self,
250 auxdata: _state.Globals,
251 display: DisplayOptions,
252 ) -> None:
253 processor_name = (
254 self.processor_name if self.processor_name is not None
255 else __.absent )
256 result = await _functions.detect(
257 auxdata, self.location, self.genus,
258 processor_name = processor_name )
259 await _render_and_print_result(
260 result, display, auxdata.exits, reveal_internals = False )
263class QueryInventoryCommand(
264 _CliCommand, decorators = ( __.standard_tyro_class, ),
265):
266 ''' Explores documentation structure and object inventory.
268 Use before content searches to:
270 - Discover available topics and object types
271 - Identify relevant search terms and filters
272 - Understand documentation scope and organization
273 '''
275 location: LocationArgument
276 term: TermArgument
277 filters: __.typx.Annotated[
278 __.cabc.Mapping[ str, __.typx.Any ],
279 __.tyro.conf.arg( prefix_name = False ),
280 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )
281 search_behaviors: __.typx.Annotated[
282 _interfaces.SearchBehaviors,
283 __.tyro.conf.arg( prefix_name = False ),
284 ] = __.dcls.field(
285 default_factory = lambda: _interfaces.SearchBehaviors( ) )
286 results_max: __.typx.Annotated[
287 int,
288 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),
289 ] = 5
290 reveal_internals: __.typx.Annotated[
291 bool,
292 __.tyro.conf.arg(
293 help = (
294 "Show internal implementation details (domain, priority, "
295 "project, version)." )
296 ),
297 ] = False
298 @intercept_errors( )
299 async def __call__(
300 self,
301 auxdata: _state.Globals,
302 display: DisplayOptions,
303 ) -> None:
304 result = await _functions.query_inventory(
305 auxdata,
306 self.location,
307 self.term,
308 search_behaviors = self.search_behaviors,
309 filters = self.filters,
310 results_max = self.results_max )
311 await _render_and_print_result(
312 result, display, auxdata.exits,
313 reveal_internals = self.reveal_internals )
316class QueryContentCommand(
317 _CliCommand, decorators = ( __.standard_tyro_class, ),
318):
319 ''' Searches documentation with flexible preview/extraction modes.
321 Workflows:
323 - Sample: Use --lines-max 5-10 to preview results and identify relevant
324 content
325 - Extract: Use --content-id from sample results to retrieve full
326 content
327 - Direct: Search with higher --lines-max for immediate full results
328 '''
330 location: LocationArgument
331 term: TermArgument
332 search_behaviors: __.typx.Annotated[
333 _interfaces.SearchBehaviors,
334 __.tyro.conf.arg( prefix_name = False ),
335 ] = __.dcls.field(
336 default_factory = lambda: _interfaces.SearchBehaviors( ) )
337 filters: __.typx.Annotated[
338 __.cabc.Mapping[ str, __.typx.Any ],
339 __.tyro.conf.arg( prefix_name = False ),
340 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )
341 results_max: ResultsMax = 10
342 lines_max: __.typx.Annotated[
343 int,
344 __.tyro.conf.arg(
345 help = (
346 "Lines per result for preview/sampling. Use 5-10 for "
347 "discovery, omit for full content extraction via "
348 "content-id." ) ),
349 ] = 40
350 content_id: __.typx.Annotated[
351 __.typx.Optional[ str ],
352 __.tyro.conf.arg(
353 help = (
354 "Extract full content for specific result. Obtain IDs from "
355 "previous query-content calls with limited lines-max." ) ),
356 ] = None
357 reveal_internals: __.typx.Annotated[
358 bool,
359 __.tyro.conf.arg(
360 help = (
361 "Show internal implementation details (domain, priority, "
362 "project, version)." )
363 ),
364 ] = False
365 @intercept_errors( )
366 async def __call__(
367 self,
368 auxdata: _state.Globals,
369 display: DisplayOptions,
370 ) -> None:
371 content_id_ = (
372 __.absent if self.content_id is None else self.content_id )
373 result = await _functions.query_content(
374 auxdata, self.location, self.term,
375 search_behaviors = self.search_behaviors,
376 filters = self.filters,
377 content_id = content_id_,
378 results_max = self.results_max,
379 lines_max = self.lines_max )
380 await _render_and_print_result(
381 result, display, auxdata.exits,
382 reveal_internals = self.reveal_internals,
383 lines_max = self.lines_max )
388class SurveyProcessorsCommand(
389 _CliCommand, decorators = ( __.standard_tyro_class, ),
390):
391 ''' List processors for specified genus and their capabilities. '''
393 genus: __.typx.Annotated[
394 _interfaces.ProcessorGenera,
395 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),
396 ]
397 name: __.typx.Annotated[
398 __.typx.Optional[ str ],
399 __.tyro.conf.arg( help = "Name of processor to describe" ),
400 ] = None
402 @intercept_errors( )
403 async def __call__(
404 self,
405 auxdata: _state.Globals,
406 display: DisplayOptions,
407 ) -> None:
408 nomargs: __.NominativeArguments = { 'genus': self.genus }
409 if self.name is not None: nomargs[ 'name' ] = self.name
410 result = await _functions.survey_processors( auxdata, **nomargs )
411 await _render_and_print_result(
412 result, display, auxdata.exits, reveal_internals = False )
416class ServeCommand(
417 _CliCommand, decorators = ( __.standard_tyro_class, ),
418):
419 ''' Starts MCP server. '''
421 port: PortArgument = None
422 transport: TransportArgument = None
423 extra_functions: __.typx.Annotated[
424 bool,
425 __.tyro.conf.arg(
426 help = "Enable extra functions (detect and survey-processors)." ),
427 ] = False
428 serve_function: __.typx.Callable[
429 [ _state.Globals ], __.cabc.Awaitable[ None ]
430 ] = _server.serve
431 async def __call__(
432 self,
433 auxdata: _state.Globals,
434 display: DisplayOptions,
435 ) -> None:
436 nomargs: __.NominativeArguments = { }
437 if self.port is not None: nomargs[ 'port' ] = self.port
438 if self.transport is not None: nomargs[ 'transport' ] = self.transport
439 nomargs[ 'extra_functions' ] = self.extra_functions
440 await self.serve_function( auxdata, **nomargs )
443class Cli( __.immut.DataclassObject, decorators = ( __.simple_tyro_class, ) ):
444 ''' MCP server CLI. '''
446 display: DisplayOptions = __.dcls.field(
447 default_factory = lambda: DisplayOptions( ) )
448 command: __.typx.Union[
449 __.typx.Annotated[
450 DetectCommand,
451 __.tyro.conf.subcommand( 'detect', prefix_name = False ),
452 ],
453 __.typx.Annotated[
454 QueryInventoryCommand,
455 __.tyro.conf.subcommand( 'query-inventory', prefix_name = False ),
456 ],
457 __.typx.Annotated[
458 QueryContentCommand,
459 __.tyro.conf.subcommand( 'query-content', prefix_name = False ),
460 ],
461 __.typx.Annotated[
462 SurveyProcessorsCommand,
463 __.tyro.conf.subcommand(
464 'survey-processors', prefix_name = False ),
465 ],
466 __.typx.Annotated[
467 ServeCommand,
468 __.tyro.conf.subcommand( 'serve', prefix_name = False ),
469 ],
470 ]
471 logfile: __.typx.Annotated[
472 __.typx.Optional[ str ],
473 __.ddoc.Doc( ''' Path to log capture file. ''' ),
474 ] = None
476 async def __call__( self ):
477 ''' Invokes command after library preparation. '''
478 nomargs = self.prepare_invocation_args( )
479 async with __.ctxl.AsyncExitStack( ) as exits:
480 auxdata = await _prepare( exits = exits, **nomargs )
481 from . import xtnsmgr
482 await xtnsmgr.register_processors( auxdata )
483 await self.command(
484 auxdata = auxdata,
485 display = self.display )
487 def prepare_invocation_args(
488 self,
489 ) -> __.cabc.Mapping[ str, __.typx.Any ]:
490 ''' Prepares arguments for initial configuration. '''
491 args: dict[ str, __.typx.Any ] = dict(
492 environment = True,
493 logfile = self.logfile,
494 )
495 return args
498def execute( ) -> None:
499 ''' Entrypoint for CLI execution. '''
500 config = (
501 __.tyro.conf.HelptextFromCommentsOff,
502 )
503 with __.warnings.catch_warnings( ):
504 __.warnings.filterwarnings(
505 'ignore',
506 message = r'Mutable type .* is used as a default value.*',
507 category = UserWarning,
508 module = 'tyro.constructors._struct_spec_dataclass' )
509 try: __.asyncio.run( __.tyro.cli( Cli, config = config )( ) )
510 except SystemExit: raise
511 except BaseException as exc:
512 __.report_exceptions( exc, _scribe )
513 raise SystemExit( 1 ) from None
526async def _prepare(
527 environment: __.typx.Annotated[
528 bool,
529 __.ddoc.Doc( ''' Whether to configure environment. ''' )
530 ],
531 exits: __.typx.Annotated[
532 __.ctxl.AsyncExitStack,
533 __.ddoc.Doc( ''' Exit stack for resource management. ''' )
534 ],
535 logfile: __.typx.Annotated[
536 __.typx.Optional[ str ],
537 __.ddoc.Doc( ''' Path to log capture file. ''' )
538 ],
539) -> __.typx.Annotated[
540 _state.Globals,
541 __.ddoc.Doc( ''' Configured global state. ''' )
542]:
543 ''' Configures application based on arguments. '''
544 nomargs: __.NominativeArguments = {
545 'environment': environment,
546 'exits': exits,
547 }
548 if logfile:
549 logfile_p = __.Path( logfile ).resolve( )
550 ( logfile_p.parent ).mkdir( parents = True, exist_ok = True )
551 logstream = exits.enter_context( logfile_p.open( 'w' ) )
552 inscription = __.appcore.inscription.Control(
553 level = 'debug', target = logstream )
554 nomargs[ 'inscription' ] = inscription
555 auxdata = await __.appcore.prepare( **nomargs )
556 content_cache, probe_cache, robots_cache = _cacheproxy.prepare( auxdata )
557 return _state.Globals(
558 application = auxdata.application,
559 configuration = auxdata.configuration,
560 directories = auxdata.directories,
561 distribution = auxdata.distribution,
562 exits = auxdata.exits,
563 content_cache = content_cache,
564 probe_cache = probe_cache,
565 robots_cache = robots_cache )