Usando a API para ler dados de uma campanha

# modules that need install : run [pip install <module>] on the command line
import requests
import pytz
# modules included in Python default libraries
import json
import csv
from datetime import datetime, timedelta
import time
import base64
import os
import sys
import argparse
def convert_to_brt(utc_string):
    utc_zone = pytz.utc
    brt_zone = pytz.timezone('America/Sao_Paulo')
    utc_dt = utc_zone.localize(datetime.fromisoformat(utc_string.replace('Z', '')))
    brt_dt = utc_dt.astimezone(brt_zone)
    return brt_dt.strftime('%d-%m-%Y %H:%M:%S')
def process_included_entities(calls, included):
    
    # Create a lookup dictionary for included entities
    entity_map = {(item['type'], item['id']): item['attributes'] for item in included}
    
    # Process each call and add related entity attributes
    for call in calls:
        attrs = call['attributes']
        relationships = call['relationships']
        
        # Convert UTC times to BRT
        attrs['started_at'] = convert_to_brt(attrs['started_at'])
        attrs['ended_at'] = convert_to_brt(attrs['ended_at'])
        if 'answered_at' in attrs:
            attrs['answered_at'] = convert_to_brt(attrs['answered_at'])
        
        # Add campaign_contact attributes
        # contact attributes are DYNAMIC, so lines mapping contact attributes to CSV fields should be changed/added according to what contact fields one wants on the CSV
        # As an EXAMPLE five contact attributes are included: "nome", "cpf", "produto" , "publico", "telefone", and they are mapped to the "contact_nome", "contact_cpf" and so on as CSV fields
        contact_data = relationships.get('campaign_contact', {}).get('data')
        if contact_data and (contact_data['type'], contact_data['id']) in entity_map:
            contact_attrs = entity_map[(contact_data['type'], contact_data['id'])]
            attrs['contact_nome'] = contact_attrs.get('nome', '')
            attrs['contact_cpf'] = contact_attrs.get('cpf', '')
            attrs['contact_produto'] = contact_attrs.get('produto', '')
            attrs['contact_publico'] = contact_attrs.get('publico', '')
            attrs['contact_telefone'] = contact_attrs.get('telefone', '')
        # Add qualification attributes if they exist
        qual_data = relationships.get('qualification', {}).get('data')
        if qual_data and (qual_data['type'], qual_data['id']) in entity_map:
            qual_attrs = entity_map[(qual_data['type'], qual_data['id'])]
            attrs['qualification_name'] = qual_attrs.get('name', '')
            attrs['qualification_cpc'] = qual_attrs.get('cpc', '')
            attrs['qualification_success'] = qual_attrs.get('success', '')
            attrs['qualification_is_system_qualification'] = qual_attrs.get('is_system_qualification', '')
        
        # Add user (agent) attributes if they exist
        agent_data = relationships.get('agent', {}).get('data')
        if agent_data and (agent_data['type'], agent_data['id']) in entity_map:
            agent_attrs = entity_map[(agent_data['type'], agent_data['id'])]
            attrs['agent_name'] = agent_attrs.get('name', '')
            attrs['agent_email'] = agent_attrs.get('email', '')
            attrs['agent_login'] = agent_attrs.get('login', '')
    return calls
def download_call_audio(audio_link, protocol, headers, base_url, output_dir="call_audios"):
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    
    filename = f"{output_dir}/{protocol}.mp3"
    
    # Check if audio file already exists locally
    if os.path.exists(filename) and os.path.getsize(filename) > 0:
        print(f"Audio for protocol {protocol} already exists and is non-empty, skipping download")
        return True
    
    full_audio_url = base_url + audio_link
    try:
        response = requests.get(full_audio_url, headers=headers)
        #sleeps for 500ms to respect API rate limit
        time.sleep(0.5)
        if response.status_code == 200:
            content_type = response.headers.get('Content-Type', '').lower()
            
            if 'application/json' in content_type:
                audio_data = response.json().get('audio', '')
                if audio_data:
                    audio_bytes = base64.b64decode(audio_data)
                    with open(filename, 'wb') as audio_file:
                        audio_file.write(audio_bytes)
                    print(f"Downloaded audio for protocol {protocol} (base64 JSON)")
                    return True
                else:
                    print(f"No audio data in JSON response for protocol {protocol}")
                    return False
            else:
                with open(filename, 'wb') as audio_file:
                    audio_file.write(response.content)
                print(f"Downloaded audio for protocol {protocol} (direct binary)")
                return True
        else:
            print(f"Failed to download audio for protocol {protocol}: Status {response.status_code}: {response.text}")
            return False
    except Exception as e:
        print(f"Error downloading audio for protocol {protocol}: {e}")
        return False
def fetch_and_write_calls(base_url, export_type, headers, start_date, end_date, download_audio=False, output_dir="call_audios"):
    if export_type not in ["completed", "missed"]:
        raise ValueError("export_type must be 'completed' or 'missed'")
    
    # Infer endpoint, include, and filename based on export_type
    if export_type == "completed":
        endpoint = "/api/v1/campaign_completed_calls"
        include = "campaign_contact,qualification,agent"
        filename = "completed_calls.csv"
        rate_limit_delay = 7  # 10/minute = 6s + 1s buffer
    else:  # export_type == "missed"
        endpoint = "/api/v1/campaign_missed_calls"
        include = "campaign_contact"
        filename = "missed_calls.csv"
        rate_limit_delay = 61  # 1/minute = 60s + 1s buffer
    
    total_calls = 0
    is_first_file = True
    csv_path = os.path.join(output_dir, filename)
    
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    
    try:
        current_date = start_date
        while current_date <= end_date:
            day_start = current_date.replace(hour=0, minute=0, second=0, microsecond=0)
            day_end = current_date.replace(hour=23, minute=59, second=59, microsecond=999999)
            
            start_str = day_start.strftime('%Y-%m-%dT%H:%M:%S.000Z')
            end_str = day_end.strftime('%Y-%m-%dT%H:%M:%S.000Z')
            
            params = {
                "filter[started_at]": f"{start_str},{end_str}",
                "page[limit]": 2000,
                "page[offset]": 0,
                "include": include
            }
            
            print(f"Processing {export_type} calls for {current_date.strftime('%Y-%m-%d')}...")
            is_first_page = True
            current_url = endpoint
            
            while current_url:
                response = requests.get(
                    base_url + endpoint if current_url == endpoint else current_url,
                    headers=headers,
                    params=params if current_url == endpoint else None
                )
                
                if response.status_code != 200:
                    print(f"Error on {current_date.strftime('%Y-%m-%d')}: Received status code {response.status_code}: {response.text}")
                    break
                    
                data = response.json()
                calls = data['data']
                
                if not calls:
                    print(f"No {export_type} calls found for {current_date.strftime('%Y-%m-%d')}")
                    break
                    
                included = data.get('included', [])
                if included:
                    calls = process_included_entities(calls, included)
                
                if download_audio and export_type == "completed":
                    for call in calls:
                        attrs = call['attributes']
                        if attrs.get('has_audio', False) and 'audio_link' in attrs:
                            download_call_audio(attrs['audio_link'], attrs['protocol'], headers, base_url, output_dir)
                        
                original_fieldnames = list(calls[0]['attributes'].keys())
                relationship_fieldnames = (
                    ['contact_nome', 'contact_cpf', 'contact_produto', 'contact_publico', 'contact_telefone', 'qualification_name','qualification_cpc', 'qualification_success', 'qualification_is_system_qualification',  'agent_name', 'agent_login', 'agent_email']
                    if export_type == "completed"
                    else ['contact_nome', 'contact_cpf', 'contact_produto', 'contact_publico', 'contact_telefone']
                )
                all_fieldnames = original_fieldnames + [f for f in relationship_fieldnames if f not in original_fieldnames]
                
                try:
                    mode = 'w' if is_first_file and is_first_page else 'a'
                    with open(csv_path, mode, newline='', encoding='utf-8') as csvfile:
                        writer = csv.DictWriter(csvfile, fieldnames=all_fieldnames, delimiter=';')
                        
                        if is_first_page and is_first_file:
                            writer.writeheader()
                            is_first_file = False
                        
                        for call in calls:
                            writer.writerow(call['attributes'])
                except PermissionError as pe:
                    print(f"Permission denied when writing to {csv_path}: {pe}. Please ensure the file is not open elsewhere.")
                    return
                except Exception as e:
                    print(f"Error writing to {csv_path}: {e}")
                    return
                
                total_calls += len(calls)
                print(f"Fetched and wrote {len(calls)} {export_type} calls for {current_date.strftime('%Y-%m-%d')}, total so far: {total_calls}")
                
                time.sleep(rate_limit_delay)
                
                next_page = data['links'].get('next')
                current_url = base_url + next_page if next_page else None
            
            current_date += timedelta(days=1)
                
    except requests.RequestException as e:
        print(f"Error: {e}")
    
    print(f"Finished processing {export_type} calls. Total calls saved: {total_calls}")
def main():
    parser = argparse.ArgumentParser(description="Fetch and write call data from Callix API.")
    parser.add_argument('-y', '--yesterday', action='store_true', help="Use yesterday's date for start and end date")
    parser.add_argument('-d', '--download', action='store_true', help="Download audio files for completed calls (default: False)")
    parser.add_argument('-m', '--missed', action='store_true', help="Also export missed calls (default: only completed calls)")
    parser.add_argument('-t', '--tenant', required=True, help="Tenant prefix for base URL (e.g., 'globo' for https://sua-conta.callix.com.br)")
    parser.add_argument('-b', '--bearer', required=True, help="Bearer token for Authorization (e.g., '14ae293d-f1f0-4f0c-a220-c4bc9438a855')")
    parser.add_argument('start_date', nargs='?', help="Start date in dd/mm/aaaa format (required if -y not used)")
    parser.add_argument('end_date', nargs='?', help="End date in dd/mm/aaaa format (required if -y not used)")
    
    args = parser.parse_args()
    base_url = f"https://{args.tenant}.callix.com.br"
    headers = {
        'Content-Type': "application/json",
        'Authorization': f"Bearer {args.bearer}"
    }
    
    if args.yesterday:
        yesterday = datetime.now(pytz.utc) - timedelta(days=1)
        start_date = yesterday.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=pytz.utc)
        end_date = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999, tzinfo=pytz.utc)
    else:
        if not args.start_date or not args.end_date:
            print("Error: Start date and end date are required unless -y is used.")
            print("Usage: python export_calls_v2.py -t <tenant> -b <bearer> [-y] [-d] [-m] [<start_date> <end_date>]")
            print("Example: python export_calls_v2.py -t globo -b 14ae293d-f1f0-4f0c-a220-c4bc9438a855 31/03/2025 03/04/2025")
            print("         python export_calls_v2.py -t globo -b 14ae293d-f1f0-4f0c-a220-c4bc9438a855 -y -d")
            print("Dates must be in dd/mm/aaaa format.")
            sys.exit(1)
        
        try:
            start_date = datetime.strptime(args.start_date, '%d/%m/%Y').replace(tzinfo=pytz.utc)
            end_date = datetime.strptime(args.end_date, '%d/%m/%Y').replace(tzinfo=pytz.utc)
        except ValueError as e:
            print(f"Error: Invalid date format. {e}")
            print("Usage: python export_calls_v2 -t <tenant> -b <bearer> [-y] [-d] [-m] [<start_date> <end_date>]")
            print("Example: python export_calls_v2.py -t globo -b 14ae293d-f1f0-4f0c-a220-c4bc9438a855 31/03/2025 03/04/2025")
            print("         python export_calls_v2.py -t globo -b 14ae293d-f1f0-4f0c-a220-c4bc9438a855 -y -d")
            print("Dates must be in dd/mm/aaaa format.")
            sys.exit(1)
    
    print("Starting to fetch and write completed calls...")
    fetch_and_write_calls(base_url, "completed", headers, start_date, end_date, download_audio=args.download)
    
    if args.missed:
        print("Starting to fetch and write missed calls...")
        fetch_and_write_calls(base_url, "missed", headers, start_date, end_date, download_audio=False)
if __name__ == "__main__":
    main()