LUKEによる日本語感情分析（ソースコードと実行結果）

REM Python をシステム領域にインストール
winget install --scope machine --id Python.Python.3.12 -e --silent
REM Python のパス設定
set "PYTHON_PATH=C:\Program Files\Python312"
set "PYTHON_SCRIPTS_PATH=C:\Program Files\Python312\Scripts"
echo "%PATH%" | find /i "%PYTHON_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_SCRIPTS_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_SCRIPTS_PATH%" /M >nul

winget install --scope machine Codeium.Windsurf -e --silent

# プログラム名: LUKE日本語感情分析プログラム
# 特徴技術名: LUKE (Language Understanding with Knowledge-based Embeddings)
# 出典: Yamada, I., Asai, A., Shindo, H., Takeda, H., & Matsumoto, Y. (2020). LUKE: Deep Contextualized Entity Representations with Entity-aware Self-attention. In Proceedings of the 2020 Conference on Empirical Methods in Natural Language Processing (EMNLP), pages 6442–6454.
# 特徴機能: エンティティ認識型自己注意機構により、文中の単語とエンティティを独立したトークンとして扱い、それぞれの文脈を考慮した表現を出力する機能
# 学習済みモデル: Mizuiro-sakura/luke-japanese-large-sentiment-analysis-wrime - LUKEの日本語版をWRIMEデータセットで8感情分類にファインチューニングしたモデル
# 方式設計:
#   関連利用技術: Transformers（Hugging Face）、PyTorch、NumPy、matplotlib、japanize-matplotlib、ruptures
#   入力と出力: 入力: テキストファイル（tkinterでファイル選択）、出力: 感情分析結果をprint()で表示、matplotlibで感情変化グラフ表示、終了時result.txt保存
#   処理手順: テキスト読み込み→文章分割→感情分類→スコア計算→変化点検出→グラフ生成
#   前処理、後処理: テキスト正規化、トークン化、ソフトマックス確率変換
#   追加処理: 閾値処理による文章抽出、感情変化点検出処理 - rupturesライブラリによる統計的変化点検出
#   調整を必要とする設定値: 閾値設定（0.0-1.0）
# 将来方策: 変化点検出の感度を文章全体の感情変化の分散から動的に決定
# その他の重要事項: 8感情順番固定（喜び0、悲しみ1、期待2、驚き3、怒り4、恐れ5、嫌悪6、信頼7）、グラフは各文章の最大感情のみ表示
# 前準備: pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
# pip install transformers matplotlib japanize-matplotlib numpy ruptures

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import numpy as np
import tkinter as tk
from tkinter import filedialog
import matplotlib.pyplot as plt
import japanize_matplotlib
import ruptures as rpt
import re
from collections import Counter
import sys

# matplotlibの設定
plt.rcParams['figure.figsize'] = (20, 12)
plt.rcParams['font.size'] = 12
plt.rcParams['axes.grid'] = True
plt.rcParams['grid.alpha'] = 0.3

# 設定値
MAX_SEQ_LENGTH = 512  # トークナイザーの最大シーケンス長
BATCH_SIZE = 8        # バッチ処理のサイズ

# 感情の色定義（RGB形式、0-1の範囲）- カラーユニバーサルデザイン対応（CANONICAL順）
EMOTION_COLORS = [
    (0.9, 0.2, 0.2),    # 明るい赤（喜び）
    (0.2, 0.4, 0.8),    # 明るい青（悲しみ）
    (0.2, 0.8, 0.2),    # 明るい緑（期待）
    (0.0, 0.8, 0.8),    # シアン（驚き）
    (0.8, 0.0, 0.0),    # 濃い赤（怒り）
    (0.6, 0.2, 0.8),    # 紫（恐れ）
    (0.8, 0.6, 0.0),    # オレンジ（嫌悪）
    (0.3, 0.6, 1.0)     # 水色（信頼）
]

print('=' * 60)
print('LUKE日本語感情分析プログラム')
print('=' * 60)
print('\n【概要】')
print('本プログラムはLUKEモデルを使用してテキストの感情を8種類に分類する')
print('文章ごとに感情を分析し、感情の変化点を統計的に検出する')
print('\n【使用方法】')
print('1. テキストファイルを選択する')
print('2. 閾値を設定する（0.0-1.0、デフォルト: 0.1）')
print('3. 分析結果が表示される')
print('4. グラフウィンドウを閉じるとプログラムを終了する')
print('\n【出力ファイル】')
print('- emotion_graph.png: 感情変化グラフ')
print('- result.txt: 詳細な分析結果')
print('=' * 60 + '\n')

# GPU/CPU自動選択
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f'使用デバイス: {device}')

# モデルとトークナイザーの読み込み
print('LUKEモデルを読み込み中...')
try:
    tokenizer = AutoTokenizer.from_pretrained('Mizuiro-sakura/luke-japanese-large-sentiment-analysis-wrime')
    model = AutoModelForSequenceClassification.from_pretrained('Mizuiro-sakura/luke-japanese-large-sentiment-analysis-wrime')
    model = model.to(device)
    model.eval()  # 評価モードに設定
    print('モデル読み込み完了\n')
except Exception as e:
    print(f'モデルの読み込みに失敗しました: {e}')
    sys.exit(1)

# ラベル動的取得と写像の初期化（CANONICAL順に統一）
CANONICAL_LABELS = ['喜び', '悲しみ', '期待', '驚き', '怒り', '恐れ', '嫌悪', '信頼']

# 英日エイリアス
LABEL_ALIAS = {
    'joy': '喜び', '喜び': '喜び',
    'sadness': '悲しみ', '悲しみ': '悲しみ',
    'anticipation': '期待', 'expectation': '期待', '期待': '期待',
    'surprise': '驚き', '驚き': '驚き',
    'anger': '怒り', '怒り': '怒り',
    'fear': '恐れ', '恐れ': '恐れ',
    'disgust': '嫌悪', '嫌悪': '嫌悪',
    'trust': '信頼', '信頼': '信頼',
}

def _normalize_label(s: str) -> str:
    s = str(s).strip()
    ascii_only = all(ord(ch) < 128 for ch in s)
    return s.lower() if ascii_only else s

try:
    id2label = model.config.id2label
except Exception as e:
    print(f'警告: id2labelを取得できなかったため暫定対応を行う: {e}')
    id2label = {i: CANONICAL_LABELS[i] for i in range(len(CANONICAL_LABELS))}

# キー整列（intキー/数字文字列キーを考慮）
_keys = list(id2label.keys())
if all(isinstance(k, int) for k in _keys):
    index_keys = sorted(_keys)
elif all(isinstance(k, str) and k.isdigit() for k in _keys):
    index_keys = sorted(_keys, key=lambda k: int(k))
else:
    try:
        index_keys = sorted(_keys, key=lambda k: int(k))
    except Exception:
        index_keys = sorted(_keys, key=lambda k: str(k))

model_labels_raw = [id2label[k] for k in index_keys]

canon_label_to_idx = {lab: i for i, lab in enumerate(CANONICAL_LABELS)}
model_idx_to_canon_idx = []
unknown_labels = []

for i, raw in enumerate(model_labels_raw):
    key = _normalize_label(raw)
    if key not in LABEL_ALIAS:
        unknown_labels.append(str(raw))
        # 警告のみで続行。暫定として位置対応で割当
        mapped_idx = i if i < len(CANONICAL_LABELS) else (len(CANONICAL_LABELS) - 1)
    else:
        canon = LABEL_ALIAS[key]
        mapped_idx = canon_label_to_idx[canon]
    model_idx_to_canon_idx.append(mapped_idx)

# 逆写像（固定順→モデル順）
canon_idx_to_model_idx = [None] * len(CANONICAL_LABELS)
for m_idx, c_idx in enumerate(model_idx_to_canon_idx):
    if 0 <= c_idx < len(CANONICAL_LABELS) and (canon_idx_to_model_idx[c_idx] is None):
        canon_idx_to_model_idx[c_idx] = m_idx
# 未割当があればフォールバックで穴埋め
for j in range(len(CANONICAL_LABELS)):
    if canon_idx_to_model_idx[j] is None:
        canon_idx_to_model_idx[j] = j if j < len(model_labels_raw) else (len(model_labels_raw) - 1)

if unknown_labels:
    print(f'警告: 未知のラベルが検出されたため暫定写像で続行する: {unknown_labels}')

# 以降の表示用ラベル（固定順）
emotion_labels = CANONICAL_LABELS

# ファイル選択
root = tk.Tk()
root.withdraw()
file_path = filedialog.askopenfilename(title='テキストファイルを選択', filetypes=[('Text files', '*.txt')])
if not file_path:
    print('ファイルが選択されていない')
    sys.exit(0)

# テキスト読み込み
try:
    with open(file_path, 'r', encoding='utf-8') as f:
        text = f.read()
    print(f'ファイル読み込み完了: {file_path}')
except Exception as e:
    print(f'ファイルの読み込みに失敗しました: {e}')
    sys.exit(1)

# 文章分割（複数の文末記号に対応）
sentences = re.split('[。！？!?]+', text)
sentences = [s.strip() for s in sentences if s.strip()]
print(f'文章数: {len(sentences)}')

if len(sentences) == 0:
    print('分析可能な文章が見つからない')
    sys.exit(0)

# 閾値設定（デフォルト0.1）
threshold_input = input('\n閾値を入力（0.0-1.0、未入力の場合は0.1）: ').strip()
try:
    threshold = float(threshold_input) if threshold_input else 0.1
    threshold = max(0.0, min(1.0, threshold))  # クランプ処理
    if threshold_input and (float(threshold_input) < 0.0 or float(threshold_input) > 1.0):
        print(f'閾値を{threshold}に調整した')
except ValueError:
    threshold = 0.1
    print('無効な入力のため、閾値を0.1に設定した')

print(f'\n感情分析を開始する（閾値: {threshold}）')
print('-' * 60)

# 感情分析実行（バッチ処理対応）
results = []       # 各文章の確率ベクトル（モデル順）
max_emotions = []  # 各文章の最大感情のインデックス（CANONICAL順）
max_scores = []    # 各文章の最大感情のスコア

# バッチ処理
for batch_start in range(0, len(sentences), BATCH_SIZE):
    batch_end = min(batch_start + BATCH_SIZE, len(sentences))
    batch_sentences = sentences[batch_start:batch_end]

    # バッチトークン化
    tokens = tokenizer(batch_sentences, truncation=True, max_length=MAX_SEQ_LENGTH,
                       padding=True, return_tensors='pt')
    tokens = {k: v.to(device) for k, v in tokens.items()}

    # バッチ推論
    with torch.no_grad():
        outputs = model(**tokens)
        batch_scores = torch.softmax(outputs.logits, dim=-1).cpu().numpy()

    # 各文章の結果
    for i, (sentence, scores) in enumerate(zip(batch_sentences, batch_scores)):
        sentence_idx = batch_start + i
        results.append(scores)  # モデル順ベクトルを保持

        # 最大感情（モデル順→固定順に写像）
        model_max_idx = int(np.argmax(scores))
        max_score = float(scores[model_max_idx])
        canonical_idx = model_idx_to_canon_idx[model_max_idx]
        max_emotions.append(canonical_idx)
        max_scores.append(max_score)

        # 結果表示
        print(f'\n文章{sentence_idx + 1}: {sentence}')

        # 閾値以上を明示（固定順で表示）
        scores_by_canon = [float(scores[canon_idx_to_model_idx[j]]) for j in range(len(emotion_labels))]
        threshold_exceeded = []
        for j, (emotion, score) in enumerate(zip(emotion_labels, scores_by_canon)):
            if score >= threshold:
                print(f'  {emotion}: {score:.3f} ★')
                threshold_exceeded.append(f'{emotion}({score:.3f})')
            else:
                print(f'  {emotion}: {score:.3f}')

        if threshold_exceeded:
            print(f'  → 閾値{threshold}以上: {", ".join(threshold_exceeded)}')

def detect_emotion_change_points_statistical(signal, max_scores):
    """PELTアルゴリズム（model="l2"）による変化点検出。
    入力信号は各文章の8次元確率ベクトル（results）である。
    max_scoresは説明用統計出力に利用する。"""
    if len(max_scores) < 3:
        return []

    # signal: shape = (n_samples, 8) を前提
    print(f'信号の形状: {signal.shape}')
    print(f'最初の文章の8感情: {signal[0]}')

    n_samples = len(max_scores)
    # ペナルティ設定：2*log(n_samples) を8で割り、8次元信号に対するスケール調整とする
    penalty = (2 * np.log(n_samples)) / 8

    print('\n変化点検出パラメータ:')
    print(f'  文章数: {n_samples}')
    print(f'  ペナルティ設定: 2*log({n_samples})/8 = {penalty:.3f}')

    print('\n文章ごとの最大感情スコア:')
    for i, score in enumerate(max_scores):
        print(f'  文章{i+1}: {score:.3f}')

    print('\n隣接文章間の最大スコア差分:')
    for i in range(len(max_scores) - 1):
        change = abs(max_scores[i+1] - max_scores[i])
        print(f'  文章{i+1}→{i+2}: {change:.3f}')

    if len(max_scores) > 1:
        max_change = max(abs(max_scores[i+1] - max_scores[i]) for i in range(len(max_scores) - 1))
        print(f'\n最大スコア差分: {max_change:.3f}')
        print(f'使用ペナルティ値: {penalty:.3f}')

    change_points = []
    try:
        algo = rpt.Pelt(model="l2").fit(signal)
        change_points_raw = algo.predict(pen=penalty)
        change_points = [cp for cp in change_points_raw[:-1]]

        print(f'  検出された変化点数: {len(change_points)}個')
        if len(change_points) > 0 and len(change_points) <= 30:
            print(f'  変化点位置（文章番号）: {change_points}')
    except Exception as e:
        print(f'  変化点検出エラー: {e}')

    return change_points


print('\n' + '=' * 60)
print('感情変化の統計分析')
print('=' * 60)

# 感情変化点の検出
signal = np.array(results)  # モデル順の生配列
change_points = detect_emotion_change_points_statistical(signal, max_scores)

# 基本統計量の表示
print(f'\n【感情スコアの統計】')
print(f'  平均スコア: {np.mean(max_scores):.3f}')
print(f'  標準偏差: {np.std(max_scores):.3f}')
print(f'  最小スコア: {np.min(max_scores):.3f}')
print(f'  最大スコア: {np.max(max_scores):.3f}')

# 最頻感情の計算（固定順インデックス）
emotion_counts = Counter(max_emotions)
max_count = max(emotion_counts.values()) if emotion_counts else 0
most_common_emotions = [emotion_labels[i] for i in range(len(emotion_labels))
                        if emotion_counts.get(i, 0) == max_count and max_count > 0]

print(f'\n【最頻感情】')
if len(most_common_emotions) == 1:
    print(f'  {most_common_emotions[0]} ({max_count}回)')
else:
    if max_count > 0:
        print(f'  {", ".join(most_common_emotions)} (各{max_count}回、同数)')
    else:
        print('  なし')

# 感情の分布を表示（0回も表示、固定順）
print(f'\n【感情の分布】')
for i, lab in enumerate(emotion_labels):
    count = emotion_counts.get(i, 0)
    percentage = (count / len(sentences)) * 100
    print(f'  {lab}: {count}回 ({percentage:.1f}%)')

# matplotlibでグラフ生成
print('\n' + '=' * 60)
print('感情変化グラフを生成している...')

fig, ax = plt.subplots(figsize=(20, 12))

# 最大感情の折れ線グラフを描画
x_values = list(range(1, len(sentences) + 1))

# 各区間を異なる色で描画（CANONICAL順インデックスに基づく）
for i in range(len(sentences)):
    if i < len(sentences) - 1:
        ax.plot([x_values[i], x_values[i+1]],
                [max_scores[i], max_scores[i+1]],
                color=EMOTION_COLORS[max_emotions[i]],
                linewidth=2.5,
                marker='o',
                markersize=8,
                markerfacecolor=EMOTION_COLORS[max_emotions[i]],
                markeredgecolor='white',
                markeredgewidth=1.5)
    else:
        # 最後の点
        ax.plot(x_values[i], max_scores[i],
                marker='o',
                markersize=8,
                color=EMOTION_COLORS[max_emotions[i]],
                markerfacecolor=EMOTION_COLORS[max_emotions[i]],
                markeredgecolor='white',
                markeredgewidth=1.5)

# 変化点を描画
if len(change_points) > 0:
    for cp in change_points:
        ax.axvline(x=cp, color='gray', linestyle='--', linewidth=2.5, alpha=0.7)
        ax.text(cp, ax.get_ylim()[1] * 0.95, f'変化点{cp}',
                color='gray', fontsize=10, ha='center', fontweight='bold')
    print(f'グラフに{len(change_points)}個の変化点を描画した')

# 感情ラベルを点の近くに表示（文章数が少ない場合）
if len(sentences) <= 30:
    for i in range(len(sentences)):
        ax.annotate(emotion_labels[max_emotions[i]],
                    (x_values[i], max_scores[i]),
                    textcoords="offset points",
                    xytext=(0, 10),
                    ha='center',
                    fontsize=9,
                    color=EMOTION_COLORS[max_emotions[i]])

# 軸の設定
ax.set_xlabel('文章番号', fontsize=14)
ax.set_ylabel('感情スコア', fontsize=14)
ax.set_title('感情変化グラフ - 文章ごとの最大感情', fontsize=18, pad=20)

# Y軸の範囲を0-1に設定
ax.set_ylim(0, 1)

# X軸の目盛り設定
if len(sentences) <= 50:
    ax.set_xticks(range(1, len(sentences) + 1, max(1, len(sentences) // 20)))
else:
    ax.set_xticks(range(1, len(sentences) + 1, max(1, len(sentences) // 10)))

# グリッドの設定
ax.grid(True, alpha=0.3, linestyle='-', linewidth=0.5)

# 凡例の作成（CANONICAL順）
from matplotlib.patches import Patch
legend_elements = [Patch(facecolor=EMOTION_COLORS[i], label=f'{emotion_labels[i]}')
                   for i in range(len(emotion_labels))]
ax.legend(handles=legend_elements, loc='upper right', fontsize=11,
          framealpha=0.9, ncol=2)

# 統計情報を表示
stats_text = f'文章数: {len(sentences)}\n'
stats_text += f'変化点数: {len(change_points)}\n'
stats_text += f'平均スコア: {np.mean(max_scores):.3f}\n'
stats_text += f'標準偏差: {np.std(max_scores):.3f}'

if len(most_common_emotions) == 1:
    stats_text += f'\n最頻感情: {most_common_emotions[0]}'
else:
    if most_common_emotions:
        stats_text += f'\n最頻感情: {", ".join(most_common_emotions)}'
    else:
        stats_text += '\n最頻感情: なし'

ax.text(0.02, 0.98, stats_text, transform=ax.transAxes,
        fontsize=10, verticalalignment='top',
        bbox=dict(boxstyle='round', facecolor='white', alpha=0.8))

# レイアウトの調整
plt.tight_layout()

# グラフの保存
try:
    plt.savefig('emotion_graph.png', dpi=150, bbox_inches='tight')
    print('emotion_graph.pngに保存した')
except Exception as e:
    print(f'グラフの保存に失敗した: {e}')

# グラフの表示
plt.show()

# 結果保存
save_success = False
try:
    with open('result.txt', 'w', encoding='utf-8') as f:
        f.write('LUKE日本語感情分析結果\n')
        f.write('=' * 60 + '\n')
        f.write(f'分析ファイル: {file_path}\n')
        f.write(f'文章数: {len(sentences)}\n')
        f.write(f'閾値: {threshold}\n')
        f.write(f'検出された変化点数: {len(change_points)}\n')

        if len(change_points) > 0 and len(change_points) <= 30:
            f.write(f'変化点位置（文章番号）: {change_points}\n')

        f.write(f'\n【統計情報】\n')
        f.write(f'平均感情スコア: {np.mean(max_scores):.3f}\n')
        f.write(f'標準偏差: {np.std(max_scores):.3f}\n')
        f.write(f'最小スコア: {np.min(max_scores):.3f}\n')
        f.write(f'最大スコア: {np.max(max_scores):.3f}\n')

        if len(most_common_emotions) == 1:
            f.write(f'最頻感情: {most_common_emotions[0]}\n')
        else:
            if most_common_emotions:
                f.write(f'最頻感情: {", ".join(most_common_emotions)}（同数）\n')
            else:
                f.write('最頻感情: なし\n')

        f.write(f'\n【感情の分布】\n')
        for i, lab in enumerate(emotion_labels):
            count = emotion_counts.get(i, 0)
            percentage = (count / len(sentences)) * 100
            f.write(f'{lab}: {count}回 ({percentage:.1f}%)\n')

        f.write(f'\n【詳細な分析結果】\n')
        f.write('=' * 60 + '\n')

        for i, (sentence, scores) in enumerate(zip(sentences, results)):
            f.write(f'\n文章{i+1}: {sentence}\n')
            scores_by_canon = [float(scores[canon_idx_to_model_idx[j]]) for j in range(len(emotion_labels))]
            for emotion, score in zip(emotion_labels, scores_by_canon):
                if score >= threshold:
                    f.write(f'  {emotion}: {score:.3f} ★\n')
                else:
                    f.write(f'  {emotion}: {score:.3f}\n')

    save_success = True
    print('result.txtに保存した')
except Exception as e:
    print(f'結果ファイルの保存に失敗した: {e}')

print('\n' + '=' * 60)
if save_success:
    print('プログラムを正常終了する')
    print('出力ファイル:')
    print('  - emotion_graph.png (感情変化グラフ)')
    print('  - result.txt (詳細な分析結果)')
else:
    print('結果ファイルの保存に失敗したが、分析は完了した')
print('=' * 60)