感情分析（かんじょうぶんせき、英: sentiment analysis）は、オピニオンマイニング（英: opinion mining）や感情AI（英: emotion AI）とも呼ばれ、自然言語処理、テキスト解析、計算言語学、バイオメトリクス (en:英語版) などを使用して、感情状態や主観的情報を体系的に識別、抽出、定量化、探究する技術である。

感情分析 - Wikipedia

感情分析は、マーケティングから顧客サービス、臨床医学に至るまで、さまざまな用途で、レビュー（英語版）やアンケート回答などの顧客の声、オンラインメディアやソーシャルメディアのコンテンツ、ヘルスケア情報などの分析に利用されている。

感情分析 - Wikipedia

RoBERTaのような深層言語モデルの登場により、たとえば記者が暗黙のうちに感情を表現することが多いニューステキストなど、より困難なデータ領域も分析できるようになった。

感情分析 - Wikipedia

⇧ とありますと。

用途が幅広いですな...

Pythonで感情分析（Sentiment Analysis）を実現するためのライブラリにはどんなものがあるのか

ChatGPTに質問したところ、

⇧ 上記のような回答が返ってきた。

ライブラリの数としては、12個の候補が提案されましたと。

No	ライブラリ	カテゴリ	ライセンス	初出年
1	VADER	機械学習（教師あり）	MIT	2014年
2	TextBlob	機械学習（教師あり）	MIT	2008年
3	Transformers (BERT, RoBERTa)	深層学習	Apache 2.0	2018年
4	Pattern	機械学習（教師あり）	MIT	2009年
5	spaCy	機械学習（教師なし）	MIT	2015年
6	DeepMoji	深層学習	MIT	2017年
7	Stanford NLP	機械学習（教師あり）	Apache 2.0	2006年
8	AllenNLP	深層学習	Apache 2.0	2017年
9	GloVe	機械学習（教師なし）	MIT	2014年
10	FastText	機械学習（教師なし）	MIT	2016年
11	TextCNN	深層学習	MIT	2014年
12	Hugging Face Datasets	深層学習	MIT	2018年

選定の基準が分からん...

「scikit-learn」とかが含まれていないのが気になりますな...

ちなみに、「RoBERTa」については、

qiita.com

⇧ 上記サイト様が詳しいです。

Pythonで感情分析（Sentiment Analysis）で映画の台詞のTOP100選出を試みるものの...

前回、

ts0818.hatenablog.com

⇧「IMSDb（Internet Movie Script Database）」というサイトで公開されている洋画の台詞を取得しましたと。

で、

アメリカ映画の名セリフベスト100（アメリカえいがのめいセリフベスト100、AFI's 100 Years...100 Movie Quotes、AFIの百年…映画百の名台詞）は、アメリカン・フィルム・インスティチュート（AFI）が「AFIアメリカ映画100年シリーズ」の一環として選出したアメリカ合衆国の映画の100の名セリフの一覧である。

アメリカ映画の名セリフベスト100 - Wikipedia

⇧ みたいなことをAIで実現させたいなと。

ChatGPTに質問して、ソースコードを生成してみました。

自分のPCの「メモリ」が貧弱で、「WSL 2（Windows Subsystem for Linux 2）」の「仮想マシン」では、「OOMK（Out of Memory Killer）」の問題が起きて処理に失敗するので、「Google Colaboratory」で試してみる。

「Google Colaboratory」の無料版ですら、「メモリ」が12GBということで、自分のPCの「メモリ」を越えていたので...

残念ながら、「Google Colaboratory」の「スペック」でも処理し切れないっぽく、動作検証で失敗する...

長時間待たせておいて、駄目でしたパターンが辛過ぎるんだが...

■/content/drive/MyDrive/Colab Notebooks/movie_script_analyzer.py

import os
import re
import time
import gc
import logging
from typing import Optional
from sklearn.feature_extraction.text import TfidfVectorizer
from google.colab import drive
from nltk.sentiment.vader import SentimentIntensityAnalyzer  # VADERのインポート
import nltk

# VADERのリソース（vader_lexicon）のダウンロード
nltk.download('vader_lexicon')

# ログの設定
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

class MovieScriptAnalyzer:
    def __init__(self, scripts_dir: str, output_file: Optional[str] = None) -> None:
        self.scripts_dir = scripts_dir
        self.output_file = output_file
        self.vectorizer = TfidfVectorizer(max_features=1000)  # TF-IDFのベクトライザーを一度だけ定義
        self.sid = SentimentIntensityAnalyzer()  # VADERインスタンスを初期化

    def preprocess_script(self, file_path: str) -> list[str]:
        """映画の台詞ファイルを前処理する"""
        start_time = time.time()  # 計測開始
        with open(file_path, encoding="utf-8") as f:
            script = f.read()

        # 改行で分割し、不要な行（空行やキャラクター名）を除去
        sentences = script.split("\n")
        processed_sentences = []
        current_sentence = []
        
        for sentence in sentences:
            sentence = sentence.strip()  # 前後の空白を除去
            
            if not sentence:
                # 空行はスキップ
                continue
            elif re.match(r'^[A-Za-z0-9]+:.*', sentence):
                # キャラクター名（例えば "JOHN:"）に続く台詞
                if current_sentence:
                    processed_sentences.append(" ".join(current_sentence))  # 現在の台詞を追加
                    current_sentence = []  # 現在の台詞をリセット
                current_sentence.append(sentence)  # 新しい台詞開始
            else:
                # 台詞の続き
                current_sentence.append(sentence)
        
        if current_sentence:
            processed_sentences.append(" ".join(current_sentence))  # 最後の台詞を追加
        
        end_time = time.time()  # 計測終了
        logging.info(f"Preprocessing {file_path} took {end_time - start_time:.2f} seconds.")
        return processed_sentences

    def extract_features(self, sentences: list[str]) -> tuple:
        """TF-IDF特徴量を抽出する"""
        start_time = time.time()  # 計測開始
        X = self.vectorizer.fit_transform(sentences)  # TF-IDF特徴量を抽出
        end_time = time.time()  # 計測終了
        logging.info(f"Feature extraction took {end_time - start_time:.2f} seconds.")
        return X

    def analyze_sentiment(self, sentences: list[str]) -> list[float]:
        """各台詞の感情分析を実施する"""
        sentiment_scores = []
        for sentence in sentences:
            score = self.sid.polarity_scores(sentence)  # VADERで感情分析
            sentiment_scores.append(score['compound'])  # compoundスコア（ポジティブ・ネガティブの全体的な評価）
        return sentiment_scores

    def rank_sentences(self, sentences: list[str], X: tuple, sentiment_scores: list[float], file_name: str) -> list[tuple[str, str, float, str]]:
        """感情スコアと重要度で台詞をランク付け"""
        start_time = time.time()  # 計測開始
        ranked_sentences = []
        
        for sentence, score, sentiment in zip(sentences, X.sum(axis=1), sentiment_scores):
            ranked_sentences.append((sentence, score.sum(), sentiment, file_name))  # ファイル名も追加
        
        # スコア順に並べ替え（降順）
        ranked_sentences.sort(key=lambda x: x[1], reverse=True)
        
        end_time = time.time()  # 計測終了
        logging.info(f"Ranking sentences took {end_time - start_time:.2f} seconds.")
        return ranked_sentences

    def analyze_scripts(self) -> list[tuple[str, str, float, str]]:
        """ディレクトリ内の全ての映画台詞を処理し、全体のTop 100を選出"""
        heap_size = 100  # 保持する最大の台詞数
        movie_sentences = {}  # 映画ごとの台詞リストを格納する辞書

        start_time = time.time()  # 計測開始
        # すべてのスクリプトファイルを一度に学習するのではなく、1つずつ処理
        for file_name in os.listdir(self.scripts_dir):
            file_path = os.path.join(self.scripts_dir, file_name)
            if os.path.isfile(file_path):
                processed_script = self.preprocess_script(file_path)
                
                # 感情分析を実施
                sentiment_scores = self.analyze_sentiment(processed_script)
                
                # 特徴量を抽出
                X = self.extract_features(processed_script)
                
                # ランク付け
                ranked_sentences = self.rank_sentences(processed_script, X, sentiment_scores, file_name)
                
                # 映画ごとの台詞を格納
                movie_sentences[file_name] = ranked_sentences

                # ガベージコレクションを呼び出して、メモリを解放
                gc.collect()

        # すべての映画から台詞を集め、ランク付けされた台詞を1つのリストにまとめる
        all_ranked_sentences = []
        for movie, sentences in movie_sentences.items():
            all_ranked_sentences.extend(sentences)

        # スコア順に並べ替える（降順）
        all_ranked_sentences.sort(key=lambda x: x[1], reverse=True)

        # Top 100を選出
        top_sentences = all_ranked_sentences[:heap_size]

        end_time = time.time()  # 計測終了
        logging.info(f"Analyzing scripts took {end_time - start_time:.2f} seconds.")
        return top_sentences

    def save_results(self, ranked_sentences: list[tuple[str, str, float, str]]) -> None:
        """結果をファイルに保存"""
        start_time = time.time()  # 計測開始
        if self.output_file:
            with open(self.output_file, "w", encoding="utf-8") as f:
                for idx, (sentence, score, sentiment, file_name) in enumerate(ranked_sentences, 1):
                    f.write(f"{idx}. {file_name}: {sentence} (Score: {score}, Sentiment: {sentiment})\n")
        end_time = time.time()  # 計測終了
        logging.info(f"Saving results took {end_time - start_time:.2f} seconds.")

    def execute(self) -> list[tuple[str, str, float, str]]:
        """解析を実行し、結果を保存する"""
        start_time = time.time()  # 計測開始
        ranked_sentences = self.analyze_scripts()
        if self.output_file:
            self.save_results(ranked_sentences)
        end_time = time.time()  # 計測終了
        logging.info(f"Execution took {end_time - start_time:.2f} seconds.")
        return ranked_sentences


# Googleドライブのマウント
drive.mount('/content/drive')

# 使用例
scripts_dir = "/content/drive/MyDrive/Colab Notebooks/movie/input/movie_scripts"  # Googleドライブ上の映画スクリプトが保存されているディレクトリ
output_file = "/content/drive/MyDrive/Colab Notebooks/movie/output/top_100_sentences.txt"  # 出力ファイル

# MovieScriptAnalyzerインスタンスの作成
analyzer = MovieScriptAnalyzer(scripts_dir, output_file)

# 解析を実行して、Top 100の台詞を取得
top_100_sentences = analyzer.execute()

# 上位100フレーズを表示
for idx, (sentence, score, sentiment, file_name) in enumerate(top_100_sentences, 1):
    print(f"{idx}. {file_name}: {sentence} (Score: {score}, Sentiment: {sentiment})")

結果

上手いこと「台詞」を選出できず。

次に「ModernBERT」を利用した実装をChatGPTにお願いしてみた。

■/content/drive/MyDrive/Colab Notebooks/movie/movie_script_analyzer_revised.py

import os
import re
import time
import gc
import logging
from typing import Optional
from google.colab import drive
import torch
from transformers import BertTokenizer, BertModel
from nltk.sentiment.vader import SentimentIntensityAnalyzer  # VADERのインポート
import nltk

# VADERのリソース（vader_lexicon）のダウンロード
nltk.download('vader_lexicon')

# ログの設定
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

class MovieScriptAnalyzer:
    def __init__(self, scripts_dir: str, output_file: Optional[str] = None) -> None:
        self.scripts_dir = scripts_dir
        self.output_file = output_file
        # Hugging Faceの認証をスキップ
        self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased', use_auth_token=None)  # BERTのトークナイザー
        self.model = BertModel.from_pretrained('bert-base-uncased', use_auth_token=None)  # BERTのモデル
        self.sid = SentimentIntensityAnalyzer()  # VADERインスタンスを初期化

    def preprocess_script(self, file_path: str) -> list[str]:
        """映画の台詞ファイルを前処理する"""
        start_time = time.time()  # 計測開始
        with open(file_path, encoding="utf-8") as f:
            script = f.read()

        sentences = script.split("\n")
        processed_sentences = []
        for sentence in sentences:
            sentence = sentence.lower()  # 小文字化
            sentence = re.sub(r"[^a-z\s]", "", sentence)  # 特殊文字の除去
            processed_sentences.append(sentence)

        end_time = time.time()  # 計測終了
        logging.info(f"Preprocessing {file_path} took {end_time - start_time:.2f} seconds.")
        return processed_sentences

    def extract_features(self, sentences: list[str]) -> torch.Tensor:
        """BERT特徴量を抽出する"""
        start_time = time.time()  # 計測開始
        inputs = self.tokenizer(sentences, return_tensors='pt', padding=True, truncation=True, max_length=512)
        with torch.no_grad():
            outputs = self.model(**inputs)
        features = outputs.last_hidden_state.mean(dim=1)  # 文全体の特徴量を得るために平均を取る
        end_time = time.time()  # 計測終了
        logging.info(f"Feature extraction took {end_time - start_time:.2f} seconds.")
        return features

    def analyze_sentiment(self, sentences: list[str]) -> list[float]:
        """各台詞の感情分析を実施する"""
        sentiment_scores = []
        for sentence in sentences:
            score = self.sid.polarity_scores(sentence)  # VADERで感情分析
            sentiment_scores.append(score['compound'])  # compoundスコア（ポジティブ・ネガティブの全体的な評価）  
        return sentiment_scores

    def rank_sentences(self, sentences: list[str], features: torch.Tensor, sentiment_scores: list[float]) -> list[tuple[str, str, float]]:
        """感情スコアと特徴量を基に台詞をランク付け"""
        start_time = time.time()  # 計測開始
        ranked_sentences = []
        
        # 各台詞の特徴量と感情スコアを元にスコア化
        for sentence, score, sentiment in zip(sentences, features.sum(axis=1), sentiment_scores):
            ranked_sentences.append((sentence, score.sum(), sentiment))  # スコアに感情分析結果を掛け合わせる
        
        # スコア順に並べ替え（降順）
        ranked_sentences.sort(key=lambda x: x[1], reverse=True)
        end_time = time.time()  # 計測終了
        logging.info(f"Ranking sentences took {end_time - start_time:.2f} seconds.")
        return ranked_sentences

    def analyze_scripts(self) -> list[tuple[str, str, float]]:
        """ディレクトリ内の全ての映画台詞を処理し、全体のTop 100を選出"""
        heap_size = 100  # 保持する最大の台詞数
        movie_sentences = {}  # 映画ごとの台詞リストを格納する辞書

        start_time = time.time()  # 計測開始
        for file_name in os.listdir(self.scripts_dir):
            file_path = os.path.join(self.scripts_dir, file_name)
            if os.path.isfile(file_path):
                processed_script = self.preprocess_script(file_path)
                
                # 感情分析を実施
                sentiment_scores = self.analyze_sentiment(processed_script)
                
                # 特徴量を抽出
                features = self.extract_features(processed_script)
                
                # ランク付け
                ranked_sentences = self.rank_sentences(processed_script, features, sentiment_scores)
                
                # 映画ごとの台詞を格納
                movie_sentences[file_name] = ranked_sentences

                # ガベージコレクションを呼び出して、メモリを解放
                gc.collect()

        # すべての映画から台詞を集め、ランク付けされた台詞を1つのリストにまとめる
        all_ranked_sentences = []
        for movie, sentences in movie_sentences.items():
            all_ranked_sentences.extend(sentences)

        # スコア順に並べ替える（降順）
        all_ranked_sentences.sort(key=lambda x: x[1], reverse=True)

        # Top 100を選出
        top_sentences = all_ranked_sentences[:heap_size]

        end_time = time.time()  # 計測終了
        logging.info(f"Analyzing scripts took {end_time - start_time:.2f} seconds.")
        return top_sentences

    def save_results(self, ranked_sentences: list[tuple[str, str, float]]) -> None:
        """結果をファイルに保存"""
        start_time = time.time()  # 計測開始
        if self.output_file:
            with open(self.output_file, "w", encoding="utf-8") as f:
                for idx, (sentence, score, sentiment) in enumerate(ranked_sentences, 1):
                    f.write(f"{idx}. {sentence} (Score: {score}, Sentiment: {sentiment})\n")
        end_time = time.time()  # 計測終了
        logging.info(f"Saving results took {end_time - start_time:.2f} seconds.")

    def execute(self) -> list[tuple[str, str, float]]:
        """解析を実行し、結果を保存する"""
        start_time = time.time()  # 計測開始
        ranked_sentences = self.analyze_scripts()
        if self.output_file:
            self.save_results(ranked_sentences)
        end_time = time.time()  # 計測終了
        logging.info(f"Execution took {end_time - start_time:.2f} seconds.")
        return ranked_sentences


# Googleドライブのマウント
drive.mount('/content/drive')

# 使用例
scripts_dir = "/content/drive/MyDrive/Colab Notebooks/movie/input/movie_scripts"  # Googleドライブ上の映画スクリプトが保存されているディレクトリ
output_file = "/content/drive/MyDrive/Colab Notebooks/movie/output/top_100_sentences.txt"  # 出力ファイル

# MovieScriptAnalyzerインスタンスの作成
analyzer = MovieScriptAnalyzer(scripts_dir, output_file)

# 解析を実行して、Top 100の台詞を取得
top_100_sentences = analyzer.execute()

# 上位100フレーズを表示
for idx, (sentence, score, sentiment) in enumerate(top_100_sentences, 1):
    print(f"{idx}. {sentence} (Score: {score}, Sentiment: {sentiment})")

結果

『使用可能な RAM をすべて使用した後で、セッションがクラッシュしました。』というエラーで失敗。マシンの「スペック」の問題で処理が完了できず。

何というか「ハードディスク」の「スペック」に依存し過ぎていて、まともな検証ができない...

学習済みモデルとかあれば良いのかもしれないんのだけど、流石にニッチ過ぎる領域なのか、公開されてい無さそうなのよね...

研究自体は、行われているようなのだけど、

www.unite.ai

https://www.unite.ai/mapping-emotion-dynamics-in-movie-dialogues/

⇧『As noted by recent research from the Swedish Media council, there are many non-textual factors that should be taken into consideration when attempting to gauge the emotional temperature of a narrative, since context, music, visual cues and unspoken temporal factors (such as silence) contribute greatly to the meaning of discourse.』

とか言われても、「インプット」として利用できない「要因」を上げられてもお手上げという気はしますな...

とりあえず、「ハードディスク」の「メモリ」の要件がネックで、仮に最適なモデルを生成する手法があったとしても、「メモリ」を大量に必要とするようなものは「OOMK（Out of Memory Killer）」の問題が起きて強制終了してしまうので、試せないという問題もあるのだが...

データサイエンスは「ハードディスク」の「スペック」の影響が大き過ぎるんよね...

良い分析結果を得るには、お金がかかるってことですな...

それにしても、AIにAI的な最適解を求めても解決できないとは皮肉ですな...

人間が人間自身のことを解明できていない構造と同じってことなんですかね？

毎度モヤモヤ感が半端ない…

今回はこのへんで。