# コード例 1: Elasticsearch インデックス設定 (Python - elasticsearch-py)
from elasticsearch import Elasticsearch
 
es = Elasticsearch(['http://localhost:9200'])
 
# 日本語検索用のインデックス設定
index_settings = {
    "settings": {
        "number_of_shards": 3,
        "number_of_replicas": 1,
        "refresh_interval": "1s",    # インデックス更新間隔
        "analysis": {
            "char_filter": {
                "normalize_filter": {
                    "type": "icu_normalizer",      # Unicode 正規化
                    "name": "nfkc_cf",
                }
            },
            "analyzer": {
                "ja_analyzer": {
                    "type": "custom",
                    "char_filter": ["normalize_filter"],
                    "tokenizer": "kuromoji_tokenizer",
                    "filter": [
                        "kuromoji_baseform",      # 活用形 → 基本形 (走った→走る)
                        "kuromoji_part_of_speech", # 助詞・助動詞の除去
                        "cjk_width",              # 全角半角統一
                        "ja_stop",                # 日本語ストップワード
                        "lowercase",              # 英字小文字化
                    ]
                },
                "ja_search_analyzer": {
                    "type": "custom",
                    "char_filter": ["normalize_filter"],
                    "tokenizer": "kuromoji_tokenizer",
                    "filter": [
                        "kuromoji_baseform",
                        "kuromoji_part_of_speech",
                        "cjk_width",
                        "ja_stop",
                        "lowercase",
                        "synonym_filter",          # 検索時のみ同義語展開
                    ]
                },
                "ja_ngram_analyzer": {
                    "type": "custom",
                    "char_filter": ["normalize_filter"],
                    "tokenizer": "ja_ngram_tokenizer",
                    "filter": ["lowercase"]
                }
            },
            "tokenizer": {
                "ja_ngram_tokenizer": {
                    "type": "ngram",
                    "min_gram": 2,
                    "max_gram": 3,
                    "token_chars": ["letter", "digit"]
                }
            },
            "filter": {
                "synonym_filter": {
                    "type": "synonym",
                    "synonyms": [
                        "PC, パソコン, コンピュータ",
                        "スマホ, スマートフォン, 携帯電話",
                        "テレビ, TV, ティーヴィー",
                    ]
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "ja_analyzer",
                "search_analyzer": "ja_search_analyzer",
                "fields": {
                    "keyword": {"type": "keyword"},  # 完全一致・ソート・集計用
                    "ngram": {                        # 部分一致検索用
                        "type": "text",
                        "analyzer": "ja_ngram_analyzer"
                    }
                }
            },
            "description": {
                "type": "text",
                "analyzer": "ja_analyzer",
                "search_analyzer": "ja_search_analyzer",
            },
            "category": {
                "type": "keyword",               # フィルタ・集計用
            },
            "tags": {
                "type": "keyword",               # 複数タグ
            },
            "price": {
                "type": "integer",               # 範囲フィルタ用
            },
            "rating": {
                "type": "float",                 # ソート・ブースト用
            },
            "review_count": {
                "type": "integer",               # 人気度スコアリング用
            },
            "created_at": {
                "type": "date",                  # 時系列フィルタ用
            },
            "updated_at": {
                "type": "date",
            },
            "location": {
                "type": "geo_point",             # 地理検索用
            },
            "suggest": {
                "type": "completion",            # オートコンプリート用
                "analyzer": "ja_analyzer",
            },
            "metadata": {
                "type": "object",
                "enabled": False,                # インデックスしない (保存のみ)
            }
        }
    }
}
 
# インデックス作成 (エイリアス経由)
index_name = "products_v1"
alias_name = "products"
 
es.indices.create(index=index_name, body=index_settings)
es.indices.put_alias(index=index_name, name=alias_name)
print(f"インデックス '{index_name}' を作成し、エイリアス '{alias_name}' を設定しました")

# コード例 2: 複合検索クエリ (商品検索API)
from typing import Optional, List, Dict, Any
from datetime import datetime
 
 
def search_products(
    query: str,
    category: Optional[str] = None,
    tags: Optional[List[str]] = None,
    min_price: Optional[int] = None,
    max_price: Optional[int] = None,
    min_rating: Optional[float] = None,
    sort_by: str = "relevance",
    page: int = 1,
    size: int = 20,
) -> Dict[str, Any]:
    """
    商品の全文検索を実行する。
 
    Args:
        query: 検索キーワード
        category: カテゴリフィルタ
        tags: タグフィルタ (AND 条件)
        min_price: 最低価格
        max_price: 最高価格
        min_rating: 最低レーティング
        sort_by: ソート順 ("relevance", "price_asc", "price_desc",
                 "rating", "newest")
        page: ページ番号 (1-indexed)
        size: 1ページあたりの件数
 
    Returns:
        検索結果 (ヒット件数、商品リスト、ファセット情報)
    """
    # --- must 句: 全文検索 ---
    must_clauses = []
    if query:
        must_clauses.append({
            "multi_match": {
                "query": query,
                "fields": [
                    "title^3",          # タイトルは3倍ブースト
                    "title.ngram^0.5",  # N-gram は低ブースト
                    "description",
                    "tags^2",
                ],
                "type": "best_fields",
                "fuzziness": "AUTO",    # タイポ許容 (3-5文字→1編集距離)
                "minimum_should_match": "75%",
            }
        })
 
    # --- filter 句: 絞り込み (スコアに影響しない) ---
    filter_clauses = []
    if category:
        filter_clauses.append({"term": {"category": category}})
    if tags:
        for tag in tags:
            filter_clauses.append({"term": {"tags": tag}})
    if min_price is not None or max_price is not None:
        price_range = {}
        if min_price is not None:
            price_range["gte"] = min_price
        if max_price is not None:
            price_range["lte"] = max_price
        filter_clauses.append({"range": {"price": price_range}})
    if min_rating is not None:
        filter_clauses.append({"range": {"rating": {"gte": min_rating}}})
 
    # --- ソート順の決定 ---
    sort_options = {
        "relevance": [{"_score": "desc"}, {"rating": "desc"}],
        "price_asc": [{"price": "asc"}, {"_score": "desc"}],
        "price_desc": [{"price": "desc"}, {"_score": "desc"}],
        "rating": [{"rating": "desc"}, {"review_count": "desc"}],
        "newest": [{"created_at": "desc"}],
    }
    sort = sort_options.get(sort_by, sort_options["relevance"])
 
    # --- クエリ本体の構築 ---
    body = {
        "query": {
            "bool": {
                "must": must_clauses,
                "filter": filter_clauses,
            }
        },
        "highlight": {
            "fields": {
                "title": {"number_of_fragments": 0},  # 全文返却
                "description": {
                    "fragment_size": 150,
                    "number_of_fragments": 3,
                },
            },
            "pre_tags": ["<mark>"],
            "post_tags": ["</mark>"],
        },
        "aggs": {
            "categories": {
                "terms": {"field": "category", "size": 20}
            },
            "price_stats": {
                "stats": {"field": "price"}
            },
            "price_ranges": {
                "range": {
                    "field": "price",
                    "ranges": [
                        {"key": "~1000円", "to": 1000},
                        {"key": "1000~5000円", "from": 1000, "to": 5000},
                        {"key": "5000~10000円", "from": 5000, "to": 10000},
                        {"key": "10000円~", "from": 10000},
                    ]
                }
            },
            "avg_rating": {
                "avg": {"field": "rating"}
            },
            "tags": {
                "terms": {"field": "tags", "size": 30}
            },
        },
        "from": (page - 1) * size,
        "size": size,
        "sort": sort,
        "_source": {
            "excludes": ["suggest", "metadata"]  # 不要フィールド除外
        },
    }
 
    result = es.search(index="products", body=body)
 
    # --- レスポンスの整形 ---
    return {
        "total": result["hits"]["total"]["value"],
        "page": page,
        "size": size,
        "items": [
            {
                "id": hit["_id"],
                "score": hit["_score"],
                "source": hit["_source"],
                "highlight": hit.get("highlight", {}),
            }
            for hit in result["hits"]["hits"]
        ],
        "facets": {
            "categories": [
                {"key": b["key"], "count": b["doc_count"]}
                for b in result["aggregations"]["categories"]["buckets"]
            ],
            "price_ranges": [
                {"key": b["key"], "count": b["doc_count"]}
                for b in result["aggregations"]["price_ranges"]["buckets"]
            ],
            "price_stats": result["aggregations"]["price_stats"],
            "avg_rating": result["aggregations"]["avg_rating"]["value"],
            "tags": [
                {"key": b["key"], "count": b["doc_count"]}
                for b in result["aggregations"]["tags"]["buckets"]
            ],
        },
    }

# コード例 3: サジェスト (オートコンプリート)
def autocomplete(prefix: str, size: int = 5, category: str = None) -> list:
    """
    入力中のキーワードに対してサジェスト候補を返す。
 
    Args:
        prefix: ユーザーの入力文字列
        size: 返すサジェスト数
        category: カテゴリで絞り込む場合
 
    Returns:
        サジェスト候補のリスト
    """
    contexts = {}
    if category:
        contexts["category"] = [category]
 
    body = {
        "suggest": {
            "product_suggest": {
                "prefix": prefix,
                "completion": {
                    "field": "suggest",
                    "size": size,
                    "fuzzy": {
                        "fuzziness": 1,       # 1文字のタイポを許容
                        "transpositions": True, # 文字の入れ替えを許容
                    },
                    "contexts": contexts if contexts else None,
                    "skip_duplicates": True,
                }
            }
        },
        "_source": ["title", "category", "price"],
    }
 
    result = es.search(index="products", body=body)
    suggestions = result["suggest"]["product_suggest"][0]["options"]
 
    return [
        {
            "text": s["text"],
            "title": s["_source"]["title"],
            "category": s["_source"]["category"],
            "price": s["_source"]["price"],
            "score": s["_score"],
        }
        for s in suggestions
    ]
 
 
# サジェストデータのインデックス時の設定
def build_suggest_input(product: dict) -> dict:
    """サジェスト用の入力データを構築する"""
    inputs = [product["name"]]
 
    # 読み仮名があれば追加 (ローマ字入力対応)
    if product.get("name_kana"):
        inputs.append(product["name_kana"])
 
    # ブランド名
    if product.get("brand"):
        inputs.append(product["brand"])
 
    # キーワード
    inputs.extend(product.get("keywords", []))
 
    return {
        "input": inputs,
        "weight": int(product.get("rating", 3.0) * product.get("review_count", 1)),
        "contexts": {
            "category": [product["category"]],
        }
    }

# コード例 4: バルクインデキシングとエラーハンドリング
from elasticsearch.helpers import bulk, BulkIndexError
from datetime import datetime
import logging
 
logger = logging.getLogger(__name__)
 
 
def index_products(products: list, chunk_size: int = 500) -> dict:
    """
    商品リストを一括でインデックスする。
 
    Args:
        products: 商品データのリスト
        chunk_size: 1回のバルクリクエストで送る件数
 
    Returns:
        成功件数とエラー件数
    """
    def generate_actions(products):
        """バルクアクション生成のジェネレータ"""
        for product in products:
            yield {
                "_index": "products",
                "_id": product["id"],
                "_source": {
                    "title": product["name"],
                    "description": product.get("description", ""),
                    "category": product["category"],
                    "tags": product.get("tags", []),
                    "price": product["price"],
                    "rating": product.get("rating", 0.0),
                    "review_count": product.get("review_count", 0),
                    "created_at": product.get("created_at",
                                              datetime.now().isoformat()),
                    "updated_at": datetime.now().isoformat(),
                    "suggest": build_suggest_input(product),
                },
            }
 
    try:
        success, errors = bulk(
            es,
            generate_actions(products),
            chunk_size=chunk_size,
            request_timeout=120,
            raise_on_error=False,     # エラーでも処理を続行
            raise_on_exception=False,
            max_retries=3,            # リトライ回数
            initial_backoff=1,        # リトライ間隔(秒)
            max_backoff=60,
        )
        if errors:
            logger.error(f"バルクインデックスエラー: {len(errors)} 件")
            for error in errors[:5]:  # 最初の5件のみログ
                logger.error(f"  {error}")
        logger.info(f"インデックス完了: 成功={success}, エラー={len(errors)}")
        return {"success": success, "errors": len(errors)}
 
    except BulkIndexError as e:
        logger.exception(f"バルクインデックス致命的エラー: {e}")
        raise
 
 
def reindex_with_zero_downtime(new_settings: dict,
                                alias: str = "products") -> str:
    """
    ダウンタイムなしでインデックスを再構築する。
 
    Args:
        new_settings: 新しいインデックス設定
        alias: エイリアス名
 
    Returns:
        新しいインデックス名
    """
    # Step 1: 現在のインデックス名を取得
    current_indices = list(es.indices.get_alias(name=alias).keys())
    current_index = current_indices[0] if current_indices else None
 
    # Step 2: 新しいインデックス名を生成
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    new_index = f"{alias}_v_{timestamp}"
 
    # Step 3: 新しいインデックスを作成
    es.indices.create(index=new_index, body=new_settings)
    logger.info(f"新インデックス '{new_index}' を作成しました")
 
    # Step 4: リインデックス
    es.reindex(
        body={
            "source": {"index": current_index},
            "dest": {"index": new_index},
        },
        wait_for_completion=True,
        request_timeout=3600,
    )
    logger.info(f"リインデックス完了: {current_index} → {new_index}")
 
    # Step 5: エイリアスを切り替え (アトミック操作)
    es.indices.update_aliases(body={
        "actions": [
            {"remove": {"index": current_index, "alias": alias}},
            {"add": {"index": new_index, "alias": alias}},
        ]
    })
    logger.info(f"エイリアス '{alias}' を '{new_index}' に切り替えました")
 
    # Step 6: 旧インデックスを削除 (任意)
    # es.indices.delete(index=current_index)
 
    return new_index

# コード例 5: Function Score Query による高度なランキング
def search_with_custom_scoring(
    query: str,
    user_location: tuple = None,   # (lat, lon)
    boost_new: bool = True,
    boost_popular: bool = True,
    page: int = 1,
    size: int = 20,
) -> dict:
    """
    BM25 スコアにビジネスロジックを組み合わせた検索。
 
    スコア = BM25 × popularity_boost × freshness_boost × distance_decay
 
    Args:
        query: 検索キーワード
        user_location: ユーザーの位置情報 (緯度, 経度)
        boost_new: 新しい商品をブーストするか
        boost_popular: 人気商品をブーストするか
        page: ページ番号
        size: 件数
    """
    functions = []
 
    # --- 人気度ブースト ---
    if boost_popular:
        functions.append({
            "field_value_factor": {
                "field": "review_count",
                "modifier": "log1p",    # log(1 + review_count)
                "factor": 0.5,
                "missing": 0,
            },
            "weight": 2,
        })
        functions.append({
            "field_value_factor": {
                "field": "rating",
                "modifier": "none",
                "factor": 1,
                "missing": 3.0,         # レビューなしは3.0扱い
            },
            "weight": 1.5,
        })
 
    # --- 新着ブースト (直近30日を優遇) ---
    if boost_new:
        functions.append({
            "gauss": {
                "created_at": {
                    "origin": "now",
                    "scale": "30d",     # 30日で半減
                    "offset": "7d",     # 7日以内は減衰なし
                    "decay": 0.5,
                }
            },
            "weight": 1.2,
        })
 
    # --- 距離ブースト (近い店舗を優遇) ---
    if user_location:
        functions.append({
            "gauss": {
                "location": {
                    "origin": {
                        "lat": user_location[0],
                        "lon": user_location[1],
                    },
                    "scale": "5km",     # 5km で半減
                    "offset": "1km",    # 1km 以内は減衰なし
                    "decay": 0.5,
                }
            },
            "weight": 1.5,
        })
 
    body = {
        "query": {
            "function_score": {
                "query": {
                    "multi_match": {
                        "query": query,
                        "fields": ["title^3", "description", "tags^2"],
                        "type": "best_fields",
                        "fuzziness": "AUTO",
                    }
                },
                "functions": functions,
                "score_mode": "multiply",   # 各関数の結果を掛け算
                "boost_mode": "multiply",   # BM25 と掛け算
                "max_boost": 10,            # ブースト上限
            }
        },
        "from": (page - 1) * size,
        "size": size,
    }
 
    return es.search(index="products", body=body)

# コード例 6: 検索ログの収集と分析クエリ
import json
from datetime import datetime
 
 
def log_search_event(
    query: str,
    user_id: str,
    results_count: int,
    clicked_ids: list = None,
    response_time_ms: int = 0,
):
    """検索イベントをログインデックスに記録する"""
    event = {
        "query": query,
        "user_id": user_id,
        "results_count": results_count,
        "clicked_ids": clicked_ids or [],
        "has_clicks": bool(clicked_ids),
        "response_time_ms": response_time_ms,
        "timestamp": datetime.now().isoformat(),
    }
    es.index(index="search_logs", body=event)
 
 
def analyze_zero_result_queries(days: int = 7) -> list:
    """
    ゼロ件ヒットのクエリを分析する。
    → 同義語辞書の追加候補を発見するのに有用。
    """
    body = {
        "query": {
            "bool": {
                "must": [
                    {"range": {"timestamp": {"gte": f"now-{days}d"}}},
                    {"term": {"results_count": 0}},
                ]
            }
        },
        "aggs": {
            "zero_result_queries": {
                "terms": {
                    "field": "query.keyword",
                    "size": 50,
                    "order": {"_count": "desc"},
                }
            }
        },
        "size": 0,
    }
    result = es.search(index="search_logs", body=body)
    return [
        {"query": b["key"], "count": b["doc_count"]}
        for b in result["aggregations"]["zero_result_queries"]["buckets"]
    ]
 
 
def calculate_ctr(days: int = 7) -> list:
    """
    クエリごとの CTR (Click Through Rate) を計算する。
    → ランキング改善の指標として使用。
    """
    body = {
        "query": {
            "range": {"timestamp": {"gte": f"now-{days}d"}}
        },
        "aggs": {
            "queries": {
                "terms": {
                    "field": "query.keyword",
                    "size": 100,
                    "min_doc_count": 10,
                },
                "aggs": {
                    "click_rate": {
                        "avg": {
                            "script": {
                                "source": "doc['has_clicks'].value ? 1 : 0"
                            }
                        }
                    },
                    "avg_response_time": {
                        "avg": {"field": "response_time_ms"}
                    }
                }
            }
        },
        "size": 0,
    }
    result = es.search(index="search_logs", body=body)
    return [
        {
            "query": b["key"],
            "searches": b["doc_count"],
            "ctr": round(b["click_rate"]["value"] * 100, 1),
            "avg_response_ms": round(
                b["avg_response_time"]["value"], 0
            ),
        }
        for b in result["aggregations"]["queries"]["buckets"]
    ]

# コード例 7: Kafka Consumer (CDC → Elasticsearch)
from confluent_kafka import Consumer, KafkaError
import json
import signal
import sys
 
 
class SearchIndexConsumer:
    """
    Kafka から CDC イベントを消費し、Elasticsearch を更新する。
    """
 
    def __init__(self, kafka_config: dict, es_client, batch_size: int = 100):
        self.consumer = Consumer(kafka_config)
        self.es = es_client
        self.batch_size = batch_size
        self.running = True
        self.buffer = []
 
        signal.signal(signal.SIGINT, self._shutdown)
        signal.signal(signal.SIGTERM, self._shutdown)
 
    def _shutdown(self, signum, frame):
        self.running = False
 
    def start(self, topics: list):
        """Consumer ループを開始する"""
        self.consumer.subscribe(topics)
 
        while self.running:
            msg = self.consumer.poll(timeout=1.0)
            if msg is None:
                # タイムアウト: バッファに溜まったものをフラッシュ
                if self.buffer:
                    self._flush_buffer()
                continue
 
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    print(f"Kafka エラー: {msg.error()}")
                    continue
 
            # メッセージを処理
            event = json.loads(msg.value().decode("utf-8"))
            action = self._build_action(event)
            if action:
                self.buffer.append(action)
 
            # バッチサイズに達したらフラッシュ
            if len(self.buffer) >= self.batch_size:
                self._flush_buffer()
                self.consumer.commit()
 
        # 終了前にバッファをフラッシュ
        if self.buffer:
            self._flush_buffer()
        self.consumer.close()
 
    def _build_action(self, event: dict) -> dict:
        """CDC イベントからバルクアクションを構築する"""
        op = event.get("op")
 
        if op in ("c", "u", "r"):  # create, update, read (snapshot)
            after = event["after"]
            return {
                "_op_type": "index",
                "_index": "products",
                "_id": after["id"],
                "_source": {
                    "title": after["name"],
                    "description": after.get("description", ""),
                    "category": after["category"],
                    "price": after["price"],
                    "rating": after.get("rating", 0.0),
                    "updated_at": after.get("updated_at"),
                },
            }
        elif op == "d":  # delete
            before = event["before"]
            return {
                "_op_type": "delete",
                "_index": "products",
                "_id": before["id"],
            }
        return None
 
    def _flush_buffer(self):
        """バッファの内容を Elasticsearch に一括書き込み"""
        if not self.buffer:
            return
 
        from elasticsearch.helpers import bulk
        try:
            success, errors = bulk(
                self.es,
                self.buffer,
                raise_on_error=False,
            )
            if errors:
                print(f"バルクエラー: {len(errors)} 件")
            print(f"フラッシュ完了: {success} 件処理")
        except Exception as e:
            print(f"バルク書き込みエラー: {e}")
        finally:
            self.buffer = []
 
 
# 使用例
if __name__ == "__main__":
    kafka_config = {
        "bootstrap.servers": "kafka:9092",
        "group.id": "search-index-consumer",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,
    }
    es_client = Elasticsearch(["http://elasticsearch:9200"])
    consumer = SearchIndexConsumer(kafka_config, es_client)
    consumer.start(["dbserver1.public.products"])

# コード例 8: クエリ書き換えパイプライン
import re
from typing import List, Tuple
 
 
class QueryRewriter:
    """検索クエリの前処理と書き換えを行う"""
 
    def __init__(self):
        self.spelling_corrections = {
            "あいふぉん": "iPhone",
            "ぐーぐる": "Google",
            "あまぞん": "Amazon",
        }
        self.query_expansions = {
            "ノートPC": ["ノートPC", "ノートパソコン", "ラップトップ"],
            "イヤホン": ["イヤホン", "イヤフォン", "ヘッドホン"],
        }
        self.stop_patterns = [
            r"を?\s*探して(い?ます|ください)?",
            r"が?\s*欲しい(です)?",
            r"おすすめ",
        ]
 
    def rewrite(self, query: str) -> dict:
        """
        クエリを分析し、書き換えた結果を返す。
 
        Returns:
            {
                "original": 元のクエリ,
                "rewritten": 書き換え後のクエリ,
                "expansions": 展開されたクエリ,
                "corrections": 修正内容,
            }
        """
        corrections = []
        rewritten = query.strip()
 
        # Step 1: 自然言語パターンの除去
        for pattern in self.stop_patterns:
            cleaned = re.sub(pattern, "", rewritten).strip()
            if cleaned != rewritten:
                corrections.append(
                    f"パターン除去: '{rewritten}' → '{cleaned}'"
                )
                rewritten = cleaned
 
        # Step 2: スペル修正
        for wrong, correct in self.spelling_corrections.items():
            if wrong in rewritten.lower():
                rewritten = rewritten.replace(wrong, correct)
                corrections.append(f"スペル修正: {wrong} → {correct}")
 
        # Step 3: クエリ展開
        expansions = [rewritten]
        for term, expanded in self.query_expansions.items():
            if term in rewritten:
                expansions = expanded
                corrections.append(f"クエリ展開: {term} → {expanded}")
 
        return {
            "original": query,
            "rewritten": rewritten,
            "expansions": expansions,
            "corrections": corrections,
        }
 
 
# 使用例
rewriter = QueryRewriter()
result = rewriter.rewrite("ノートPCを探しています")
# → {
#     "original": "ノートPCを探しています",
#     "rewritten": "ノートPC",
#     "expansions": ["ノートPC", "ノートパソコン", "ラップトップ"],
#     "corrections": [
#         "パターン除去: 'ノートPCを探しています' → 'ノートPC'",
#         "クエリ展開: ノートPC → ['ノートPC', 'ノートパソコン', 'ラップトップ']"
#     ]
# }

-- NG: LIKE 検索はインデックスが効かない
SELECT * FROM products
WHERE name LIKE '%東京%' OR description LIKE '%東京%';
-- → フルテーブルスキャン
-- → 100万行で2-5秒、スケールしない
-- → 形態素解析なし、ランキングなし
 
-- NG: MySQL FULLTEXT も CJK (日本語) に弱い
SELECT * FROM products
WHERE MATCH(name, description)
AGAINST('東京の観光' IN BOOLEAN MODE);
-- → 形態素解析なし (ngram のみ)
-- → 精度が低い、同義語非対応
-- → ファセット検索不可

# OK: 専用の検索エンジンを使う
#
# アーキテクチャ:
#   DB (PostgreSQL) --- CDC (Debezium) ---> Kafka ---> Elasticsearch
#
# メリット:
#   - 形態素解析 (kuromoji) で日本語を正確にトークナイズ
#   - BM25 ランキングで関連性の高い結果を上位に
#   - ファセット検索、サジェスト、ハイライト全対応
#   - 水平スケーリング可能 (シャーディング)
#
# DB は SSOT (Single Source of Truth) として維持し、
# Elasticsearch は検索専用のリードモデルとして位置づける
 
result = es.search(
    index="products",
    body={
        "query": {
            "multi_match": {
                "query": "東京の観光",
                "fields": ["title^3", "description"],
                "analyzer": "ja_search_analyzer",
            }
        },
        "highlight": {"fields": {"title": {}, "description": {}}},
        "aggs": {"categories": {"terms": {"field": "category"}}},
    }
)

# NG: マッピングを定義せずにドキュメントをインデックス
es.index(index="products", body={
    "name": "テスト商品",        # → text + keyword (両方にインデックス)
    "price": 1000,              # → long (integer で十分)
    "description": "テスト説明",  # → text + keyword (keyword は不要)
    "internal_notes": "社内メモ", # → text + keyword (検索対象外なのにインデックス)
    "created_at": "2024-01-15",  # → date (正しいが偶然)
})
# 問題点:
#   - 全フィールドが text + keyword 双方にインデックスされる
#   - ストレージが2倍以上
#   - インデキシング速度が低下
#   - 不要なフィールドまで検索対象になる

# OK: 明示的なマッピングを事前設計
# (3.1 節のインデックスマッピング設計を参照)
#
# 設計原則:
#   - 検索対象フィールド: text + 適切なアナライザー
#   - フィルタ/ソート/集計用: keyword
#   - 数値: integer / float (最小限の型を選ぶ)
#   - 検索不要フィールド: enabled: false
#   - dynamic: "strict" で未知フィールドを拒否
 
index_settings = {
    "mappings": {
        "dynamic": "strict",  # 未定義フィールドはエラーにする
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "ja_analyzer",
            },
            "price": {"type": "integer"},
            "metadata": {
                "type": "object",
                "enabled": False,   # インデックスしない
            },
        }
    }
}

# NG: インデックス時に同義語展開
index_settings = {
    "settings": {
        "analysis": {
            "analyzer": {
                "my_analyzer": {
                    "tokenizer": "kuromoji_tokenizer",
                    "filter": ["synonym_filter"],  # インデックス時に同義語展開
                }
            },
            "filter": {
                "synonym_filter": {
                    "type": "synonym",
                    "synonyms": ["PC, パソコン, コンピュータ"]
                }
            }
        }
    }
}
# 問題点:
#   - 同義語辞書を更新するたびに全文書の再インデックスが必要
#   - インデックスサイズが膨張する
#   - 100万文書の再インデックスに数時間かかることも

# OK: 検索時のみ同義語展開
index_settings = {
    "settings": {
        "analysis": {
            "analyzer": {
                "ja_index_analyzer": {     # インデックス時
                    "tokenizer": "kuromoji_tokenizer",
                    "filter": ["kuromoji_baseform", "lowercase"]
                    # ← 同義語フィルタなし
                },
                "ja_search_analyzer": {    # 検索時
                    "tokenizer": "kuromoji_tokenizer",
                    "filter": [
                        "kuromoji_baseform",
                        "lowercase",
                        "synonym_filter",  # ← 検索時のみ同義語展開
                    ]
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "ja_index_analyzer",       # インデックス時
                "search_analyzer": "ja_search_analyzer" # 検索時
            }
        }
    }
}
# メリット:
#   - 同義語辞書の更新は analyzer の reload のみ
#   - 再インデックス不要
#   - インデックスサイズが小さい

"""
前提:
- 商品検索システム (ECサイト)
- 100万件の商品データ
- 検索ログから以下の問題が判明:
  1. "パソコン" で検索しても "PC" がヒットしない
  2. "iphon" (タイポ) で検索すると0件
  3. 長い商品説明の方が短いタイトル一致より上位に来る
  4. 新着商品が埋もれてしまう
  5. "東京スカイツリー" が "東京" と "スカイツリー" に分割される
 
課題:
(a) 問題1を解決する同義語設定を書け
(b) 問題2を解決するファジー検索設定を書け
(c) 問題3を解決するフィールドブースト設定を書け
(d) 問題4を解決する Function Score 設定を書け
(e) 問題5を解決するユーザー辞書設定を書け
"""

# (a) 同義語設定
synonym_settings = {
    "filter": {
        "synonym_filter": {
            "type": "synonym",
            "synonyms": [
                "PC, パソコン, コンピュータ, パーソナルコンピュータ",
                "スマホ, スマートフォン, 携帯電話",
                "テレビ, TV, ティーヴィー",
            ]
        }
    }
}
 
# (b) ファジー検索 (fuzziness: "AUTO" は長さに応じて自動調整)
fuzzy_query = {
    "multi_match": {
        "query": "iphon",
        "fields": ["title^3", "description"],
        "fuzziness": "AUTO",        # 3-5文字: 編集距離1, 6文字以上: 距離2
        "prefix_length": 1,         # 先頭1文字は一致必須
        "max_expansions": 50,
    }
}
 
# (c) フィールドブースト
boosted_query = {
    "multi_match": {
        "query": "ノートパソコン",
        "fields": [
            "title^5",          # タイトル完全一致を最重視
            "title.ngram^1",    # タイトル部分一致
            "description^1",    # 説明文は低ウェイト
        ],
        "type": "best_fields",
    }
}
 
# (d) 新着ブースト (Function Score)
freshness_query = {
    "function_score": {
        "query": {"match_all": {}},
        "functions": [
            {
                "gauss": {
                    "created_at": {
                        "origin": "now",
                        "scale": "14d",
                        "decay": 0.5,
                    }
                },
                "weight": 2,
            }
        ],
        "boost_mode": "multiply",
    }
}
 
# (e) ユーザー辞書 (kuromoji)
# userdict.txt に以下を追加:
# 東京スカイツリー,東京スカイツリー,トウキョウスカイツリー,カスタム名詞
user_dict_settings = {
    "tokenizer": {
        "kuromoji_user_dict": {
            "type": "kuromoji_tokenizer",
            "user_dictionary": "userdict.txt",
            "mode": "search",   # search モードで複合語を分割
        }
    }
}