Spaces:
Sleeping
Sleeping
| """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 | |