مدخل إلى الأتمتة

# محاكاة مفاهيم الأتمتة — Automation Core Concepts Demo
import time
import json
from dataclasses import dataclass
from typing import Callable

@dataclass
class Node:
    name: str
    fn: Callable[[dict], dict]
    node_type: str = "action"

class Workflow:
    """محاكاة بسيطة لـ n8n Workflow"""
    def __init__(self, name: str):
        self.name = name
        self.nodes: list[Node] = []

    def add_node(self, node: Node) -> "Workflow":
        self.nodes.append(node)
        return self

    def run(self, trigger_data: dict) -> dict:
        icons = {"trigger": "⚡", "transform": "🔄", "action": "▶", "filter": "🔀"}
        print(f"\n🚀 [{self.name}] بدأ")
        print(f"   المحفّز: {trigger_data.get('subject', trigger_data)}")
        data = trigger_data.copy()
        for node in self.nodes:
            icon = icons.get(node.node_type, "▶")
            print(f"  {icon} {node.name}...", end=" ")
            data = node.fn(data)
            print("✅")
        print(f"✅ [{self.name}] اكتمل")
        return data

# ─── تعريف الـ Nodes ──────────────────────────────────────
def check_priority(data: dict) -> dict:
    keywords = ["عاجل", "urgent", "ASAP", "طارئ"]
    data["is_urgent"] = any(k in data.get("subject", "") for k in keywords)
    return data

def classify_email(data: dict) -> dict:
    s = data.get("subject", "").lower()
    data["category"] = (
        "billing"   if any(w in s for w in ["فاتورة", "invoice"]) else
        "technical" if any(w in s for w in ["خطأ", "error", "bug"]) else
        "sales"     if any(w in s for w in ["طلب", "order", "سعر"]) else
        "general"
    )
    return data

def generate_reply(data: dict) -> dict:
    replies = {
        "billing":   "سيتولى فريق الفوترة الرد خلال 24 ساعة.",
        "technical": "سيتواصل معك مهندس الدعم خلال ساعتين.",
        "sales":     "سيتواصل معك فريق المبيعات قريباً.",
        "general":   "شكراً لرسالتك. سنرد قريباً.",
    }
    data["auto_reply"] = replies.get(data["category"], replies["general"])
    return data

def log_to_sheet(data: dict) -> dict:
    print(f"\n     📊 تسجيل: {data['category']} | {'عاجل' if data['is_urgent'] else 'عادي'}")
    return {**data, "logged": True}

def notify_slack(data: dict) -> dict:
    if data.get("is_urgent"):
        print(f"\n     📢 Slack #urgent: بريد عاجل من {data.get('from')}")
    return data

# ─── بناء سير العمل ───────────────────────────────────────
workflow = (
    Workflow("معالجة البريد الإلكتروني")
    .add_node(Node("تحديد الأولوية",   check_priority,  "transform"))
    .add_node(Node("تصنيف البريد",     classify_email,  "transform"))
    .add_node(Node("توليد الرد",       generate_reply,  "transform"))
    .add_node(Node("تسجيل في Sheets",  log_to_sheet,    "action"))
    .add_node(Node("إرسال Slack",      notify_slack,    "action"))
)

# ─── تشغيل ────────────────────────────────────────────────
emails = [
    {"from": "client@co.com",      "subject": "مشكلة عاجلة في الخادم"},
    {"from": "billing@vendor.com", "subject": "فاتورة شهر يونيو"},
    {"from": "friend@email.com",   "subject": "مرحباً كيف حالك"},
]

for email in emails:
    result = workflow.run(email)
    print(f"   الرد: {result['auto_reply']}\n")