config/config_loader.py

"""
Configuration loader and validator for search engine configurations.
This module handles loading, parsing, and validating YAML configuration files
that define how search should be executed (NOT how data should be indexed).
索引结构由 mappings/search_products.json 定义。
此配置只定义搜索行为：字段权重、搜索域、查询策略等。
"""
import yaml
import os
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from pathlib import Path
@dataclass
class IndexConfig:
    """Configuration for an index domain (e.g., default, title, brand)."""
    name: str
    label: str
    fields: List[str]  # List of field names to include in this search domain
    boost: float = 1.0
    example: Optional[str] = None
@dataclass
class QueryConfig:
    """Configuration for query processing."""
    supported_languages: List[str] = field(default_factory=lambda: ["zh", "en"])
    default_language: str = "zh"
    
    # Feature flags
    enable_translation: bool = True
    enable_text_embedding: bool = True
    enable_query_rewrite: bool = True
    
    # Query rewrite dictionary (loaded from external file)
    rewrite_dictionary: Dict[str, str] = field(default_factory=dict)
    
    # Translation settings
    translation_service: str = "deepl"
    translation_api_key: Optional[str] = None
    translation_glossary_id: Optional[str] = None
    translation_context: str = "e-commerce product search"
    
    # Embedding field names
    text_embedding_field: Optional[str] = "title_embedding"
    image_embedding_field: Optional[str] = None
    
    # Embedding disable thresholds (disable vector search for short queries)
    embedding_disable_chinese_char_limit: int = 4
    embedding_disable_english_word_limit: int = 3
    
    # Source fields configuration
    source_fields: Optional[List[str]] = None
@dataclass
class SPUConfig:
    """Configuration for SPU aggregation."""
    enabled: bool = False
    spu_field: Optional[str] = None
    inner_hits_size: int = 3
    # 配置哪些option维度参与检索（进索引、以及在线搜索）
    searchable_option_dimensions: List[str] = field(default_factory=lambda: ['option1', 'option2', 'option3'])
@dataclass
class FunctionScoreConfig:
    """Function Score配置（ES层打分规则）"""
    score_mode: str = "sum"
    boost_mode: str = "multiply"
    functions: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class RankingConfig:
    """Configuration for ranking expressions."""
    expression: str = "bm25()"
    description: str = "Default BM25 ranking"
@dataclass
class RerankConfig:
    """本地重排配置（当前禁用）"""
    enabled: bool = False
    expression: str = ""
    description: str = ""
@dataclass
class SearchConfig:
    """Complete configuration for search engine (multi-tenant)."""
    
    # 字段权重配置（用于搜索）
    field_boosts: Dict[str, float]
    
    # Index structure (query domains)
    indexes: List[IndexConfig]
    
    # Query processing
    query_config: QueryConfig
    
    # Ranking configuration
    ranking: RankingConfig
    
    # Function Score configuration (ES层打分)
    function_score: FunctionScoreConfig
    
    # Rerank configuration (本地重排)
    rerank: RerankConfig
    
    # SPU configuration
    spu_config: SPUConfig
    
    # ES index settings
    es_index_name: str
    es_settings: Dict[str, Any] = field(default_factory=dict)
class ConfigurationError(Exception):
    """Raised when configuration validation fails."""
    pass
class ConfigLoader:
    """Loads and validates unified search engine configuration from YAML file."""
    
    def __init__(self, config_file: Optional[Path] = None):
        """
        Initialize config loader.
        
        Args:
            config_file: Path to config YAML file (defaults to config/config.yaml)
        """
        if config_file is None:
            config_file = Path(__file__).parent / "config.yaml"
        self.config_file = Path(config_file)
    
    def _load_rewrite_dictionary(self) -> Dict[str, str]:
        """Load query rewrite dictionary from external file."""
        rewrite_file = Path(__file__).parent / "rewrite_dictionary.txt"
        rewrite_dict = {}
        
        if not rewrite_file.exists():
            return rewrite_dict
        
        try:
            with open(rewrite_file, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith('#'):
                        continue
                    
                    parts = line.split('\t')
                    if len(parts) >= 2:
                        original = parts[0].strip()
                        replacement = parts[1].strip()
                        if original and replacement:
                            rewrite_dict[original] = replacement
        except Exception as e:
            print(f"Warning: Failed to load rewrite dictionary: {e}")
        
        return rewrite_dict
    
    def load_config(self, validate: bool = True) -> SearchConfig:
        """
        Load unified configuration from YAML file.
        
        Args:
            validate: Whether to validate configuration after loading
        
        Returns:
            SearchConfig object
        
        Raises:
            ConfigurationError: If config file not found, invalid, or validation fails
        """
        if not self.config_file.exists():
            raise ConfigurationError(f"Configuration file not found: {self.config_file}")
        
        try:
            with open(self.config_file, 'r', encoding='utf-8') as f:
                config_data = yaml.safe_load(f)
        except yaml.YAMLError as e:
            raise ConfigurationError(f"Invalid YAML in {self.config_file}: {e}")
        
        config = self._parse_config(config_data)
        
        # Auto-validate configuration
        if validate:
            errors = self.validate_config(config)
            if errors:
                error_msg = "Configuration validation failed:\n" + "\n".join(f"  - {err}" for err in errors)
                raise ConfigurationError(error_msg)
        
        return config
    
    def _parse_config(self, config_data: Dict[str, Any]) -> SearchConfig:
        """Parse configuration dictionary into SearchConfig object."""
        
        # Parse field_boosts
        field_boosts = config_data.get("field_boosts", {})
        if not isinstance(field_boosts, dict):
            raise ConfigurationError("field_boosts must be a dictionary")
        
        # Parse indexes
        indexes = []
        for index_data in config_data.get("indexes", []):
            indexes.append(self._parse_index_config(index_data))
        
        # Parse query config
        query_config_data = config_data.get("query_config", {})
        
        # Load rewrite dictionary from external file
        rewrite_dictionary = self._load_rewrite_dictionary()
        
        # Parse embedding disable thresholds
        embedding_thresholds = query_config_data.get("embedding_disable_thresholds", {})
        
        query_config = QueryConfig(
            supported_languages=query_config_data.get("supported_languages") or ["zh", "en"],
            default_language=query_config_data.get("default_language") or "zh",
            enable_translation=query_config_data.get("enable_translation", True),
            enable_text_embedding=query_config_data.get("enable_text_embedding", True),
            enable_query_rewrite=query_config_data.get("enable_query_rewrite", True),
            rewrite_dictionary=rewrite_dictionary,
            translation_api_key=query_config_data.get("translation_api_key"),
            translation_service=query_config_data.get("translation_service") or "deepl",
            translation_glossary_id=query_config_data.get("translation_glossary_id"),
            translation_context=query_config_data.get("translation_context") or "e-commerce product search",
            text_embedding_field=query_config_data.get("text_embedding_field"),
            image_embedding_field=query_config_data.get("image_embedding_field"),
            embedding_disable_chinese_char_limit=embedding_thresholds.get("chinese_char_limit", 4),
            embedding_disable_english_word_limit=embedding_thresholds.get("english_word_limit", 3),
            source_fields=query_config_data.get("source_fields")
        )
        
        # Parse ranking config
        ranking_data = config_data.get("ranking", {})
        ranking = RankingConfig(
            expression=ranking_data.get("expression") or "bm25() + 0.2*text_embedding_relevance()",
            description=ranking_data.get("description") or "Default BM25 + text embedding ranking"
        )
        
        # Parse Function Score configuration
        fs_data = config_data.get("function_score", {})
        function_score = FunctionScoreConfig(
            score_mode=fs_data.get("score_mode") or "sum",
            boost_mode=fs_data.get("boost_mode") or "multiply",
            functions=fs_data.get("functions") or []
        )
        
        # Parse Rerank configuration
        rerank_data = config_data.get("rerank", {})
        rerank = RerankConfig(
            enabled=rerank_data.get("enabled", False),
            expression=rerank_data.get("expression") or "",
            description=rerank_data.get("description") or ""
        )
        
        # Parse SPU config
        spu_data = config_data.get("spu_config", {})
        spu_config = SPUConfig(
            enabled=spu_data.get("enabled", False),
            spu_field=spu_data.get("spu_field"),
            inner_hits_size=spu_data.get("inner_hits_size", 3),
            searchable_option_dimensions=spu_data.get("searchable_option_dimensions", ['option1', 'option2', 'option3'])
        )
        
        return SearchConfig(
            field_boosts=field_boosts,
            indexes=indexes,
            query_config=query_config,
            ranking=ranking,
            function_score=function_score,
            rerank=rerank,
            spu_config=spu_config,
            es_index_name=config_data.get("es_index_name", "search_products"),
            es_settings=config_data.get("es_settings", {})
        )
    
    def _parse_index_config(self, index_data: Dict[str, Any]) -> IndexConfig:
        """Parse index configuration from dictionary."""
        return IndexConfig(
            name=index_data["name"],
            label=index_data.get("label", index_data["name"]),
            fields=index_data.get("fields", []),
            boost=index_data.get("boost", 1.0),
            example=index_data.get("example")
        )
    
    def validate_config(self, config: SearchConfig) -> List[str]:
        """
        Validate configuration for common errors.
        
        Args:
            config: SearchConfig to validate
        
        Returns:
            List of error messages (empty if valid)
        """
        errors = []
        
        # Validate es_index_name
        if not config.es_index_name:
            errors.append("es_index_name is required")
        
        # Validate field_boosts
        if not config.field_boosts:
            errors.append("field_boosts is empty")
        
        for field_name, boost in config.field_boosts.items():
            if not isinstance(boost, (int, float)):
                errors.append(f"field_boosts['{field_name}']: boost must be a number, got {type(boost).__name__}")
            elif boost < 0:
                errors.append(f"field_boosts['{field_name}']: boost must be non-negative")
        
        # Validate indexes
        if not config.indexes:
            errors.append("At least one index domain must be defined")
        
        index_names = set()
        for index in config.indexes:
            # Check for duplicate index names
            if index.name in index_names:
                errors.append(f"Duplicate index name: {index.name}")
            index_names.add(index.name)
            
            # Validate fields in index
            if not index.fields:
                errors.append(f"Index '{index.name}': fields list is empty")
        
        # Validate SPU config
        if config.spu_config.enabled:
            if not config.spu_config.spu_field:
                errors.append("SPU aggregation enabled but no spu_field specified")
        
        # Validate query config
        if not config.query_config.supported_languages:
            errors.append("At least one supported language must be specified")
        
        if config.query_config.default_language not in config.query_config.supported_languages:
            errors.append(
                f"Default language '{config.query_config.default_language}' "
                f"not in supported languages: {config.query_config.supported_languages}"
            )
        
        return errors
    
    def to_dict(self, config: SearchConfig) -> Dict[str, Any]:
        """Convert SearchConfig to dictionary representation."""
        
        # Build query_config dict
        query_config_dict = {
            "supported_languages": config.query_config.supported_languages,
            "default_language": config.query_config.default_language,
            "enable_translation": config.query_config.enable_translation,
            "enable_text_embedding": config.query_config.enable_text_embedding,
            "enable_query_rewrite": config.query_config.enable_query_rewrite,
            "translation_service": config.query_config.translation_service,
            "text_embedding_field": config.query_config.text_embedding_field,
            "image_embedding_field": config.query_config.image_embedding_field,
            "embedding_disable_thresholds": {
                "chinese_char_limit": config.query_config.embedding_disable_chinese_char_limit,
                "english_word_limit": config.query_config.embedding_disable_english_word_limit
            },
            "source_fields": config.query_config.source_fields
        }
        
        return {
            "es_index_name": config.es_index_name,
            "es_settings": config.es_settings,
            "field_boosts": config.field_boosts,
            "indexes": [self._index_to_dict(index) for index in config.indexes],
            "query_config": query_config_dict,
            "ranking": {
                "expression": config.ranking.expression,
                "description": config.ranking.description
            },
            "function_score": {
                "score_mode": config.function_score.score_mode,
                "boost_mode": config.function_score.boost_mode,
                "functions": config.function_score.functions
            },
            "rerank": {
                "enabled": config.rerank.enabled,
                "expression": config.rerank.expression,
                "description": config.rerank.description
            },
            "spu_config": {
                "enabled": config.spu_config.enabled,
                "spu_field": config.spu_config.spu_field,
                "inner_hits_size": config.spu_config.inner_hits_size,
                "searchable_option_dimensions": config.spu_config.searchable_option_dimensions
            }
        }
    
    def _index_to_dict(self, index: IndexConfig) -> Dict[str, Any]:
        """Convert IndexConfig to dictionary."""
        result = {
            "name": index.name,
            "label": index.label,
            "fields": index.fields,
            "boost": index.boost
        }
        
        if index.example:
            result["example"] = index.example
        
        return result
def load_tenant_config(tenant_id: Optional[str] = None) -> SearchConfig:
    """
    Load tenant configuration (backward compatibility wrapper).
    
    Args:
        tenant_id: Ignored (kept for backward compatibility)
    
    Returns:
        SearchConfig loaded from config/config.yaml
    """
    loader = ConfigLoader()
    return loader.load_config()