api/result_formatter.py

"""
Result formatter for converting ES internal format to external-friendly format.
"""
from typing import List, Dict, Any, Optional
from .models import SpuResult, SkuResult, FacetResult, FacetValue
class ResultFormatter:
    """Formats ES search results to external-friendly format."""
    @staticmethod
    def format_search_results(
        es_hits: List[Dict[str, Any]],
        max_score: float = 1.0,
        language: str = "zh"
    ) -> List[SpuResult]:
        """
        Convert ES hits to SpuResult list.
        Args:
            es_hits: List of ES hit dictionaries (with _id, _score, _source)
            max_score: Maximum score (unused, kept for compatibility)
        Returns:
            List of SpuResult objects
        """
        results = []
        lang = (language or "zh").lower()
        if lang not in ("zh", "en"):
            lang = "en"
        def pick_lang_field(src: Dict[str, Any], base: str) -> Optional[str]:
            """从 *_zh / *_en 字段中按语言选择一个值，若目标语言缺失则回退到另一种。"""
            zh_val = src.get(f"{base}_zh")
            en_val = src.get(f"{base}_en")
            if lang == "zh":
                return zh_val or en_val
            else:
                return en_val or zh_val
        for hit in es_hits:
            source = hit.get('_source', {})
            score = hit.get('_score')
            
            # Use original ES score directly (no normalization)
            # Handle None score (can happen with certain query types or when score is explicitly null)
            if score is None:
                relevance_score = 0.0
            else:
                try:
                    relevance_score = float(score)
                except (ValueError, TypeError):
                    relevance_score = 0.0
            # Multi-language fields
            title = pick_lang_field(source, "title")
            brief = pick_lang_field(source, "brief")
            description = pick_lang_field(source, "description")
            vendor = pick_lang_field(source, "vendor")
            category_path = pick_lang_field(source, "category_path")
            category_name = pick_lang_field(source, "category_name")
            # Extract SKUs
            skus = []
            skus_data = source.get('skus', [])
            if isinstance(skus_data, list):
                for sku_entry in skus_data:
                    sku = SkuResult(
                        sku_id=str(sku_entry.get('sku_id', '')),
                        title=sku_entry.get('title'),
                        price=sku_entry.get('price'),
                        compare_at_price=sku_entry.get('compare_at_price'),
                        sku=sku_entry.get('sku'),
                        stock=sku_entry.get('stock', 0),
                        options=sku_entry.get('options')
                    )
                    skus.append(sku)
            # Determine in_stock (any sku has stock > 0)
            in_stock = any(sku.stock > 0 for sku in skus) if skus else True
            # Build SpuResult
            spu = SpuResult(
                spu_id=str(source.get('spu_id', '')),
                title=title,
                brief=brief,
                handle=source.get('handle'),
                description=description,
                vendor=vendor,
                category=category_name,
                category_path=category_path,
                category_name=category_name,
                category_id=source.get('category_id'),
                category_level=source.get('category_level'),
                category1_name=source.get('category1_name'),
                category2_name=source.get('category2_name'),
                category3_name=source.get('category3_name'),
                tags=source.get('tags'),
                price=source.get('min_price'),
                compare_at_price=source.get('compare_at_price'),
                currency="USD",  # Default currency
                image_url=source.get('image_url'),
                in_stock=in_stock,
                sku_prices=source.get('sku_prices'),
                sku_weights=source.get('sku_weights'),
                sku_weight_units=source.get('sku_weight_units'),
                total_inventory=source.get('total_inventory'),
                option1_name=source.get('option1_name'),
                option2_name=source.get('option2_name'),
                option3_name=source.get('option3_name'),
                specifications=source.get('specifications'),
                skus=skus,
                relevance_score=relevance_score
            )
            results.append(spu)
        return results
    @staticmethod
    def format_facets(
        es_aggregations: Dict[str, Any],
        facet_configs: Optional[List[Any]] = None
    ) -> List[FacetResult]:
        """
        Format ES aggregations to FacetResult list.
        支持：
        1. 普通terms聚合
        2. range聚合
        3. specifications嵌套聚合（按name分组，然后按value聚合）
        Args:
            es_aggregations: ES aggregations response
            facet_configs: Facet configurations (optional)
        Returns:
            List of FacetResult objects
        """
        facets = []
        for field_name, agg_data in es_aggregations.items():
            display_field = field_name[:-6] if field_name.endswith("_facet") else field_name
            
            # 处理specifications嵌套分面（所有name）
            if field_name == "specifications_facet" and 'by_name' in agg_data:
                # specifications嵌套聚合：按name分组，每个name下有value_counts
                by_name_agg = agg_data['by_name']
                if 'buckets' in by_name_agg:
                    for name_bucket in by_name_agg['buckets']:
                        name = name_bucket['key']
                        value_counts = name_bucket.get('value_counts', {})
                        
                        values = []
                        if 'buckets' in value_counts:
                            for value_bucket in value_counts['buckets']:
                                value = FacetValue(
                                    value=value_bucket['key'],
                                    label=str(value_bucket['key']),
                                    count=value_bucket['doc_count'],
                                    selected=False
                                )
                                values.append(value)
                        
                        # 为每个name创建一个分面结果
                        facet = FacetResult(
                            field=f"specifications.{name}",
                            label=str(name),  # 使用name作为label，如"颜色"、"尺寸"
                            type="terms",
                            values=values,
                            total_count=name_bucket['doc_count']
                        )
                        facets.append(facet)
                continue
            
            # 处理specifications嵌套分面（指定name）
            if field_name.startswith("specifications_") and field_name.endswith("_facet") and 'filter_by_name' in agg_data:
                # 提取name（从 "specifications_颜色_facet" 提取 "颜色"）
                name = field_name[len("specifications_"):-len("_facet")]
                filter_by_name_agg = agg_data.get('filter_by_name', {})
                value_counts = filter_by_name_agg.get('value_counts', {})
                
                values = []
                if 'buckets' in value_counts:
                    for value_bucket in value_counts['buckets']:
                        value = FacetValue(
                            value=value_bucket['key'],
                            label=str(value_bucket['key']),
                            count=value_bucket['doc_count'],
                            selected=False
                        )
                        values.append(value)
                
                # 创建分面结果
                facet = FacetResult(
                    field=f"specifications.{name}",
                    label=str(name),
                    type="terms",
                    values=values,
                    total_count=filter_by_name_agg.get('doc_count', 0)
                )
                facets.append(facet)
                continue
            
            # Handle terms aggregation
            if 'buckets' in agg_data:
                values = []
                for bucket in agg_data['buckets']:
                    value = FacetValue(
                        value=bucket['key'],
                        label=bucket.get('key_as_string', str(bucket['key'])),
                        count=bucket['doc_count'],
                        selected=False
                    )
                    values.append(value)
                facet = FacetResult(
                    field=display_field,
                    label=display_field,  # Can be enhanced with field labels
                    type="terms",
                    values=values,
                    total_count=agg_data.get('sum_other_doc_count', 0) + len(values)
                )
                facets.append(facet)
            # Handle range aggregation
            elif 'buckets' in agg_data and any('from' in b or 'to' in b for b in agg_data['buckets']):
                values = []
                for bucket in agg_data['buckets']:
                    range_key = bucket.get('key', '')
                    value = FacetValue(
                        value=range_key,
                        label=range_key,
                        count=bucket['doc_count'],
                        selected=False
                    )
                    values.append(value)
                facet = FacetResult(
                    field=display_field,
                    label=display_field,
                    type="range",
                    values=values
                )
                facets.append(facet)
        return facets
    @staticmethod
    def generate_suggestions(
        query: str,
        results: List[SpuResult]
    ) -> List[str]:
        """
        Generate search suggestions.
        Args:
            query: Original search query
            results: Search results
        Returns:
            List of suggestion strings (currently returns empty list)
        """
        # TODO: Implement suggestion generation logic
        return []
    @staticmethod
    def generate_related_searches(
        query: str,
        results: List[SpuResult]
    ) -> List[str]:
        """
        Generate related searches.
        Args:
            query: Original search query
            results: Search results
        Returns:
            List of related search strings (currently returns empty list)
        """
        # TODO: Implement related search generation logic
        return []