Marketing Agency Report Factory

How It Works

1. 1
   Initialize Platform Connections

   Load OAuth tokens and API credentials from secure storage, validate connection health with test requests to Google Analytics, Meta Ads, and Google Ads APIs

   requests library with OAuth2Session
2. 2
   Query Platform APIs

   Execute parallel data collection requests for previous day's metrics including sessions, conversions, ad spend, and performance data with proper date range formatting

   Google Analytics Reporting API v4, Meta Marketing API, Google Ads API
3. 3
   Normalize Data Schemas

   Transform platform-specific response formats into unified data structures with standardized field names, data types, and client identifiers

   pandas for data transformation
4. 4
   Store in Airtable

   Batch upload normalized records to designated Airtable bases using upsert operations to handle duplicate prevention and data versioning

   Airtable API with pyairtable
5. 5
   Log Collection Results

   Record collection success rates, data volumes, and any API errors to monitoring dashboard with timestamp and client-specific metrics

   structured logging with JSON format

Implementation

# Data Collector Implementation

## File Structure
```
data_collector/
├── main.py              # Entry point and orchestration
├── collectors/
│   ├── google_analytics.py
│   ├── meta_ads.py
│   └── google_ads.py
├── utils/
│   ├── auth_manager.py  # OAuth token management
│   ├── data_normalizer.py
│   └── airtable_client.py
├── config/
│   ├── credentials.json  # Encrypted credential storage
│   └── client_mappings.json
└── requirements.txt
```

## Key Functions
```python
# main.py
def run_collection_cycle():
    clients = load_client_configs()
    for client in clients:
        try:
            ga_data = collect_google_analytics(client)
            meta_data = collect_meta_ads(client)
            gads_data = collect_google_ads(client)
            
            normalized = normalize_data_schemas(ga_data, meta_data, gads_data)
            store_in_airtable(client.base_id, normalized)
            
        except Exception as e:
            log_collection_error(client.id, str(e))

# collectors/google_analytics.py
def collect_google_analytics(client):
    service = build_analytics_service(client.credentials)
    request = {
        'reportRequests': [{
            'viewId': client.ga_view_id,
            'dateRanges': [{'startDate': 'yesterday', 'endDate': 'yesterday'}],
            'metrics': [{'expression': 'ga:sessions'}, {'expression': 'ga:conversions'}]
        }]
    }
    return service.reports().batchGet(body=request).execute()
```

## Environment Variables
```
AIRTABLE_API_KEY=key_xxxxxx
GOOGLE_CLIENT_ID=xxxx.apps.googleusercontent.com
GOOGLE_CLIENT_SECRET=xxxxxx
META_APP_ID=xxxxxx
META_APP_SECRET=xxxxxx
GOOGLE_ADS_DEVELOPER_TOKEN=xxxxxx
```

## Cron Setup
```bash
# Add to crontab
0 3 * * * cd /path/to/data_collector && python main.py >> /var/log/data_collector.log 2>&1
```

Data Flow

Prerequisites

• -OAuth 2.0 tokens for all integrated platforms
• -Airtable base with properly configured field types
• -Network access to external APIs
• -Secure credential storage mechanism

Error Handling