"""Blueprint that handles research center queries in both directions.""" from __future__ import annotations import difflib from typing import Any, Dict, List, Optional from .base import AnalysisContext, Blueprint, BlueprintResult, Fact from .faculty_profile import _lookup_record from ..data.utils import canonicalize_name class CenterBlueprint(Blueprint): """Return center information - works for both faculty and center queries. Automatically detects the query type: - If faculty name: returns centers they lead - If center name: returns faculty who lead it """ name = "center" def run(self, context: AnalysisContext, **kwargs: Any) -> BlueprintResult: # Check what kind of input we have faculty_name = (kwargs.get("faculty") or kwargs.get("name") or "").strip() center_name = (kwargs.get("center") or "").strip() # If explicitly given a center, look up its leadership if center_name: return self._lookup_center_leadership(context, center_name, kwargs) # Otherwise, try to detect if input is faculty or center if faculty_name: # First try as faculty faculty_result = self._lookup_faculty_centers(context, faculty_name, kwargs) if faculty_result.facts or "not found" not in " ".join(faculty_result.notes).lower(): return faculty_result # If no faculty found, try as center name center_result = self._lookup_center_leadership(context, faculty_name, kwargs) if center_result.facts: return center_result # Neither worked - return faculty not found message return faculty_result return BlueprintResult( self.name, kwargs, facts=[], notes=["Please provide a faculty name or center name."] ) def _lookup_faculty_centers( self, context: AnalysisContext, target_name: str, kwargs: Dict[str, Any] ) -> BlueprintResult: """Look up what centers a faculty member leads.""" record, lookup_notes, office_row = _lookup_record(context, target_name) if record is None: if office_row is None: return BlueprintResult( self.name, kwargs, facts=[], notes=[f"Faculty '{target_name}' not found in roster."], ) return BlueprintResult( self.name, kwargs, facts=[], notes=[f"No roster entry for '{target_name}'; cannot determine center leadership."], ) centers_entity = context.catalog.try_get("centers") centers_origin = centers_entity.origin if centers_entity else None matches = context.catalog.resolve_relationship("faculty_to_centers", record) facts: List[Fact] = [] for center in matches: facts.append( Fact( subject=record["Name"], predicate="leads_center", value=center.get("Name"), source=centers_origin, confidence=0.9, ) ) notes = list(lookup_notes) if not facts: notes.append(f"{record['Name']} is not currently listed as a center leader.") return BlueprintResult(self.name, kwargs, facts=facts, notes=notes) def _lookup_center_leadership( self, context: AnalysisContext, target_name: str, kwargs: Dict[str, Any] ) -> BlueprintResult: """Look up who leads a research center.""" record, lookup_notes = self._resolve_center(context, target_name) if record is None: return BlueprintResult( self.name, kwargs, facts=[], notes=[f"Center '{target_name}' not found in catalog."], ) centers = context.catalog.try_get("centers") centers_origin = centers.origin if centers else None matches = context.catalog.resolve_relationship("center_to_faculty_leads", record) facts: List[Fact] = [] for faculty in matches: facts.append( Fact( subject=record.get("Name", target_name), predicate="led_by", value=faculty.get("Name"), source=centers_origin, confidence=0.9, ) ) notes: List[str] = list(lookup_notes) if not facts: # Include leadership text if available leadership = record.get("Leadership") if leadership: notes.append(f"Leadership listed as: {leadership}") else: notes.append(f"No leadership assignments found for {record.get('Name', target_name)}.") return BlueprintResult(self.name, kwargs, facts=facts, notes=notes) def _resolve_center( self, context: AnalysisContext, target: str ) -> tuple[Optional[Dict[str, Any]], List[str]]: """Find a center by name with fuzzy matching.""" centers = context.catalog.try_get("centers") if not centers: return None, ["Centers dataset not available."] canonical_target = canonicalize_name(target) notes: List[str] = [] def _matches(name: str) -> bool: canonical_name = canonicalize_name(name) if canonical_name == canonical_target: return True if canonical_target and canonical_target in canonical_name: return True if canonical_name and canonical_name in canonical_target: return True return False # Exact or substring match for row in centers.records: if _matches(row.get("Name", "")): return row, notes # Fuzzy match names = [row.get("Name", "") for row in centers.records if row.get("Name")] closest = difflib.get_close_matches(target, names, n=1, cutoff=0.6) if closest: row = next((r for r in centers.records if r.get("Name") == closest[0]), None) if row: notes.append(f"Showing results for '{closest[0]}' (closest match).") return row, notes return None, notes