أول سير عمل في n8n

# محاكاة n8n Workflow Nodes بـ Python
# Simulating n8n Workflow Nodes in Python
import json
import time
import urllib.request
from dataclasses import dataclass
from typing import Any

# ─── Node Classes ─────────────────────────────────────────
class HTTPRequestNode:
    """محاكاة HTTP Request Node في n8n"""
    def __init__(self, url: str, method: str = "GET"):
        self.url = url
        self.method = method

    def execute(self, _: dict) -> dict:
        req = urllib.request.Request(self.url, method=self.method)
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                data = json.loads(resp.read())
                print(f"     🌐 HTTP {self.method} → {resp.status}")
                return {"success": True, "data": data, "status": resp.status}
        except Exception as e:
            print(f"     ⚠️  HTTP Error: {e}")
            return {"success": False, "error": str(e), "data": {}}

class SetNode:
    """محاكاة Set Node — يضيف أو يعدّل حقولاً"""
    def __init__(self, mappings: dict[str, Any]):
        self.mappings = mappings

    def execute(self, data: dict) -> dict:
        result = data.copy()
        for key, value in self.mappings.items():
            result[key] = value(data) if callable(value) else value
        print(f"     🔄 Set: أضاف {list(self.mappings.keys())}")
        return result

class IFNode:
    """محاكاة IF Node — يشعّب سير العمل"""
    def __init__(self, condition: callable, label: str = ""):
        self.condition = condition
        self.label = label

    def execute(self, data: dict) -> tuple:
        passed = self.condition(data)
        marker = "✅ True" if passed else "❌ False"
        print(f"     🔀 IF [{self.label}]: {marker}")
        return (data, None) if passed else (None, data)

# ─── بناء Workflow ────────────────────────────────────────
print("🏗️  سير العمل: جلب بيانات → معالجة → تصفية → إشعار")
print("="*55)

# الـ Nodes
http_node = HTTPRequestNode("https://jsonplaceholder.typicode.com/users/1")

set_node = SetNode({
    "processed_at":  lambda _: time.strftime("%Y-%m-%d %H:%M"),
    "name_upper":    lambda d: d.get("data", {}).get("name", "").upper(),
    "email_domain":  lambda d: d.get("data", {}).get("email", "@").split("@")[-1],
    "is_important":  lambda d: d.get("data", {}).get("id", 0) <= 5,
})

filter_node = IFNode(
    condition=lambda d: d.get("is_important", False),
    label="المستخدم مهم؟",
)

# ─── تشغيل ────────────────────────────────────────────────
print("\n▶ تشغيل سير العمل...")
t0 = time.time()

step1 = http_node.execute({})
step2 = set_node.execute(step1)
passed, rejected = filter_node.execute(step2)

elapsed = round((time.time() - t0) * 1000)

if passed:
    print(f"\n✅ اكتمل في {elapsed}ms")
    print(f"   الاسم:    {passed.get('name_upper')}")
    print(f"   الدومين:  {passed.get('email_domain')}")
    print(f"   الوقت:    {passed.get('processed_at')}")
    print(f"   الحالة:   مهم → سيُرسل إشعار Slack")
else:
    print(f"\n⏭️  غير مهم — تخطّي الإشعار ({elapsed}ms)")

# ─── محاكاة Executions Log ───────────────────────────────
print("\n📋 Executions Log:")
for i, node_name in enumerate(["HTTP Request", "Set Fields", "IF Filter"], 1):
    status = "✅" if not (i == 3 and rejected) else "⏭️"
    print(f"   {i}. {status} {node_name}")