同義語キャッシュの生成スクリプトを作成

config/files.json

@@ -2,7 +2,10 @@
     "sudachi": {
         "sudachi_config": "scripts/sudachi.json",
         "user_dict": "resources/user_dict.csv",
-        "user_dict_generated": "data/generated/user.dic"
+        "user_dict_generated": "data/generated/user.dic",
+        "synonyms": "resources/synonyms_DO_NOT_EDIT.txt",
+        "synonyms_cache": "data/generated/synonyms_cache.json",
+        "stopwords": "resources/stopwords.json"
     },
     "projects": {
         "original_csv": "resources/test_data.csv",

config/search_model.json

@@ -0,0 +1,30 @@
+{
+    "target_pos_l1": [
+        "名詞", "動詞"
+    ],
+    "target_fields": [
+        "title",
+        "organization",
+        "description",
+        "prComment",
+        "prCommentLong",
+        "reading"
+    ],
+    "synonyms": {
+        "enable": true,
+        "sources": {
+            "sudachi": true,
+            "custom_json": "resources/synonyms_custom.json"
+        },
+        "limits": {
+            "max_expansions_per_term": 4,
+            "max_query_variants": 5,
+            "min_char_len": 2
+        },
+        "banlist": [
+            "部",
+            "会",
+            "サークル"
+        ]
+    }
+}

scripts/3_build_synonyms_from_sudachi.py

@@ -0,0 +1,171 @@
+import os
+import sys
+import json
+from collections import defaultdict
+from tqdm import tqdm
+import pandas as pd
+
+from sudachipy import dictionary, tokenizer
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from utils.logger import setup_logger
+from utils.json import get_file_path_from_config, field_getter, json_dumps
+from utils.io import LineIteratorIO, comment_filtered_lines
+import schemas.projects as projects_schema
+
+# --- ロギングの設定 ---
+log = setup_logger(__name__)
+
+# --- 設定 ---
+input_file = get_file_path_from_config("projects.projects_json")
+output_file = get_file_path_from_config("sudachi.synonyms_cache")
+sudachi_config_file = get_file_path_from_config("sudachi.sudachi_config")
+
+try:
+    search_model = field_getter("config/search_model.json")
+    target_pos_l1 = search_model("target_pos_l1")
+    target_fields = search_model("target_fields")
+    ban_list = search_model("synonyms.banlist")
+    stopwords_file = get_file_path_from_config("sudachi.stopwords")
+    with open(stopwords_file, encoding="utf-8") as f:
+        stopwords = set(json.load(f))
+    synonyms_file = get_file_path_from_config("sudachi.synonyms")
+except FileNotFoundError as e:
+    log.error(f"設定ファイルが見つかりません: {e}")
+    sys.exit(1)
+except KeyError as e:
+    log.error(f"設定ファイルに必要なキーが見つかりません: {e}")
+    sys.exit(1)
+except Exception as e:
+    log.error(f"予期しないエラーが発生しました: {e}")
+    sys.exit(1)
+
+
+# --- SudachiPy初期化 ---
+tokenizer_obj = dictionary.Dictionary(config_path=sudachi_config_file).create()
+mode = tokenizer.Tokenizer.SplitMode.A
+
+
+def tokenize(text: str) -> list[str]:
+    """テキストをトークン化し、ストップワードを除去する。"""
+    if not text:
+        return []
+    tokens = []
+    for m in tokenizer_obj.tokenize(text, mode):
+        if m.normalized_form().strip() == "":
+            continue
+        pos = m.part_of_speech()
+        if pos[0] not in target_pos_l1:
+            continue
+        base = m.normalized_form().lower()
+        if base in stopwords or base in ban_list:
+            continue
+        tokens.append(base)
+    return tokens
+
+
+def get_corpus_vocab(projects: list[projects_schema.Project]) -> set[str]:
+    """プロジェクト全体から語彙セットを構築する"""
+    vocab = set()
+    log.info("語彙セットを構築中...")
+    for project in tqdm(projects):
+        for field in target_fields:
+            text = getattr(project, field, "")
+            if not text:
+                continue
+            tokens = tokenize(str(text))
+            vocab.update(tokens)
+    return vocab
+
+
+def find_and_parse_synonyms_text() -> dict[str, list[str]]:
+    """同義語辞書をpandasを使ってパースする"""
+    log.info("同義語辞書をパース中...")
+    log.info(f"同義語辞書のパス: {synonyms_file}")
+
+    try:
+        # ヘッダーなしのCSVとして読み込む
+        # 必要なのはグループID(0)と見出し語(8)
+        GROUP_ID_COL_IDX = 0
+        TERM_COL_IDX = 8
+
+        stream = LineIteratorIO(comment_filtered_lines(synonyms_file))
+        df = pd.read_csv(
+            stream,
+            header=None,
+            usecols=[GROUP_ID_COL_IDX, TERM_COL_IDX],
+            names=["group_id", "term"],
+            on_bad_lines="skip",
+            encoding="utf-8",
+            engine="c",
+            sep=",",
+        )
+
+    except Exception as e:
+        log.error(f"同義語辞書のパース中にエラーが発生しました: {e}")
+        raise e
+
+    synonym_groups = defaultdict(list)
+    # group_idでグループ化し、各グループ内の単語リストを作成
+    for group_id, group_df in df.groupby("group_id"):
+        terms = group_df["term"].tolist()
+        if len(terms) < 2:
+            continue
+        for i, term in enumerate(terms):
+            synonyms = [t for j, t in enumerate(terms) if i != j]
+            synonym_groups[term].extend(synonyms)
+
+    return synonym_groups
+
+
+def main():
+    log.info("同義語キャッシュの生成を開始します")
+
+    with open(input_file, encoding="utf-8") as f:
+        try:
+            project_dicts = json.load(f)
+            projects = [projects_schema.Project(**item) for item in project_dicts]
+        except json.JSONDecodeError as e:
+            log.error(f"JSONデコードエラー: {e}")
+            sys.exit(1)
+        except FileNotFoundError:
+            log.error(f"ファイルが見つかりません: {input_file}")
+            sys.exit(1)
+        except Exception as e:
+            log.error(f"予期しないエラーが発生しました: {e}")
+            sys.exit(1)
+
+    # 1. コーパスの語彙を構築
+    vocab = get_corpus_vocab(projects)
+    log.info(f"コーパスの語彙数: {len(vocab)}")
+
+    # 2. Sudachiの同義語辞書をパース
+    try:
+        synonym_groups = find_and_parse_synonyms_text()
+    except Exception as e:
+        log.error(e)
+        sys.exit(1)
+
+    # 3. コーパスに存在する単語に絞ってキャッシュを作成
+    synonym_cache = {}
+    log.info("同義語キャッシュを構築中...")
+    for term in tqdm(vocab):
+        if term in ban_list:
+            continue
+        if term in synonym_groups:
+            valid_synonyms = [
+                s for s in synonym_groups[term] if s in vocab and s not in ban_list
+            ]
+            if valid_synonyms:
+                synonym_cache[term] = list(set(valid_synonyms))  # 重複除去
+
+    # 4. キャッシュを保存
+    json_dumps(synonym_cache, output_file)
+
+    log.info(f"同義語キャッシュの生成が完了しました: {output_file}")
+    log.info(f"キャッシュされた単語数: {len(synonym_cache)}")
+
+
+if __name__ == "__main__":
+    main()

utils/io.py

@@ -0,0 +1,38 @@
+import io
+
+
+def comment_filtered_lines(path: str):
+    with open(path, encoding="utf-8") as f:
+        for line in f:
+            if not line.lstrip().startswith("#"):
+                yield line
+
+
+class LineIteratorIO(io.TextIOBase):
+    def __init__(self, iterator):
+        self._it = iter(iterator)
+        self._buf = ""
+
+    def readable(self):
+        return True
+
+    def read(self, size=-1):
+        # size<0 のときはEOFまで貯めて返す
+        if size is None or size < 0:
+            try:
+                for chunk in self._it:
+                    self._buf += chunk
+            except StopIteration:
+                pass
+            out, self._buf = self._buf, ""
+            return out
+
+        # size 指定ありのときは、必要分だけバッファを満たす
+        while len(self._buf) < size:
+            try:
+                self._buf += next(self._it)
+            except StopIteration:
+                break
+
+        out, self._buf = self._buf[:size], self._buf[size:]
+        return out

utils/json.py

@@ -14,6 +14,35 @@ def _get_nested(d: dict[str, Any], keys: list[str]) -> Any:
     return cur
 
 
+def _get_field_from_json(
+    path: str | Path, key: str, default: str | None = None, sep: str = "."
+) -> Any:
+    """
+    JSONファイルからフィールドを取得する。
+    階層構造がある場合は`sep`で区切って指定する。
+
+    example: `get_file_path_from_config("projects.original_csv")`
+    """
+    with open(path, "r", encoding="utf-8") as f:
+        try:
+            config: dict[str, Any] = json.load(f)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"JSONデコードエラー: {e}")
+        except FileNotFoundError:
+            raise ValueError(f"ファイルが見つかりません: {path}")
+
+    keys = key.split(sep) if sep in key else [key]
+
+    try:
+        value = _get_nested(config, keys)
+    except KeyError:
+        if default is not None:
+            return default
+        raise KeyError(f"Key '{key}' not found in {path}")
+
+    return value
+
+
 def get_file_path_from_config(
     key: str, default: str | None = None, sep: str = "."
 ) -> str:
@@ -43,6 +72,13 @@ def get_file_path_from_config(
     raise TypeError(f"Value at '{key}' is not a string: {type(value).__name__}")
 
 
+def field_getter(path: str | Path):
+    def getter(key: str, default: str | None = None, sep: str = ".") -> Any:
+        return _get_field_from_json(path, key, default, sep)
+
+    return getter
+
+
 def _write_atomic(path: Path, data: str) -> None:
     temp = path.with_suffix(path.suffix + ".tmp")
     temp.write_bytes(data)