for Marketing Data Analysis vs Customer Support: Which Wins?

import os
import time
import logging
from amplitude import Amplitude, BaseEvent
from amplitude.exception import AmplitudeException
from typing import List, Dict, Any

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("amplitude_events.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Initialize Amplitude client with retry config
# Benchmark environment: Amplitude Python SDK 3.2.1, Python 3.11.4, AWS c6g.4xlarge
# Amplitude Python SDK repo: https://github.com/amplitude/amplitude-python
AMPLITUDE_API_KEY = os.getenv("AMPLITUDE_API_KEY", "test-api-key-12345")
client = Amplitude(AMPLITUDE_API_KEY)
client.configuration.retry_handler.max_retries = 3
client.configuration.retry_handler.retry_delay_ms = 500
client.configuration.batch_handler.batch_size = 100
client.configuration.batch_handler.flush_interval_ms = 5000

def track_marketing_events(events: List[Dict[str, Any]]) -> int:
    """
    Track marketing attribution events with Amplitude, returns success count.
    Includes error handling for rate limits, network errors, and invalid events.
    """
    success_count = 0
    for event_data in events:
        try:
            # Validate required fields per Amplitude schema
            if not all(k in event_data for k in ["user_id", "event_type", "timestamp"]):
                logger.warning(f"Invalid event missing required fields: {event_data}")
                continue

            event = BaseEvent(
                user_id=event_data["user_id"],
                event_type=event_data["event_type"],
                time=int(event_data["timestamp"]),
                event_properties=event_data.get("properties", {}),
                user_properties=event_data.get("user_properties", {})
            )

            # Send event with batching (auto-handled by client)
            client.track(event)
            success_count += 1
            logger.debug(f"Tracked event {event_data['event_type']} for user {event_data['user_id']}")

        except AmplitudeException as e:
            logger.error(f"Amplitude API error: {str(e)} | Event: {event_data}")
            if e.status_code == 429:  # Rate limit
                logger.info("Rate limited, sleeping 2s")
                time.sleep(2)
        except ConnectionError as e:
            logger.error(f"Network error: {str(e)} | Event: {event_data}")
            time.sleep(1)
        except Exception as e:
            logger.error(f"Unexpected error: {str(e)} | Event: {event_data}")

    # Flush remaining batched events
    try:
        client.flush()
        logger.info(f"Flushed batch, total success: {success_count}")
    except Exception as e:
        logger.error(f"Flush failed: {str(e)}")

    return success_count

if __name__ == "__main__":
    # Test event batch: 50 marketing events (signups, ad clicks, page views)
    test_events = [
        {
            "user_id": f"user_{i}",
            "event_type": "marketing_email_click",
            "timestamp": int(time.time() * 1000) - (i * 1000),
            "properties": {"campaign_id": "summer_sale_2024", "channel": "email"},
            "user_properties": {"signup_source": "email"}
        } for i in range(50)
    ]

    start = time.time()
    success = track_marketing_events(test_events)
    elapsed = time.time() - start

    print(f"Tracked {success}/{len(test_events)} events in {elapsed:.2f}s")
    print(f"Throughput: {success/elapsed:.2f} events/sec")

import os
import time
import logging
from typing import List, Dict, Any
from zendesk import Zendesk, ZendeskException
from zendesk.models.ticket import Ticket, TicketPriority, TicketStatus

# Configure logging for support ticket audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("zendesk_tickets.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Initialize Zendesk client
# Benchmark environment: Zendesk Python SDK 2.1.0, Python 3.11.4, AWS c6g.4xlarge
# Zendesk SDK repo: https://github.com/zendesk/python-zendesk
ZENDESK_SUBDOMAIN = os.getenv("ZENDESK_SUBDOMAIN", "test-subdomain")
ZENDESK_EMAIL = os.getenv("ZENDESK_EMAIL", "test@company.com")
ZENDESK_API_TOKEN = os.getenv("ZENDESK_API_TOKEN", "test-token-12345")

client = Zendesk(
    subdomain=ZENDESK_SUBDOMAIN,
    email=ZENDESK_EMAIL,
    token=ZENDESK_API_TOKEN,
    timeout=10
)

def create_support_tickets(tickets: List[Dict[str, Any]]) -> int:
    """
    Create customer support tickets in Zendesk, returns success count.
    Handles rate limits, validation errors, and network failures.
    """
    success_count = 0
    for ticket_data in tickets:
        try:
            # Validate required fields per Zendesk API schema
            required = ["requester_email", "subject", "description"]
            if not all(k in ticket_data for k in required):
                logger.warning(f"Invalid ticket missing required fields: {ticket_data}")
                continue

            # Map priority string to Zendesk enum
            priority_map = {
                "low": TicketPriority.LOW,
                "normal": TicketPriority.NORMAL,
                "high": TicketPriority.HIGH,
                "urgent": TicketPriority.URGENT
            }
            priority = priority_map.get(ticket_data.get("priority", "normal"), TicketPriority.NORMAL)

            ticket = Ticket(
                requester_email=ticket_data["requester_email"],
                subject=ticket_data["subject"],
                description=ticket_data["description"],
                priority=priority,
                status=TicketStatus.OPEN,
                tags=ticket_data.get("tags", []),
                custom_fields=ticket_data.get("custom_fields", {})
            )

            # Create ticket via API
            created = client.tickets.create(ticket)
            success_count += 1
            logger.debug(f"Created ticket {created.id} for {ticket_data['requester_email']}")

        except ZendeskException as e:
            logger.error(f"Zendesk API error: {str(e)} | Ticket: {ticket_data}")
            if e.response.status_code == 429:  # Rate limit
                retry_after = int(e.response.headers.get("Retry-After", 2))
                logger.info(f"Rate limited, sleeping {retry_after}s")
                time.sleep(retry_after)
        except ConnectionError as e:
            logger.error(f"Network error: {str(e)} | Ticket: {ticket_data}")
            time.sleep(1)
        except Exception as e:
            logger.error(f"Unexpected error: {str(e)} | Ticket: {ticket_data}")

    logger.info(f"Total tickets created: {success_count}/{len(tickets)}")
    return success_count

if __name__ == "__main__":
    # Test ticket batch: 30 support tickets (refunds, bugs, onboarding)
    test_tickets = [
        {
            "requester_email": f"customer_{i}@example.com",
            "subject": f"Refund request for order #{1000 + i}",
            "description": f"Customer {i} requests refund for delayed order.",
            "priority": "high" if i % 3 == 0 else "normal",
            "tags": ["refund", "order_issue"],
            "custom_fields": {"order_id": f"{1000 + i}", "refund_amount": f"${49.99 + i}"}
        } for i in range(30)
    ]

    start = time.time()
    success = create_support_tickets(test_tickets)
    elapsed = time.time() - start

    print(f"Created {success}/{len(test_tickets)} tickets in {elapsed:.2f}s")
    print(f"Throughput: {success/elapsed:.2f} tickets/sec")

import os
import time
import logging
from typing import List, Dict, Any
from amplitude import Amplitude
from zendesk import Zendesk
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

# Configure logging for ETL pipelines
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("reverse_etl.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Initialize clients (repos: https://github.com/amplitude/amplitude-python, https://github.com/zendesk/python-zendesk)
amplitude_client = Amplitude(os.getenv("AMPLITUDE_API_KEY", "test-key"))
zendesk_client = Zendesk(
    subdomain=os.getenv("ZENDESK_SUBDOMAIN", "test-subdomain"),
    email=os.getenv("ZENDESK_EMAIL", "test@company.com"),
    token=os.getenv("ZENDESK_API_TOKEN", "test-token")
)

# PostgreSQL connection for data warehouse (benchmark: PG 16.2, AWS RDS c6g.2xlarge)
DB_CONNECTION_STRING = os.getenv("DB_CONNECTION_STRING", "postgresql://user:pass@localhost:5432/company_db")
engine = create_engine(DB_CONNECTION_STRING, pool_size=20, max_overflow=10)

def sync_marketing_support_data(start_date: str, end_date: str) -> Dict[str, int]:
    """
    Sync Amplitude marketing events and Zendesk tickets to unified data warehouse.
    Returns counts of synced records per tool.
    """
    sync_stats = {"amplitude_events": 0, "zendesk_tickets": 0, "errors": 0}

    # 1. Fetch Amplitude events via Export API
    try:
        logger.info(f"Fetching Amplitude events from {start_date} to {end_date}")
        # Note: Amplitude Export API requires date in YYYYMMDD format
        amplitude_events = amplitude_client.export(
            start_date=start_date.replace("-", ""),
            end_date=end_date.replace("-", "")
        )

        # Batch insert into PG
        with engine.begin() as conn:
            for event in amplitude_events:
                try:
                    stmt = text("""
                        INSERT INTO unified_events (user_id, event_type, timestamp, properties, source)
                        VALUES (:user_id, :event_type, to_timestamp(:ts/1000), :properties, 'amplitude')
                        ON CONFLICT (event_id) DO NOTHING
                    """)
                    conn.execute(stmt, {
                        "user_id": event.get("user_id"),
                        "event_type": event.get("event_type"),
                        "ts": event.get("time"),
                        "properties": event.get("event_properties", {})
                    })
                    sync_stats["amplitude_events"] += 1
                except Exception as e:
                    logger.error(f"Amplitude event insert failed: {str(e)}")
                    sync_stats["errors"] += 1
        logger.info(f"Synced {sync_stats['amplitude_events']} Amplitude events")

    except Exception as e:
        logger.error(f"Amplitude export failed: {str(e)}")
        sync_stats["errors"] += 1

    # 2. Fetch Zendesk tickets via API
    try:
        logger.info(f"Fetching Zendesk tickets from {start_date} to {end_date}")
        tickets = zendesk_client.tickets.list(
            start_time=int(time.mktime(time.strptime(start_date, "%Y-%m-%d"))),
            end_time=int(time.mktime(time.strptime(end_date, "%Y-%m-%d")))
        )

        with engine.begin() as conn:
            for ticket in tickets:
                try:
                    stmt = text("""
                        INSERT INTO unified_tickets (ticket_id, requester_email, subject, status, priority, created_at, source)
                        VALUES (:ticket_id, :requester_email, :subject, :status, :priority, to_timestamp(:created), 'zendesk')
                        ON CONFLICT (ticket_id) DO NOTHING
                    """)
                    conn.execute(stmt, {
                        "ticket_id": ticket.id,
                        "requester_email": ticket.requester_email,
                        "subject": ticket.subject,
                        "status": ticket.status.value,
                        "priority": ticket.priority.value,
                        "created": ticket.created_at.timestamp()
                    })
                    sync_stats["zendesk_tickets"] += 1
                except Exception as e:
                    logger.error(f"Zendesk ticket insert failed: {str(e)}")
                    sync_stats["errors"] += 1
        logger.info(f"Synced {sync_stats['zendesk_tickets']} Zendesk tickets")

    except Exception as e:
        logger.error(f"Zendesk ticket fetch failed: {str(e)}")
        sync_stats["errors"] += 1

    return sync_stats

if __name__ == "__main__":
    start = time.time()
    stats = sync_marketing_support_data("2024-06-01", "2024-06-07")
    elapsed = time.time() - start

    print(f"Sync complete in {elapsed:.2f}s")
    print(f"Amplitude events: {stats['amplitude_events']}")
    print(f"Zendesk tickets: {stats['zendesk_tickets']}")
    print(f"Errors: {stats['errors']}")
    print(f"Total throughput: {(stats['amplitude_events'] + stats['zendesk_tickets'])/elapsed:.2f} records/sec")

# Amplitude batch config snippet
client.configuration.batch_handler.batch_size = 100
client.configuration.batch_handler.flush_interval_ms = 5000  # 5 seconds

# Incremental sync snippet for Amplitude
cursor = get_last_sync_cursor()  # Load from warehouse
events = amplitude_client.export(start_date=cursor, end_date=now())
# Process and insert events, update cursor

# Zendesk rate limit retry snippet
except ZendeskException as e:
    if e.response.status_code == 429:
        retry_after = int(e.response.headers.get("Retry-After", 2))
        time.sleep(retry_after)