#!/usr/bin/env python3

"""

Complete working code for uploading audience data to AskRally via API.

This is the full implementation referenced in the guide article.

"""

import json

import os

import requests

from typing import Dict, List, Any, Optional

class AskRallyUploader:

"""Complete AskRally API uploader with two-step process."""



def __init__(self, api_key: str):

"""Initialize with API key."""

self.api_key = api_key

self.base_url = "https://api.askrally.com/api/v1"

self.headers = {

'Authorization': f'Bearer {api_key}',

'Content-Type': 'application/json',

'Accept': 'application/json'

}



def create_audience(self, name: str, description: str = "") -> Dict[str, Any]:

"""

Step 1: Create empty audience to get audience ID.



Args:

name: Name of the audience

description: Description of the audience



Returns:

API response containing audience_id

"""

endpoint = f"{self.base_url}/audiences"



payload = {

"description": description,

"name": name,

"personas": [] # Empty array required

}



print(f"🚀 Creating audience: {name}")

print(f"📡 POST {endpoint}")



response = requests.post(endpoint, headers=self.headers, json=payload)

response.raise_for_status()



audience_data = response.json()

print(f"✅ Audience created successfully!")

print(f" - ID: {audience_data.get('audience_id', 'N/A')}")



return audience_data



def update_audience_with_personas(self, audience_id: str, personas: List[Dict[str, Any]],

name: str, description: str) -> Dict[str, Any]:

"""

Step 2: Update audience with personas.



Args:

audience_id: ID from create_audience step

personas: List of persona objects

name: Audience name

description: Audience description



Returns:

API response from update

"""

endpoint = f"{self.base_url}/audiences/{audience_id}"



print(f"🔄 Updating audience {audience_id} with {len(personas)} personas")

print(f"📡 PUT {endpoint}")



# Transform personas for AskRally API format

api_personas = []

for i, persona in enumerate(personas):

api_persona = persona.copy()



# Add required id field if not present

if 'id' not in api_persona:

api_persona['id'] = f"persona_{i+1}"



# Convert manual_memories to base_memories for API

if 'manual_memories' in api_persona:

api_persona['base_memories'] = api_persona.pop('manual_memories')



api_personas.append(api_persona)



# Full audience payload for PUT request

payload = {

"description": description,

"name": name,

"personas": api_personas

}



response = requests.put(endpoint, headers=self.headers, json=payload)

response.raise_for_status()



update_data = response.json()

print(f"✅ Audience updated successfully!")

print(f" - Personas added: {len(personas)}")



return update_data



def upload_audience_file(self, file_path: str) -> Dict[str, Any]:

"""

Complete workflow: Load file and execute two-step upload.



Args:

file_path: Path to audience JSON file



Returns:

Complete results from both API calls

"""

print(f"📁 Loading audience file: {file_path}")



# Load file

with open(file_path, 'r', encoding='utf-8') as f:

audience_data = json.load(f)



name = audience_data['name']

description = audience_data.get('description', '')

personas = audience_data['personas']



print(f"📊 File contains:")

print(f" - Name: {name}")

print(f" - Personas: {len(personas)}")

print(f" - Total memories: {sum(len(p.get('manual_memories', [])) for p in personas)}")



print("\n" + "="*50)

print("STEP 1: Creating Empty Audience")

print("="*50)



# Step 1: Create empty audience

audience_result = self.create_audience(name, description)

audience_id = audience_result.get('audience_id')



if not audience_id:

raise ValueError("Failed to get audience_id from creation response")



print("\n" + "="*50)

print("STEP 2: Uploading Personas")

print("="*50)



# Step 2: Update with personas

update_result = self.update_audience_with_personas(audience_id, personas, name, description)



print("\n" + "="*50)

print("🎉 UPLOAD COMPLETED SUCCESSFULLY!")

print("="*50)

print(f"✅ Audience ID: {audience_id}")

print(f"✅ Audience Name: {name}")

print(f"✅ Personas Uploaded: {len(personas)}")

print(f"✅ Total Memories: {sum(len(p.get('manual_memories', [])) for p in personas)}")



return {

'audience_creation': audience_result,

'personas_upload': update_result,

'audience_id': audience_id,

'audience_name': name,

'personas_count': len(personas),

'total_memories': sum(len(p.get('manual_memories', [])) for p in personas)

}

def main():

"""Example usage of the uploader."""

import sys



if len(sys.argv) != 3:

print("Usage: python upload_to_askrally.py <api_key> <audience_file>")

print("\nExample:")

print(" python upload_to_askrally.py rally_sk_abc123... audience.json")

sys.exit(1)



api_key = sys.argv[1]

file_path = sys.argv[2]



try:

uploader = AskRallyUploader(api_key)

result = uploader.upload_audience_file(file_path)



print(f"\n📝 Upload completed!")

print(f" Audience ID: {result['audience_id']}")

print(f" You can now view this audience in your AskRally dashboard")



except Exception as e:

print(f"\n❌ Upload failed: {e}")

sys.exit(1)

if __name__ == "__main__":

main()

# Alternative usage examples:

# Example 1: Direct instantiation

"""

uploader = AskRallyUploader("rally_sk_your_api_key_here")

result = uploader.upload_audience_file("audience.json")

print(f"Uploaded to audience: {result['audience_id']}")

"""

# Example 2: Environment variable

"""

api_key = os.getenv('ASKRALLY_API_KEY')

uploader = AskRallyUploader(api_key)

result = uploader.upload_audience_file("audience.json")

"""

# Example 3: Just the two API calls

"""

uploader = AskRallyUploader("rally_sk_your_api_key_here")

# Step 1: Create empty audience

audience_result = uploader.create_audience("Test Audience", "Test description")

audience_id = audience_result['audience_id']

# Step 2: Add personas

personas = [{"name": "Test Person", "age": 30, "manual_memories": []}]

update_result = uploader.update_audience_with_personas(

audience_id, personas, "Test Audience", "Test description"

)

"""