getresponse_client.py

#!/usr/bin/env python3
"""
GetResponse API Integration Class
================================

A comprehensive Python class for interacting with the GetResponse API v3.
This class provides methods to manage contacts, campaigns, and perform
various email marketing operations.

Features:
- Add contacts to campaigns/lists
- Remove contacts from campaigns
- Search and manage contacts
- Get campaign information
- Error handling and logging
- Environment variable integration

Requirements:
- Python 3.6+
- requests library
- python-dotenv for environment variables

Setup:
1. Install required packages: pip install requests python-dotenv
2. Create a .env file with: GETRESPONSE_API_KEY=your_api_key_here
3. Import and use the GetResponseClient class

Author: Agent Zero
Date: 2025-11-06
"""

import os
import json
import requests
from typing import Dict, List, Optional, Union
from dotenv import load_dotenv


class GetResponseClient:
    """
    A versatile client for interacting with the GetResponse API v3.
    
    This class provides comprehensive methods to manage contacts, campaigns,
    and perform email marketing operations through the GetResponse API.
    
    Attributes:
        base_url (str): The base URL for GetResponse API
        api_key (str): The API key for authentication
        headers (dict): HTTP headers including authentication
    """
    
    def __init__(self, api_key: Optional[str] = None, base_url: str = "https://api.getresponse.com/v3"):
        """
        Initialize the GetResponse client.
        
        Args:
            api_key (str, optional): GetResponse API key. If not provided,
                will attempt to load from environment variable GETRESPONSE_API_KEY.
            base_url (str): Base URL for GetResponse API. Defaults to retail API.
                For MAX platform users, use appropriate URL.
                
        Raises:
            ValueError: If API key is not provided and not found in environment.
        """
        load_dotenv()  # Load environment variables from .env file
        
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key or os.getenv('GETRESPONSE_API_KEY')
        
        if not self.api_key:
            raise ValueError(
                "API key is required. Either pass it directly or set GETRESPONSE_API_KEY "
                "environment variable or in .env file."
            )
        
        self.headers = {
            'X-Auth-Token': f'api-key {self.api_key}',
            'Content-Type': 'application/json'
        }
    
    def _make_request(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict] = None,
        params: Optional[Dict] = None
    ) -> Dict:
        """
        Make HTTP request to GetResponse API (v3).
        Handles empty 202/204 responses gracefully.
        """
        url = f"{self.base_url}{endpoint}"

        try:
            response = requests.request(
                method=method,
                url=url,
                headers=self.headers,
                json=data,
                params=params,
                timeout=30
            )

            # Erfolgreiche Antworten ohne Body akzeptieren
            if response.status_code in (200, 201, 202, 204):
                if not response.text.strip():
                    return {"success": True, "code": response.status_code, "message": "No content"}
                try:
                    return response.json()
                except ValueError:
                    # Kein valider JSON-Body, aber trotzdem Erfolg
                    return {"success": True, "code": response.status_code, "raw": response.text}

            # Bekannte Fehlercodes mit spezifischer Meldung
            if response.status_code == 400:
                raise ValueError(f"Bad Request: {response.text}")
            if response.status_code == 401:
                raise ValueError("Unauthorized: Invalid API key")
            if response.status_code == 404:
                raise ValueError(f"Resource not found: {endpoint}")
            if response.status_code == 429:
                raise ValueError("Rate limit exceeded: Too many requests")

            # Alle anderen Statuscodes
            try:
                err = response.json()
            except ValueError:
                err = response.text
            raise ValueError(f"API Error {response.status_code}: {err}")

        except requests.exceptions.Timeout:
            raise ValueError("Request timeout: API took too long to respond")
        except requests.exceptions.ConnectionError:
            raise ValueError("Connection error: Unable to connect to GetResponse API")
        except requests.exceptions.RequestException as e:
            raise ValueError(f"Request failed: {e}")

    
    def get_campaigns(self) -> List[Dict]:
        """
        Get all campaigns (lists) from GetResponse account.
        
        Returns:
            List[Dict]: List of campaign objects with details
            
        Example:
            >>> client = GetResponseClient(api_key='your_key')
            >>> campaigns = client.get_campaigns()
            >>> for campaign in campaigns:
            ...     print(f"{campaign['name']}: {campaign['campaignId']}")
        """
        response = self._make_request('GET', '/campaigns')
        return response
    
    def get_campaign_by_id(self, campaign_id: str) -> Dict:
        """
        Retrieve detailed information about a specific campaign by its ID.

        Args:
            campaign_id (str): Unique campaign ID (e.g., "MV2gN")

        Returns:
            dict: Campaign details including subscription settings, fromField, replyTo, etc.

        Raises:
            ValueError: If campaign not found or API error occurs
        """
        if not campaign_id:
            raise ValueError("Campaign ID is required")

        endpoint = f"/campaigns/{campaign_id}"
        response = self._make_request("GET", endpoint)

        # Validate structure
        if not isinstance(response, dict) or "campaignId" not in response:
            raise ValueError(f"Invalid campaign response for ID {campaign_id}: {response}")

        return response

    
    def get_campaign_by_name(self, campaign_name: str) -> Optional[Dict]:
        """
        Get a specific campaign by name.
        
        Args:
            campaign_name (str): Name of the campaign to find
            
        Returns:
            dict: Campaign object if found, None otherwise
        """
        campaigns = self.get_campaigns()
        for campaign in campaigns:
            if campaign.get('name') == campaign_name:
                return campaign
        return None
    
    def add_contact(self, email: str, campaign_id: str, name: Optional[str] = None, 
                   day_of_cycle: Optional[int] = None, custom_fields: Optional[Dict] = None) -> Dict:
        """
        Add a contact to a specific campaign (list).
        
        Args:
            email (str): Contact's email address
            campaign_id (str): ID of the campaign to add contact to
            name (str, optional): Contact's name
            day_of_cycle (int, optional): Day in autoresponder cycle
            custom_fields (dict, optional): Custom field values
            
        Returns:
            dict: Response from API with contact details
            
        Example:
            >>> client = GetResponseClient(api_key='your_key')
            >>> result = client.add_contact(
            ...     email='john@example.com',
            ...     campaign_id='abc123',
            ...     name='John Doe',
            ...     day_of_cycle=0
            ... )
        """
        data = {
            'email': email,
            'campaign': {'campaignId': campaign_id}
        }
        
        if name:
            data['name'] = name
        
        if day_of_cycle is not None:
            data['dayOfCycle'] = day_of_cycle
        
        if custom_fields:
            data['customFieldValues'] = [
                {'customFieldId': field_id, 'value': [value]}
                for field_id, value in custom_fields.items()
            ]
        
        return self._make_request('POST', '/contacts', data=data)
    
    def get_contact(self, contact_id: str) -> Dict:
        """
        Get contact details by ID.
        
        Args:
            contact_id (str): ID of the contact to retrieve
            
        Returns:
            dict: Contact object with details
        """
        return self._make_request('GET', f'/contacts/{contact_id}')
    
    def search_contacts(self, query: Optional[str] = None, campaign_id: Optional[str] = None, 
                       email: Optional[str] = None, limit: int = 100) -> List[Dict]:
        """
        Search for contacts based on various criteria.
        
        Args:
            query (str, optional): Search query for contact name or email
            campaign_id (str, optional): Filter by campaign ID
            email (str, optional): Filter by exact email address
            limit (int): Maximum number of results to return
            
        Returns:
            List[Dict]: List of matching contacts
        """
        params = {'perPage': limit}
        
        if query:
            params['query[name]'] = query
            params['query[email]'] = query
        
        if campaign_id:
            params['query[campaignId]'] = campaign_id
        
        if email:
            params['query[email]'] = email
        
        return self._make_request('GET', '/contacts', params=params)
    
    def get_contact_by_email(self, email: str, campaign_id: Optional[str] = None) -> Optional[Dict]:
        """
        Get a specific contact by email address.
        
        Args:
            email (str): Email address to search for
            campaign_id (str, optional): Optional campaign filter
            
        Returns:
            dict: Contact object if found, None otherwise
        """
        contacts = self.search_contacts(email=email, campaign_id=campaign_id, limit=1)
        return contacts[0] if contacts else None
    
    def remove_contact(self, contact_id: str) -> Dict:
        """
        Remove (delete) a contact from GetResponse.
        
        Args:
            contact_id (str): ID of the contact to remove
            
        Returns:
            dict: Success message
            
        Example:
            >>> client = GetResponseClient(api_key='your_key')
            >>> result = client.remove_contact('contact123')
        """
        return self._make_request('DELETE', f'/contacts/{contact_id}')
    
    def remove_contact_by_email(self, email: str, campaign_id: Optional[str] = None) -> bool:
        """
        Remove a contact by email address.
        
        Args:
            email (str): Email address of contact to remove
            campaign_id (str, optional): Optional campaign filter
            
        Returns:
            bool: True if contact was found and removed, False otherwise
        """
        contact = self.get_contact_by_email(email, campaign_id)
        if contact:
            self.remove_contact(contact['contactId'])
            return True
        return False
    
    def update_contact(self, contact_id: str, name: Optional[str] = None, 
                      campaign_id: Optional[str] = None, custom_fields: Optional[Dict] = None) -> Dict:
        """
        Update contact information.
        
        Args:
            contact_id (str): ID of contact to update
            name (str, optional): New name for contact
            campaign_id (str, optional): New campaign ID
            custom_fields (dict, optional): Updated custom fields
            
        Returns:
            dict: Updated contact object
        """
        data = {}
        
        if name:
            data['name'] = name
        
        if campaign_id:
            data['campaign'] = {'campaignId': campaign_id}
        
        if custom_fields:
            data['customFieldValues'] = [
                {'customFieldId': field_id, 'value': [value]}
                for field_id, value in custom_fields.items()
            ]
        
        return self._make_request('POST', f'/contacts/{contact_id}', data=data)
    
    def get_account_info(self) -> Dict:
        """
        Get account information and limits.
        
        Returns:
            dict: Account details and usage information
        """
        return self._make_request('GET', '/accounts')
    
    def test_connection(self) -> bool:
        """
        Test API connection and authentication.
        
        Returns:
            bool: True if connection successful, False otherwise
        """
        try:
            self.get_account_info()
            return True
        except Exception:
            return False