report-review

integrate_data.py (5568B)

---

 import glob
 import json
 import os
 from typing import Any, Dict, List, Tuple
 
 
 def integrate_and_format_data(data_pairs: List[Dict[str, str]], output_file_path: str):
     """
     Typstファイルと構造化JSONファイルを統合し、学習用のJSON Lines形式に変換する。
 
     Args:
         data_pairs: TypstファイルパスとJSONファイルパスの辞書のリスト。
         output_file_path: 出力するJSON Linesファイルパス。
     """
 
     formatted_data = []
     processed_count = 0
 
     for pair in data_pairs:
         typst_path = pair["typst_file"]
         json_path = pair["json_file"]
 
         if not os.path.exists(typst_path) or not os.path.exists(json_path):
             print(
                 f"警告: ファイルペアが見つかりません - Typst: {typst_path}, JSON: {json_path}"
             )
             continue
 
         try:
             # 1. Typstファイルの内容を読み込む
             with open(typst_path, "r", encoding="utf-8") as f:
                 typst_content = f.read()
 
             # 2. 構造化されたJSONのフィードバックリストを読み込む
             with open(json_path, "r", encoding="utf-8") as f:
                 feedback_items = json.load(f)
 
             # 3. フィードバックリストをJSON文字列に変換
             #    改行やスペースを除去し、整形された単一のJSON文字列にする
             feedback_output = json.dumps(
                 feedback_items, ensure_ascii=False, separators=(",", ":")
             )
 
             # 4. LLM学習用のInstruction/Outputペアを作成
             learning_example = {"instruction": typst_content, "output": feedback_output}
 
             formatted_data.append(learning_example)
             processed_count += 1
             print(f"統合完了 ({processed_count}件目): {typst_path} と {json_path}")
 
         except Exception as e:
             print(f"エラー処理: {typst_path} / {json_path} - {e}")
 
     # 5. すべてのデータをJSON Lines形式でファイルに書き出す
     with open(output_file_path, "w", encoding="utf-8") as outfile:
         for entry in formatted_data:
             json.dump(entry, outfile, ensure_ascii=False)
             outfile.write("\n")
 
     print("\n--- 処理完了 ---")
     print(
         f"合計 {len(formatted_data)} 件の学習データが {output_file_path} に出力されました。"
     )
 
 
 def find_file_pairs(
     reports_dir: str, structured_dir: str, typst_base: str, json_base: str
 ) -> List[Dict[str, str]]:
     """
     構造化JSONファイルと連番に基づいて、対応するTypstファイルを検索し、ペアのリストを返す。
     """
     data_pairs = []
     # 例: structured-feedback/structured_data_*.json にマッチ
     search_pattern = os.path.join(structured_dir, f"{json_base}_*.json")
     json_files = glob.glob(search_pattern)
 
     for json_file_path in json_files:
         try:
             # structured_data_NNN.json から NNN を抽出
             base_filename = os.path.basename(json_file_path)
             parts = base_filename.replace(f"{json_base}_", "").split(".")
             if len(parts) < 2 or not parts[0].isdigit():
                 continue
 
             sequence_num_str = parts[0]
 
             # 対応するTypstファイル名を構築 (例: reports/report001.typ)
             typst_file_name = f"{typst_base}{sequence_num_str}.typ"
             typst_file_path = os.path.join(reports_dir, typst_file_name)
 
             if os.path.exists(typst_file_path):
                 data_pairs.append(
                     {
                         "typst_file": typst_file_path,
                         "json_file": json_file_path,
                     }
                 )
             else:
                 print(
                     f"警告: JSONファイル {base_filename} に対応するTypstファイル "
                     f"({typst_file_name}) が {reports_dir} に見つかりませんでした。スキップします。"
                 )
         except Exception as e:
             print(f"ファイルペアリング中にエラーが発生しました ({json_file_path}): {e}")
 
     return data_pairs
 
 
 if __name__ == "__main__":
     # === 設定箇所 (ディレクトリとファイル名のベース) ===
 
     # Typstファイルが格納されているディレクトリ
     REPORTS_DIR = "reports"
     # 構造化JSONファイルが格納されているディレクトリ
     STRUCTURED_DIR = "structured-feedback"
     # Typstファイルのベース名 (例: report001.typ の 'report')
     TYPST_BASE_NAME = "report"
     # 構造化JSONファイルのベース名 (例: structured_data_001.json の 'structured_data')
     JSON_BASE_NAME = "structured_data"
 
     OUTPUT_FILE_NAME = "typst_feedback_dataset.jsonl"
 
     # === スクリプト実行 ===
 
     # 構造化JSONファイルに基づいて、対応するTypstファイルを自動的に検索・ペアリング
     file_pairs = find_file_pairs(
         reports_dir=REPORTS_DIR,
         structured_dir=STRUCTURED_DIR,
         typst_base=TYPST_BASE_NAME,
         json_base=JSON_BASE_NAME,
     )
 
     if not file_pairs:
         print(
             "エラー: 処理対象のファイルペアが見つかりませんでした。ディレクトリとファイル名を確認してください。"
         )
     else:
         integrate_and_format_data(file_pairs, OUTPUT_FILE_NAME)