backend/core/agent/utils.py

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", "")