Kafka

#!/usr/bin/env python3
"""
Kafka to MoEngage Consumer
Reads events from Kafka and sends to MoEngage using standard JSON format
"""
import os
import json
import requests
import time
import base64
from dotenv import load_dotenv
from confluent_kafka import Consumer, KafkaError

load_dotenv()

# Configuration
MOENGAGE_PARTNER_NAME = os.getenv('MOENGAGE_PARTNER_NAME')
MOENGAGE_DATA_CENTER = os.getenv('MOENGAGE_DATA_CENTER')
MOENGAGE_CONFIG_NAME = os.getenv('MOENGAGE_CONFIG_NAME')
MOENGAGE_APP_ID = os.getenv('MOENGAGE_APP_ID')
MOENGAGE_API_SECRET = os.getenv('MOENGAGE_API_SECRET')

KAFKA_BOOTSTRAP_SERVERS = os.getenv('KAFKA_BOOTSTRAP_SERVERS')
KAFKA_TOPICS = os.getenv('KAFKA_TOPICS', 'events').split(',')
KAFKA_GROUP_ID = os.getenv('KAFKA_GROUP_ID')
KAFKA_SECURITY_PROTOCOL = os.getenv('KAFKA_SECURITY_PROTOCOL', 'PLAINTEXT')
KAFKA_SASL_MECHANISM = os.getenv('KAFKA_SASL_MECHANISM', 'PLAIN')
KAFKA_API_KEY = os.getenv('KAFKA_API_KEY')
KAFKA_API_SECRET = os.getenv('KAFKA_API_SECRET')

LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')

# MoEngage endpoint with configName
MOENGAGE_ENDPOINT = (
    f'https://api-{MOENGAGE_DATA_CENTER}.moengage.com/v1/partner/'
    f'{MOENGAGE_PARTNER_NAME}/events/?configName={MOENGAGE_CONFIG_NAME}'
)

def log(level, msg):
    """Simple logging"""
    if level in ['ERROR', 'WARNING', 'INFO', 'DEBUG']:
        print(f"[{level}] {msg}")

def create_basic_auth():
    """Create Basic Auth header"""
    credentials = f"{MOENGAGE_APP_ID}:{MOENGAGE_API_SECRET}"
    encoded = base64.b64encode(credentials.encode()).decode()
    return f"Basic {encoded}"

def send_to_moengage(kafka_message):
    """Send Kafka message to MoEngage with retries"""
    headers = {
        'Content-Type': 'application/json',
        'Authorization': create_basic_auth()
    }

    for attempt in range(3):
        try:
            response = requests.post(
                MOENGAGE_ENDPOINT,
                json=kafka_message,
                headers=headers,
                timeout=10
            )

            if response.status_code in [200, 201, 202, 204, 207]:
                log('INFO', f"Event sent successfully (status {response.status_code})")
                return True
            elif response.status_code >= 500:
                log('WARNING', f"Server error {response.status_code}, retrying...")
                time.sleep(2 ** attempt)
                continue
            else:
                log('ERROR', f"API error {response.status_code}: {response.text}")
                return False

        except Exception as e:
            log('ERROR', f"Send failed (attempt {attempt+1}): {e}")
            if attempt < 2:
                continue
            return False
    return False

# ... main consumer loop continues here (see source for full script) ...