import chardet import os import spacy from collections import defaultdict import xml.sax.saxutils import pykakasi import csv import re import pandas as pd # --- 設定 --- # テキストファイルが置かれているディレクトリ名 INPUT_DIR = "既存のファイルへのパスを教えて" # 解析結果を出力するXMLファイル名 OUTPUT_XML = "新しいファイルへのパスを考えて/analysis_results_V2.xml" # ファイル名とメタデータ（話者情報など）の対応表CSVファイル CORRESPONDENCE_CSV = "対応表へのパスを教えて/Correspondence.xlsx" def analyze_and_generate_xml(): """ downloadフォルダ内のテキストファイルを解析し、指定のXMLフォーマットで出力するメイン関数 """ print("spaCyとGiNZAモデルを読み込んでいます...") try: nlp = spacy.load('ja_ginza') except OSError: print("エラー: 'ja_ginza'モデルが見つかりません。") print("次のコマンドでインストールしてください: python -m spacy download ja_ginza") return print("モデルの読み込みが完了しました。") # --- ここから変更 (PandasでExcel対応表の読み込み) --- print(f"対応表Excel '{CORRESPONDENCE_CSV}' を読み込んでいます...") metadata_map = {} # (Task, textID) をキーとする辞書 try: # sheet_name=0 で「1枚目のシート」を指定 # dtype=str ですべての列を「文字列」として読み込む (重要: Task, textIDが数値にならないように) # keep_default_na=False で空セルを '' (空文字) として読み込む (NaN になるのを防ぐ) df = pd.read_excel(CORRESPONDENCE_CSV, sheet_name=0, dtype=str, keep_default_na=False) # 1行ずつ辞書に変換して処理 for index, row in df.iterrows(): # Task列やtextID列が空の行はスキップ if not row.get('Task') or not row.get('textID'): continue key = (row['Task'], row['textID']) # '0' のような値も保持 metadata_map[key] = { "Task": row.get('Task', ''), "textID": row.get('textID', ''), "personID": row.get('personID', ''), "personAge": row.get('personAge', ''), "language": row.get('language', ''), "personFirstLang": row.get('personFirstLang', ''), "personSecondLang": row.get('personSecondLang', '') } print(f"対応表データを {len(metadata_map)} 件読み込みました。") except FileNotFoundError: print(f"エラー: 対応表Excel '{CORRESPONDENCE_CSV}' が見つかりません。") print("スクリプト上部の CORRESPONDENCE_CSV のパス設定を確認してください。") return except Exception as e: print(f"エラー: Excelファイルの読み込みに失敗しました: {e}") print("ライブラリ 'pandas' と 'openpyxl' が正しくインストールされているか確認してください。") return #--- ローマ字変換 --- print("ローマ字変換器を初期化しています...") # 初期化（古いAPIの警告が出るかもしれませんが、これが正しい方法です） kks = pykakasi.kakasi() print("初期化が完了しました。") # --- ここまで追加 --- # 入力ディレクトリが存在しない場合は作成 if not os.path.exists(INPUT_DIR): print(f"'{INPUT_DIR}' ディレクトリを作成しました。解析したい.txtファイルをこの中に入れてください。") os.makedirs(INPUT_DIR) return files_to_process = [f for f in os.listdir(INPUT_DIR) if f.endswith(".txt")] if not files_to_process: print(f"警告: '{INPUT_DIR}' フォルダ内に解析対象の.txtファイルが見つかりません。") return # XMLを文字列として構築するためのリスト output_lines = ['', ''] print(f"{len(files_to_process)}個のファイルを処理します...") for filename in files_to_process: print(f" - ファイル '{filename}' を解析中...") filepath = os.path.join(INPUT_DIR, filename) match = re.search(r'Task(\d+)ID(\d+)\.txt', filename, re.IGNORECASE) metadata = None if match: # Task092 -> 92, ID00001 -> 1 のように正規化 task_id_str = str(int(match.group(1))) text_id_str = str(int(match.group(2))) key = (task_id_str, text_id_str) metadata = metadata_map.get(key) if metadata: print(f" -> メタデータを発見 (Task={task_id_str}, textID={text_id_str})") else: print(f" -> 警告: '{CORRESPONDENCE_CSV}' 内に対応するメタデータが見つかりません (Key: {key})") else: print(f" -> 警告: ファイル名 '{filename}' の形式が 'Task...ID....txt' ではありません。メタデータを紐付けできません。") try: # 1. ファイルを「バイナリモード(rb)」で読み込む with open(filepath, "rb") as f: raw_data = f.read() # 2. chardetでエンコーディングを推定 result = chardet.detect(raw_data) encoding = result['encoding'] if not encoding: raise Exception("エンコーディングを検出できませんでした。") # 3. 推定したエンコーディングでデコード（テキストに変換） text = raw_data.decode(encoding) except Exception as e: print(f" エラー: ファイル '{filename}' の読み込みに失敗しました: {e}") continue doc = nlp(text) # タグを追加 output_lines.append(f' ') # 文ごとにループ for sent_i, sent in enumerate(doc.sents, 1): # タグを追加 (インデント: 12スペース) sent_attrs = [f'sentId="{sent_i}"'] if metadata: # CSVから読み込んだ属性を追加 (安全のため quoteattr を使用) sent_attrs.append(f'Task={xml.sax.saxutils.quoteattr(metadata["Task"])}') sent_attrs.append(f'textID={xml.sax.saxutils.quoteattr(metadata["textID"])}') sent_attrs.append(f'personID={xml.sax.saxutils.quoteattr(metadata["personID"])}') sent_attrs.append(f'personAge={xml.sax.saxutils.quoteattr(metadata["personAge"])}') sent_attrs.append(f'language={xml.sax.saxutils.quoteattr(metadata["language"])}') sent_attrs.append(f'personFirstLang={xml.sax.saxutils.quoteattr(metadata["personFirstLang"])}') sent_attrs.append(f'personSecondLang={xml.sax.saxutils.quoteattr(metadata["personSecondLang"])}') attr_string = " ".join(sent_attrs) output_lines.append(f' ') # --- ▲ここまで変更▲ --- # --- udToを計算するための事前処理 --- # 1. 各単語がどの単語から係られているか（子要素の一覧）をマップに保存 children_map = defaultdict(list) for token in sent: # ROOT(根)以外の場合、親トークンに自分(子)を追加 if token.head is not token: children_map[token.head.i].append(token.i) # 2. 文内でのインデックスを管理するためのマップ (doc全体index -> 文中index) # wordIdは1から始まる token_map = {token.i: idx for idx, token in enumerate(sent, 1)} # 単語（トークン）ごとにループ (インデント: 12スペース) for token in sent: word_id = token_map[token.i] # udFrom: 親のwordIdを取得 # ROOTの場合は自分自身を指す head_id = token_map[token.head.i] # udTo: 子のwordIdリストを取得し、'/'で連結 child_indices = children_map.get(token.i, []) child_word_ids = [str(token_map[child_idx]) for child_idx in sorted(child_indices)] ud_to = "/".join(child_word_ids) # ローマ字に変換 (新しいAPI) # kks.convert() は辞書のリストを返します (例: [{'orig': '...','hepburn': '...'}]) result = kks.convert(token.text) # 各辞書から 'hepburn' の値を取り出し、連結して一つの文字列にします romaji_text = "".join([item['hepburn'] for item in result]) # 属性を辞書で定義 attrs = { "wordId": str(word_id), "pos": token.pos_, "lemma": token.lemma_, "tag": token.tag_, "udType": token.dep_, "udFrom": str(head_id), "udTo": ud_to, "original": token.text } # 属性を "key=value" の文字列に変換 attr_str = " ".join(f'{k}={xml.sax.saxutils.quoteattr(v)}' for k, v in attrs.items()) # タグを生成 escaped_romaji = xml.sax.saxutils.escape(romaji_text) # ローマ字をエスケープ output_lines.append(f' {escaped_romaji}') # タグの中身をローマ字に変更 # は for token ループの外 (インデント: 12スペース) output_lines.append(' ') # は for sent ループの外 (インデント: 8スペース) output_lines.append(' ') # は for filename ループの外 (インデント: 4スペース) output_lines.append('') # 最終的なXML文字列をファイルに書き出す try: with open(OUTPUT_XML, "w", encoding="utf-8") as f: f.write("\n".join(output_lines)) print(f"\n解析が完了し、結果を '{OUTPUT_XML}' に保存しました。") except Exception as e: print(f"エラー: XMLファイルの書き込みに失敗しました: {e}") if __name__ == "__main__": analyze_and_generate_xml()