monish563/NU-KIOSK-API

1"""Source adapters that load structured data into the kiosk catalog."""2 3from __future__ import annotations4 5import csv6from abc import ABC, abstractmethod7from dataclasses import dataclass, field8from pathlib import Path9from typing import Any, Callable, Dict, Iterable, List, Optional10 11from .utils import canonicalize_name12 13 14@dataclass15class EntityDefinition:16    """Specification for registering an entity in the data catalog."""17 18    name: str19    records: List[Dict[str, Any]]20    key_field: Optional[str] = None21    origin: Optional[str] = None22    normalizer: Optional[Callable[[str], str]] = None23 24 25@dataclass26class SourceResult:27    """Payload returned by a data source."""28 29    entities: List[EntityDefinition] = field(default_factory=list)30    metadata: Dict[str, Any] = field(default_factory=dict)31 32 33class DataSource(ABC):34    """Interface for ingesting structured data into the catalog."""35 36    def __init__(self, name: str) -> None:37        self.name = name38 39    @abstractmethod40    def load(self) -> SourceResult:41        raise NotImplementedError42 43 44class CSVSource(DataSource):45    """Loads a CSV file into an entity definition."""46 47    def __init__(48        self,49        name: str,50        path: Path,51        entity_name: str,52        *,53        key_field: Optional[str] = None,54        normalizer: Optional[Callable[[str], str]] = None,55    ) -> None:56        super().__init__(name)57        self.path = path58        self.entity_name = entity_name59        self.key_field = key_field60        self.normalizer = normalizer61 62    def load(self) -> SourceResult:63        if not self.path.exists():64            return SourceResult()65        records = self._read_csv(self.path)66        entity = EntityDefinition(67            name=self.entity_name,68            records=records,69            key_field=self.key_field,70            origin=str(self.path),71            normalizer=self.normalizer,72        )73        return SourceResult(entities=[entity])74 75    @staticmethod76    def _read_csv(path: Path) -> List[Dict[str, Any]]:77        with path.open(newline="", encoding="utf-8-sig") as handle:78            reader = csv.DictReader(handle)79            return [dict(row) for row in reader]80 81 82class FeedListSource(DataSource):83    """Loads newline-delimited feed URLs into catalog metadata."""84 85    def __init__(self, name: str, path: Path, metadata_key: str) -> None:86        super().__init__(name)87        self.path = path88        self.metadata_key = metadata_key89 90    def load(self) -> SourceResult:91        if not self.path.exists():92            return SourceResult()93        urls = [94            line.strip()95            for line in self.path.read_text(encoding="utf-8").splitlines()96            if line.strip()97        ]98        return SourceResult(metadata={self.metadata_key: {"urls": urls}})99 100 101def default_sources(base_dir: Path, *, name_normalizer: Optional[Callable[[str], str]] = None) -> List[DataSource]:102    """103    Produce the default set of data sources used by the backend.104 105    Additional sources (e.g., TA office hours) can be appended to this list106    without modifying the rest of the pipeline.107    """108 109    base_dir = base_dir.resolve()110    normalizer = name_normalizer or canonicalize_name111    sources: List[DataSource] = [112        CSVSource(113            name="faculty_roster",114            path=base_dir / "faculty_2.csv",115            entity_name="faculty",116            key_field="Name",117            normalizer=normalizer,118        ),119        CSVSource(120            name="faculty_offices",121            path=base_dir / "Faculty.csv",122            entity_name="faculty_offices",123            key_field="Assignee Name",124            normalizer=normalizer,125        ),126        CSVSource(127            name="staff_roster",128            path=base_dir / "staff.csv",129            entity_name="staff",130            key_field="Name",131            normalizer=normalizer,132        ),133        CSVSource(134            name="students_roster",135            path=base_dir / "students.csv",136            entity_name="students",137            key_field="Name",138            normalizer=normalizer,139        ),140        CSVSource(141            name="office_hours",142            path=base_dir / "CS Office Hours Room Reservations.csv",143            entity_name="office_hours",144            key_field="Course Name",145            normalizer=normalizer,146        ),147        CSVSource(148            name="centers_catalog",149            path=base_dir / "centers.csv",150            entity_name="centers",151            key_field="Name",152            normalizer=normalizer,153        ),154        CSVSource(155            name="mudd_seating",156            path=base_dir / "Mudd Seating Sample.csv",157            entity_name="mudd_seating",158            key_field="Student/Visitor",159            normalizer=normalizer,160        ),161        FeedListSource(162            name="event_feeds",163            path=base_dir / "feed.txt",164            metadata_key="event_feeds",165        ),166    ]167    return sources168