Mutual_Fund/nl2sql/semantics.py

"""基金业务语义目录：把自然语言术语映射为受权威 Schema 约束的数据库概念。"""
from __future__ import annotations

import json
import hashlib
from functools import lru_cache
from pathlib import Path
from typing import Any

DEFAULT_CATALOG_PATH = Path(__file__).with_name("semantic_catalog.json")


@lru_cache(maxsize=8)
def _load_catalog_cached(path: str) -> dict[str, Any]:
    payload = json.loads(Path(path).read_text(encoding="utf-8"))
    return validate_semantic_catalog(payload)


def load_semantic_catalog(path: str | Path | None = None) -> dict[str, Any]:
    """加载可替换的基金业务语义目录。"""
    return _load_catalog_cached(str(Path(path or DEFAULT_CATALOG_PATH).resolve()))


def validate_semantic_catalog(catalog: dict[str, Any]) -> dict[str, Any]:
    """校验语义目录结构，保证目录错误在加载阶段暴露。"""
    if not isinstance(catalog, dict) or not str(catalog.get("version", "")).strip():
        raise ValueError("语义目录必须包含 version")
    terms = catalog.get("terms")
    if not isinstance(terms, list):
        raise ValueError("语义目录必须包含 terms 数组")
    for item in terms:
        if not isinstance(item, dict) or not str(item.get("term", "")).strip():
            raise ValueError("语义目录 term 无效")
        if "enabled" in item and not isinstance(item["enabled"], bool):
            raise ValueError("语义目录 enabled 必须是布尔值")
        for key in ("aliases", "tables", "fields"):
            if not isinstance(item.get(key), list) or not item[key]:
                raise ValueError(f"语义目录 {key} 无效")
    relationships = catalog.get("relationships", [])
    if not isinstance(relationships, list):
        raise ValueError("语义目录 relationships 必须是数组")
    for item in relationships:
        if not isinstance(item, dict) or not all(
            str(item.get(key, "")).strip() for key in ("left", "right", "meaning")
        ):
            raise ValueError("语义目录关联关系无效")
    return catalog


def clear_semantic_catalog_cache() -> None:
    """清理目录缓存，使下一次请求重新读取文件。"""
    _load_catalog_cached.cache_clear()


def refresh_semantic_catalog(path: str | Path | None = None) -> dict[str, Any]:
    """校验并刷新语义目录；新目录无效时保留当前缓存。"""
    catalog_path = Path(path or DEFAULT_CATALOG_PATH).resolve()
    payload = json.loads(catalog_path.read_text(encoding="utf-8"))
    catalog = validate_semantic_catalog(payload)
    previous = get_semantic_catalog_info(catalog_path)
    _load_catalog_cached.cache_clear()
    _load_catalog_cached(str(catalog_path))
    current = _build_semantic_catalog_info(catalog, catalog_path)
    return {
        **current,
        "previous_version": previous["version"],
        "changed": previous["digest"] != current["digest"],
    }


def _build_semantic_catalog_info(catalog: dict[str, Any], path: Path) -> dict[str, Any]:
    """生成不含业务明细的目录摘要。"""
    digest = hashlib.sha256(
        json.dumps(catalog, ensure_ascii=False, sort_keys=True).encode("utf-8")
    ).hexdigest()
    enabled_count = sum(item.get("enabled", True) for item in catalog["terms"])
    return {
        "version": catalog["version"],
        "term_count": len(catalog["terms"]),
        "enabled_term_count": enabled_count,
        "disabled_term_count": len(catalog["terms"]) - enabled_count,
        "relationship_count": len(catalog.get("relationships", [])),
        "digest": digest,
        "path": str(path),
    }


def get_semantic_catalog_info(path: str | Path | None = None) -> dict[str, Any]:
    """返回不含业务明细的语义目录摘要。"""
    catalog = load_semantic_catalog(path)
    return _build_semantic_catalog_info(catalog, Path(path or DEFAULT_CATALOG_PATH).resolve())


def resolve_semantics(question: str, *, catalog: dict[str, Any] | None = None) -> dict[str, Any]:
    """根据问题匹配业务术语，返回未经过 Schema 过滤的候选语义。"""
    text = (question or "").strip()
    catalog = catalog or load_semantic_catalog()
    matched = [
        item
        for item in catalog["terms"]
        if item.get("enabled", True)
        if any(alias in text for alias in item.get("aliases", []))
    ]
    tables = sorted({table for item in matched for table in item["tables"]})
    fields = sorted({field for item in matched for field in item["fields"]})
    metrics = [
        {
            "term": item["term"],
            "field": item["fields"][0],
            "hint": item["metric_hint"],
        }
        for item in matched
        if item.get("metric_hint")
    ]
    value_hints = [
        {
            "term": item["term"],
            "field": item["fields"][0],
            "hint": item["value_hint"],
        }
        for item in matched
        if item.get("value_hint")
    ]
    return {
        "terms": [item["term"] for item in matched],
        "tables": tables,
        "fields": fields,
        "metrics": metrics,
        "value_hints": value_hints,
        "relationships": list(catalog.get("relationships", [])),
    }


def build_semantic_context(question: str, schema: dict[str, Any]) -> dict[str, Any]:
    """只保留当前权威 Schema 中存在的表、字段和关联，避免语义目录越权扩张。"""
    resolved = resolve_semantics(question)
    schema_tables = {
        str(item.get("table_name"))
        for item in schema.get("tables", [])
        if item.get("table_name")
    }
    schema_fields = {
        (str(item.get("table_name")), str(item.get("field_name")))
        for item in schema.get("columns", [])
        if item.get("table_name") and item.get("field_name")
    }
    tables = sorted(set(resolved["tables"]) & schema_tables)
    fields = sorted(
        {
            field
            for table, field in schema_fields
            if table in tables and field in set(resolved["fields"])
        }
    )
    metrics = [
        metric
        for metric in resolved["metrics"]
        if any(metric["field"] == field for _, field in schema_fields if _ in tables)
    ]
    value_hints = [
        hint
        for hint in resolved["value_hints"]
        if any(hint["field"] == field for _, field in schema_fields if _ in tables)
    ]
    relationships = [
        relation
        for relation in resolved["relationships"]
        if relation["left"].split(".", 1)[0] in tables
        and relation["right"].split(".", 1)[0] in tables
    ]
    return {
        "terms": resolved["terms"],
        "tables": tables,
        "fields": fields,
        "metrics": metrics,
        "value_hints": value_hints,
        "relationships": relationships,
    }