軽いリファクタリング

scripts/3_build_synonyms_from_sudachi.py

@@ -12,7 +12,7 @@ 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
+from schemas.projects import Project
 
 # --- ロギングの設定 ---
 log = setup_logger(__name__)
@@ -65,7 +65,7 @@ def tokenize(text: str) -> list[str]:
     return tokens
 
 
-def get_corpus_vocab(projects: list[projects_schema.Project]) -> set[str]:
+def get_corpus_vocab(projects: list[Project]) -> set[str]:
     """プロジェクト全体から語彙セットを構築する"""
     vocab = set()
     log.info("語彙セットを構築中...")
@@ -125,7 +125,7 @@ def main():
     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]
+            projects = [Project(**item) for item in project_dicts]
         except json.JSONDecodeError as e:
             log.error(f"JSONデコードエラー: {e}")
             sys.exit(1)

scripts/4_prepare_bm25f_meta.py

@@ -1,6 +1,7 @@
 import os
 import sys
 import json
+import math
 from collections import defaultdict
 from typing import Dict, List
 
@@ -37,7 +38,9 @@ def build_tokenizer(sudachi_config_path: str):
     return tok, mode
 
 
-def tokenize(text: str, tok, mode, target_pos_l1: List[str], ban_list: set, stopwords: set) -> List[str]:
+def tokenize(
+    text: str, tok, mode, target_pos_l1: List[str], ban_list: set, stopwords: set
+) -> List[str]:
     if not text:
         return []
     out: List[str] = []
@@ -57,13 +60,19 @@ def tokenize(text: str, tok, mode, target_pos_l1: List[str], ban_list: set, stop
 def main():
     log.info("BM25Fメタデータ(bm25_meta.json)を生成します")
     try:
-        target_pos_l1, target_fields, ban_list, stopwords, sudachi_config_path = load_configs()
+        target_pos_l1, target_fields, ban_list, stopwords, sudachi_config_path = (
+            load_configs()
+        )
     except Exception as e:
         log.error(f"設定の読み込みに失敗しました: {e}")
         sys.exit(1)
 
-    projects_path = get_file_path_from_config("projects.projects_json", "data/generated/projects.json")
-    output_path = get_file_path_from_config("bm25.bm25_meta", "data/generated/bm25_meta.json")
+    projects_path = get_file_path_from_config(
+        "projects.projects_json", "data/generated/projects.json"
+    )
+    output_path = get_file_path_from_config(
+        "bm25.bm25_meta", "data/generated/bm25_meta.json"
+    )
 
     try:
         with open(projects_path, encoding="utf-8") as f:
@@ -94,13 +103,14 @@ def main():
     # IDF 計算（BM25で一般的な +0.5 smoothing と +1 オフセット）
     idf: Dict[str, float] = {}
     for term, dfi in df.items():
-        idf_val = max(0.0, ( ( (N - dfi + 0.5) / (dfi + 0.5) ) ))
+        idf_val = max(0.0, ((N - dfi + 0.5) / (dfi + 0.5)))
         # 数値安定化のためlog1p
-        import math
-
         idf[term] = math.log1p(idf_val)
 
-    avg_len = {field: (field_token_lens_sum[field] / N if N > 0 else 0.0) for field in target_fields}
+    avg_len = {
+        field: (field_token_lens_sum[field] / N if N > 0 else 0.0)
+        for field in target_fields
+    }
 
     meta = {
         "N": N,

scripts/5_prepare_tf_token.py

@@ -10,7 +10,7 @@ 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
-import schemas.tf_token as tf_schema
+from schemas import tf_token
 
 log = setup_logger(__name__)
 
@@ -38,7 +38,9 @@ def build_tokenizer(sudachi_config_path: str):
     return tok, mode
 
 
-def tokenize(text: str, tok, mode, target_pos_l1: List[str], ban_list: set, stopwords: set) -> List[str]:
+def tokenize(
+    text: str, tok, mode, target_pos_l1: List[str], ban_list: set, stopwords: set
+) -> List[str]:
     if not text:
         return []
     out: List[str] = []
@@ -58,13 +60,19 @@ def tokenize(text: str, tok, mode, target_pos_l1: List[str], ban_list: set, stop
 def main():
     log.info("フィールド別TF/トークン(tf_token.json)を生成します")
     try:
-        target_pos_l1, target_fields, ban_list, stopwords, sudachi_config_path = load_configs()
+        target_pos_l1, target_fields, ban_list, stopwords, sudachi_config_path = (
+            load_configs()
+        )
     except Exception as e:
         log.error(f"設定の読み込みに失敗しました: {e}")
         sys.exit(1)
 
-    projects_path = get_file_path_from_config("projects.projects_json", "data/generated/projects.json")
-    output_path = get_file_path_from_config("bm25.tf_token", "data/generated/tf_token.json")
+    projects_path = get_file_path_from_config(
+        "projects.projects_json", "data/generated/projects.json"
+    )
+    output_path = get_file_path_from_config(
+        "bm25.tf_token", "data/generated/tf_token.json"
+    )
 
     try:
         with open(projects_path, encoding="utf-8") as f:
@@ -75,13 +83,13 @@ def main():
 
     tok, mode = build_tokenizer(sudachi_config_path)
 
-    results: List[tf_schema.Project] = []
+    results: List[tf_token.Project] = []
 
     for p in projects:
         project_id = p.get("projectId")
 
         # 各フィールドのトークン化とTF
-        field_objs: Dict[str, tf_schema.TfOfField] = {}
+        field_objs: Dict[str, tf_token.TfOfField] = {}
         doc_tf_counter: Counter = Counter()
         doc_token_set: set = set()
 
@@ -89,15 +97,15 @@ def main():
             text = p.get(field) or ""
             toks = tokenize(str(text), tok, mode, target_pos_l1, ban_list, stopwords)
             tf = Counter(toks)
-            field_objs[field] = tf_schema.TfOfField(len=len(toks), tf=dict(tf))
+            field_objs[field] = tf_token.TfOfField(len=len(toks), tf=dict(tf))
             doc_tf_counter.update(tf)
             doc_token_set.update(tf.keys())
 
         # スキーマ Fields へ詰める（未定義フィールドは長さ0/空dictで埋める）
-        def get_field(name: str) -> tf_schema.TfOfField:
-            return field_objs.get(name, tf_schema.TfOfField(len=0, tf={}))
+        def get_field(name: str) -> tf_token.TfOfField:
+            return field_objs.get(name, tf_token.TfOfField(len=0, tf={}))
 
-        fields_obj = tf_schema.Fields(
+        fields_obj = tf_token.Fields(
             title=get_field("title"),
             organization=get_field("organization"),
             reading=get_field("reading"),
@@ -106,7 +114,7 @@ def main():
             prCommentLong=get_field("prCommentLong"),
         )
 
-        project_entry = tf_schema.Project(
+        project_entry = tf_token.Project(
             projectId=project_id,
             fields=fields_obj,
             tf=dict(doc_tf_counter),