| from __future__ import annotations |
|
|
| import json |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional |
|
|
|
|
| class QuestionSupportBank: |
| def __init__(self, data_path: Optional[str] = None) -> None: |
| base_dir = Path(__file__).resolve().parent |
| self.data_path = Path(data_path) if data_path else base_dir / "data" / "question_support_bank.jsonl" |
| self._loaded = False |
| self._by_id: Dict[str, Dict[str, Any]] = {} |
| self._by_text: Dict[str, Dict[str, Any]] = {} |
|
|
| def _normalize(self, text: Optional[str]) -> str: |
| return " ".join((text or "").strip().lower().split()) |
|
|
| def load(self) -> None: |
| self._by_id = {} |
| self._by_text = {} |
|
|
| if self.data_path.exists(): |
| with self.data_path.open("r", encoding="utf-8") as handle: |
| for raw_line in handle: |
| line = raw_line.strip() |
| if not line: |
| continue |
| try: |
| item = json.loads(line) |
| except json.JSONDecodeError: |
| continue |
| self._store_item(item) |
|
|
| self._loaded = True |
|
|
| def _ensure_loaded(self) -> None: |
| if not self._loaded: |
| self.load() |
|
|
| def _store_item(self, item: Dict[str, Any]) -> None: |
| if not isinstance(item, dict): |
| return |
| qid = str(item.get("question_id") or "").strip() |
| qtext = self._normalize(item.get("question_text") or item.get("stem") or "") |
| if qid: |
| self._by_id[qid] = item |
| if qtext: |
| self._by_text[qtext] = item |
|
|
| def get(self, question_id: Optional[str] = None, question_text: Optional[str] = None) -> Optional[Dict[str, Any]]: |
| self._ensure_loaded() |
| qid = str(question_id or "").strip() |
| if qid and qid in self._by_id: |
| return dict(self._by_id[qid]) |
|
|
| qtext = self._normalize(question_text) |
| if qtext and qtext in self._by_text: |
| return dict(self._by_text[qtext]) |
| return None |
|
|
| def upsert(self, item: Dict[str, Any]) -> None: |
| self._ensure_loaded() |
| self._store_item(item) |
|
|
| def all_items(self) -> List[Dict[str, Any]]: |
| self._ensure_loaded() |
| return [dict(v) for v in self._by_id.values()] |
|
|
|
|
| question_support_bank = QuestionSupportBank() |
|
|
