مشروع: أتمتة ردود البريد الإلكتروني

#!/usr/bin/env python3
"""
مشروع: بوت أتمتة ردود البريد الإلكتروني
Project: Email Response Automation Bot

Architecture: Emails → Claude AI → Route → Auto-reply + Log
"""

import os
import json
import time
import anthropic
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class Email:
    id: str
    from_addr: str
    subject: str
    body: str
    received_at: str = field(
        default_factory=lambda: datetime.now().strftime("%H:%M:%S")
    )

@dataclass
class EmailAnalysis:
    category: str = "general"
    priority: str = "low"
    sentiment: str = "neutral"
    auto_reply: str = ""
    requires_human: bool = False
    summary: str = ""
    tokens_used: int = 0

class EmailAutomationBot:
    """
    بوت أتمتة كامل لمعالجة البريد الإلكتروني بـ Claude
    Full email automation bot powered by Claude AI
    """

    SYSTEM_PROMPT = """أنت نظام ذكي لمعالجة البريد الإلكتروني.
حلّل كل بريد وارد وأعد JSON بهذا الشكل بالضبط:
{
  "category": "technical|billing|sales|hr|general|spam",
  "priority": "high|normal|low",
  "sentiment": "positive|neutral|negative|angry",
  "auto_reply": "رد مهني ومناسب بالعربية (2-3 جمل)",
  "requires_human": false,
  "summary": "ملخص في جملة واحدة"
}
requires_human = true فقط للمشاكل التقنية العميقة أو طلبات استرداد كبيرة."""

    def __init__(self):
        self.claude = anthropic.Anthropic(
            api_key=os.environ.get("ANTHROPIC_API_KEY", "demo")
        )
        self.processed: list[tuple[Email, EmailAnalysis]] = []
        self.stats = {"total": 0, "auto": 0, "human": 0, "errors": 0, "tokens": 0}

    # ─── Nodes (كل دالة = Node واحد في n8n) ──────────────

    def gmail_trigger(self, emails: list[Email]) -> list[Email]:
        """Node 1: Gmail Trigger"""
        print(f"⚡ Gmail Trigger: {len(emails)} بريد وارد")
        return emails

    def claude_node(self, email: Email) -> EmailAnalysis:
        """Node 2: HTTP Request → Claude API"""
        prompt = (
            f"من: {email.from_addr}\n"
            f"الموضوع: {email.subject}\n"
            f"المحتوى:\n{email.body}"
        )
        msg = self.claude.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=600,
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": prompt}],
        )
        raw = msg.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1].replace("json", "").strip()

        data = json.loads(raw)
        tokens = msg.usage.input_tokens + msg.usage.output_tokens
        self.stats["tokens"] += tokens

        return EmailAnalysis(
            category=data.get("category", "general"),
            priority=data.get("priority", "low"),
            sentiment=data.get("sentiment", "neutral"),
            auto_reply=data.get("auto_reply", ""),
            requires_human=data.get("requires_human", False),
            summary=data.get("summary", ""),
            tokens_used=tokens,
        )

    def router_node(self, analysis: EmailAnalysis) -> str:
        """Node 3: Switch Router"""
        if analysis.priority == "high":
            return "urgent"
        elif not analysis.requires_human:
            return "auto_reply"
        return "human_queue"

    def slack_alert_node(self, email: Email, analysis: EmailAnalysis):
        """Node 4a: Slack Alert"""
        print(f"  📢 Slack #urgent @here:")
        print(f"     من: {email.from_addr}")
        print(f"     الملخص: {analysis.summary}")

    def auto_reply_node(self, email: Email, analysis: EmailAnalysis):
        """Node 4b: Gmail Auto Reply"""
        print(f"  📧 Auto Reply → {email.from_addr}")
        print(f"     '{analysis.auto_reply[:70]}...'")

    def human_queue_node(self, email: Email, analysis: EmailAnalysis):
        """Node 4c: Human Review Queue"""
        print(f"  🔔 Human Queue [{analysis.category}]: {email.subject[:40]}")

    def sheets_node(self, email: Email, analysis: EmailAnalysis):
        """Node 5: Google Sheets"""
        row = (f"{email.from_addr} | {analysis.category} | "
               f"{analysis.priority} | {analysis.tokens_used}t")
        print(f"  📊 Sheets: {row}")

    # ─── تشغيل سير العمل ──────────────────────────────────

    def process_single(self, email: Email):
        t0 = time.time()
        self.stats["total"] += 1
        try:
            print(f"\n{'─'*55}")
            print(f"📨 [{email.id}] {email.subject[:48]}")

            analysis = self.claude_node(email)
            route = self.router_node(analysis)

            p_icon = {"high": "🔴", "normal": "🟡", "low": "🟢"}.get(
                analysis.priority, "⚪"
            )
            print(f"  {p_icon} {analysis.category} | {analysis.sentiment} | {route}")

            if route == "urgent":
                self.slack_alert_node(email, analysis)
                self.stats["human"] += 1
            elif route == "auto_reply":
                self.auto_reply_node(email, analysis)
                self.stats["auto"] += 1
            else:
                self.human_queue_node(email, analysis)
                self.stats["human"] += 1

            self.sheets_node(email, analysis)
            self.processed.append((email, analysis))

        except Exception as e:
            self.stats["errors"] += 1
            print(f"  ⚠️  Error: {e}")

        print(f"  ⏱️  {round((time.time()-t0)*1000)}ms")

    def run(self, emails: list[Email]):
        print("🤖 Email Automation Bot — بدأ")
        print("="*55)
        triggered = self.gmail_trigger(emails)
        for email in triggered:
            self.process_single(email)

        total = self.stats["total"]
        auto_pct = self.stats["auto"] / max(total, 1) * 100
        print(f"\n{'='*55}")
        print(f"📊 الإحصائيات:")
        print(f"   إجمالي: {total} | تلقائي: {self.stats['auto']} | بشري: {self.stats['human']}")
        print(f"   إجمالي Tokens: {self.stats['tokens']:,}")
        print(f"   معدل الأتمتة: {auto_pct:.0f}%")
        print("\n✅ مشروع الأتمتة مكتمل! 🎉")

# ─── تشغيل المشروع ────────────────────────────────────────
bot = EmailAutomationBot()

sample_emails = [
    Email("E001", "angry@enterprise.com",
          "الموقع معطل منذ 3 ساعات ونفقد الطلبات!",
          "نظامنا لا يعمل والمبيعات تتوقف. هذا غير مقبول تماماً."),
    Email("E002", "happy@user.com",
          "شكراً على الخدمة الرائعة",
          "المنتج رائع. أريد ترقية خطتي للخطة المتقدمة."),
    Email("E003", "info@company.com",
          "استفسار عن أسعار الخطة المؤسسية",
          "نحن فريق 200 موظف. ما هي الأسعار والعقود السنوية؟"),
    Email("E004", "cfo@bigcorp.com",
          "طلب استرداد مبلغ 12000 دولار بسبب الانقطاع",
          "بسبب الانقطاع الذي استمر 48 ساعة نطالب باسترداد كامل رسوم الشهر."),
]

bot.run(sample_emails)