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