ソフトウェアエンジニアの技術ブログ：Software engineer tech blog

"""
契約リスク自動分析＆要約ツール - Webアプリケーション
Gradioを使用したユーザーインターフェース
"""

import gradio as gr
from contract_analyzer import ContractAnalyzer
import os
import json


def analyze_contract_pdf(pdf_file):
    """
    契約書PDFを分析する関数
    Args:
        pdf_file: アップロードされたPDFファイル
    Returns:
        分析結果のテキスト
    """
    if pdf_file is None:
        return "PDFファイルをアップロードしてください。"

    try:
        # 分析器を初期化（初回のみ時間がかかる）
        analyzer = ContractAnalyzer()

        # PDFファイルのパスを取得
        pdf_path = pdf_file.name

        # 契約書を分析
        result = analyzer.analyze_contract(pdf_path)

        # 平易な要約を生成
        summary = analyzer.generate_plain_summary(result)

        # 詳細情報をJSON形式で追加
        detailed_json = json.dumps(result, ensure_ascii=False, indent=2)

        # 結果を組み合わせて返す
        output = f"{summary}\n\n" + "="*50 + "\n"
        output += "\n【詳細情報（JSON形式）】\n"
        output += detailed_json

        return output

    except Exception as e:
        return f"エラーが発生しました: {str(e)}\n\n詳細: {type(e).__name__}"


def create_demo():
    """Gradioデモを作成"""

    with gr.Blocks(title="契約リスク自動分析ツール") as demo:
        gr.Markdown(
            """
            # 契約リスク自動分析＆要約ツール

            ## 機能
            - **PDF契約書のアップロード**: 契約書PDFを簡単にアップロード
            - **AI自動分析**: Hugging Face LLMを使用して自動分析
            - **リスク抽出**: リスク条項を自動検出
            - **義務抽出**: 義務事項を明確化
            - **期限管理**: 重要な期限・納期を抽出
            - **平易な要約**: 専門用語を避けた分かりやすい説明

            ## 使い方
            1. 下のボックスに契約書PDF（日本語または英語）をアップロード
            2. 「分析開始」ボタンをクリック
            3. 数秒～数十秒で分析結果が表示されます

            **⚠️ 注意**: 初回実行時は、AIモデルのダウンロードに数分かかる場合があります。
            """
        )

        with gr.Row():
            with gr.Column():
                pdf_input = gr.File(
                    label="契約書PDF",
                    file_types=[".pdf"],
                    type="filepath"
                )
                analyze_btn = gr.Button("分析開始", variant="primary", size="lg")

            with gr.Column():
                output_text = gr.Textbox(
                    label="分析結果",
                    lines=30,
                    max_lines=50,
                    show_copy_button=True
                )

        # ボタンクリック時の処理
        analyze_btn.click(
            fn=analyze_contract_pdf,
            inputs=[pdf_input],
            outputs=[output_text]
        )

        gr.Markdown(
            """
            ---
            ### システム情報
            - **使用モデル**:
              - 要約: facebook/bart-large-cnn
              - テキスト生成: google/flan-t5-base
            - **処理時間**: 通常5〜30秒（モデル読み込み済みの場合）
            - **対応言語**: 日本語、英語

            ### ヒント
            - PDFのテキストが抽出可能である必要があります（画像のみのPDFは非対応）
            - より正確な分析のため、契約書は明瞭な文章で記載されている必要があります
            - 大きなファイルの場合、処理に時間がかかることがあります
            """
        )

    return demo


if __name__ == "__main__":
    print("契約リスク自動分析ツールを起動中...")
    print("ブラウザが自動的に開きます...")

    demo = create_demo()

    # アプリケーション起動
    demo.launch(
        share=False,  # 公開リンクを生成しない（ローカルのみ）
        server_name="0.0.0.0",  # すべてのネットワークインターフェースでリッスン
        server_port=7860,
        show_error=True
    )

"""
契約リスク自動分析ツール - コア機能
PDFから契約書を読み込み、Hugging Face LLMを使用してリスク分析を実行
"""

import PyPDF2
from typing import Dict, List
from transformers import pipeline
import re


class ContractAnalyzer:
    """契約書分析クラス"""

    def __init__(self, model_name: str = "facebook/bart-large-cnn"):
        """
        初期化
        Args:
            model_name: 使用するHugging Faceモデル名
        """
        print(f"モデルを読み込み中: {model_name}")
        # 要約用のパイプラインを初期化
        self.summarizer = pipeline("summarization", model=model_name)

        # テキスト生成用のパイプライン（リスク分析用）
        # より軽量なモデルを使用
        try:
            self.text_generator = pipeline(
                "text2text-generation",
                model="google/flan-t5-base"
            )
        except:
            # フォールバック: 要約モデルを使用
            self.text_generator = self.summarizer

    def extract_text_from_pdf(self, pdf_path: str) -> str:
        """
        PDFファイルからテキストを抽出
        Args:
            pdf_path: PDFファイルのパス
        Returns:
            抽出されたテキスト
        """
        try:
            with open(pdf_path, 'rb') as file:
                pdf_reader = PyPDF2.PdfReader(file)
                text = ""
                for page in pdf_reader.pages:
                    text += page.extract_text() + "\n"
                return text
        except Exception as e:
            raise Exception(f"PDF読み込みエラー: {str(e)}")

    def extract_risk_clauses(self, text: str) -> List[str]:
        """
        リスク条項を抽出
        Args:
            text: 契約書テキスト
        Returns:
            リスク条項のリスト
        """
        risk_keywords = [
            "損害賠償", "違約金", "解除", "終了", "責任", "義務",
            "禁止", "制限", "罰則", "ペナルティ", "補償",
            "liability", "penalty", "termination", "obligation",
            "prohibited", "restricted", "damages", "breach"
        ]

        # 文章を分割
        sentences = re.split(r'[。\.\n]', text)
        risk_clauses = []

        for sentence in sentences:
            sentence = sentence.strip()
            if not sentence:
                continue
            # リスクキーワードを含む文章を抽出
            if any(keyword in sentence.lower() for keyword in risk_keywords):
                risk_clauses.append(sentence)

        return risk_clauses

    def extract_obligations(self, text: str) -> List[str]:
        """
        義務条項を抽出
        Args:
            text: 契約書テキスト
        Returns:
            義務条項のリスト
        """
        obligation_patterns = [
            r"(甲|乙|当事者|受注者|発注者|契約者)(?:は|が).*(?:しなければならない|する義務|する責任)",
            r"shall\s+.*",
            r"must\s+.*",
            r"obligated to\s+.*"
        ]

        sentences = re.split(r'[。\.\n]', text)
        obligations = []

        for sentence in sentences:
            sentence = sentence.strip()
            if not sentence:
                continue
            for pattern in obligation_patterns:
                if re.search(pattern, sentence, re.IGNORECASE):
                    obligations.append(sentence)
                    break

        return obligations

    def extract_deadlines(self, text: str) -> List[str]:
        """
        期限・日付を抽出
        Args:
            text: 契約書テキスト
        Returns:
            期限に関する条項のリスト
        """
        deadline_patterns = [
            r"\d{4}年\d{1,2}月\d{1,2}日",
            r"\d{1,2}日以内",
            r"\d+(?:日|週間|ヶ月|年)(?:以内|前|後|まで)",
            r"期限|締切|納期|有効期限",
            r"\d{4}-\d{2}-\d{2}",
            r"within\s+\d+\s+(?:days|weeks|months|years)",
            r"deadline|due date|expiration"
        ]

        sentences = re.split(r'[。\.\n]', text)
        deadlines = []

        for sentence in sentences:
            sentence = sentence.strip()
            if not sentence:
                continue
            for pattern in deadline_patterns:
                if re.search(pattern, sentence, re.IGNORECASE):
                    deadlines.append(sentence)
                    break

        return deadlines

    def summarize_text(self, text: str, max_length: int = 150) -> str:
        """
        テキストを要約
        Args:
            text: 要約するテキスト
            max_length: 要約の最大長
        Returns:
            要約されたテキスト
        """
        if not text or len(text.strip()) == 0:
            return "（テキストが空です）"

        # テキストが短すぎる場合はそのまま返す
        if len(text) < 100:
            return text

        try:
            # 長すぎるテキストは分割
            chunk_size = 1024
            if len(text) > chunk_size:
                text = text[:chunk_size]

            summary = self.summarizer(
                text,
                max_length=max_length,
                min_length=30,
                do_sample=False
            )
            return summary[0]['summary_text']
        except Exception as e:
            # エラー時は先頭部分を返す
            return f"要約エラー: {str(e)}\n\n元のテキスト（抜粋）:\n{text[:300]}..."

    def analyze_contract(self, pdf_path: str) -> Dict:
        """
        契約書を総合的に分析
        Args:
            pdf_path: PDFファイルのパス
        Returns:
            分析結果の辞書
        """
        print("PDF読み込み中...")
        text = self.extract_text_from_pdf(pdf_path)

        print("リスク条項を抽出中...")
        risk_clauses = self.extract_risk_clauses(text)

        print("義務条項を抽出中...")
        obligations = self.extract_obligations(text)

        print("期限情報を抽出中...")
        deadlines = self.extract_deadlines(text)

        print("要約を生成中...")
        # 全体の要約
        overall_summary = self.summarize_text(text[:2000], max_length=200)

        # リスク条項の要約（上位5件）
        risk_summary = []
        for clause in risk_clauses[:5]:
            try:
                summary = self.summarize_text(clause, max_length=100)
                risk_summary.append(summary)
            except:
                risk_summary.append(clause[:150] + "...")

        result = {
            "全体要約": overall_summary,
            "リスク条項": {
                "件数": len(risk_clauses),
                "主要なリスク": risk_clauses[:10],  # 上位10件
                "要約": risk_summary
            },
            "義務条項": {
                "件数": len(obligations),
                "主要な義務": obligations[:10]
            },
            "期限・納期": {
                "件数": len(deadlines),
                "主要な期限": deadlines[:10]
            },
            "リスクレベル": self._calculate_risk_level(
                len(risk_clauses),
                len(obligations),
                len(deadlines)
            )
        }

        return result

    def _calculate_risk_level(
        self,
        risk_count: int,
        obligation_count: int,
        deadline_count: int
    ) -> str:
        """
        リスクレベルを計算
        Args:
            risk_count: リスク条項数
            obligation_count: 義務条項数
            deadline_count: 期限条項数
        Returns:
            リスクレベル（高/中/低）
        """
        total_score = risk_count * 2 + obligation_count + deadline_count * 1.5

        if total_score > 50:
            return "高 - 詳細な法的レビューを推奨"
        elif total_score > 20:
            return "中 - 重要条項の確認を推奨"
        else:
            return "低 - 標準的な契約書"

    def generate_plain_summary(self, analysis_result: Dict) -> str:
        """
        平易な言葉で分析結果を要約
        Args:
            analysis_result: analyze_contractの結果
        Returns:
            平易な要約文
        """
        summary_lines = []
        summary_lines.append("=== 契約書分析結果 ===\n")

        summary_lines.append(f"【総合リスクレベル】{analysis_result['リスクレベル']}\n")

        summary_lines.append(f"\n【全体要約】\n{analysis_result['全体要約']}\n")

        summary_lines.append(f"\n【リスク条項】（{analysis_result['リスク条項']['件数']}件検出）")
        if analysis_result['リスク条項']['主要なリスク']:
            summary_lines.append("主な注意点:")
            for i, risk in enumerate(analysis_result['リスク条項']['主要なリスク'][:5], 1):
                summary_lines.append(f"  {i}. {risk}")

        summary_lines.append(f"\n【義務事項】（{analysis_result['義務条項']['件数']}件検出）")
        if analysis_result['義務条項']['主要な義務']:
            summary_lines.append("主な義務:")
            for i, obligation in enumerate(analysis_result['義務条項']['主要な義務'][:5], 1):
                summary_lines.append(f"  {i}. {obligation}")

        summary_lines.append(f"\n【期限・納期】（{analysis_result['期限・納期']['件数']}件検出）")
        if analysis_result['期限・納期']['主要な期限']:
            summary_lines.append("重要な期限:")
            for i, deadline in enumerate(analysis_result['期限・納期']['主要な期限'][:5], 1):
                summary_lines.append(f"  {i}. {deadline}")

        return "\n".join(summary_lines)


if __name__ == "__main__":
    # テスト用
    analyzer = ContractAnalyzer()
    print("契約リスク分析ツールが初期化されました")

# 契約リスク自動分析＆要約ツール - 必要パッケージ

# PDF処理
PyPDF2>=3.0.0

# Hugging Face Transformers（LLM）
transformers>=4.35.0
torch>=2.0.0
sentencepiece>=0.1.99

# Webインターフェース
gradio>=4.0.0

# データ処理
numpy>=1.24.0