أتمتة سير العمل بـ n8n وMake.com

// Example n8n workflow for lead processing
{
  "nodes": [
    {
      "name": "Webhook Trigger",
      "type": "n8n-nodes-base.webhook",
      "parameters": {
        "path": "lead-webhook",
        "httpMethod": "POST"
      }
    },
    {
      "name": "Data Validation",
      "type": "n8n-nodes-base.function",
      "parameters": {
        "functionCode": `
          const data = $input.first().json;
          
          // Validate required fields
          if (!data.email || !data.name) {
            throw new Error('Missing required fields');
          }
          
          // Clean and format data
          return {
            email: data.email.toLowerCase().trim(),
            name: data.name.trim(),
            phone: data.phone ? data.phone.replace(/[^0-9]/g, '') : null,
            source: data.source || 'website',
            timestamp: new Date().toISOString()
          };
        `
      }
    },
    {
      "name": "CRM Integration",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "url": "https://api.crm.com/contacts",
        "method": "POST",
        "headers": {
          "Authorization": "Bearer {{$credentials.crmApiKey}}",
          "Content-Type": "application/json"
        },
        "body": "={{$json}}"
      }
    }
  ]
}

{
  "scenario": {
    "name": "Booking Management Workflow",
    "modules": [
      {
        "name": "Booking Webhook",
        "type": "webhook",
        "url": "https://hook.eu1.make.com/booking-trigger"
      },
      {
        "name": "Data Processing",
        "type": "data-store",
        "operation": "create",
        "fields": {
          "booking_id": "{{1.booking_id}}",
          "customer_email": "{{1.customer_email}}",
          "check_in": "{{1.check_in_date}}",
          "check_out": "{{1.check_out_date}}",
          "status": "confirmed"
        }
      },
      {
        "name": "Send Confirmation Email",
        "type": "email",
        "to": "{{2.customer_email}}",
        "subject": "Booking Confirmation - {{2.booking_id}}",
        "body": "Your booking has been confirmed..."
      }
    ]
  }
}

# Python script for complex data processing
import requests
import json
from datetime import datetime
 
class LeadProcessor:
    def __init__(self):
        self.crm_api = "https://api.crm.com"
        self.email_api = "https://api.email.com"
        self.analytics_api = "https://api.analytics.com"
    
    def process_lead(self, lead_data):
        # Step 1: Data enrichment
        enriched_data = self.enrich_lead_data(lead_data)
        
        # Step 2: Lead scoring
        score = self.calculate_lead_score(enriched_data)
        
        # Step 3: CRM integration
        crm_response = self.create_crm_contact(enriched_data, score)
        
        # Step 4: Automated follow-up
        if score > 70:
            self.schedule_immediate_followup(crm_response['contact_id'])
        else:
            self.schedule_nurture_sequence(crm_response['contact_id'])
        
        # Step 5: Analytics tracking
        self.track_lead_conversion(lead_data['source'], score)
        
        return crm_response
    
    def enrich_lead_data(self, lead):
        # Company information lookup
        company_info = self.get_company_data(lead['email'])
        
        # Social media profiles
        social_profiles = self.find_social_profiles(lead['email'])
        
        # Geographic data
        location_data = self.get_location_data(lead['ip_address'])
        
        return {
            **lead,
            'company': company_info,
            'social_profiles': social_profiles,
            'location': location_data,
            'enriched_at': datetime.now().isoformat()
        }

# n8n workflow configuration
workflow:
  name: "Content Generation Pipeline"
  nodes:
    - name: "Content Trigger"
      type: "schedule"
      parameters:
        cron: "0 9 * * 1"  # Every Monday at 9 AM
    
    - name: "Topic Research"
      type: "http-request"
      parameters:
        url: "https://api.content-research.com/trending"
        method: "GET"
    
    - name: "AI Content Generation"
      type: "openai"
      parameters:
        model: "gpt-4"
        prompt: |
          Create engaging content about: {{$json.topic}}
          Target audience: Real estate professionals
          Tone: Professional but approachable
          Length: 500-800 words
          Include: Practical tips and actionable advice
    
    - name: "Content Optimization"
      type: "function"
      parameters:
        code: |
          const content = $input.first().json;
          
          // SEO optimization
          const seoScore = calculateSEOScore(content);
          
          // Readability check
          const readabilityScore = calculateReadability(content);
          
          // Social media adaptation
          const socialMediaPosts = adaptForSocialMedia(content);
          
          return {
            original_content: content,
            seo_score: seoScore,
            readability_score: readabilityScore,
            social_posts: socialMediaPosts,
            optimized_content: optimizeContent(content)
          };
    
    - name: "Multi-Channel Publishing"
      type: "parallel"
      branches:
        - name: "Blog Post"
          type: "wordpress"
          parameters:
            title: "{{$json.optimized_content.title}}"
            content: "{{$json.optimized_content.body}}"
            status: "publish"
        
        - name: "Social Media"
          type: "social-media"
          parameters:
            platforms: ["linkedin", "twitter", "facebook"]
            posts: "{{$json.social_posts}}"
        
        - name: "Email Newsletter"
          type: "email-marketing"
          parameters:
            subject: "{{$json.optimized_content.title}}"
            content: "{{$json.optimized_content.body}}"
            list: "subscribers"

// Make.com scenario for booking management
const bookingWorkflow = {
  trigger: {
    type: "webhook",
    url: "https://hook.eu1.make.com/booking-webhook"
  },
  
  modules: [
    {
      name: "Validate Booking Data",
      type: "data-store",
      operation: "create",
      conditions: [
        {
          field: "email",
          operator: "is_not_empty"
        },
        {
          field: "check_in_date",
          operator: "is_not_empty"
        }
      ]
    },
    
    {
      name: "Check Availability",
      type: "http-request",
      url: "https://api.property-management.com/availability",
      method: "POST",
      body: {
        property_id: "{{1.property_id}}",
        check_in: "{{1.check_in_date}}",
        check_out: "{{1.check_out_date}}"
      }
    },
    
    {
      name: "Create Booking",
      type: "http-request",
      url: "https://api.booking-system.com/bookings",
      method: "POST",
      body: {
        customer: "{{1}}",
        property: "{{2.property}}",
        dates: "{{1.check_in_date}} to {{1.check_out_date}}",
        status: "confirmed"
      }
    },
    
    {
      name: "Send Confirmation",
      type: "email",
      to: "{{1.email}}",
      subject: "Booking Confirmation - {{3.booking_id}}",
      template: "booking-confirmation",
      variables: {
        booking_id: "{{3.booking_id}}",
        property_name: "{{3.property_name}}",
        check_in: "{{1.check_in_date}}",
        check_out: "{{1.check_out_date}}"
      }
    },
    
    {
      name: "Schedule Reminders",
      type: "schedule",
      delay: "7 days",
      action: "send_reminder",
      booking_id: "{{3.booking_id}}"
    }
  ]
};

# Error handling in automation workflows
class WorkflowMonitor:
    def __init__(self):
        self.alert_system = AlertSystem()
        self.logger = WorkflowLogger()
    
    def handle_workflow_error(self, error, context):
        # Log the error
        self.logger.log_error(error, context)
        
        # Determine error severity
        severity = self.assess_error_severity(error)
        
        # Send appropriate alerts
        if severity == "critical":
            self.alert_system.send_immediate_alert(error, context)
        elif severity == "warning":
            self.alert_system.send_daily_summary(error, context)
        
        # Attempt recovery
        if self.can_recover(error):
            return self.attempt_recovery(error, context)
        
        return False
    
    def assess_error_severity(self, error):
        critical_errors = [
            "database_connection_failed",
            "payment_processing_failed",
            "email_delivery_failed"
        ]
        
        if any(critical in str(error).lower() for critical in critical_errors):
            return "critical"
        elif "timeout" in str(error).lower():
            return "warning"
        else:
            return "info"

// Data validation in n8n workflows
const validateData = (data) => {
  const errors = [];
  
  // Email validation
  if (!data.email || !isValidEmail(data.email)) {
    errors.push('Invalid email address');
  }
  
  // Phone validation
  if (data.phone && !isValidPhone(data.phone)) {
    errors.push('Invalid phone number');
  }
  
  // Date validation
  if (data.check_in_date && !isValidDate(data.check_in_date)) {
    errors.push('Invalid check-in date');
  }
  
  // Business logic validation
  if (data.check_in_date && data.check_out_date) {
    if (new Date(data.check_in_date) >= new Date(data.check_out_date)) {
      errors.push('Check-out date must be after check-in date');
    }
  }
  
  return {
    isValid: errors.length === 0,
    errors: errors,
    data: errors.length === 0 ? cleanData(data) : null
  };
};

# Caching and performance optimization
import redis
from functools import lru_cache
import asyncio
 
class WorkflowOptimizer:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.cache_ttl = 3600  # 1 hour
    
    @lru_cache(maxsize=1000)
    async def get_cached_data(self, key):
        # Check Redis cache first
        cached = self.redis_client.get(key)
        if cached:
            return json.loads(cached)
        
        # Fetch from source if not cached
        data = await self.fetch_from_source(key)
        
        # Cache the result
        self.redis_client.setex(key, self.cache_ttl, json.dumps(data))
        
        return data
    
    async def batch_process(self, items, batch_size=100):
        """Process items in batches for better performance"""
        results = []
        
        for i in range(0, len(items), batch_size):
            batch = items[i:i + batch_size]
            batch_results = await asyncio.gather(*[
                self.process_item(item) for item in batch
            ])
            results.extend(batch_results)
        
        return results