Coverage for sources/librovore/results.py: 33%
258 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''' Results structures.
23 Search results, inventory objects, content documents, etc....
24'''
27from . import __
28from . import exceptions as _exceptions
31_CONTENT_PREVIEW_LIMIT = 100
34class ResultBase( __.immut.DataclassProtocol, __.typx.Protocol ):
35 ''' Base protocol for all result objects with rendering methods. '''
37 @__.abc.abstractmethod
38 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
39 ''' Renders result as JSON-compatible dictionary. '''
40 raise NotImplementedError
42 @__.abc.abstractmethod
43 def render_as_markdown(
44 self, /, *,
45 reveal_internals: bool = False,
46 ) -> tuple[ str, ... ]:
47 ''' Renders result as Markdown lines for display. '''
48 raise NotImplementedError
51class InventoryObject( ResultBase ):
52 ''' Universal inventory object with complete source attribution.
54 Represents a single documentation object from any inventory source
55 with standardized fields and format-specific metadata container.
56 '''
58 name: __.typx.Annotated[
59 str,
60 __.ddoc.Doc( "Primary object identifier from inventory source." ),
61 ]
62 uri: __.typx.Annotated[
63 str,
64 __.ddoc.Doc( "Relative URI to object documentation content." ),
65 ]
66 inventory_type: __.typx.Annotated[
67 str,
68 __.ddoc.Doc(
69 "Inventory format identifier (e.g., sphinx_objects_inv)." ),
70 ]
71 location_url: __.typx.Annotated[
72 str, __.ddoc.Doc(
73 "Complete URL to inventory location for attribution." )
74 ]
75 display_name: __.typx.Annotated[
76 __.typx.Optional[ str ],
77 __.ddoc.Doc( "Human-readable name if different from name." ),
78 ] = None
79 specifics: __.typx.Annotated[
80 __.immut.Dictionary[ str, __.typx.Any ],
81 __.ddoc.Doc(
82 "Format-specific metadata (domain, role, priority, etc.)." ),
83 ] = __.dcls.field( default_factory = lambda: __.immut.Dictionary( ) )
86 @property
87 def effective_display_name( self ) -> str:
88 ''' Effective display name. Might be same as name. '''
89 if self.display_name is not None:
90 return self.display_name
91 return self.name
93 @__.abc.abstractmethod
94 def render_specifics_json(
95 self, /, *,
96 reveal_internals: bool = False,
97 ) -> __.immut.Dictionary[ str, __.typx.Any ]:
98 ''' Renders specifics for JSON output. '''
99 raise NotImplementedError
101 @__.abc.abstractmethod
102 def render_specifics_markdown(
103 self, /, *,
104 reveal_internals: __.typx.Annotated[
105 bool,
106 __.ddoc.Doc( '''
107 Controls whether implementation-specific details (internal
108 field names, version numbers, priority scores) are included.
109 When False, only user-facing information is shown.
110 ''' ),
111 ] = False,
112 ) -> tuple[ str, ... ]:
113 ''' Renders specifics as Markdown lines for CLI display. '''
114 raise NotImplementedError
116 def render_as_json(
117 self, /, *,
118 reveal_internals: bool = False,
119 ) -> __.immut.Dictionary[ str, __.typx.Any ]:
120 ''' Renders complete object as JSON-compatible dictionary. '''
121 base = __.immut.Dictionary[
122 str, __.typx.Any
123 ](
124 name = self.name,
125 uri = self.uri,
126 inventory_type = self.inventory_type,
127 location_url = self.location_url,
128 display_name = self.display_name,
129 effective_display_name = self.effective_display_name,
130 )
131 formatted_specifics = self.render_specifics_json(
132 reveal_internals = reveal_internals )
133 result_dict = dict( base )
134 result_dict.update( dict( formatted_specifics ) )
135 return __.immut.Dictionary[ str, __.typx.Any ]( result_dict )
137 def render_as_markdown(
138 self, /, *,
139 reveal_internals: __.typx.Annotated[
140 bool,
141 __.ddoc.Doc( "Controls whether internal details are shown." ),
142 ] = False,
143 ) -> tuple[ str, ... ]:
144 ''' Renders complete object as Markdown lines for display. '''
145 lines = [ f"### `{self.effective_display_name}`" ]
146 lines.append( f"- **URI:** {self.uri}" )
147 lines.append( f"- **Type:** {self.inventory_type}" )
148 lines.append( f"- **Location:** {self.location_url}" )
149 specifics_lines = self.render_specifics_markdown(
150 reveal_internals = reveal_internals )
151 lines.extend( specifics_lines )
152 return tuple( lines )
155class ContentDocument( ResultBase ):
156 ''' Documentation content with extracted metadata and content ID. '''
158 inventory_object: __.typx.Annotated[
159 InventoryObject,
160 __.ddoc.Doc( "Location inventory object for this content." ),
161 ]
162 content_id: __.typx.Annotated[
163 str,
164 __.ddoc.Doc( "Deterministic identifier for content retrieval." ),
165 ]
166 description: __.typx.Annotated[
167 str,
168 __.ddoc.Doc( "Extracted object description or summary." ),
169 ] = ''
170 documentation_url: __.typx.Annotated[
171 str,
172 __.ddoc.Doc( "Complete URL to full documentation page." ),
173 ] = ''
174 extraction_metadata: __.typx.Annotated[
175 __.immut.Dictionary[ str, __.typx.Any ],
176 __.ddoc.Doc( "Metadata from structure processor extraction." ),
177 ] = __.dcls.field( default_factory = lambda: __.immut.Dictionary( ) )
179 @property
180 def has_meaningful_content( self ) -> bool:
181 ''' Returns True if document contains useful extracted content. '''
182 return bool( self.description )
184 def render_as_json(
185 self, /, *,
186 lines_max: __.typx.Optional[ int ] = None,
187 ) -> __.immut.Dictionary[ str, __.typx.Any ]:
188 ''' Renders complete document as JSON-compatible dictionary. '''
189 description = self.description
190 if lines_max is not None:
191 desc_lines = description.split( '\n' )
192 if len( desc_lines ) > lines_max:
193 desc_lines = desc_lines[ :lines_max ]
194 desc_lines.append( "..." )
195 description = '\n'.join( desc_lines )
196 return __.immut.Dictionary[
197 str, __.typx.Any
198 ](
199 inventory_object = dict( self.inventory_object.render_as_json( ) ),
200 content_id = self.content_id,
201 description = description,
202 documentation_url = self.documentation_url,
203 extraction_metadata = dict( self.extraction_metadata ),
204 has_meaningful_content = self.has_meaningful_content,
205 )
207 def render_as_markdown(
208 self, /, *,
209 reveal_internals: __.typx.Annotated[
210 bool,
211 __.ddoc.Doc( "Controls whether internal details are shown." ),
212 ] = False,
213 lines_max: __.typx.Annotated[
214 __.typx.Optional[ int ],
215 __.ddoc.Doc( "Maximum lines to display for description." ),
216 ] = None,
217 include_title: __.typx.Annotated[
218 bool,
219 __.ddoc.Doc( "Whether to include document title header." ),
220 ] = True,
221 ) -> tuple[ str, ... ]:
222 ''' Renders complete document as Markdown lines for display. '''
223 lines: list[ str ] = [ ]
224 if include_title:
225 lines.append(
226 f"### `{self.inventory_object.effective_display_name}`" )
227 metadata_lines: list[ str ] = [ ]
228 if self.documentation_url:
229 metadata_lines.append( f"- **URL:** {self.documentation_url}" )
230 metadata_lines.append( f"- **Content ID:** `{self.content_id}`" )
231 if metadata_lines:
232 lines.extend( metadata_lines )
233 inventory_lines = self.inventory_object.render_specifics_markdown(
234 reveal_internals = reveal_internals )
235 if inventory_lines:
236 lines.extend( inventory_lines )
237 if self.description:
238 lines.append( "" )
239 description = self.description
240 if lines_max is not None:
241 desc_lines = description.split( '\n' )
242 if len( desc_lines ) > lines_max:
243 desc_lines = desc_lines[ :lines_max ]
244 desc_lines.append( "..." )
245 description = '\n'.join( desc_lines )
246 lines.append( description )
247 return tuple( lines )
250class InventoryLocationInfo( __.immut.DataclassObject ):
251 ''' Information about detected inventory location and processor. '''
253 inventory_type: __.typx.Annotated[
254 str,
255 __.ddoc.Doc( "Inventory format type identifier." ),
256 ]
257 location_url: __.typx.Annotated[
258 str,
259 __.ddoc.Doc( "Complete URL to inventory location." ),
260 ]
261 processor_name: __.typx.Annotated[
262 str,
263 __.ddoc.Doc( "Name of processor handling this location." ),
264 ]
265 confidence: __.typx.Annotated[
266 float,
267 __.ddoc.Doc( "Detection confidence score (0.0-1.0)." ),
268 ]
269 object_count: __.typx.Annotated[
270 int,
271 __.ddoc.Doc( "Total objects available in this inventory." ),
272 ]
274 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
275 ''' Renders location info as JSON-compatible dictionary. '''
276 return __.immut.Dictionary(
277 inventory_type = self.inventory_type,
278 location_url = self.location_url,
279 processor_name = self.processor_name,
280 confidence = self.confidence,
281 object_count = self.object_count,
282 )
285class SearchMetadata( __.immut.DataclassObject ):
286 ''' Search operation metadata and performance statistics. '''
288 results_count: __.typx.Annotated[
289 int,
290 __.ddoc.Doc( "Number of results returned to user." ),
291 ]
292 results_max: __.typx.Annotated[
293 int,
294 __.ddoc.Doc( "Maximum results requested by user." ),
295 ]
296 matches_total: __.typx.Annotated[
297 __.typx.Optional[ int ],
298 __.ddoc.Doc( "Total matching objects before limit applied." ),
299 ] = None
300 search_time_ms: __.typx.Annotated[
301 __.typx.Optional[ int ],
302 __.ddoc.Doc( "Search execution time in milliseconds." ),
303 ] = None
305 @property
306 def results_truncated( self ) -> bool:
307 ''' Returns True if results were limited by results_max. '''
308 if self.matches_total is None:
309 return False
310 return self.results_count < self.matches_total
312 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
313 ''' Renders search metadata as JSON-compatible dictionary. '''
314 return __.immut.Dictionary(
315 results_count = self.results_count,
316 results_max = self.results_max,
317 matches_total = self.matches_total,
318 search_time_ms = self.search_time_ms,
319 results_truncated = self.results_truncated,
320 )
323class SearchResult( ResultBase ):
324 ''' Search result with inventory object and match metadata. '''
326 inventory_object: __.typx.Annotated[
327 InventoryObject,
328 __.ddoc.Doc( "Matched inventory object with metadata." ),
329 ]
330 score: __.typx.Annotated[
331 float,
332 __.ddoc.Doc( "Search relevance score (0.0-1.0)." ),
333 ]
334 match_reasons: __.typx.Annotated[
335 tuple[ str, ... ],
336 __.ddoc.Doc( "Detailed reasons for search match." ),
337 ]
339 @classmethod
340 def from_inventory_object(
341 cls,
342 inventory_object: InventoryObject, *,
343 score: float,
344 match_reasons: __.cabc.Sequence[ str ],
345 ) -> __.typx.Self:
346 ''' Produces search result from inventory object with scoring. '''
347 return cls(
348 inventory_object = inventory_object,
349 score = score,
350 match_reasons = tuple( match_reasons ) )
352 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
353 ''' Renders search result as JSON-compatible dictionary. '''
354 return __.immut.Dictionary[
355 str, __.typx.Any
356 ](
357 inventory_object = dict( self.inventory_object.render_as_json( ) ),
358 score = self.score,
359 match_reasons = list( self.match_reasons ),
360 )
362 def render_as_markdown(
363 self, /, *,
364 reveal_internals: __.typx.Annotated[
365 bool,
366 __.ddoc.Doc( "Controls whether internal details are shown." ),
367 ] = False,
368 ) -> tuple[ str, ... ]:
369 ''' Renders search result as Markdown lines for display. '''
370 title = "### `{name}` (Score: {score:.2f})".format(
371 name = self.inventory_object.effective_display_name,
372 score = self.score )
373 lines = [ title ]
374 if reveal_internals and self.match_reasons:
375 reasons = ', '.join( self.match_reasons )
376 lines.append( "- **Match reasons:** {reasons}".format(
377 reasons = reasons ) )
378 inventory_lines = self.inventory_object.render_as_markdown(
379 reveal_internals = reveal_internals )
380 lines.extend( inventory_lines[ 1: ] ) # Skip duplicate title line
381 return tuple( lines )
384class ContentQueryResult( ResultBase ):
385 ''' Complete result structure for content queries. '''
387 location: __.typx.Annotated[
388 str,
389 __.ddoc.Doc( "Primary location URL for this query." ),
390 ]
391 term: __.typx.Annotated[
392 str,
393 __.ddoc.Doc( "Search term used for this query." ),
394 ]
395 documents: __.typx.Annotated[
396 tuple[ ContentDocument, ... ],
397 __.ddoc.Doc( "Documentation content for matching objects." ) ]
398 search_metadata: __.typx.Annotated[
399 SearchMetadata,
400 __.ddoc.Doc( "Search execution and result metadata." ),
401 ]
402 inventory_locations: __.typx.Annotated[
403 tuple[ InventoryLocationInfo, ... ],
404 __.ddoc.Doc( "Information about inventory locations used." ),
405 ]
407 def render_as_json(
408 self, /, *,
409 lines_max: __.typx.Optional[ int ] = None,
410 ) -> __.immut.Dictionary[ str, __.typx.Any ]:
411 ''' Renders content query result as JSON-compatible dictionary. '''
412 documents_json = [
413 dict( doc.render_as_json( lines_max = lines_max ) )
414 for doc in self.documents ]
415 locations_json = [
416 dict( loc.render_as_json( ) ) for loc in self.inventory_locations ]
417 return __.immut.Dictionary[
418 str, __.typx.Any
419 ](
420 location = self.location,
421 term = self.term,
422 documents = documents_json,
423 search_metadata = dict( self.search_metadata.render_as_json( ) ),
424 inventory_locations = locations_json,
425 )
427 def render_as_markdown(
428 self, /, *,
429 reveal_internals: __.typx.Annotated[
430 bool,
431 __.ddoc.Doc( "Controls whether internal details are shown." ),
432 ] = False,
433 lines_max: __.typx.Annotated[
434 __.typx.Optional[ int ],
435 __.ddoc.Doc( "Maximum lines to display per content result." ),
436 ] = None,
437 ) -> tuple[ str, ... ]:
438 ''' Renders content query result as Markdown lines for display. '''
439 title = "# Content Query Results"
440 if lines_max is not None:
441 title += " (truncated)"
442 lines = [ title ]
443 lines.append( "- **Term:** {term}".format( term = self.term ) )
444 if reveal_internals:
445 lines.append( "- **Location:** {location}".format(
446 location = self.location ) )
447 lines.append( "- **Results:** {count} of {max}".format(
448 count = self.search_metadata.results_count,
449 max = self.search_metadata.results_max ) )
450 if self.documents:
451 lines.append( "" )
452 lines.append( "## Documents" )
453 for index, doc in enumerate( self.documents, 1 ):
454 separator = "\n📄 ── Document {} ──────────────────── 📄\n"
455 lines.append( separator.format( index ) )
456 doc_lines = doc.render_as_markdown(
457 reveal_internals = reveal_internals,
458 lines_max = lines_max,
459 include_title = False )
460 lines.extend( doc_lines )
461 return tuple( lines )
464class InventoryQueryResult( ResultBase ):
465 ''' Complete result structure for inventory queries. '''
467 location: __.typx.Annotated[
468 str,
469 __.ddoc.Doc( "Primary location URL for this query." ),
470 ]
471 term: __.typx.Annotated[
472 str,
473 __.ddoc.Doc( "Search term used for this query." ),
474 ]
475 objects: __.typx.Annotated[
476 tuple[ InventoryObject, ... ],
477 __.ddoc.Doc( "Inventory objects matching search criteria." ),
478 ]
479 search_metadata: __.typx.Annotated[
480 SearchMetadata,
481 __.ddoc.Doc( "Search execution and result metadata." ),
482 ]
483 inventory_locations: __.typx.Annotated[
484 tuple[ InventoryLocationInfo, ... ],
485 __.ddoc.Doc( "Information about inventory locations used." ),
486 ]
488 def render_as_json(
489 self, /, *,
490 reveal_internals: bool = False,
491 ) -> __.immut.Dictionary[ str, __.typx.Any ]:
492 ''' Renders inventory query result as JSON-compatible dictionary. '''
493 objects_json = [
494 dict( obj.render_as_json( reveal_internals = reveal_internals ) )
495 for obj in self.objects ]
496 locations_json = [
497 dict( loc.render_as_json( ) ) for loc in self.inventory_locations ]
498 return __.immut.Dictionary[
499 str, __.typx.Any
500 ](
501 location = self.location,
502 term = self.term,
503 objects = objects_json,
504 search_metadata = dict( self.search_metadata.render_as_json( ) ),
505 inventory_locations = locations_json,
506 )
508 def render_as_markdown(
509 self, /, *,
510 reveal_internals: __.typx.Annotated[
511 bool,
512 __.ddoc.Doc( "Controls whether internal details are shown." ),
513 ] = False,
514 ) -> tuple[ str, ... ]:
515 ''' Renders inventory query result as Markdown lines for display. '''
516 lines = [ "# Inventory Query Results" ]
517 lines.append( "- **Term:** {term}".format( term = self.term ) )
518 if reveal_internals:
519 lines.append( "- **Location:** {location}".format(
520 location = self.location ) )
521 lines.append( "- **Results:** {count} of {max}".format(
522 count = self.search_metadata.results_count,
523 max = self.search_metadata.results_max ) )
524 if self.objects:
525 lines.append( "" )
526 lines.append( "## Objects" )
527 for index, obj in enumerate( self.objects, 1 ):
528 separator = "\n📦 ── Object {} ─────────────────────── 📦\n"
529 lines.append( separator.format( index ) )
530 obj_lines = obj.render_as_markdown(
531 reveal_internals = reveal_internals )
532 lines.extend( obj_lines )
533 return tuple( lines )
536class Detection( __.immut.DataclassObject ):
537 ''' Processor detection information with confidence scoring. '''
539 processor_name: __.typx.Annotated[
540 str,
541 __.ddoc.Doc( "Name of the processor that can handle this location." ),
542 ]
543 confidence: __.typx.Annotated[
544 float,
545 __.ddoc.Doc( "Detection confidence score (0.0-1.0)." ),
546 ]
547 processor_type: __.typx.Annotated[
548 str,
549 __.ddoc.Doc( "Type of processor (inventory, structure)." ),
550 ]
551 detection_metadata: __.typx.Annotated[
552 __.immut.Dictionary[ str, __.typx.Any ],
553 __.ddoc.Doc( "Processor-specific detection metadata." ),
554 ] = __.dcls.field( default_factory = lambda: __.immut.Dictionary( ) )
556 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
557 ''' Renders detection as JSON-compatible dictionary. '''
558 return __.immut.Dictionary[
559 str, __.typx.Any
560 ](
561 processor_name = self.processor_name,
562 confidence = self.confidence,
563 processor_type = self.processor_type,
564 detection_metadata = dict( self.detection_metadata ),
565 )
568class DetectionsResult( ResultBase ):
569 ''' Detection results with processor selection and timing metadata. '''
571 source: __.typx.Annotated[
572 str,
573 __.ddoc.Doc( "Primary location URL for detection operation." ),
574 ]
575 detections: __.typx.Annotated[
576 tuple[ Detection, ... ],
577 __.ddoc.Doc( "All processor detections found for location." ),
578 ]
579 detection_optimal: __.typx.Annotated[
580 __.typx.Optional[ Detection ],
581 __.ddoc.Doc( "Best detection result based on confidence scoring." ),
582 ]
583 time_detection_ms: __.typx.Annotated[
584 int,
585 __.ddoc.Doc( "Detection operation time in milliseconds." ),
586 ]
589 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
590 ''' Renders detection results as JSON-compatible dictionary. '''
591 detections_json = [
592 dict( detection.render_as_json( ) )
593 for detection in self.detections ]
594 return __.immut.Dictionary[
595 str, __.typx.Any
596 ](
597 source = self.source,
598 detections = detections_json,
599 detection_optimal = (
600 dict( self.detection_optimal.render_as_json( ) )
601 if self.detection_optimal else None ),
602 time_detection_ms = self.time_detection_ms,
603 )
605 def render_as_markdown(
606 self, /, *,
607 reveal_internals: __.typx.Annotated[
608 bool,
609 __.ddoc.Doc( "Controls whether internal details are shown." ),
610 ] = False,
611 ) -> tuple[ str, ... ]:
612 ''' Renders detection results as Markdown lines for display. '''
613 lines = [ "# Detection Results" ]
614 if reveal_internals:
615 lines.append( "- **Source:** {source}".format(
616 source = self.source ) )
617 lines.append( "- **Detection time:** {time}ms".format(
618 time = self.time_detection_ms ) )
619 if self.detection_optimal:
620 lines.append( "- **Optimal processor:** {name} ({type})".format(
621 name = self.detection_optimal.processor_name,
622 type = self.detection_optimal.processor_type ) )
623 lines.append( "- **Confidence:** {conf:.2f}".format(
624 conf = self.detection_optimal.confidence ) )
625 else:
626 lines.append( "- **No optimal processor found**" )
627 if reveal_internals and self.detections:
628 lines.append( "" )
629 lines.append( "## All Detections" )
630 detection_lines = [
631 "- **{name}** ({type}): {conf:.2f}".format(
632 name = detection.processor_name,
633 type = detection.processor_type,
634 conf = detection.confidence )
635 for detection in self.detections ]
636 lines.extend( detection_lines )
637 return tuple( lines )
640class ProcessorInfo( ResultBase ):
641 ''' Information about a processor and its capabilities. '''
643 processor_name: __.typx.Annotated[
644 str,
645 __.ddoc.Doc( "Name of the processor for identification." ),
646 ]
647 processor_type: __.typx.Annotated[
648 str,
649 __.ddoc.Doc( "Type of processor (inventory, structure)." ),
650 ]
651 capabilities: __.typx.Annotated[
652 __.typx.Any, # Will be _interfaces.ProcessorCapabilities after import
653 __.ddoc.Doc( "Complete capability description for processor." ),
654 ]
656 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
657 ''' Renders processor info as JSON-compatible dictionary. '''
658 return __.immut.Dictionary[
659 str, __.typx.Any
660 ](
661 processor_name = self.processor_name,
662 processor_type = self.processor_type,
663 capabilities = self.capabilities.render_as_json( ),
664 )
666 def render_as_markdown(
667 self, /, *,
668 reveal_internals: __.typx.Annotated[
669 bool,
670 __.ddoc.Doc( "Controls whether internal details are shown." ),
671 ] = False,
672 ) -> tuple[ str, ... ]:
673 ''' Renders processor info as Markdown lines for display. '''
674 lines = [ f"### `{self.processor_name}` ({self.processor_type})" ]
675 if reveal_internals:
676 capabilities_lines = self.capabilities.render_as_markdown( )
677 lines.extend( capabilities_lines )
678 return tuple( lines )
681class ProcessorsSurveyResult( ResultBase ):
682 ''' Survey results listing available processors and capabilities. '''
684 genus: __.typx.Annotated[
685 __.typx.Any, # Will be _interfaces.ProcessorGenera after import
686 __.ddoc.Doc(
687 "Processor genus that was surveyed (inventory or structure)." ),
688 ]
689 filter_name: __.typx.Annotated[
690 __.typx.Optional[ str ],
691 __.ddoc.Doc( "Optional processor name filter applied to survey." ),
692 ] = None
693 processors: __.typx.Annotated[
694 tuple[ ProcessorInfo, ... ],
695 __.ddoc.Doc( "Available processors matching survey criteria." ),
696 ]
697 survey_time_ms: __.typx.Annotated[
698 int,
699 __.ddoc.Doc( "Survey operation time in milliseconds." ),
700 ]
702 def render_as_json( self ) -> __.immut.Dictionary[ str, __.typx.Any ]:
703 ''' Renders survey results as JSON-compatible dictionary. '''
704 processors_json = [
705 dict( processor.render_as_json( ) )
706 for processor in self.processors ]
707 return __.immut.Dictionary[
708 str, __.typx.Any
709 ](
710 genus = (
711 self.genus.value if hasattr( self.genus, 'value' )
712 else str( self.genus ) ),
713 filter_name = self.filter_name,
714 processors = processors_json,
715 survey_time_ms = self.survey_time_ms,
716 )
718 def render_as_markdown(
719 self, /, *,
720 reveal_internals: __.typx.Annotated[
721 bool,
722 __.ddoc.Doc( "Controls whether internal details are shown." ),
723 ] = False,
724 ) -> tuple[ str, ... ]:
725 ''' Renders survey results as Markdown lines for display. '''
726 genus_name = (
727 self.genus.value if hasattr( self.genus, 'value' )
728 else str( self.genus ) )
729 title = f"# Processor Survey Results ({genus_name})"
730 lines = [ title ]
731 if reveal_internals:
732 lines.append( f"- **Survey time:** {self.survey_time_ms}ms" )
733 if self.filter_name:
734 lines.append( f"- **Filter:** {self.filter_name}" )
735 lines.append( f"- **Processors found:** {len( self.processors )}" )
736 if self.processors:
737 lines.append( "" )
738 for i, processor in enumerate( self.processors, 1 ):
739 lines.append( f"📦 ── Processor {i} ──────────" )
740 processor_lines = processor.render_as_markdown(
741 reveal_internals = reveal_internals )
742 lines.extend( processor_lines )
743 if i < len( self.processors ):
744 lines.append( "" )
745 return tuple( lines )
748def parse_content_id( content_id: str ) -> tuple[ str, str ]:
749 ''' Parses content identifier back to location and name components.
751 Returns tuple of (location, name) extracted from content_id.
752 Raises ContentIdInvalidity if content_id is malformed or cannot be
753 decoded.
754 '''
755 try:
756 identifier_source = __.base64.b64decode(
757 content_id.encode( 'ascii' ) ).decode( 'utf-8' )
758 except Exception as exc:
759 raise _exceptions.ContentIdInvalidity(
760 content_id, "Base64 decoding failed" ) from exc
761 if ':' not in identifier_source:
762 raise _exceptions.ContentIdInvalidity(
763 content_id, "Missing location:object separator" )
764 location, name = identifier_source.rsplit( ':', 1 )
765 return location, name
768def produce_content_id( location: str, name: str ) -> str:
769 ''' Produces deterministic content identifier for browse-then-extract.
771 Uses base64 encoding of location + ":" + name to create stable,
772 debuggable identifiers that maintain stateless operation.
773 '''
774 identifier_source = f"{location}:{name}"
775 return __.base64.b64encode(
776 identifier_source.encode( 'utf-8' ) ).decode( 'ascii' )
784ContentDocuments: __.typx.TypeAlias = __.cabc.Sequence[ ContentDocument ]
785InventoryObjects: __.typx.TypeAlias = __.cabc.Sequence[ InventoryObject ]
786SearchResults: __.typx.TypeAlias = __.cabc.Sequence[ SearchResult ]