import json import re import os from typing import Any, Dict, List from pathlib import Path from dotenv import load_dotenv ROOT_ENV_PATH = Path(__file__).resolve().parents[3] / ".env" load_dotenv(dotenv_path=ROOT_ENV_PATH) def _env(name: str, default: str = "") -> str: value = os.getenv(name, default) if (value is None or not str(value).strip()) and ROOT_ENV_PATH.exists(): pattern = re.compile(rf"^\s*{re.escape(name)}\s*=\s*(.*)\s*$") for line in ROOT_ENV_PATH.read_text(encoding="utf-8-sig").splitlines(): match = pattern.match(line) if match: value = match.group(1).strip().strip('"').strip("'") break return str(value or "") def _extract_json(text: str) -> Dict[str, Any]: text = text.strip() try: return json.loads(text) except Exception: match = re.search(r"\{.*\}", text, re.DOTALL) if match: try: return json.loads(match.group(0)) except: pass return {} def _active_query(state: Dict[str, Any]) -> str: return state.get("rewritten_query") or state.get("query", "")