Coverage for sources/librovore/cli.py: 34%
178 statements
« prev ^ index » next coverage.py v7.10.5, created at 2025-08-29 01:14 +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
34_scribe = __.acquire_scribe( __name__ )
37GroupByArgument: __.typx.TypeAlias = __.typx.Annotated[
38 __.typx.Optional[ str ],
39 __.tyro.conf.arg( help = __.access_doctab( 'group by argument' ) ),
40]
41IncludeSnippets: __.typx.TypeAlias = __.typx.Annotated[
42 bool,
43 __.tyro.conf.arg( help = __.access_doctab( 'include snippets argument' ) ),
44]
45PortArgument: __.typx.TypeAlias = __.typx.Annotated[
46 __.typx.Optional[ int ],
47 __.tyro.conf.arg( help = __.access_doctab( 'server port argument' ) ),
48]
49TermArgument: __.typx.TypeAlias = __.typx.Annotated[
50 __.tyro.conf.Positional[ str ],
51 __.tyro.conf.arg( help = __.access_doctab( 'term argument' ) ),
52]
53ResultsMax: __.typx.TypeAlias = __.typx.Annotated[
54 int,
55 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),
56]
57LocationArgument: __.typx.TypeAlias = __.typx.Annotated[
58 __.tyro.conf.Positional[ str ],
59 __.tyro.conf.arg( help = __.access_doctab( 'location argument' ) ),
60]
61TransportArgument: __.typx.TypeAlias = __.typx.Annotated[
62 __.typx.Optional[ str ],
63 __.tyro.conf.arg( help = __.access_doctab( 'transport argument' ) ),
64]
67_search_behaviors_default = _interfaces.SearchBehaviors( )
68_filters_default = __.immut.Dictionary[ str, __.typx.Any ]( )
70_MARKDOWN_OBJECT_LIMIT = 10
71_MARKDOWN_CONTENT_LIMIT = 200
75class _CliCommand(
76 __.immut.DataclassProtocol, __.typx.Protocol,
77 decorators = ( __.typx.runtime_checkable, ),
78):
79 ''' CLI command. '''
81 @__.abc.abstractmethod
82 async def __call__(
83 self,
84 auxdata: _state.Globals,
85 display: __.DisplayTarget,
86 display_format: _interfaces.DisplayFormat,
87 ) -> None:
88 ''' Executes command with global state. '''
89 raise NotImplementedError
92class DetectCommand(
93 _CliCommand, decorators = ( __.standard_tyro_class, ),
94):
95 ''' Detect which processors can handle a documentation source. '''
97 location: LocationArgument
98 genus: __.typx.Annotated[
99 _interfaces.ProcessorGenera,
100 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),
101 ]
102 processor_name: __.typx.Annotated[
103 __.typx.Optional[ str ],
104 __.tyro.conf.arg( help = "Specific processor to use." ),
105 ] = None
107 async def __call__(
108 self,
109 auxdata: _state.Globals,
110 display: __.DisplayTarget,
111 display_format: _interfaces.DisplayFormat,
112 ) -> None:
113 stream = await display.provide_stream( )
114 processor_name = (
115 self.processor_name if self.processor_name is not None
116 else __.absent )
117 try:
118 result = await _functions.detect(
119 auxdata, self.location, self.genus,
120 processor_name = processor_name )
121 except Exception as exc:
122 _scribe.error( "detect failed: %s", exc )
123 print( _format_cli_exception( exc ), file = stream )
124 raise SystemExit( 1 ) from None
125 match display_format:
126 case _interfaces.DisplayFormat.JSON:
127 if isinstance( result, _results.ErrorResponse ):
128 serialized = dict( result.render_as_json( ) )
129 else:
130 serialized = dict( result.render_as_json( ) )
131 output = __.json.dumps( serialized, indent = 2 )
132 case _interfaces.DisplayFormat.Markdown:
133 if isinstance( result, _results.ErrorResponse ):
134 lines = result.render_as_markdown(
135 reveal_internals = True )
136 else:
137 lines = result.render_as_markdown(
138 reveal_internals = True )
139 output = '\n'.join( lines )
140 print( output, file = stream )
143class QueryInventoryCommand(
144 _CliCommand, decorators = ( __.standard_tyro_class, ),
145):
146 ''' Searches object inventory by name with fuzzy matching. '''
148 location: LocationArgument
149 term: TermArgument
150 details: __.typx.Annotated[
151 _interfaces.InventoryQueryDetails,
152 __.tyro.conf.arg(
153 help = __.access_doctab( 'query details argument' ) ),
154 ] = _interfaces.InventoryQueryDetails.Documentation
155 filters: __.typx.Annotated[
156 __.cabc.Mapping[ str, __.typx.Any ],
157 __.tyro.conf.arg( prefix_name = False ),
158 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )
159 search_behaviors: __.typx.Annotated[
160 _interfaces.SearchBehaviors,
161 __.tyro.conf.arg( prefix_name = False ),
162 ] = __.dcls.field(
163 default_factory = lambda: _interfaces.SearchBehaviors( ) )
164 results_max: __.typx.Annotated[
165 int,
166 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),
167 ] = 5
169 async def __call__(
170 self,
171 auxdata: _state.Globals,
172 display: __.DisplayTarget,
173 display_format: _interfaces.DisplayFormat,
174 ) -> None:
175 stream = await display.provide_stream( )
176 try:
177 result = await _functions.query_inventory(
178 auxdata,
179 self.location,
180 self.term,
181 search_behaviors = self.search_behaviors,
182 filters = self.filters,
183 results_max = self.results_max,
184 details = self.details )
185 except Exception as exc:
186 _scribe.error( "query-inventory failed: %s", exc )
187 print( _format_cli_exception( exc ), file = stream )
188 raise SystemExit( 1 ) from None
189 match display_format:
190 case _interfaces.DisplayFormat.JSON:
191 output = __.json.dumps(
192 dict( result.render_as_json( ) ), indent = 2 )
193 case _interfaces.DisplayFormat.Markdown:
194 lines = result.render_as_markdown(
195 reveal_internals = True )
196 output = '\n'.join( lines )
197 print( output, file = stream )
200class QueryContentCommand(
201 _CliCommand, decorators = ( __.standard_tyro_class, ),
202):
203 ''' Searches documentation content with relevance ranking and snippets. '''
205 location: LocationArgument
206 term: TermArgument
207 search_behaviors: __.typx.Annotated[
208 _interfaces.SearchBehaviors,
209 __.tyro.conf.arg( prefix_name = False ),
210 ] = __.dcls.field(
211 default_factory = lambda: _interfaces.SearchBehaviors( ) )
212 filters: __.typx.Annotated[
213 __.cabc.Mapping[ str, __.typx.Any ],
214 __.tyro.conf.arg( prefix_name = False ),
215 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )
216 include_snippets: IncludeSnippets = True
217 results_max: ResultsMax = 10
218 lines_max: __.typx.Annotated[
219 int,
220 __.tyro.conf.arg(
221 help = "Maximum number of lines to display per result." ),
222 ] = 40
224 async def __call__(
225 self,
226 auxdata: _state.Globals,
227 display: __.DisplayTarget,
228 display_format: _interfaces.DisplayFormat,
229 ) -> None:
230 stream = await display.provide_stream( )
231 try:
232 result = await _functions.query_content(
233 auxdata, self.location, self.term,
234 search_behaviors = self.search_behaviors,
235 filters = self.filters,
236 results_max = self.results_max,
237 include_snippets = self.include_snippets )
238 except Exception as exc:
239 _scribe.error( "query-content failed: %s", exc )
240 print( _format_cli_exception( exc ), file = stream )
241 raise SystemExit( 1 ) from None
242 match display_format:
243 case _interfaces.DisplayFormat.JSON:
244 output = __.json.dumps(
245 dict( result.render_as_json( ) ), indent = 2 )
246 case _interfaces.DisplayFormat.Markdown:
247 if isinstance( result, _results.ContentQueryResult ):
248 lines = result.render_as_markdown(
249 reveal_internals = True, lines_max = self.lines_max )
250 else:
251 lines = result.render_as_markdown(
252 reveal_internals = True )
253 output = '\n'.join( lines )
254 print( output, file = stream )
259class SurveyProcessorsCommand(
260 _CliCommand, decorators = ( __.standard_tyro_class, ),
261):
262 ''' List processors for specified genus and their capabilities. '''
264 genus: __.typx.Annotated[
265 _interfaces.ProcessorGenera,
266 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),
267 ]
268 name: __.typx.Annotated[
269 __.typx.Optional[ str ],
270 __.tyro.conf.arg( help = "Name of processor to describe" ),
271 ] = None
273 async def __call__(
274 self,
275 auxdata: _state.Globals,
276 display: __.DisplayTarget,
277 display_format: _interfaces.DisplayFormat,
278 ) -> None:
279 stream = await display.provide_stream( )
280 nomargs: __.NominativeArguments = { 'genus': self.genus }
281 if self.name is not None: nomargs[ 'name' ] = self.name
282 try:
283 result = await _functions.survey_processors( auxdata, **nomargs )
284 except Exception as exc:
285 _scribe.error( "survey-processors failed: %s", exc )
286 print( _format_cli_exception( exc ), file = stream )
287 raise SystemExit( 1 ) from None
288 match display_format:
289 case _interfaces.DisplayFormat.JSON:
290 output = __.json.dumps(
291 dict( result.render_as_json( ) ), indent = 2 )
292 case _interfaces.DisplayFormat.Markdown:
293 lines = result.render_as_markdown(
294 reveal_internals = False )
295 output = '\n'.join( lines )
296 print( output, file = stream )
300class ServeCommand(
301 _CliCommand, decorators = ( __.standard_tyro_class, ),
302):
303 ''' Starts MCP server. '''
305 port: PortArgument = None
306 transport: TransportArgument = None
307 extra_functions: __.typx.Annotated[
308 bool,
309 __.tyro.conf.arg(
310 help = "Enable extra functions (detect and survey-processors)." ),
311 ] = False
312 serve_function: __.typx.Callable[
313 [ _state.Globals ], __.cabc.Awaitable[ None ]
314 ] = _server.serve
315 async def __call__(
316 self,
317 auxdata: _state.Globals,
318 display: __.DisplayTarget,
319 display_format: _interfaces.DisplayFormat,
320 ) -> None:
321 nomargs: __.NominativeArguments = { }
322 if self.port is not None: nomargs[ 'port' ] = self.port
323 if self.transport is not None: nomargs[ 'transport' ] = self.transport
324 nomargs[ 'extra_functions' ] = self.extra_functions
325 await self.serve_function( auxdata, **nomargs )
328class Cli( __.immut.DataclassObject, decorators = ( __.simple_tyro_class, ) ):
329 ''' MCP server CLI. '''
331 display: __.DisplayTarget
332 display_format: __.typx.Annotated[
333 _interfaces.DisplayFormat,
334 __.tyro.conf.arg( help = "Output format for command results." ),
335 ] = _interfaces.DisplayFormat.Markdown
336 command: __.typx.Union[
337 __.typx.Annotated[
338 DetectCommand,
339 __.tyro.conf.subcommand( 'detect', prefix_name = False ),
340 ],
341 __.typx.Annotated[
342 QueryInventoryCommand,
343 __.tyro.conf.subcommand( 'query-inventory', prefix_name = False ),
344 ],
345 __.typx.Annotated[
346 QueryContentCommand,
347 __.tyro.conf.subcommand( 'query-content', prefix_name = False ),
348 ],
349 __.typx.Annotated[
350 SurveyProcessorsCommand,
351 __.tyro.conf.subcommand(
352 'survey-processors', prefix_name = False ),
353 ],
354 __.typx.Annotated[
355 ServeCommand,
356 __.tyro.conf.subcommand( 'serve', prefix_name = False ),
357 ],
358 ]
359 logfile: __.typx.Annotated[
360 __.typx.Optional[ str ],
361 __.ddoc.Doc( ''' Path to log capture file. ''' ),
362 ] = None
364 async def __call__( self ):
365 ''' Invokes command after library preparation. '''
366 nomargs = self.prepare_invocation_args( )
367 async with __.ctxl.AsyncExitStack( ) as exits:
368 auxdata = await _prepare( exits = exits, **nomargs )
369 from . import xtnsmgr
370 await xtnsmgr.register_processors( auxdata )
371 await self.command(
372 auxdata = auxdata,
373 display = self.display,
374 display_format = self.display_format )
376 def prepare_invocation_args(
377 self,
378 ) -> __.cabc.Mapping[ str, __.typx.Any ]:
379 ''' Prepares arguments for initial configuration. '''
380 args: dict[ str, __.typx.Any ] = dict(
381 environment = True,
382 logfile = self.logfile,
383 )
384 return args
387def execute( ) -> None:
388 ''' Entrypoint for CLI execution. '''
389 config = (
390 __.tyro.conf.HelptextFromCommentsOff,
391 )
392 with __.warnings.catch_warnings( ):
393 __.warnings.filterwarnings(
394 'ignore',
395 message = r'Mutable type .* is used as a default value.*',
396 category = UserWarning,
397 module = 'tyro.constructors._struct_spec_dataclass' )
398 try: __.asyncio.run( __.tyro.cli( Cli, config = config )( ) )
399 except SystemExit: raise
400 except BaseException as exc:
401 __.report_exceptions( exc, _scribe )
402 raise SystemExit( 1 ) from None
413def _format_cli_exception( exc: Exception ) -> str: # noqa: PLR0911
414 ''' Formats exceptions for user-friendly CLI output. '''
415 match exc:
416 case _exceptions.ProcessorInavailability( ):
417 return (
418 f"❌ No processor found to handle source: {exc.source}\n"
419 f"💡 Verify this is a Sphinx documentation site" )
420 case _exceptions.InventoryInaccessibility( ):
421 return (
422 f"❌ Cannot access documentation inventory: {exc.source}\n"
423 f"💡 Check URL accessibility and network connection" )
424 case _exceptions.DocumentationContentAbsence( ):
425 return (
426 f"❌ Documentation structure not recognized: {exc.url}\n"
427 f"💡 This may be an unsupported Sphinx theme" )
428 case _exceptions.DocumentationObjectAbsence( ):
429 return (
430 f"❌ Object '{exc.object_id}' not found in page: {exc.url}\n"
431 f"💡 Verify the object name and try a broader search" )
432 case _exceptions.InventoryInvalidity( ):
433 return (
434 f"❌ Invalid documentation inventory: {exc.source}\n"
435 f"💡 The documentation site may be corrupted" )
436 case _exceptions.DocumentationInaccessibility( ):
437 return (
438 f"❌ Documentation inaccessible: {exc.url}\n"
439 f"💡 Check URL accessibility and network connection" )
440 case _:
441 return f"❌ Unexpected error: {exc}"
444async def _prepare(
445 environment: __.typx.Annotated[
446 bool,
447 __.ddoc.Doc( ''' Whether to configure environment. ''' )
448 ],
449 exits: __.typx.Annotated[
450 __.ctxl.AsyncExitStack,
451 __.ddoc.Doc( ''' Exit stack for resource management. ''' )
452 ],
453 logfile: __.typx.Annotated[
454 __.typx.Optional[ str ],
455 __.ddoc.Doc( ''' Path to log capture file. ''' )
456 ],
457) -> __.typx.Annotated[
458 _state.Globals,
459 __.ddoc.Doc( ''' Configured global state. ''' )
460]:
461 ''' Configures application based on arguments. '''
462 nomargs: __.NominativeArguments = {
463 'environment': environment,
464 'exits': exits,
465 }
466 if logfile:
467 logfile_p = __.Path( logfile ).resolve( )
468 ( logfile_p.parent ).mkdir( parents = True, exist_ok = True )
469 logstream = exits.enter_context( logfile_p.open( 'w' ) )
470 inscription = __.appcore.inscription.Control(
471 level = 'debug', target = logstream )
472 nomargs[ 'inscription' ] = inscription
473 auxdata = await __.appcore.prepare( **nomargs )
474 content_cache, probe_cache, robots_cache = _cacheproxy.prepare( auxdata )
475 return _state.Globals(
476 application = auxdata.application,
477 configuration = auxdata.configuration,
478 directories = auxdata.directories,
479 distribution = auxdata.distribution,
480 exits = auxdata.exits,
481 content_cache = content_cache,
482 probe_cache = probe_cache,
483 robots_cache = robots_cache )