1use crate::graph::PythonGraphSource;
9use crate::graph_support::PythonUsageSource;
10use crate::imports::resolve_fqn_candidates;
11use crate::syntax::{
12 python_deferred_annotation_identifier_ranges, python_deferred_annotation_tree,
13 python_node_is_in_annotation,
14};
15use crate::usage_index::usage_seeds;
16use brokk_bifrost_core::analyzer::fq_name::{FqName, SegmentKind, segment_interner};
17use brokk_bifrost_core::analyzer::symbol_path::parse_symbol_path;
18use brokk_bifrost_core::analyzer::usages::model::{ExportEntry, ImportBinder, ImportKind};
19use brokk_bifrost_core::analyzer::usages::{ImportEdge, ImportEdgeKind};
20use brokk_bifrost_core::analyzer::{
21 CodeUnit, CodeUnitIndex, DefinitionLanguageScope, Language, ProjectFile, Range,
22 RelationalDefinitionQuery, RelationalDefinitionQuestion, RelationalDefinitionValue,
23 RelationalName,
24};
25use brokk_bifrost_core::hash::HashSet;
26use std::collections::BTreeSet;
27use tree_sitter::Node;
28
29const MAX_MEMBER_OWNER_FRONTIER: usize = 512;
30
31pub(crate) fn resolved_member_declarations(
35 graph: &PythonGraphSource<'_>,
36 receiver: &CodeUnit,
37 member: &str,
38) -> Vec<CodeUnit> {
39 let mut visited = HashSet::default();
40 let mut frontier = vec![receiver.clone()];
41 while !frontier.is_empty() && visited.len() < MAX_MEMBER_OWNER_FRONTIER {
42 let mut declarations = Vec::new();
43 let mut next = Vec::new();
44 for owner in frontier {
45 if !visited.insert(owner.clone()) {
46 continue;
47 }
48 let fqn = format!("{}.{member}", owner.fq_name());
49 declarations.extend(
50 graph
51 .index
52 .definitions(&fqn)
53 .filter(|candidate| graph.index.parent_of(candidate).as_ref() == Some(&owner)),
54 );
55 if let Some(hierarchy) = graph.hierarchy {
56 next.extend(hierarchy.get_direct_ancestors(&owner));
57 }
58 }
59 declarations.sort();
60 declarations.dedup();
61 if !declarations.is_empty() {
62 return declarations;
63 }
64 next.sort();
65 next.dedup();
66 frontier = next;
67 }
68 Vec::new()
69}
70
71pub fn infer_export_names(python: &dyn PythonUsageSource, target: &CodeUnit) -> BTreeSet<String> {
72 if target_owner_code_unit(python, target).is_some() {
73 let owner_name = top_level_identifier(python, target);
74 let owner_exports =
75 infer_export_names_for_local(python, target, target.source(), &owner_name);
76 if !owner_exports.is_empty() {
77 return owner_exports;
78 }
79 }
80
81 infer_export_names_for_local(python, target, target.source(), target.identifier())
82}
83
84pub fn infer_usage_seeds(
85 python: &dyn PythonUsageSource,
86 target: &CodeUnit,
87 seed_names: BTreeSet<String>,
88) -> BTreeSet<(ProjectFile, String)> {
89 let mut seeds = BTreeSet::new();
90 for seed_name in &seed_names {
91 seeds.extend(usage_seeds(python, target.source(), seed_name));
92 }
93 if seeds.is_empty()
94 && seed_names.contains(target.identifier())
95 && is_module_level_target_identifier(python, target, target.source(), target.identifier())
96 {
97 seeds.insert((target.source().clone(), target.identifier().to_string()));
98 }
99 seeds
100}
101
102fn infer_export_names_for_local(
103 python: &dyn PythonUsageSource,
104 target: &CodeUnit,
105 file: &ProjectFile,
106 local_name: &str,
107) -> BTreeSet<String> {
108 let index = python.export_index_of(file);
109 let mut export_names = BTreeSet::new();
110 if index.exports_by_name.contains_key(local_name) {
111 export_names.insert(local_name.to_string());
112 }
113 for (export_name, entry) in &index.exports_by_name {
114 if matches!(entry, ExportEntry::Local { local_name: name } if name == local_name) {
115 export_names.insert(export_name.clone());
116 }
117 }
118 if export_names.is_empty()
119 && is_module_level_target_identifier(python, target, file, local_name)
120 {
121 export_names.insert(local_name.to_string());
122 }
123 export_names
124}
125
126fn is_module_level_target_identifier(
127 python: &dyn PythonUsageSource,
128 target: &CodeUnit,
129 file: &ProjectFile,
130 local_name: &str,
131) -> bool {
132 target.source() == file
133 && target.identifier() == local_name
134 && python
135 .parent_of(target)
136 .is_some_and(|parent| parent.is_module() && parent.source() == file)
137}
138
139pub fn top_level_identifier(index: &dyn CodeUnitIndex, target: &CodeUnit) -> String {
140 let mut current = target.clone();
141 while let Some(parent) = index.parent_of(¤t) {
142 if parent.is_module() {
143 break;
144 }
145 current = parent;
146 }
147 current.identifier().to_string()
148}
149
150pub fn member_name(index: &dyn CodeUnitIndex, target: &CodeUnit) -> Option<String> {
151 target_owner_code_unit(index, target).map(|_| target.identifier().to_string())
152}
153
154pub fn target_owner_code_unit(index: &dyn CodeUnitIndex, target: &CodeUnit) -> Option<CodeUnit> {
155 index
156 .parent_of(target)
157 .filter(|parent| parent.source() == target.source() && parent.is_class())
158}
159
160pub fn resolve_receiver_type(
161 graph: &PythonGraphSource<'_>,
162 python: &dyn PythonUsageSource,
163 file: &ProjectFile,
164 raw_type: &str,
165 target_self_file: bool,
166) -> Option<CodeUnit> {
167 let raw_type = raw_type.trim();
168 if raw_type.is_empty() || raw_type.contains('.') || raw_type.contains('|') {
169 return None;
170 }
171
172 if let Some(binding) = python.import_binder_of(file).bindings.get(raw_type)
173 && binding.kind == ImportKind::Named
174 && let Some(imported) = binding.imported_name.as_ref()
175 {
176 let fqn = format!("{}.{}", binding.module_specifier, imported);
177 if let Some(class) =
178 resolve_fqn_candidates(python, &fqn, |name| graph.index.definitions(name).collect())
179 .into_iter()
180 .find(CodeUnit::is_class)
181 {
182 return Some(class);
183 }
184 }
185
186 if let Some(provider) = graph.imports
187 && let Some(imported) = provider
188 .imported_code_units_of(file)
189 .iter()
190 .find(|code_unit| code_unit.identifier() == raw_type && code_unit.is_class())
191 {
192 return Some(imported.clone());
193 }
194
195 graph
196 .index
197 .declarations(file)
198 .into_iter()
199 .find(|code_unit| code_unit.identifier() == raw_type && code_unit.is_class())
200 .or_else(|| {
201 if !target_self_file {
202 return None;
203 }
204 resolve_indexed_receiver_type(graph, file, raw_type)
205 })
206}
207
208fn resolve_bare_annotation_symbol(
209 graph: &PythonGraphSource<'_>,
210 python: &dyn PythonUsageSource,
211 file: &ProjectFile,
212 source: &str,
213 node: Node<'_>,
214 raw_symbol: &str,
215) -> Vec<CodeUnit> {
216 let raw_symbol = raw_symbol.trim();
217 if raw_symbol.is_empty() {
218 return Vec::new();
219 }
220
221 if let Some(owner) = annotation_scope_owner_class(graph, file, source, node) {
222 let owner_candidates: Vec<_> = exact_owner_annotation_members(graph, &owner, raw_symbol)
223 .into_iter()
224 .filter(|candidate| !candidate.is_function())
225 .collect();
226 if !owner_candidates.is_empty() {
227 return owner_candidates;
228 }
229 }
230
231 let mut candidates = Vec::new();
232 if let Some(binding) = python.import_binder_of(file).bindings.get(raw_symbol)
233 && binding.kind == ImportKind::Named
234 && let Some(imported) = binding.imported_name.as_ref()
235 {
236 let fqn = format!("{}.{}", binding.module_specifier, imported);
237 let mut imported_candidates =
238 resolve_fqn_candidates(python, &fqn, |name| graph.index.definitions(name).collect());
239 imported_candidates.retain(|candidate| {
240 !candidate.is_module() || candidate.fq_name() != binding.module_specifier
241 });
242 candidates.extend(imported_candidates);
243 }
244
245 candidates.extend(
246 graph
247 .index
248 .top_level_declarations(file)
249 .into_iter()
250 .filter(|code_unit| {
251 !code_unit.is_module()
252 && code_unit.identifier() == raw_symbol
253 && graph
254 .index
255 .parent_of(code_unit)
256 .is_some_and(|parent| parent.is_module())
257 }),
258 );
259
260 candidates.retain(|candidate| !candidate.is_function());
261 candidates.sort();
262 candidates.dedup();
263 candidates
264}
265
266pub fn annotation_reference_candidates(
273 graph: &PythonGraphSource<'_>,
274 python: &dyn PythonUsageSource,
275 file: &ProjectFile,
276 source: &str,
277 node: Node<'_>,
278 target_self_file: bool,
279) -> Option<Vec<CodeUnit>> {
280 if !is_annotation_reference_node(node) {
281 return None;
282 }
283
284 let mut candidates = match node.kind() {
285 "identifier" => {
286 let mut candidates = resolve_bare_annotation_symbol(
287 graph,
288 python,
289 file,
290 source,
291 node,
292 node_text(node, source),
293 );
294 if candidates.is_empty() {
295 candidates.extend(resolve_receiver_type(
296 graph,
297 python,
298 file,
299 node_text(node, source),
300 target_self_file,
301 ));
302 }
303 candidates
304 }
305 "string_content" => {
306 let Some(string) = node.parent() else {
307 return Some(Vec::new());
308 };
309 let Some(ranges) = python_deferred_annotation_identifier_ranges(string, source, None)
310 else {
311 return Some(Vec::new());
312 };
313 let mut candidates = Vec::new();
314 for range in ranges {
315 let Some(symbol) = source.get(range.start_byte..range.end_byte) else {
316 continue;
317 };
318 let mut symbol_candidates =
319 resolve_bare_annotation_symbol(graph, python, file, source, node, symbol);
320 if symbol_candidates.is_empty() {
321 symbol_candidates.extend(resolve_receiver_type(
322 graph,
323 python,
324 file,
325 symbol,
326 target_self_file,
327 ));
328 }
329 candidates.extend(symbol_candidates);
330 }
331 candidates
332 }
333 "attribute" => resolve_annotation_attribute_types(graph, python, file, source, node),
334 _ => Vec::new(),
335 };
336 candidates.sort();
337 candidates.dedup();
338 Some(candidates)
339}
340
341#[allow(clippy::too_many_arguments)]
345pub fn annotation_reference_candidates_at_focus(
346 graph: &PythonGraphSource<'_>,
347 python: &dyn PythonUsageSource,
348 file: &ProjectFile,
349 source: &str,
350 node: Node<'_>,
351 focus_start: usize,
352 focus_end: usize,
353 target_self_file: bool,
354) -> Option<Vec<CodeUnit>> {
355 if !is_annotation_reference_node(node) {
356 return None;
357 }
358
359 let deferred_tree = if node.kind() == "string_content" {
360 Some(python_deferred_annotation_tree(
361 node.parent()?,
362 source,
363 None,
364 )?)
365 } else {
366 None
367 };
368 let search_root = deferred_tree.as_ref().map_or(node, |tree| tree.root_node());
369 let focused = search_root.descendant_for_byte_range(focus_start, focus_end)?;
370 if focused.kind() != "identifier"
371 || focused.start_byte() != focus_start
372 || focused.end_byte() != focus_end
373 {
374 return Some(Vec::new());
375 }
376
377 let mut path = focused;
378 while let Some(parent) = path.parent() {
379 if parent.kind() != "attribute" {
380 break;
381 }
382 path = parent;
383 }
384 if path.kind() == "attribute" {
385 return Some(focused_annotation_attribute_candidates(
386 graph, python, file, source, node, path, focused,
387 ));
388 }
389
390 let symbol = node_text(focused, source);
391 let mut candidates = resolve_bare_annotation_symbol(graph, python, file, source, node, symbol);
392 if candidates.is_empty() {
393 candidates.extend(resolve_receiver_type(
394 graph,
395 python,
396 file,
397 symbol,
398 target_self_file,
399 ));
400 }
401 candidates.sort();
402 candidates.dedup();
403 Some(candidates)
404}
405
406fn focused_annotation_attribute_candidates(
407 graph: &PythonGraphSource<'_>,
408 python: &dyn PythonUsageSource,
409 file: &ProjectFile,
410 source: &str,
411 scope_node: Node<'_>,
412 path: Node<'_>,
413 focused: Node<'_>,
414) -> Vec<CodeUnit> {
415 let Some((root, attributes)) = annotation_attribute_chain(path) else {
416 return Vec::new();
417 };
418 if root.id() == focused.id() {
419 return resolve_bare_annotation_symbol(
420 graph,
421 python,
422 file,
423 source,
424 scope_node,
425 node_text(root, source),
426 );
427 }
428
429 if attributes
430 .last()
431 .is_some_and(|node| node.id() == focused.id())
432 {
433 let candidates = namespace_qualified_declarations(graph, python, file, source, path);
434 if !candidates.is_empty() {
435 return candidates;
436 }
437 }
438
439 let owners: Vec<_> = resolve_bare_annotation_symbol(
440 graph,
441 python,
442 file,
443 source,
444 scope_node,
445 node_text(root, source),
446 )
447 .into_iter()
448 .filter(CodeUnit::is_class)
449 .collect();
450 let [owner] = owners.as_slice() else {
451 return Vec::new();
452 };
453 let mut owner = owner.clone();
454 for attribute in attributes {
455 let candidates = exact_nested_annotation_class(graph, &owner, node_text(attribute, source));
456 if attribute.id() == focused.id() {
457 return candidates;
458 }
459 let [next] = candidates.as_slice() else {
460 return Vec::new();
461 };
462 owner = next.clone();
463 }
464 Vec::new()
465}
466
467pub fn annotation_class_qualifier_site<'tree>(
470 graph: &PythonGraphSource<'_>,
471 python: &dyn PythonUsageSource,
472 file: &ProjectFile,
473 source: &str,
474 node: Node<'tree>,
475 target: &CodeUnit,
476) -> Option<Node<'tree>> {
477 if node.kind() != "attribute" || !target.is_class() || !is_annotation_reference_node(node) {
478 return None;
479 }
480
481 let (root, attributes) = annotation_attribute_chain(node)?;
482 let owners: Vec<_> =
483 resolve_bare_annotation_symbol(graph, python, file, source, root, node_text(root, source))
484 .into_iter()
485 .filter(CodeUnit::is_class)
486 .collect();
487 let [owner] = owners.as_slice() else {
488 return None;
489 };
490 let mut owner = owner.clone();
491 if &owner == target {
492 return Some(root);
493 }
494
495 let qualifier_count = attributes.len().saturating_sub(1);
498 for attribute in attributes.into_iter().take(qualifier_count) {
499 let next_candidates =
500 exact_nested_annotation_class(graph, &owner, node_text(attribute, source));
501 let [next] = next_candidates.as_slice() else {
502 return None;
503 };
504 owner = next.clone();
505 if &owner == target {
506 return Some(attribute);
507 }
508 }
509
510 None
511}
512
513fn resolve_annotation_attribute_types(
514 graph: &PythonGraphSource<'_>,
515 python: &dyn PythonUsageSource,
516 file: &ProjectFile,
517 source: &str,
518 node: Node<'_>,
519) -> Vec<CodeUnit> {
520 let mut candidates = namespace_qualified_declarations(graph, python, file, source, node);
525 let Some((root, attributes)) = annotation_attribute_chain(node) else {
526 return candidates;
527 };
528 let root_text = node_text(root, source);
529 let owners: Vec<_> =
530 resolve_bare_annotation_symbol(graph, python, file, source, root, root_text)
531 .into_iter()
532 .filter(CodeUnit::is_class)
533 .collect();
534 let [owner] = owners.as_slice() else {
535 return candidates;
536 };
537 let mut owner = owner.clone();
538
539 for attribute in attributes {
540 let segment = node_text(attribute, source);
541 let next_candidates = exact_nested_annotation_class(graph, &owner, segment);
542 let [next] = next_candidates.as_slice() else {
543 return candidates;
544 };
545 owner = next.clone();
546 }
547
548 candidates.push(owner);
549 candidates
550}
551
552fn annotation_attribute_chain(node: Node<'_>) -> Option<(Node<'_>, Vec<Node<'_>>)> {
553 let mut attributes = Vec::new();
554 let mut current = node;
555 while current.kind() == "attribute" {
556 attributes.push(current.child_by_field_name("attribute")?);
557 current = current.child_by_field_name("object")?;
558 }
559 if current.kind() != "identifier" || attributes.is_empty() {
560 return None;
561 }
562 attributes.reverse();
563 Some((current, attributes))
564}
565
566fn exact_nested_annotation_class(
567 graph: &PythonGraphSource<'_>,
568 owner: &CodeUnit,
569 segment: &str,
570) -> Vec<CodeUnit> {
571 let mut candidates: Vec<_> = exact_owner_annotation_members(graph, owner, segment)
572 .into_iter()
573 .filter(CodeUnit::is_class)
574 .collect();
575 candidates.sort();
576 candidates.dedup();
577 candidates
578}
579
580fn exact_owner_annotation_members(
581 graph: &PythonGraphSource<'_>,
582 owner: &CodeUnit,
583 segment: &str,
584) -> Vec<CodeUnit> {
585 let mut candidates: Vec<_> = graph
586 .index
587 .declarations(owner.source())
588 .into_iter()
589 .filter(|unit| {
590 unit.identifier() == segment
591 && graph
592 .index
593 .parent_of(unit)
594 .is_some_and(|parent| parent.fq_name() == owner.fq_name())
595 })
596 .collect();
597 candidates.sort();
598 candidates.dedup();
599 candidates
600}
601
602fn annotation_scope_owner_class(
603 graph: &PythonGraphSource<'_>,
604 file: &ProjectFile,
605 source: &str,
606 node: Node<'_>,
607) -> Option<CodeUnit> {
608 if !annotation_expression_is_class_scoped(node) {
609 return None;
610 }
611 let range = Range {
612 start_byte: node.start_byte(),
613 end_byte: node.end_byte(),
614 start_line: 0,
615 end_line: 0,
616 };
617 if let Some(enclosing) = graph.index.enclosing_code_unit(file, &range) {
618 if enclosing.is_class() {
619 return Some(enclosing);
620 }
621 if let Some(owner) = target_owner_code_unit(graph.index, &enclosing) {
622 return Some(owner);
623 }
624 }
625 structural_annotation_owner_class(graph, file, source, node)
626}
627
628fn annotation_expression_is_class_scoped(node: Node<'_>) -> bool {
629 let site_start = node.start_byte();
630 let site_end = node.end_byte();
631 let mut current = node;
632 while let Some(parent) = current.parent() {
633 if matches!(parent.kind(), "function_definition" | "lambda")
634 && parent
635 .child_by_field_name("body")
636 .is_some_and(|body| body.start_byte() <= site_start && site_end <= body.end_byte())
637 {
638 return false;
639 }
640 if parent.kind() == "class_definition" {
641 return true;
642 }
643 current = parent;
644 }
645 false
646}
647
648fn structural_annotation_owner_class(
649 graph: &PythonGraphSource<'_>,
650 file: &ProjectFile,
651 source: &str,
652 node: Node<'_>,
653) -> Option<CodeUnit> {
654 let mut current = node;
655 while let Some(parent) = current.parent() {
656 if parent.kind() == "class_definition" {
657 let name = node_text(parent.child_by_field_name("name")?, source).trim();
658 if name.is_empty() {
659 return None;
660 }
661 let class_range = Range {
662 start_byte: parent.start_byte(),
663 end_byte: parent.end_byte(),
664 start_line: 0,
665 end_line: 0,
666 };
667 let mut matches: Vec<_> = graph
668 .index
669 .declarations(file)
670 .into_iter()
671 .filter(|unit| unit.is_class() && unit.identifier() == name)
672 .filter(|unit| {
673 graph
674 .index
675 .ranges(unit)
676 .into_iter()
677 .any(|range| range.contains(&class_range))
678 })
679 .collect();
680 matches.sort();
681 matches.dedup();
682 let [owner] = matches.as_slice() else {
683 return None;
684 };
685 return Some(owner.clone());
686 }
687 current = parent;
688 }
689 None
690}
691
692fn is_annotation_reference_node(node: Node<'_>) -> bool {
693 if !matches!(node.kind(), "identifier" | "attribute" | "string_content") {
694 return false;
695 }
696 python_node_is_in_annotation(node)
697}
698
699pub fn resolve_constructor_types(
704 graph: &PythonGraphSource<'_>,
705 python: &dyn PythonUsageSource,
706 file: &ProjectFile,
707 source: &str,
708 function: Node<'_>,
709) -> Vec<CodeUnit> {
710 let candidates = match function.kind() {
711 "identifier" => {
712 let local = node_text(function, source);
713 if local.is_empty() {
714 return Vec::new();
715 }
716 let binder = python.import_binder_of(file);
717 let fqn = match binder.bindings.get(local) {
718 Some(binding) if binding.kind == ImportKind::Named => binding
719 .imported_name
720 .as_ref()
721 .map(|imported| format!("{}.{}", binding.module_specifier, imported)),
722 _ => graph
723 .index
724 .declarations(file)
725 .into_iter()
726 .find(|unit| unit.is_class() && unit.identifier() == local)
727 .map(|unit| unit.fq_name()),
728 };
729 let Some(fqn) = fqn else {
730 return Vec::new();
731 };
732 resolve_fqn_candidates(python, &fqn, |name| graph.index.definitions(name).collect())
733 }
734 "attribute" => namespace_qualified_declarations(graph, python, file, source, function),
735 _ => Vec::new(),
736 };
737 let mut classes: Vec<CodeUnit> = candidates.into_iter().filter(CodeUnit::is_class).collect();
740 classes.sort();
741 classes.dedup();
742 classes
743}
744
745pub fn resolve_callable_parameter_default_types(
751 graph: &PythonGraphSource<'_>,
752 python: &dyn PythonUsageSource,
753 file: &ProjectFile,
754 source: &str,
755 reference: Node<'_>,
756 local_name: &str,
757) -> Vec<CodeUnit> {
758 let site_start = reference.start_byte();
759 let site_end = reference.end_byte();
760 let mut current = reference;
761 while let Some(parent) = current.parent() {
762 current = parent;
763 if !matches!(current.kind(), "function_definition" | "lambda") {
764 continue;
765 }
766 if current
767 .child_by_field_name("body")
768 .is_none_or(|body| !(body.start_byte() <= site_start && site_end <= body.end_byte()))
769 {
770 continue;
771 }
772 let Some(parameters) = current.child_by_field_name("parameters") else {
773 return Vec::new();
774 };
775 let mut cursor = parameters.walk();
776 for parameter in parameters.named_children(&mut cursor) {
777 let name = if parameter.kind() == "identifier" {
778 Some(parameter)
779 } else {
780 parameter.child_by_field_name("name")
781 };
782 if name.is_none_or(|name| node_text(name, source) != local_name) {
783 continue;
784 }
785 let Some(value) = parameter.child_by_field_name("value") else {
786 return Vec::new();
787 };
788 return resolve_constructor_types(graph, python, file, source, value);
789 }
790 return Vec::new();
791 }
792 Vec::new()
793}
794
795fn namespace_qualified_declarations(
802 graph: &PythonGraphSource<'_>,
803 python: &dyn PythonUsageSource,
804 file: &ProjectFile,
805 source: &str,
806 node: Node<'_>,
807) -> Vec<CodeUnit> {
808 let binder = python.import_binder_of(file);
809 let Some(fqn) = namespace_constructor_fqn(&binder, source, node) else {
810 return Vec::new();
811 };
812 let mut candidates =
813 resolve_fqn_candidates(python, &fqn, |name| graph.index.definitions(name).collect());
814 candidates.sort();
815 candidates.dedup();
816 candidates
817}
818
819fn namespace_constructor_fqn(
820 binder: &ImportBinder,
821 source: &str,
822 function: Node<'_>,
823) -> Option<String> {
824 let mut attributes = Vec::new();
825 let mut current = function;
826 while current.kind() == "attribute" {
827 let attribute = current.child_by_field_name("attribute")?;
828 let text = node_text(attribute, source);
829 if text.is_empty() {
830 return None;
831 }
832 attributes.push(text);
833 current = current.child_by_field_name("object")?;
834 }
835 if current.kind() != "identifier" {
836 return None;
837 }
838 let root = node_text(current, source);
839 let binding = binder.bindings.get(root)?;
840 if binding.kind != ImportKind::Namespace {
841 return None;
842 }
843 let mut fqn = binding.module_specifier.clone();
844 for attribute in attributes.into_iter().rev() {
845 fqn.push('.');
846 fqn.push_str(attribute);
847 }
848 Some(fqn)
849}
850
851fn node_text<'a>(node: Node<'_>, source: &'a str) -> &'a str {
852 brokk_bifrost_core::analyzer::common::node_source_text(node, source)
853}
854
855fn resolve_indexed_receiver_type(
856 graph: &PythonGraphSource<'_>,
857 file: &ProjectFile,
858 raw_type: &str,
859) -> Option<CodeUnit> {
860 let package_matches = module_fq_for_file(graph.index, file).map_or_else(Vec::new, |module| {
861 relational_definitions(
862 graph,
863 RelationalName::stable(module),
864 RelationalDefinitionQuery::PackageTypes {
865 simple_name: raw_type.to_string(),
866 },
867 )
868 });
869 let mut root_name = FqName::new();
870 root_name.push(segment_interner().intern(raw_type, SegmentKind::Type));
871 let exact_matches = relational_definitions(
872 graph,
873 RelationalName::stable(root_name.clone()),
874 RelationalDefinitionQuery::ExactName,
875 );
876 let normalized_matches = relational_definitions(
877 graph,
878 RelationalName::stable(root_name),
879 RelationalDefinitionQuery::NormalizedName,
880 );
881 package_matches
882 .into_iter()
883 .chain(exact_matches)
884 .chain(normalized_matches)
885 .find(|code_unit| code_unit.identifier() == raw_type && code_unit.is_class())
886}
887
888fn relational_definitions(
889 graph: &PythonGraphSource<'_>,
890 name: RelationalName,
891 query: RelationalDefinitionQuery,
892) -> Vec<CodeUnit> {
893 let question = RelationalDefinitionQuestion {
894 language_scope: DefinitionLanguageScope::Workspace,
895 name,
896 query,
897 };
898 match graph.definitions.ask(&question) {
899 RelationalDefinitionValue::Definitions(units) => units,
900 _ => panic!("definition question returned the wrong result shape"),
901 }
902}
903
904fn module_fq_for_file(index: &dyn CodeUnitIndex, file: &ProjectFile) -> Option<FqName> {
905 index
906 .declarations(file)
907 .into_iter()
908 .find(|code_unit| code_unit.is_module())
909 .map(|code_unit| code_unit.fq().clone())
910 .or_else(|| {
911 index
912 .declarations(file)
913 .into_iter()
914 .find(|code_unit| !code_unit.package_name().is_empty())
915 .map(|code_unit| code_unit.package_fq())
916 })
917}
918
919pub fn normalized_receiver_type(annotation: &str) -> Option<String> {
920 let annotation = unwrap_python_string_annotation(annotation.trim());
921 let annotation = unwrap_supported_receiver_wrapper(annotation);
922 if annotation.is_empty()
923 || annotation.contains('|')
924 || annotation.contains('[')
925 || annotation.contains(']')
926 || annotation.contains(',')
927 || annotation.contains('(')
928 || annotation.contains(')')
929 || annotation.contains('{')
930 || annotation.contains('}')
931 || annotation.contains(':')
932 {
933 return None;
934 }
935 Some(annotation.to_string())
936}
937
938fn unwrap_python_string_annotation(annotation: &str) -> &str {
939 if annotation.len() >= 2 {
940 let bytes = annotation.as_bytes();
941 let first = bytes[0];
942 let last = bytes[annotation.len() - 1];
943 if (first == b'\'' || first == b'"') && first == last {
944 return annotation[1..annotation.len() - 1].trim();
945 }
946 }
947 annotation
948}
949
950fn unwrap_supported_receiver_wrapper(annotation: &str) -> &str {
951 let mut current = annotation.trim();
952 loop {
953 let next = current
954 .strip_prefix("Optional[")
955 .or_else(|| current.strip_prefix("typing.Optional["))
956 .and_then(|inner| inner.strip_suffix(']'))
957 .map(str::trim);
958 let Some(unwrapped) = next else {
959 return current;
960 };
961 current = unwrapped;
962 }
963}
964
965pub fn receiver_annotation_matches_target(
966 annotation: &str,
967 edges: &[ImportEdge],
968 target_short: &str,
969 target_self_file: bool,
970) -> bool {
971 let annotation = annotation.trim();
972 if annotation.is_empty() {
973 return false;
974 }
975 if annotation.contains('|')
976 || annotation.contains('[')
977 || annotation.contains(']')
978 || annotation.contains(',')
979 || annotation.contains('(')
980 || annotation.contains(')')
981 {
982 return false;
983 }
984 if annotation == target_short {
985 return target_self_file || edges.iter().any(|edge| edge.local_name == target_short);
986 }
987
988 let segments = parse_symbol_path(Language::Python, annotation);
994 let Some((member, qualifier_parts)) = segments.split_last() else {
995 return false;
996 };
997 if qualifier_parts.is_empty() {
998 return false;
999 }
1000 let qualifier = qualifier_parts.join(".");
1001 let member = member.as_str();
1002 if member != target_short {
1003 return false;
1004 }
1005 edges.iter().any(|edge| {
1006 matches!(edge.kind, ImportEdgeKind::Namespace)
1007 && (edge.local_name == qualifier
1008 || qualifier.ends_with(&format!(".{}", edge.local_name)))
1009 })
1010}
1011
1012