Swarm Logs & API History

import requests
import json
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

headers = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json"
}

def get_swarm_logs():
    """Get all API request logs"""
    response = requests.get(
        f"{BASE_URL}/v1/swarm/logs",
        headers=headers
    )

    if response.status_code == 200:
        return response.json()
    else:
        print(f"Error: {response.status_code} - {response.text}")
        return None

# Get logs
logs_data = get_swarm_logs()
if logs_data:
    print("✅ Logs retrieved successfully!")
    print(f"Total logs: {len(logs_data.get('logs', []))}")
    print(json.dumps(logs_data, indent=2))

from datetime import datetime, timedelta, timezone
from collections import Counter

def get_log_usage(log):
    """Extract the usage dict from a log entry's data payload."""
    data = log.get('data') or {}
    if isinstance(data, dict):
        return data.get('usage') or {}
    return {}

def get_log_cost(log):
    """Extract total cost from a log entry (agent or swarm completion)."""
    usage = get_log_usage(log)
    if 'total_cost' in usage:
        return usage['total_cost'] or 0
    # Swarm completions nest the cost under billing_info
    return (usage.get('billing_info') or {}).get('total_cost', 0)

def analyze_logs(logs_data):
    """Analyze API usage logs"""
    if not logs_data or not logs_data.get('logs'):
        print("No logs available for analysis")
        return

    logs = logs_data['logs']

    # Basic statistics
    total_requests = len(logs)

    print("📊 API Usage Analysis")
    print("=" * 50)
    print(f"Total Logged Requests: {total_requests}")
    print()

    # Category usage (e.g. "completion")
    category_counts = Counter(log.get('category', 'unknown') for log in logs)
    print("🔗 Category Usage:")
    for category, count in category_counts.most_common():
        print(f"  {category}: {count} requests")
    print()

    # Cost analysis
    total_cost = sum(get_log_cost(log) for log in logs)
    total_tokens = sum(get_log_usage(log).get('total_tokens', 0) for log in logs)

    print("💰 Cost Analysis:")
    print(f"  Total Cost: ${total_cost:.4f}")
    print(f"  Total Tokens: {total_tokens}")
    avg_cost = total_cost / total_requests if total_requests else 0
    print(f"  Average Cost per Request: ${avg_cost:.4f}")
    print()

    # Execution time analysis (present on swarm completion logs)
    execution_times = [
        log['data'].get('execution_time')
        for log in logs
        if isinstance(log.get('data'), dict) and log['data'].get('execution_time')
    ]
    if execution_times:
        avg_execution_time = sum(execution_times) / len(execution_times)
        max_execution_time = max(execution_times)
        min_execution_time = min(execution_times)

        print("⏱️  Execution Time Analysis:")
        print(f"  Average: {avg_execution_time:.2f}s")
        print(f"  Max: {max_execution_time:.2f}s")
        print(f"  Min: {min_execution_time:.2f}s")

def filter_logs_by_date(logs_data, days=7):
    """Filter logs by date range"""
    if not logs_data or not logs_data.get('logs'):
        return logs_data

    cutoff_date = datetime.now(timezone.utc) - timedelta(days=days)
    filtered_logs = []

    for log in logs_data['logs']:
        log_timestamp = datetime.fromisoformat(log['created_at'].replace('Z', '+00:00'))
        if log_timestamp >= cutoff_date:
            filtered_logs.append(log)

    return {
        **logs_data,
        'logs': filtered_logs,
        'count': len(filtered_logs)
    }

def filter_logs_by_category(logs_data, categories):
    """Filter logs by category (e.g. ["completion"])"""
    if not logs_data or not logs_data.get('logs'):
        return logs_data

    filtered_logs = [
        log for log in logs_data['logs']
        if log.get('category') in categories
    ]

    return {
        **logs_data,
        'logs': filtered_logs,
        'count': len(filtered_logs)
    }

# Example usage
logs_data = get_swarm_logs()
if logs_data:
    # Analyze all logs
    analyze_logs(logs_data)

    # Filter for last 7 days
    recent_logs = filter_logs_by_date(logs_data, days=7)
    print(f"\n📅 Recent Logs (7 days): {recent_logs['count']} requests")

    # Filter for completions
    completion_logs = filter_logs_by_category(logs_data, ["completion"])
    print(f"✅ Completion Logs: {completion_logs['count']} requests")

import csv
import json
from datetime import datetime

def export_logs_to_csv(logs_data, filename=None):
    """Export logs to CSV format"""
    if not logs_data or not logs_data.get('logs'):
        print("No logs to export")
        return

    if not filename:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"swarm_logs_{timestamp}.csv"

    logs = logs_data['logs']

    # Define CSV columns
    fieldnames = [
        'id', 'created_at', 'category',
        'job_id', 'name', 'input_tokens', 'output_tokens',
        'total_tokens', 'total_cost'
    ]

    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for log in logs:
            # Flatten the nested data payload
            data = log.get('data') or {}
            if not isinstance(data, dict):
                data = {}
            usage = data.get('usage') or {}
            total_cost = usage.get('total_cost')
            if total_cost is None:
                total_cost = (usage.get('billing_info') or {}).get('total_cost', '')
            row = {
                'id': log.get('id', ''),
                'created_at': log.get('created_at', ''),
                'category': log.get('category', ''),
                'job_id': data.get('job_id', ''),
                'name': data.get('name', data.get('swarm_name', '')),
                'input_tokens': usage.get('input_tokens', ''),
                'output_tokens': usage.get('output_tokens', ''),
                'total_tokens': usage.get('total_tokens', ''),
                'total_cost': total_cost
            }
            writer.writerow(row)

    print(f"✅ Logs exported to {filename}")
    return filename

def export_logs_to_json(logs_data, filename=None):
    """Export logs to JSON format"""
    if not logs_data:
        print("No logs to export")
        return

    if not filename:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"swarm_logs_{timestamp}.json"

    with open(filename, 'w', encoding='utf-8') as jsonfile:
        json.dump(logs_data, jsonfile, indent=2, ensure_ascii=False)

    print(f"✅ Logs exported to {filename}")
    return filename

def create_log_backup(logs_data, compress=True):
    """Create a compressed backup of logs"""
    import gzip

    if not logs_data:
        return

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"swarm_logs_backup_{timestamp}.json"

    if compress:
        filename += '.gz'
        with gzip.open(filename, 'wt', encoding='utf-8') as f:
            json.dump(logs_data, f, indent=2, ensure_ascii=False)
    else:
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(logs_data, f, indent=2, ensure_ascii=False)

    print(f"✅ Backup created: {filename}")
    return filename

# Example usage
logs_data = get_swarm_logs()
if logs_data:
    # Export to different formats
    export_logs_to_csv(logs_data)
    export_logs_to_json(logs_data)
    create_log_backup(logs_data, compress=True)

import time
from datetime import datetime, timedelta, timezone

class LogMonitor:
    def __init__(self, check_interval=300):  # 5 minutes default
        self.check_interval = check_interval
        self.last_log_count = 0
        self.cost_accumulator = 0

    def monitor_logs(self):
        """Monitor logs continuously"""
        print("🚀 Starting log monitoring... (Press Ctrl+C to stop)")

        try:
            while True:
                logs_data = get_swarm_logs()

                if logs_data:
                    self.analyze_recent_activity(logs_data)
                    self.check_for_anomalies(logs_data)

                time.sleep(self.check_interval)

        except KeyboardInterrupt:
            print("\n⏹️  Monitoring stopped")
            self.generate_monitoring_report()

    def analyze_recent_activity(self, logs_data):
        """Analyze recent API activity"""
        if not logs_data.get('logs'):
            return

        current_count = len(logs_data.get('logs', []))

        if self.last_log_count > 0:
            new_logs = current_count - self.last_log_count
            if new_logs > 0:
                print(f"📈 {new_logs} new requests in the last {self.check_interval}s")

        self.last_log_count = current_count

        # Analyze recent logs (last hour)
        recent_logs = self.get_recent_logs(logs_data, hours=1)

        if recent_logs:
            # Cost analysis (uses the get_log_cost helper defined above)
            recent_cost = sum(get_log_cost(log) for log in recent_logs)
            self.cost_accumulator += recent_cost

            print(f"💵 Recent cost: ${recent_cost:.4f}")
            print(f"💵 Total accumulated cost: ${self.cost_accumulator:.4f}")

    def check_for_anomalies(self, logs_data):
        """Check for unusual patterns or anomalies"""
        if not logs_data.get('logs'):
            return

        recent_logs = self.get_recent_logs(logs_data, hours=1)

        if recent_logs:
            # Check for unusual execution times (swarm completion logs)
            execution_times = [
                log['data'].get('execution_time')
                for log in recent_logs
                if isinstance(log.get('data'), dict) and log['data'].get('execution_time')
            ]
            if execution_times:
                avg_execution_time = sum(execution_times) / len(execution_times)
                if avg_execution_time > 60:  # More than 60 seconds average
                    print(f"🐢 Slow swarm runs detected: {avg_execution_time:.2f}s average")

    def get_recent_logs(self, logs_data, hours=1):
        """Get logs from the last N hours"""
        if not logs_data.get('logs'):
            return []

        cutoff_time = datetime.now(timezone.utc) - timedelta(hours=hours)
        recent_logs = []

        for log in logs_data['logs']:
            log_time = datetime.fromisoformat(log['created_at'].replace('Z', '+00:00'))
            if log_time >= cutoff_time:
                recent_logs.append(log)

        return recent_logs

    def generate_monitoring_report(self):
        """Generate a final monitoring report"""
        print("\n📊 Monitoring Report")
        print("=" * 50)
        print(f"Total Accumulated Cost: ${self.cost_accumulator:.4f}")
        print(f"Monitoring Duration: {self.check_interval}s intervals")

# Usage
monitor = LogMonitor(check_interval=300)  # Check every 5 minutes
monitor.monitor_logs()