# メールパイプラインの詳細データフロー定義
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from enum import Enum
 
 
class ProcessingStage(Enum):
    """処理ステージの定義"""
    RECEIVED = "received"
    PREPROCESSED = "preprocessed"
    CLASSIFIED = "classified"
    ROUTED = "routed"
    ACTION_TAKEN = "action_taken"
    FEEDBACK_COLLECTED = "feedback_collected"
 
 
@dataclass
class EmailMessage:
    """メールメッセージの統一データモデル"""
    message_id: str
    thread_id: str
    sender: str
    sender_name: str
    recipients: list[str]
    cc: list[str] = field(default_factory=list)
    subject: str = ""
    body_text: str = ""
    body_html: str = ""
    attachments: list[dict] = field(default_factory=list)
    headers: dict = field(default_factory=dict)
    received_at: datetime = field(default_factory=datetime.now)
    labels: list[str] = field(default_factory=list)
 
    # AI処理結果
    stage: ProcessingStage = ProcessingStage.RECEIVED
    classification: Optional[dict] = None
    sentiment: Optional[str] = None
    priority: Optional[str] = None
    suggested_reply: Optional[str] = None
    action_taken: Optional[str] = None
    processing_time_ms: int = 0
    ai_cost_usd: float = 0.0
 
    def to_ai_context(self) -> str:
        """AI処理用のコンテキスト文字列に変換"""
        return (
            f"From: {self.sender_name} <{self.sender}>\n"
            f"To: {', '.join(self.recipients)}\n"
            f"CC: {', '.join(self.cc)}\n"
            f"Subject: {self.subject}\n"
            f"Date: {self.received_at.isoformat()}\n"
            f"---\n"
            f"{self.body_text}"
        )
 
    def estimated_tokens(self) -> int:
        """トークン数の概算（日本語は1文字≒1.5トークン）"""
        text_length = len(self.body_text)
        return int(text_length * 1.5)
 
 
@dataclass
class ProcessingResult:
    """パイプライン処理結果"""
    message_id: str
    success: bool
    stage: ProcessingStage
    classification: Optional[dict] = None
    reply_draft: Optional[str] = None
    action: Optional[str] = None
    error: Optional[str] = None
    processing_time_ms: int = 0
    tokens_used: int = 0
    cost_usd: float = 0.0
    confidence: float = 0.0

import anthropic
from dataclasses import dataclass
from enum import Enum
 
class EmailCategory(Enum):
    BILLING = "billing"
    TECHNICAL = "technical"
    SALES = "sales"
    PARTNERSHIP = "partnership"
    SPAM = "spam"
    PERSONAL = "personal"
    NEWSLETTER = "newsletter"
    SUPPORT = "support"
    COMPLAINT = "complaint"
    FEEDBACK = "feedback"
    INTERNAL = "internal"
 
class Priority(Enum):
    URGENT = "urgent"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
 
@dataclass
class EmailAnalysis:
    category: EmailCategory
    priority: Priority
    sentiment: str  # positive/neutral/negative
    intent: str
    summary: str
    suggested_action: str
    confidence: float
    language: str = "ja"
    topics: list[str] = None
    entities: list[dict] = None
 
class EmailClassifier:
    """AIメール分類エンジン"""
 
    CLASSIFICATION_PROMPT = """
以下のメールを分析し、JSON形式で結果を返してください。
 
分析項目:
- category: billing / technical / sales / partnership / spam / personal / newsletter / support / complaint / feedback / internal
- priority: urgent / high / medium / low
- sentiment: positive / neutral / negative
- intent: 1文で送信者の意図
- summary: 50文字以内の要約
- suggested_action: 推奨アクション
- confidence: 0.0-1.0の信頼度
- language: メールの言語コード（ja / en / zh 等）
- topics: 関連トピックのリスト（最大3つ）
- entities: 抽出したエンティティ（人名、会社名、金額、日付等）
 
判断基準:
- urgent: 即座の対応が必要（障害報告、請求トラブル、セキュリティインシデント等）
- high: 今日中に対応すべき（重要顧客、締切案件、クレーム初回対応）
- medium: 2-3日以内に対応（一般問い合わせ、機能要望）
- low: 対応不要 or いつでも可（ニュースレター、広告、FYI共有）
 
メール:
From: {sender}
Subject: {subject}
Date: {date}
Body:
{body}
"""
 
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.vip_list = set()
        self.rules = []
        self.classification_cache = {}
        self.stats = {
            "total_classified": 0,
            "cache_hits": 0,
            "rule_based": 0,
            "ai_classified": 0,
        }
 
    def add_vip(self, email: str):
        """VIPリスト追加"""
        self.vip_list.add(email.lower())
 
    def add_rule(self, condition: callable, result: EmailAnalysis):
        """カスタムルール追加"""
        self.rules.append({"condition": condition, "result": result})
 
    def classify(self, email: dict) -> EmailAnalysis:
        """メール分類"""
        self.stats["total_classified"] += 1
 
        # ルールベースの前処理
        if self._is_obvious_spam(email):
            self.stats["rule_based"] += 1
            return EmailAnalysis(
                category=EmailCategory.SPAM,
                priority=Priority.LOW,
                sentiment="neutral",
                intent="スパム",
                summary="スパムメール",
                suggested_action="自動削除",
                confidence=0.99
            )
 
        # カスタムルールチェック
        for rule in self.rules:
            if rule"condition":
                self.stats["rule_based"] += 1
                return rule["result"]
 
        # ニュースレター自動検出
        if self._is_newsletter(email):
            self.stats["rule_based"] += 1
            return EmailAnalysis(
                category=EmailCategory.NEWSLETTER,
                priority=Priority.LOW,
                sentiment="neutral",
                intent="情報配信",
                summary=f"ニュースレター: {email['subject'][:30]}",
                suggested_action="既読にしてアーカイブ",
                confidence=0.95
            )
 
        # AI分析
        self.stats["ai_classified"] += 1
        prompt = self.CLASSIFICATION_PROMPT.format(**email)
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=512,
            messages=[{"role": "user", "content": prompt}]
        )
 
        result = self._parse_response(response.content[0].text)
 
        # VIP補正
        if email["sender"].lower() in self.vip_list:
            if result.priority != Priority.URGENT:
                result.priority = Priority.HIGH
 
        return result
 
    def classify_batch(self, emails: list[dict]) -> list[EmailAnalysis]:
        """バッチ分類（コスト効率化）"""
        results = []
        ai_batch = []
 
        # ルールベースで処理可能なものを先に処理
        for email in emails:
            if self._is_obvious_spam(email):
                results.append((email["message_id"], EmailAnalysis(
                    category=EmailCategory.SPAM,
                    priority=Priority.LOW,
                    sentiment="neutral",
                    intent="スパム",
                    summary="スパムメール",
                    suggested_action="自動削除",
                    confidence=0.99
                )))
            elif self._is_newsletter(email):
                results.append((email["message_id"], EmailAnalysis(
                    category=EmailCategory.NEWSLETTER,
                    priority=Priority.LOW,
                    sentiment="neutral",
                    intent="情報配信",
                    summary=f"ニュースレター: {email['subject'][:30]}",
                    suggested_action="アーカイブ",
                    confidence=0.95
                )))
            else:
                ai_batch.append(email)
 
        # AI分類が必要なものをバッチ処理
        if ai_batch:
            batch_prompt = self._build_batch_prompt(ai_batch)
            response = self.client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=2048,
                messages=[{"role": "user", "content": batch_prompt}]
            )
            batch_results = self._parse_batch_response(
                response.content[0].text, ai_batch
            )
            results.extend(batch_results)
 
        return results
 
    def _build_batch_prompt(self, emails: list[dict]) -> str:
        """バッチ分類用プロンプト生成"""
        email_texts = []
        for i, email in enumerate(emails):
            email_texts.append(
                f"--- メール {i+1} (ID: {email['message_id']}) ---\n"
                f"From: {email['sender']}\n"
                f"Subject: {email['subject']}\n"
                f"Body: {email['body'][:500]}\n"
            )
 
        return (
            "以下の複数メールをそれぞれ分類してください。\n"
            "各メールについて category, priority, sentiment, summary を"
            "JSON配列で返してください。\n\n"
            + "\n".join(email_texts)
        )
 
    def _is_obvious_spam(self, email: dict) -> bool:
        """ルールベーススパム判定"""
        spam_keywords = [
            "当選", "無料プレゼント", "今すぐクリック",
            "unsubscribe", "配信停止", "出会い系",
            "儲かる", "副業で月収", "限定オファー"
        ]
        subject_body = f"{email['subject']} {email['body']}".lower()
        match_count = sum(1 for kw in spam_keywords if kw in subject_body)
        return match_count >= 2
 
    def _is_newsletter(self, email: dict) -> bool:
        """ニュースレター判定"""
        newsletter_indicators = [
            "list-unsubscribe" in str(email.get("headers", {})).lower(),
            "noreply" in email["sender"].lower(),
            "newsletter" in email["sender"].lower(),
            "mail.substack.com" in email["sender"].lower(),
            "配信" in email.get("subject", ""),
        ]
        return sum(newsletter_indicators) >= 2
 
    def _parse_response(self, text: str) -> EmailAnalysis:
        """レスポンスパース"""
        import json
        try:
            start = text.index("{")
            end = text.rindex("}") + 1
            data = json.loads(text[start:end])
            return EmailAnalysis(
                category=EmailCategory(data["category"]),
                priority=Priority(data["priority"]),
                sentiment=data["sentiment"],
                intent=data["intent"],
                summary=data["summary"],
                suggested_action=data["suggested_action"],
                confidence=data.get("confidence", 0.8),
                language=data.get("language", "ja"),
                topics=data.get("topics", []),
                entities=data.get("entities", [])
            )
        except Exception:
            return EmailAnalysis(
                category=EmailCategory.PERSONAL,
                priority=Priority.MEDIUM,
                sentiment="neutral",
                intent="分類不能",
                summary="AI分類に失敗",
                suggested_action="手動確認",
                confidence=0.0
            )
 
    def _parse_batch_response(self, text, emails):
        """バッチレスポンスのパース"""
        import json
        results = []
        try:
            start = text.index("[")
            end = text.rindex("]") + 1
            data_list = json.loads(text[start:end])
            for i, data in enumerate(data_list):
                msg_id = emails[i]["message_id"]
                analysis = EmailAnalysis(
                    category=EmailCategory(data.get("category", "personal")),
                    priority=Priority(data.get("priority", "medium")),
                    sentiment=data.get("sentiment", "neutral"),
                    intent=data.get("intent", ""),
                    summary=data.get("summary", ""),
                    suggested_action=data.get("suggested_action", "手動確認"),
                    confidence=data.get("confidence", 0.7)
                )
                results.append((msg_id, analysis))
        except Exception:
            for email in emails:
                results.append((email["message_id"], EmailAnalysis(
                    category=EmailCategory.PERSONAL,
                    priority=Priority.MEDIUM,
                    sentiment="neutral",
                    intent="バッチ分類失敗",
                    summary="手動確認が必要",
                    suggested_action="手動確認",
                    confidence=0.0
                )))
        return results
 
    def get_stats(self) -> dict:
        """分類統計情報の取得"""
        total = self.stats["total_classified"]
        if total == 0:
            return self.stats
        return {
            **self.stats,
            "rule_based_pct": round(
                self.stats["rule_based"] / total * 100, 1
            ),
            "ai_classified_pct": round(
                self.stats["ai_classified"] / total * 100, 1
            ),
        }

class HybridClassificationStrategy:
    """ルールベース + ML + LLM のハイブリッド分類戦略"""
 
    def __init__(self, classifier: EmailClassifier):
        self.classifier = classifier
        self.ml_model = None  # 学習済みモデル（scikit-learn等）
        self.confidence_threshold = 0.85
 
    def classify(self, email: dict) -> EmailAnalysis:
        """3段階のフォールバック分類"""
 
        # Stage 1: ルールベース（コスト0、即時）
        rule_result = self._rule_based_classify(email)
        if rule_result and rule_result.confidence >= 0.95:
            rule_result.suggested_action += " [ルールベース判定]"
            return rule_result
 
        # Stage 2: MLモデル（コスト極小、高速）
        if self.ml_model:
            ml_result = self._ml_classify(email)
            if ml_result and ml_result.confidence >= self.confidence_threshold:
                ml_result.suggested_action += " [ML判定]"
                return ml_result
 
        # Stage 3: LLM（コスト高、高精度）
        llm_result = self.classifier.classify(email)
        llm_result.suggested_action += " [LLM判定]"
        return llm_result
 
    def _rule_based_classify(self, email: dict) -> EmailAnalysis | None:
        """ルールベース分類"""
        subject = email.get("subject", "").lower()
        sender = email.get("sender", "").lower()
        body = email.get("body", "").lower()
 
        # 請求関連
        billing_keywords = ["請求", "invoice", "支払い", "料金", "プラン変更"]
        if any(kw in subject or kw in body[:200] for kw in billing_keywords):
            return EmailAnalysis(
                category=EmailCategory.BILLING,
                priority=Priority.HIGH,
                sentiment="neutral",
                intent="請求関連の問い合わせ",
                summary=f"請求関連: {email['subject'][:30]}",
                suggested_action="請求チームに転送",
                confidence=0.92
            )
 
        # 障害報告
        incident_keywords = [
            "障害", "ダウン", "エラー", "使えない",
            "動かない", "バグ", "500エラー", "タイムアウト"
        ]
        if any(kw in subject or kw in body[:300] for kw in incident_keywords):
            return EmailAnalysis(
                category=EmailCategory.TECHNICAL,
                priority=Priority.URGENT,
                sentiment="negative",
                intent="障害またはバグの報告",
                summary=f"障害報告: {email['subject'][:30]}",
                suggested_action="エンジニアリングチームに即時エスカレーション",
                confidence=0.90
            )
 
        # 解約・退会
        churn_keywords = ["解約", "退会", "キャンセル", "やめたい", "解除"]
        if any(kw in subject or kw in body[:300] for kw in churn_keywords):
            return EmailAnalysis(
                category=EmailCategory.SUPPORT,
                priority=Priority.HIGH,
                sentiment="negative",
                intent="解約・退会の意思表示",
                summary=f"解約要望: {email['subject'][:30]}",
                suggested_action="カスタマーサクセスに転送（リテンション対応）",
                confidence=0.93
            )
 
        return None
 
    def _ml_classify(self, email: dict) -> EmailAnalysis | None:
        """MLモデルによる分類（事前学習済み）"""
        if not self.ml_model:
            return None
 
        features = self._extract_features(email)
        prediction = self.ml_model.predict([features])
        confidence = max(self.ml_model.predict_proba([features])[0])
 
        if confidence < self.confidence_threshold:
            return None
 
        category_map = {
            0: EmailCategory.BILLING,
            1: EmailCategory.TECHNICAL,
            2: EmailCategory.SALES,
            3: EmailCategory.SUPPORT,
            4: EmailCategory.SPAM,
            5: EmailCategory.NEWSLETTER,
            6: EmailCategory.PERSONAL,
        }
 
        return EmailAnalysis(
            category=category_map.get(prediction[0], EmailCategory.PERSONAL),
            priority=Priority.MEDIUM,
            sentiment="neutral",
            intent="ML分類による判定",
            summary=email["subject"][:50],
            suggested_action="自動分類済み",
            confidence=float(confidence)
        )
 
    def _extract_features(self, email: dict) -> list:
        """メールから特徴量を抽出"""
        subject = email.get("subject", "")
        body = email.get("body", "")
        return [
            len(subject),
            len(body),
            body.count("?"),
            body.count("!"),
            1 if "noreply" in email.get("sender", "") else 0,
            len(email.get("cc", [])),
            len(email.get("attachments", [])),
        ]

# 各カテゴリの詳細定義と自動アクションマッピング
CATEGORY_CONFIG = {
    "billing": {
        "display_name": "請求・支払い",
        "description": "料金、請求書、プラン変更、返金に関するメール",
        "auto_actions": [
            "請求チームのSlackチャネルに通知",
            "CRMに問い合わせチケット作成",
            "顧客の契約情報を自動取得して添付",
        ],
        "sla_hours": 4,
        "escalation_after_hours": 8,
        "template_replies": {
            "price_inquiry": "料金に関するお問い合わせありがとうございます...",
            "refund_request": "ご返金のご要望を承りました...",
            "plan_change": "プラン変更のお手続きについてご案内します...",
        },
        "keywords_ja": ["請求", "料金", "支払い", "返金", "プラン", "契約"],
        "keywords_en": ["billing", "invoice", "payment", "refund", "plan"],
    },
    "technical": {
        "display_name": "技術サポート",
        "description": "バグ報告、機能の使い方、API関連の技術的な質問",
        "auto_actions": [
            "技術サポートキューに追加",
            "関連するFAQ/ドキュメントを自動検索",
            "エラーログとの照合（直近24時間）",
        ],
        "sla_hours": 8,
        "escalation_after_hours": 24,
        "template_replies": {
            "bug_report": "バグのご報告ありがとうございます...",
            "how_to": "ご質問の機能についてご説明します...",
            "api_question": "API仕様についてお答えします...",
        },
        "keywords_ja": ["エラー", "バグ", "動かない", "使い方", "API"],
        "keywords_en": ["error", "bug", "not working", "how to", "API"],
    },
    "sales": {
        "display_name": "営業・セールス",
        "description": "新規問い合わせ、デモ依頼、見積もり要求",
        "auto_actions": [
            "CRMにリード登録",
            "営業チームにSlack通知",
            "会社情報を自動リサーチ",
        ],
        "sla_hours": 2,
        "escalation_after_hours": 4,
        "template_replies": {
            "demo_request": "デモのご要望ありがとうございます...",
            "pricing_inquiry": "お見積りについてご回答します...",
            "trial_request": "トライアルのご案内をいたします...",
        },
        "keywords_ja": ["導入", "検討", "デモ", "見積", "価格"],
        "keywords_en": ["demo", "pricing", "trial", "enterprise", "quote"],
    },
    "complaint": {
        "display_name": "クレーム・苦情",
        "description": "不満、苦情、改善要求の強い表現を含むメール",
        "auto_actions": [
            "マネージャーに即時通知",
            "過去のやり取り履歴を自動取得",
            "クレーム管理システムにエスカレーション登録",
        ],
        "sla_hours": 1,
        "escalation_after_hours": 2,
        "template_replies": {
            "service_complaint": "ご不便をおかけし誠に申し訳ございません...",
            "quality_issue": "品質に関するご指摘、真摯に受け止めます...",
        },
        "keywords_ja": ["不満", "ひどい", "最悪", "怒り", "訴える", "消費者庁"],
        "keywords_en": ["terrible", "worst", "unacceptable", "lawsuit"],
    },
}

class AutoReplyEngine:
    """AI自動返信エンジン"""
 
    REPLY_PROMPT = """
あなたは{company_name}のカスタマーサポート担当です。
以下のメールに対する返信ドラフトを作成してください。
 
トーン: {tone}
言語: 日本語
署名: {signature}
 
ルール:
- 丁寧だが簡潔に
- 具体的な解決策を提示
- 不明点は正直に伝え、確認する旨を記載
- 個人情報は含めない
- 1つのメールにつき1つの明確なアクションを提案
- 顧客の名前で呼びかける（分かる場合）
 
元メール:
From: {sender_name} <{sender_email}>
Subject: {subject}
Body:
{body}
 
過去のやり取り（あれば）:
{thread_history}
 
ナレッジベース参照（あれば）:
{knowledge_context}
"""
 
    def __init__(self, api_key: str, company_name: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.company_name = company_name
        self.templates = {}
        self.knowledge_base = None
        self.tone_configs = {
            "professional": {
                "description": "ビジネス標準の丁寧なトーン",
                "examples": ["お世話になっております", "ご確認いただけますと幸いです"],
                "avoid": ["了解です", "OK", "〜っす"],
            },
            "friendly": {
                "description": "親しみやすく柔らかいトーン",
                "examples": ["いつもありがとうございます！", "お気軽にご連絡ください"],
                "avoid": ["拝啓", "敬具", "小職"],
            },
            "formal": {
                "description": "非常にフォーマルなトーン（公式謝罪等）",
                "examples": ["誠に申し訳ございません", "深くお詫び申し上げます"],
                "avoid": ["すみません", "ごめんなさい"],
            },
            "empathetic": {
                "description": "共感を示すトーン（クレーム対応等）",
                "examples": [
                    "ご不便をおかけして申し訳ございません",
                    "お気持ちは十分に理解しております"
                ],
                "avoid": ["しかしながら", "ですが"],
            },
        }
 
    def generate_reply(self, email: dict,
                       tone: str = "professional",
                       thread_history: str = "",
                       knowledge_context: str = "") -> dict:
        """返信ドラフト生成"""
        signature = self._get_signature(email.get("assigned_to"))
 
        prompt = self.REPLY_PROMPT.format(
            company_name=self.company_name,
            tone=self._get_tone_instruction(tone),
            signature=signature,
            sender_name=email.get("sender_name", ""),
            sender_email=email["sender"],
            subject=email["subject"],
            body=email["body"],
            thread_history=thread_history or "なし",
            knowledge_context=knowledge_context or "なし"
        )
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )
 
        reply_text = response.content[0].text
        confidence = self._assess_confidence(reply_text, email)
 
        return {
            "subject": f"Re: {email['subject']}",
            "body": reply_text,
            "status": self._determine_status(confidence, email),
            "confidence": confidence,
            "tone": tone,
            "tokens_used": (
                response.usage.input_tokens + response.usage.output_tokens
            ),
            "model": "claude-sonnet-4-20250514",
        }
 
    def generate_multiple_drafts(
        self, email: dict, tones: list[str] = None
    ) -> list[dict]:
        """複数トーンのドラフトを同時生成"""
        if tones is None:
            tones = ["professional", "friendly"]
 
        drafts = []
        for tone in tones:
            draft = self.generate_reply(email, tone=tone)
            draft["tone_label"] = self.tone_configs[tone]["description"]
            drafts.append(draft)
 
        return drafts
 
    def _get_tone_instruction(self, tone: str) -> str:
        """トーン設定の詳細指示を生成"""
        config = self.tone_configs.get(tone, self.tone_configs["professional"])
        return (
            f"{config['description']}\n"
            f"使用例: {', '.join(config['examples'])}\n"
            f"避ける表現: {', '.join(config['avoid'])}"
        )
 
    def _assess_confidence(self, reply: str, email: dict) -> float:
        """返信の信頼度判定"""
        score = 0.8
 
        # 長さチェック
        if len(reply) < 50:
            score -= 0.2  # 短すぎる
        elif len(reply) > 2000:
            score -= 0.1  # 長すぎる
 
        # 不確実性の表現
        uncertain_phrases = ["確認", "お調べ", "分かりかねます", "不明"]
        uncertainty_count = sum(
            1 for phrase in uncertain_phrases if phrase in reply
        )
        score -= uncertainty_count * 0.05
 
        # 緊急案件は人間確認推奨
        if email.get("priority") == "urgent":
            score -= 0.15
 
        # クレーム対応は慎重に
        if email.get("category") == "complaint":
            score -= 0.2
 
        # 金額に言及している場合は慎重に
        import re
        if re.search(r'[¥$€]\d+|円|ドル', reply):
            score -= 0.1
 
        return max(0.0, min(1.0, score))
 
    def _determine_status(self, confidence: float, email: dict) -> str:
        """信頼度に基づくステータス決定"""
        category = email.get("category", "")
        priority = email.get("priority", "medium")
 
        # 高リスクカテゴリは常にドラフト
        if category in ["complaint", "billing"]:
            return "draft_review_required"
 
        # 緊急は人間レビュー必須
        if priority == "urgent":
            return "draft_review_required"
 
        # 信頼度に基づく判定
        if confidence >= 0.9:
            return "ready_to_send"
        elif confidence >= 0.7:
            return "draft"
        else:
            return "needs_human_review"
 
    def _get_signature(self, assigned_to: str = None) -> str:
        return f"""
--
{self.company_name} カスタマーサポート
{assigned_to or "担当者"}
"""

class KnowledgeAugmentedReplyEngine:
    """ナレッジベース連携の返信エンジン（RAG方式）"""
 
    def __init__(self, reply_engine: AutoReplyEngine, vector_store):
        self.reply_engine = reply_engine
        self.vector_store = vector_store  # Pinecone, Qdrant, pgvector等
 
    def generate_reply_with_knowledge(
        self, email: dict, top_k: int = 5
    ) -> dict:
        """ナレッジベースを参照した返信生成"""
 
        # 1. 関連ナレッジの検索
        query = f"{email['subject']} {email['body'][:300]}"
        relevant_docs = self.vector_store.similarity_search(
            query=query,
            top_k=top_k,
            filter={"type": {"$in": ["faq", "doc", "past_reply"]}}
        )
 
        # 2. コンテキスト構築
        knowledge_context = self._build_context(relevant_docs)
 
        # 3. 過去の類似対応の取得
        past_replies = self._find_similar_past_replies(email)
 
        # 4. ナレッジ付き返信生成
        reply = self.reply_engine.generate_reply(
            email=email,
            knowledge_context=knowledge_context,
            thread_history=past_replies
        )
 
        # 5. 参照元情報の付与
        reply["references"] = [
            {
                "title": doc.metadata.get("title", ""),
                "url": doc.metadata.get("url", ""),
                "relevance_score": doc.metadata.get("score", 0),
            }
            for doc in relevant_docs
        ]
 
        return reply
 
    def _build_context(self, docs: list) -> str:
        """検索結果からコンテキスト文字列を構築"""
        context_parts = []
        for i, doc in enumerate(docs, 1):
            context_parts.append(
                f"[参考{i}] {doc.metadata.get('title', '無題')}\n"
                f"{doc.page_content[:500]}\n"
            )
        return "\n---\n".join(context_parts)
 
    def _find_similar_past_replies(self, email: dict) -> str:
        """過去の類似メール対応を検索"""
        results = self.vector_store.similarity_search(
            query=email["body"][:200],
            top_k=3,
            filter={"type": "past_reply", "rating": {"$gte": 4}}
        )
        if not results:
            return "過去の類似対応なし"
 
        past = []
        for r in results:
            past.append(
                f"[過去の対応例]\n"
                f"元メール: {r.metadata.get('original_subject', '')}\n"
                f"返信: {r.page_content[:300]}..."
            )
        return "\n\n".join(past)
 
    def feedback_loop(self, reply_id: str, rating: int, comment: str = ""):
        """返信品質のフィードバックを記録し学習に活用"""
        self.vector_store.update_metadata(
            id=reply_id,
            metadata={
                "rating": rating,
                "feedback_comment": comment,
                "feedback_at": datetime.now().isoformat(),
            }
        )
 
        # 高評価の返信をナレッジベースに追加
        if rating >= 4:
            reply_data = self.vector_store.get(reply_id)
            self.vector_store.upsert(
                id=f"past_reply_{reply_id}",
                content=reply_data["content"],
                metadata={
                    "type": "past_reply",
                    "rating": rating,
                    "category": reply_data.get("category", ""),
                    "original_subject": reply_data.get("subject", ""),
                }
            )

class ThreadSummarizer:
    """メールスレッド要約エンジン"""
 
    SUMMARY_PROMPT = """
以下のメールスレッドを分析し、構造化された要約を作成してください。
 
出力形式:
1. 概要（2-3文）
2. 参加者と役割
3. 経緯（時系列）
4. 現在のステータス
5. 未解決事項
6. 必要なアクション
 
スレッド:
{thread}
"""
 
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
 
    def summarize_thread(self, messages: list[dict]) -> dict:
        """スレッド全体を要約"""
        thread_text = self._format_thread(messages)
 
        # 長いスレッドはチャンク処理
        if len(thread_text) > 10000:
            return self._summarize_long_thread(messages)
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": self.SUMMARY_PROMPT.format(thread=thread_text)
            }]
        )
        return {
            "summary": response.content[0].text,
            "message_count": len(messages),
            "participants": list({m["sender"] for m in messages}),
            "date_range": {
                "start": messages[0]["date"],
                "end": messages[-1]["date"],
            },
        }
 
    def _summarize_long_thread(self, messages: list[dict]) -> dict:
        """長いスレッドの段階的要約"""
        # Phase 1: 各メールを個別要約
        individual_summaries = []
        for msg in messages:
            summary = self._summarize_single(msg)
            individual_summaries.append(summary)
 
        # Phase 2: 個別要約を統合
        combined = "\n".join(
            f"[{s['date']}] {s['from']}: {s['summary']}"
            for s in individual_summaries
        )
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"以下のメール要約を統合して、最終的な要約を作成してください:\n{combined}"
            }]
        )
        return {
            "summary": response.content[0].text,
            "method": "hierarchical",
            "message_count": len(messages),
        }
 
    def _summarize_single(self, msg: dict) -> dict:
        """個別メールの要約"""
        response = self.client.messages.create(
            model="claude-haiku-4-20250514",  # 個別要約はHaikuで十分
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": (
                    f"以下のメールを1-2文で要約:\n"
                    f"From: {msg['sender']}\n"
                    f"Date: {msg['date']}\n"
                    f"Body: {msg['body'][:1000]}"
                )
            }]
        )
        return {
            "from": msg["sender"],
            "date": msg["date"],
            "summary": response.content[0].text,
        }
 
    def _format_thread(self, messages: list[dict]) -> str:
        """スレッドをテキスト形式に整形"""
        parts = []
        for msg in messages:
            parts.append(
                f"---\nFrom: {msg['sender']}\n"
                f"Date: {msg['date']}\n"
                f"Subject: {msg['subject']}\n\n"
                f"{msg['body']}\n"
            )
        return "\n".join(parts)
 
    def generate_action_items(self, thread_summary: str) -> list[dict]:
        """要約からアクションアイテムを抽出"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=512,
            messages=[{
                "role": "user",
                "content": (
                    "以下のメールスレッド要約からアクションアイテムを抽出し、"
                    "JSON配列で返してください。\n"
                    "各アイテム: {\"task\": \"タスク\", \"assignee\": \"担当者\", "
                    "\"deadline\": \"期限\", \"priority\": \"high/medium/low\"}\n\n"
                    f"要約:\n{thread_summary}"
                )
            }]
        )
        import json
        try:
            text = response.content[0].text
            start = text.index("[")
            end = text.rindex("]") + 1
            return json.loads(text[start:end])
        except Exception:
            return []

class MeetingSummarizer:
    """会議録AI要約"""
 
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
 
    def summarize_meeting(self, transcript: str,
                          meeting_type: str = "general") -> dict:
        """会議録を構造化要約"""
        type_instructions = {
            "general": "一般的な会議の要約を作成",
            "standup": "デイリースタンドアップの要点を抽出",
            "sprint_review": "スプリントレビューの成果と課題を整理",
            "retrospective": "振り返りのKeep/Problem/Tryを整理",
            "sales": "商談の進捗と次のアクションを明確化",
            "one_on_one": "1on1の議論内容とフォローアップ事項を整理",
        }
 
        instruction = type_instructions.get(meeting_type, type_instructions["general"])
 
        prompt = f"""
以下の会議録を分析し、JSON形式で要約してください。
会議タイプ: {meeting_type}
指示: {instruction}
 
出力形式:
{{
  "title": "会議タイトル",
  "type": "{meeting_type}",
  "date": "日付",
  "participants": ["参加者リスト"],
  "duration": "所要時間",
  "summary": "3行要約",
  "key_points": ["主要論点"],
  "decisions": ["決定事項"],
  "action_items": [
    {{"task": "タスク内容", "assignee": "担当者", "deadline": "期限"}}
  ],
  "open_issues": ["未解決事項"],
  "risks": ["リスクや懸念事項"],
  "next_meeting": "次回予定"
}}
 
会議録:
{transcript}
"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )
        return self._parse_json(response.content[0].text)
 
    def summarize_from_audio(self, audio_path: str) -> dict:
        """音声ファイルから会議要約を生成するパイプライン"""
        # Step 1: 音声→テキスト変換（Whisper等）
        transcript = self._transcribe_audio(audio_path)
 
        # Step 2: 話者分離（Diarization）
        diarized = self._diarize_transcript(transcript)
 
        # Step 3: AI要約
        return self.summarize_meeting(diarized)
 
    def _transcribe_audio(self, audio_path: str) -> str:
        """音声ファイルの文字起こし"""
        import openai
        client = openai.OpenAI()
        with open(audio_path, "rb") as audio_file:
            transcript = client.audio.transcriptions.create(
                model="whisper-1",
                file=audio_file,
                language="ja",
                response_format="verbose_json",
            )
        return transcript.text
 
    def _diarize_transcript(self, transcript: str) -> str:
        """話者分離（簡易実装）"""
        # 実際にはpyannote-audio等を使用
        return transcript
 
    def _parse_json(self, text: str) -> dict:
        """JSONパース（エラー耐性あり）"""
        import json
        try:
            start = text.index("{")
            end = text.rindex("}") + 1
            return json.loads(text[start:end])
        except Exception:
            return {"raw_summary": text, "parse_error": True}

class SlackSummarizer:
    """Slackチャネル/スレッドの要約エンジン"""
 
    def __init__(self, api_key: str, slack_token: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.slack_token = slack_token
 
    def summarize_channel(
        self, channel_id: str, hours: int = 24
    ) -> dict:
        """チャネルの直近N時間の要約"""
        import requests
        from datetime import datetime, timedelta
 
        oldest = (datetime.now() - timedelta(hours=hours)).timestamp()
 
        response = requests.get(
            "https://slack.com/api/conversations.history",
            headers={"Authorization": f"Bearer {self.slack_token}"},
            params={
                "channel": channel_id,
                "oldest": str(oldest),
                "limit": 200,
            }
        )
        messages = response.json().get("messages", [])
 
        if not messages:
            return {"summary": "この期間のメッセージはありません"}
 
        # メッセージの整形
        formatted = []
        for msg in reversed(messages):  # 時系列順に
            user = msg.get("user", "unknown")
            text = msg.get("text", "")
            ts = datetime.fromtimestamp(
                float(msg["ts"])
            ).strftime("%H:%M")
            formatted.append(f"[{ts}] {user}: {text}")
 
        thread_text = "\n".join(formatted)
 
        # AI要約
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": (
                    f"以下のSlackチャネルのメッセージを要約してください。\n"
                    f"期間: 直近{hours}時間\n"
                    f"出力: 概要、主要な議論、決定事項、アクションアイテム\n\n"
                    f"メッセージ:\n{thread_text}"
                )
            }]
        )
 
        return {
            "summary": response.content[0].text,
            "channel_id": channel_id,
            "period_hours": hours,
            "message_count": len(messages),
            "unique_participants": len({m.get("user") for m in messages}),
        }
 
    def daily_digest(self, channel_ids: list[str]) -> str:
        """複数チャネルのデイリーダイジェスト生成"""
        digests = []
        for ch_id in channel_ids:
            summary = self.summarize_channel(ch_id, hours=24)
            digests.append(summary)
 
        # 全チャネルの統合ダイジェスト
        combined = "\n\n".join(
            f"## #{d.get('channel_name', d['channel_id'])}\n"
            f"メッセージ数: {d['message_count']}\n"
            f"{d['summary']}"
            for d in digests
        )
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": (
                    "以下の各チャネルの要約を統合して、"
                    "チーム全体のデイリーダイジェストを作成してください。\n"
                    "重要度順に整理し、注意が必要な事項を冒頭に配置。\n\n"
                    f"{combined}"
                )
            }]
        )
 
        return response.content[0].text

import base64
from email.mime.text import MIMEText
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
 
 
class GmailIntegration:
    """Gmail APIとの統合クラス"""
 
    def __init__(self, credentials_path: str):
        self.creds = Credentials.from_authorized_user_file(
            credentials_path,
            scopes=[
                "https://www.googleapis.com/auth/gmail.readonly",
                "https://www.googleapis.com/auth/gmail.send",
                "https://www.googleapis.com/auth/gmail.modify",
            ]
        )
        self.service = build("gmail", "v1", credentials=self.creds)
 
    def fetch_unread(self, max_results: int = 50) -> list[dict]:
        """未読メールの取得"""
        results = self.service.users().messages().list(
            userId="me",
            q="is:unread",
            maxResults=max_results
        ).execute()
 
        messages = results.get("messages", [])
        emails = []
 
        for msg_ref in messages:
            msg = self.service.users().messages().get(
                userId="me",
                id=msg_ref["id"],
                format="full"
            ).execute()
 
            email_data = self._parse_message(msg)
            emails.append(email_data)
 
        return emails
 
    def _parse_message(self, msg: dict) -> dict:
        """Gmailメッセージのパース"""
        headers = {
            h["name"].lower(): h["value"]
            for h in msg["payload"]["headers"]
        }
 
        body = ""
        if "parts" in msg["payload"]:
            for part in msg["payload"]["parts"]:
                if part["mimeType"] == "text/plain":
                    data = part["body"].get("data", "")
                    body = base64.urlsafe_b64decode(data).decode("utf-8")
                    break
        elif "body" in msg["payload"]:
            data = msg["payload"]["body"].get("data", "")
            if data:
                body = base64.urlsafe_b64decode(data).decode("utf-8")
 
        return {
            "message_id": msg["id"],
            "thread_id": msg["threadId"],
            "sender": headers.get("from", ""),
            "sender_name": self._extract_name(headers.get("from", "")),
            "recipients": headers.get("to", "").split(","),
            "subject": headers.get("subject", ""),
            "date": headers.get("date", ""),
            "body": body,
            "labels": msg.get("labelIds", []),
            "headers": headers,
        }
 
    def _extract_name(self, from_header: str) -> str:
        """From ヘッダーから名前を抽出"""
        if "<" in from_header:
            return from_header.split("<")[0].strip().strip('"')
        return from_header
 
    def send_reply(self, original_msg_id: str,
                   thread_id: str,
                   to: str,
                   subject: str,
                   body: str) -> dict:
        """返信メールの送信"""
        message = MIMEText(body, "plain", "utf-8")
        message["to"] = to
        message["subject"] = subject
        message["In-Reply-To"] = original_msg_id
        message["References"] = original_msg_id
 
        raw = base64.urlsafe_b64encode(
            message.as_bytes()
        ).decode("utf-8")
 
        result = self.service.users().messages().send(
            userId="me",
            body={
                "raw": raw,
                "threadId": thread_id,
            }
        ).execute()
 
        return result
 
    def add_label(self, message_id: str, label_name: str):
        """メールにラベルを追加"""
        # ラベルID取得（既存 or 新規作成）
        label_id = self._get_or_create_label(label_name)
 
        self.service.users().messages().modify(
            userId="me",
            id=message_id,
            body={
                "addLabelIds": [label_id],
            }
        ).execute()
 
    def _get_or_create_label(self, label_name: str) -> str:
        """ラベルの取得または作成"""
        labels = self.service.users().labels().list(
            userId="me"
        ).execute()
 
        for label in labels.get("labels", []):
            if label["name"] == label_name:
                return label["id"]
 
        # ラベル新規作成
        new_label = self.service.users().labels().create(
            userId="me",
            body={
                "name": label_name,
                "labelListVisibility": "labelShow",
                "messageListVisibility": "show",
            }
        ).execute()
 
        return new_label["id"]
 
    def mark_as_read(self, message_id: str):
        """既読にする"""
        self.service.users().messages().modify(
            userId="me",
            id=message_id,
            body={"removeLabelIds": ["UNREAD"]}
        ).execute()

{
  "name": "AI Email Automation Pipeline",
  "nodes": [
    {
      "name": "Gmail Trigger",
      "type": "n8n-nodes-base.gmailTrigger",
      "parameters": {
        "pollTimes": { "item": [{ "mode": "everyMinute", "minute": 5 }] },
        "filters": { "readStatus": "unread" }
      },
      "position": [100, 300]
    },
    {
      "name": "Extract Email Data",
      "type": "n8n-nodes-base.set",
      "parameters": {
        "values": {
          "string": [
            { "name": "sender", "value": "={{ $json.from }}" },
            { "name": "subject", "value": "={{ $json.subject }}" },
            { "name": "body", "value": "={{ $json.textPlain }}" }
          ]
        }
      },
      "position": [300, 300]
    },
    {
      "name": "AI Classification",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "method": "POST",
        "url": "https://api.anthropic.com/v1/messages",
        "headers": {
          "x-api-key": "={{ $credentials.anthropicApiKey }}",
          "content-type": "application/json",
          "anthropic-version": "2023-06-01"
        },
        "body": "={{ JSON.stringify({ model: 'claude-sonnet-4-20250514', max_tokens: 512, messages: [{ role: 'user', content: 'メールを分類: ' + $json.body }] }) }}"
      },
      "position": [500, 300]
    },
    {
      "name": "Route by Category",
      "type": "n8n-nodes-base.switch",
      "parameters": {
        "rules": [
          { "value": "spam", "output": 0 },
          { "value": "support", "output": 1 },
          { "value": "sales", "output": 2 },
          { "value": "billing", "output": 3 }
        ]
      },
      "position": [700, 300]
    },
    {
      "name": "Slack Notification",
      "type": "n8n-nodes-base.slack",
      "parameters": {
        "channel": "#support-alerts",
        "text": "新規メール: {{ $json.subject }} from {{ $json.sender }}"
      },
      "position": [900, 200]
    },
    {
      "name": "CRM Update",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "method": "POST",
        "url": "https://api.hubspot.com/crm/v3/objects/contacts",
        "body": "={{ JSON.stringify({ properties: { email: $json.sender } }) }}"
      },
      "position": [900, 400]
    }
  ]
}

import msal
import requests
 
 
class OutlookIntegration:
    """Microsoft Graph APIとのOutlook統合"""
 
    GRAPH_API_URL = "https://graph.microsoft.com/v1.0"
 
    def __init__(self, client_id: str, client_secret: str, tenant_id: str):
        self.app = msal.ConfidentialClientApplication(
            client_id,
            authority=f"https://login.microsoftonline.com/{tenant_id}",
            client_credential=client_secret,
        )
        self._token = None
 
    def _get_token(self) -> str:
        """アクセストークン取得"""
        if self._token:
            return self._token
 
        result = self.app.acquire_token_for_client(
            scopes=["https://graph.microsoft.com/.default"]
        )
        self._token = result["access_token"]
        return self._token
 
    def _headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json",
        }
 
    def fetch_messages(self, user_id: str,
                       folder: str = "inbox",
                       unread_only: bool = True,
                       top: int = 50) -> list[dict]:
        """メッセージ取得"""
        params = {
            "$top": top,
            "$orderby": "receivedDateTime desc",
            "$select": "id,subject,from,toRecipients,body,receivedDateTime,isRead",
        }
        if unread_only:
            params["$filter"] = "isRead eq false"
 
        response = requests.get(
            f"{self.GRAPH_API_URL}/users/{user_id}/mailFolders/{folder}/messages",
            headers=self._headers(),
            params=params,
        )
        response.raise_for_status()
 
        messages = response.json().get("value", [])
        return [self._normalize_message(msg) for msg in messages]
 
    def _normalize_message(self, msg: dict) -> dict:
        """Graph APIレスポンスを共通形式に変換"""
        sender = msg.get("from", {}).get("emailAddress", {})
        return {
            "message_id": msg["id"],
            "sender": sender.get("address", ""),
            "sender_name": sender.get("name", ""),
            "subject": msg.get("subject", ""),
            "body": msg.get("body", {}).get("content", ""),
            "date": msg.get("receivedDateTime", ""),
            "is_read": msg.get("isRead", False),
            "recipients": [
                r["emailAddress"]["address"]
                for r in msg.get("toRecipients", [])
            ],
        }
 
    def send_reply(self, user_id: str, message_id: str,
                   reply_body: str) -> dict:
        """返信送信"""
        response = requests.post(
            f"{self.GRAPH_API_URL}/users/{user_id}/messages/{message_id}/reply",
            headers=self._headers(),
            json={
                "message": {
                    "body": {
                        "contentType": "Text",
                        "content": reply_body,
                    }
                }
            }
        )
        response.raise_for_status()
        return {"status": "sent", "message_id": message_id}
 
    def create_category(self, user_id: str, display_name: str,
                        color: str = "preset0") -> dict:
        """カテゴリ作成"""
        response = requests.post(
            f"{self.GRAPH_API_URL}/users/{user_id}/outlook/masterCategories",
            headers=self._headers(),
            json={
                "displayName": display_name,
                "color": color,
            }
        )
        return response.json()

import re
from typing import Tuple
 
 
class PIIMasker:
    """個人情報の自動マスキング"""
 
    PATTERNS = {
        "email": {
            "pattern": r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
            "replacement": "[EMAIL_MASKED]",
            "description": "メールアドレス",
        },
        "phone_jp": {
            "pattern": r'0\d{1,4}-?\d{1,4}-?\d{4}',
            "replacement": "[PHONE_MASKED]",
            "description": "日本の電話番号",
        },
        "credit_card": {
            "pattern": r'\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}',
            "replacement": "[CARD_MASKED]",
            "description": "クレジットカード番号",
        },
        "my_number": {
            "pattern": r'\d{4}\s?\d{4}\s?\d{4}',
            "replacement": "[MYNUMBER_MASKED]",
            "description": "マイナンバー",
        },
        "postal_code": {
            "pattern": r'〒?\d{3}-?\d{4}',
            "replacement": "[POSTAL_MASKED]",
            "description": "郵便番号",
        },
        "ip_address": {
            "pattern": r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',
            "replacement": "[IP_MASKED]",
            "description": "IPアドレス",
        },
        "bank_account": {
            "pattern": r'[普通|当座]\s?\d{7,8}',
            "replacement": "[ACCOUNT_MASKED]",
            "description": "銀行口座番号",
        },
    }
 
    def mask(self, text: str,
             categories: list[str] = None) -> Tuple[str, list[dict]]:
        """テキスト中のPIIをマスキング"""
        masked_text = text
        detections = []
 
        patterns_to_check = (
            {k: v for k, v in self.PATTERNS.items() if k in categories}
            if categories
            else self.PATTERNS
        )
 
        for pii_type, config in patterns_to_check.items():
            matches = list(re.finditer(config["pattern"], masked_text))
            for match in matches:
                detections.append({
                    "type": pii_type,
                    "description": config["description"],
                    "position": match.span(),
                    "original_length": len(match.group()),
                })
            masked_text = re.sub(
                config["pattern"],
                config["replacement"],
                masked_text
            )
 
        return masked_text, detections
 
    def unmask(self, masked_text: str,
               original_text: str,
               detections: list[dict]) -> str:
        """マスキング解除（権限チェック後に使用）"""
        result = masked_text
        for det in reversed(detections):
            start, end = det["position"]
            original_value = original_text[start:end]
            result = result.replace(
                self.PATTERNS[det["type"]]["replacement"],
                original_value,
                1
            )
        return result
 
 
# 使用例
masker = PIIMasker()
text = "山田太郎様 (taro@example.com) へ請求書を送付。電話: 090-1234-5678"
masked, detections = masker.mask(text)
# → "山田太郎様 ([EMAIL_MASKED]) へ請求書を送付。電話: [PHONE_MASKED]"