kamidso-back/kamidso/native_connectors/semantic/catalog.py

"""Load Didar semantic contracts shipped beside the native provider."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, List, Optional

from kamidso.native_connectors.errors import CatalogError
from kamidso.native_connectors.flags import (
    didar_connector_enabled,
    native_connectors_v1_enabled,
)
from kamidso.native_connectors.semantic.text import normalize_text

_CACHE: Optional[Dict[str, Any]] = None

_BASIS_DEFAULTS = {
    "sales_by_owner": "won_at",
    "sales_over_time": "won_at",
}

_DISCLOSURE = {
    "sales_summary": "مبنای بازه: تاریخ موفق‌شدن",
    "won_deals": "مبنای بازه: تاریخ موفق‌شدن",
    "sales_by_owner": "مبنای بازه: تاریخ موفق‌شدن",
    "sales_over_time": "مبنای بازه: تاریخ موفق‌شدن",
    "lost_deals": "مبنای بازه: تاریخ ناموفق‌شدن",
    "deals_created": "مبنای بازه: تاریخ ثبت",
    "pending_pipeline_snapshot": "مبنای گزارش: وضعیت فعلی",
}


def provider_dir() -> Path:
    return Path(__file__).resolve().parents[1] / "providers" / "didar"


def reset_semantic_cache() -> None:
    global _CACHE
    _CACHE = None


def _read(path: Path) -> Dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise CatalogError(f"Cannot read {path.name}: {exc}") from exc
    if not isinstance(data, dict):
        raise CatalogError(f"{path.name} must be an object.")
    return data


def load_bundle() -> Dict[str, Any]:
    global _CACHE
    if _CACHE is not None:
        return _CACHE
    directory = provider_dir()
    loaded = {
        "reports": _read(directory / "reports.json"),
        "dates": _read(directory / "date_rules.json"),
        "money": _read(directory / "money_rules.json"),
        "semantics": _read(directory / "semantics.json"),
        "enums": _read(directory / "enums.json"),
    }
    _CACHE = loaded
    return loaded


def operations() -> List[Dict[str, Any]]:
    raw = load_bundle()["reports"].get("semantic_operations") or []
    if not isinstance(raw, list):
        return []
    return [item for item in raw if isinstance(item, dict) and item.get("name")]


def _didar_ready(connector_type: str) -> bool:
    return (
        connector_type == "didar"
        and native_connectors_v1_enabled()
        and didar_connector_enabled()
    )


def find_operation(connector_type: str, name: str) -> Optional[Dict[str, Any]]:
    if not _didar_ready(connector_type):
        return None
    for operation in operations():
        if operation.get("name") == name:
            return operation
    return None


def enabled_operations(connector_type: str) -> List[Dict[str, Any]]:
    if not _didar_ready(connector_type):
        return []
    return [operation for operation in operations() if operation.get("enabled")]


def operation_enabled(name: str) -> bool:
    for operation in operations():
        if operation.get("name") == name:
            return bool(operation.get("enabled"))
    return False


def date_basis_for(name: str) -> str:
    for operation in operations():
        if operation.get("name") == name and operation.get("date_basis"):
            return str(operation["date_basis"])
    return _BASIS_DEFAULTS.get(name, "")


def disclosure_for(name: str) -> str:
    return _DISCLOSURE.get(name, "")


def verification_cap() -> int:
    try:
        value = load_bundle()["dates"]["didar_search_deals_strategy"]["won_at"]["fallback"][
            "max_records"
        ]
        return int(value)
    except (KeyError, TypeError, ValueError):
        return 5000


def _month_names() -> List[str]:
    months = load_bundle()["dates"].get("jalali_months") or {}
    if not isinstance(months, dict):
        return []
    return [str(name) for name in months]


def _relative_periods() -> List[str]:
    parser = load_bundle()["dates"].get("parser_requirements") or {}
    raw = parser.get("accept_relative_periods") if isinstance(parser, dict) else []
    if not isinstance(raw, list):
        return []
    return [normalize_text(str(item)) for item in raw]


def has_period(text: str) -> bool:
    normalized = normalize_text(text)
    for phrase in _relative_periods():
        if phrase and phrase in normalized:
            return True
    tokens = set(normalized.split())
    return any(month in tokens for month in _month_names())


def route_question(text: str) -> Optional[str]:
    """Map a management question to a semantic operation name."""
    normalized = normalize_text(text)
    if "لید" in normalized or "سرنخ" in normalized:
        return "lead_count"
    if "محصول" in normalized:
        return "sales_by_product"
    if "کارشناس" in normalized or "به تفکیک مالک" in normalized:
        return "sales_by_owner"
    if "روند" in normalized:
        return "sales_over_time"
    if "جاری" in normalized:
        return "pending_pipeline_snapshot"
    if "ناموفق" in normalized:
        return "lost_deals"
    if "موفق" in normalized:
        if has_period(normalized):
            return "won_deals"
        return None
    if "فروش" in normalized:
        return "sales_summary"
    if "معاملات" in normalized or "معامله" in normalized:
        return "deals_created"
    return None