#!/usr/bin/env python3 """ Carbon Intelligence™ Meta Ads Export Script Exportateur de données Meta Ads (Facebook + Instagram) pour Carbon Intelligence™ Version 1.0.0 | Production-Ready Uses Meta Marketing API (Graph API) v19.0+ for comprehensive ad performance reporting by campaign, device, geography, placements, creative, format, and video metrics. Architecture: - CONFIG: Global configuration and API settings - Export functions: One per dimension/breakdown - Metadata tracking: Account info, period, version - CSV consolidation: All dimensions in single file for CI import """ import os import sys import json import csv import logging import requests from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple from urllib.parse import urljoin import time # ============================================================================ # CONFIG SECTION # ============================================================================ # Meta Marketing API Configuration APP_ID = os.getenv('META_APP_ID', 'YOUR_APP_ID') APP_SECRET = os.getenv('META_APP_SECRET', 'YOUR_APP_SECRET') ACCESS_TOKEN = os.getenv('META_ACCESS_TOKEN', 'YOUR_ACCESS_TOKEN') AD_ACCOUNT_ID = os.getenv('META_AD_ACCOUNT_ID', 'act_YOUR_ACCOUNT_ID') # API Settings API_VERSION = 'v19.0' BASE_URL = f'https://graph.facebook.com/{API_VERSION}/' INSIGHTS_ENDPOINT = f'{AD_ACCOUNT_ID}/insights' # Export Settings OUTPUT_DIR = os.getenv('CI_OUTPUT_DIR', './ci_exports') LOOKBACK_DAYS = int(os.getenv('CI_LOOKBACK_DAYS', '90')) MIN_IMPRESSIONS = int(os.getenv('CI_MIN_IMPRESSIONS', '1')) # Logging Configuration logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', handlers=[ logging.FileHandler(os.path.join(OUTPUT_DIR, 'carbon-intelligence-meta.log')), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger('Carbon-Intelligence-Meta') # Create output directory if needed os.makedirs(OUTPUT_DIR, exist_ok=True) # ============================================================================ # UTILITY FUNCTIONS # ============================================================================ def validate_config() -> bool: """Validate required configuration parameters.""" if APP_ID == 'YOUR_APP_ID' or ACCESS_TOKEN == 'YOUR_ACCESS_TOKEN': logger.error('Missing required configuration. Set META_APP_ID, META_APP_SECRET, META_ACCESS_TOKEN') return False if not AD_ACCOUNT_ID.startswith('act_'): logger.error(f'Invalid AD_ACCOUNT_ID format. Expected act_XXXXXXXXX, got {AD_ACCOUNT_ID}') return False return True def build_params(fields: List[str], breakdowns: Optional[List[str]] = None, date_preset: str = 'last_90_days', time_increment: int = 7) -> Dict: """Build Meta API request parameters.""" params = { 'access_token': ACCESS_TOKEN, 'fields': ','.join(fields), 'date_preset': date_preset, 'time_increment': time_increment, 'limit': 100 } if breakdowns: params['breakdowns'] = ','.join(breakdowns) return params def api_request(endpoint: str, params: Dict, retries: int = 3) -> Optional[Dict]: """Make API request with retry logic and error handling.""" url = urljoin(BASE_URL, endpoint) attempt = 0 while attempt < retries: try: response = requests.get(url, params=params, timeout=30) if response.status_code == 200: return response.json() elif response.status_code == 429: # Rate limited retry_after = int(response.headers.get('Retry-After', 60)) logger.warning(f'Rate limited. Waiting {retry_after} seconds...') time.sleep(retry_after) attempt += 1 elif response.status_code == 401: logger.error('Authentication failed. Check ACCESS_TOKEN.') return None elif response.status_code == 403: logger.error(f'Permission denied. Check account access: {response.text}') return None else: logger.error(f'API request failed [{response.status_code}]: {response.text}') return None except requests.exceptions.RequestException as e: logger.error(f'Request exception: {e}') attempt += 1 if attempt < retries: time.sleep(5) return None def paginate_results(endpoint: str, params: Dict) -> List[Dict]: """Fetch all paginated results from Meta API.""" all_data = [] url = urljoin(BASE_URL, endpoint) while url: response = requests.get(url, params=params, timeout=30) if response.status_code != 200: logger.error(f'Pagination failed: {response.text}') break data = response.json() all_data.extend(data.get('data', [])) # Get next page URL paging = data.get('paging', {}) url = paging.get('cursors', {}).get('after') if url: params['after'] = url return all_data def filter_by_impressions(records: List[Dict], min_impressions: int) -> List[Dict]: """Filter records by minimum impressions threshold.""" return [r for r in records if int(r.get('impressions', 0)) >= min_impressions] # ============================================================================ # EXPORT FUNCTIONS # ============================================================================ def export_ci_campaigns() -> List[Dict]: """Export CI_Campaigns: Campaign + Week breakdown with core metrics.""" logger.info('Exporting CI_Campaigns...') fields = [ 'campaign_id', 'campaign_name', 'date_start', 'date_stop', 'impressions', 'clicks', 'spend', 'reach', 'frequency', 'cpc', 'cpm', 'ctr', 'conversions', 'purchase_roas', 'conversion_rate_ranking' ] params = build_params(fields, time_increment=7, date_preset='last_90_days') data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) records = filter_by_impressions(records, MIN_IMPRESSIONS) # Add CI metadata for record in records: record['ci_dimension'] = 'campaign_week' record['ci_source'] = 'meta_ads' logger.info(f'Exported {len(records)} campaign records') return records def export_ci_device() -> List[Dict]: """Export CI_Device: Device Platform breakdown (mobile, desktop, tablet).""" logger.info('Exporting CI_Device...') fields = [ 'device_platform', 'impressions', 'clicks', 'spend', 'video_views', 'reach', 'cpc', 'cpm', 'ctr', 'conversions' ] params = build_params(fields, breakdowns=['device_platform'], date_preset='last_90_days') data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) records = filter_by_impressions(records, MIN_IMPRESSIONS) for record in records: record['ci_dimension'] = 'device_platform' record['ci_source'] = 'meta_ads' logger.info(f'Exported {len(records)} device records') return records def export_ci_geo() -> List[Dict]: """Export CI_Geo: Country breakdown.""" logger.info('Exporting CI_Geo...') fields = [ 'country', 'impressions', 'clicks', 'spend', 'reach', 'cpc', 'cpm', 'ctr', 'conversions', 'conversion_rate_ranking' ] params = build_params(fields, breakdowns=['country'], date_preset='last_90_days') data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) records = filter_by_impressions(records, MIN_IMPRESSIONS) for record in records: record['ci_dimension'] = 'country' record['ci_source'] = 'meta_ads' logger.info(f'Exported {len(records)} geo records') return records def export_ci_placements() -> List[Dict]: """Export CI_Placements: Publisher platform + Platform position breakdown.""" logger.info('Exporting CI_Placements...') fields = [ 'publisher_platform', 'platform_position', 'impressions', 'clicks', 'spend', 'reach', 'frequency', 'cpc', 'cpm', 'ctr', 'video_views' ] params = build_params( fields, breakdowns=['publisher_platform', 'platform_position'], date_preset='last_90_days' ) data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) records = filter_by_impressions(records, MIN_IMPRESSIONS) for record in records: record['ci_dimension'] = 'placement' record['ci_source'] = 'meta_ads' logger.info(f'Exported {len(records)} placement records') return records def export_ci_creative_size() -> List[Dict]: """Export CI_CreativeSize: Ad-level metrics with creative dimensions.""" logger.info('Exporting CI_CreativeSize...') fields = [ 'ad_id', 'ad_name', 'creative', 'impressions', 'clicks', 'spend', 'cpc', 'cpm', 'ctr', 'reach', 'video_views' ] params = build_params(fields, time_increment=1, date_preset='last_90_days') params['level'] = 'ad' # Get ad-level data data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) records = filter_by_impressions(records, MIN_IMPRESSIONS) # Extract creative dimensions where available for record in records: record['ci_dimension'] = 'creative_size' record['ci_source'] = 'meta_ads' # Creative object may contain media_type and dimensions if 'creative' in record and isinstance(record['creative'], dict): record['creative_type'] = record['creative'].get('media_type', 'unknown') logger.info(f'Exported {len(records)} creative records') return records def export_ci_ad_format() -> List[Dict]: """Export CI_AdFormat: Publisher platform + Ad format breakdown.""" logger.info('Exporting CI_AdFormat...') fields = [ 'publisher_platform', 'impression_device', 'impressions', 'clicks', 'spend', 'cpc', 'cpm', 'ctr', 'reach', 'video_views', 'conversions' ] params = build_params( fields, breakdowns=['publisher_platform'], date_preset='last_90_days' ) data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) records = filter_by_impressions(records, MIN_IMPRESSIONS) for record in records: record['ci_dimension'] = 'ad_format' record['ci_source'] = 'meta_ads' logger.info(f'Exported {len(records)} ad format records') return records def export_ci_video() -> List[Dict]: """Export CI_Video: Video-specific metrics.""" logger.info('Exporting CI_Video...') fields = [ 'campaign_id', 'campaign_name', 'date_start', 'date_stop', 'video_views', 'video_play_actions', 'video_p25_watched_actions', 'video_p50_watched_actions', 'video_p75_watched_actions', 'video_p100_watched_actions', 'video_avg_time_watched_actions', 'impressions', 'spend' ] params = build_params(fields, time_increment=7, date_preset='last_90_days') data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) # Filter for video campaigns that have video metrics records = [r for r in records if int(r.get('video_views', 0)) > 0] records = filter_by_impressions(records, MIN_IMPRESSIONS) for record in records: record['ci_dimension'] = 'video' record['ci_source'] = 'meta_ads' logger.info(f'Exported {len(records)} video records') return records def export_ci_imp_device() -> List[Dict]: """Export CI_ImpDevice: Impression Device breakdown (CRITICAL for carbon tracking).""" logger.info('Exporting CI_ImpDevice...') fields = [ 'impression_device', 'impressions', 'clicks', 'spend', 'reach', 'frequency', 'cpc', 'cpm', 'ctr', 'video_views', 'conversions' ] params = build_params( fields, breakdowns=['impression_device'], date_preset='last_90_days' ) data = api_request(INSIGHTS_ENDPOINT, params) if not data: return [] records = data.get('data', []) records = filter_by_impressions(records, MIN_IMPRESSIONS) for record in records: record['ci_dimension'] = 'impression_device' record['ci_source'] = 'meta_ads' # Normalize device type for carbon emissions calculation device = record.get('impression_device', '').lower() if 'iphone' in device or 'ipad' in device: record['device_category'] = 'apple_mobile' elif 'android' in device: record['device_category'] = 'android_mobile' elif 'desktop' in device: record['device_category'] = 'desktop' else: record['device_category'] = 'other' logger.info(f'Exported {len(records)} impression device records') return records def export_ci_metadata() -> Dict: """Export CI_Metadata: Account info, period, version, methodology.""" logger.info('Exporting CI_Metadata...') # Get account details account_endpoint = f'{AD_ACCOUNT_ID}' account_params = { 'access_token': ACCESS_TOKEN, 'fields': 'id,name,account_status,currency,timezone_name' } account_data = api_request(account_endpoint, account_params) end_date = datetime.now().date() start_date = end_date - timedelta(days=LOOKBACK_DAYS) metadata = { 'ci_export_version': '1.0.0', 'ci_source': 'meta_ads', 'ci_platform': 'facebook_instagram', 'export_timestamp': datetime.now().isoformat(), 'account_id': AD_ACCOUNT_ID, 'account_name': account_data.get('name', 'Unknown') if account_data else 'Unknown', 'account_currency': account_data.get('currency', 'USD') if account_data else 'USD', 'report_period_start': start_date.isoformat(), 'report_period_end': end_date.isoformat(), 'lookback_days': LOOKBACK_DAYS, 'min_impressions_threshold': MIN_IMPRESSIONS, 'api_version': API_VERSION, 'methodology': { 'ci_campaigns': 'Weekly aggregation by campaign_id with core performance metrics', 'ci_device': 'Breakdown by device_platform (mobile, desktop, tablet)', 'ci_geo': 'Breakdown by country (ISO 3166-1 alpha-2 codes)', 'ci_placements': 'Cross-breakdown by publisher_platform and platform_position', 'ci_creative_size': 'Ad-level metrics with creative asset data', 'ci_ad_format': 'Breakdown by publisher_platform and impression_device', 'ci_video': 'Video completion metrics (p25, p50, p75, p100) and play actions', 'ci_imp_device': 'Actual impression device used (iPhone, Android, Desktop, etc.)', 'carbon_tracking': 'Device category mapping enables carbon footprint calculation' }, 'dimensions_included': [ 'campaign_week', 'device_platform', 'country', 'placement', 'creative_size', 'ad_format', 'video', 'impression_device' ] } logger.info('Metadata exported') return metadata # ============================================================================ # CSV CONSOLIDATION # ============================================================================ def export_ci_csv_consolidated(all_exports: Dict[str, List[Dict]]) -> str: """Create consolidated CSV file with all dimensions for CI import.""" logger.info('Creating consolidated CSV export...') csv_filename = os.path.join(OUTPUT_DIR, 'ci_meta_ads_consolidated.csv') # Define master field list (union of all possible fields from exports) master_fields = set() for export_list in all_exports.values(): for record in export_list: master_fields.update(record.keys()) # Add standard CI fields master_fields.update([ 'ci_dimension', 'ci_source', 'ci_export_version', 'device_category' ]) # Sort fields for consistency fieldnames = sorted(list(master_fields)) # Write consolidated CSV with open(csv_filename, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=fieldnames, restval='') writer.writeheader() for export_list in all_exports.values(): for record in export_list: # Ensure all fields exist (fill missing with empty string) row = {field: str(record.get(field, '')) for field in fieldnames} row['ci_export_version'] = '1.0.0' writer.writerow(row) logger.info(f'Consolidated CSV written to {csv_filename}') return csv_filename # ============================================================================ # MAIN EXPORT ORCHESTRATION # ============================================================================ def main(): """Main orchestration function - runs all export functions.""" logger.info('=' * 80) logger.info('Carbon Intelligence™ Meta Ads Export - v1.0.0') logger.info('=' * 80) # Validate configuration if not validate_config(): logger.error('Configuration validation failed. Exiting.') return 1 logger.info(f'Configuration: Account={AD_ACCOUNT_ID}, Lookback={LOOKBACK_DAYS} days') logger.info(f'Output directory: {OUTPUT_DIR}') try: # Execute all export functions exports = { 'campaigns': export_ci_campaigns(), 'device': export_ci_device(), 'geo': export_ci_geo(), 'placements': export_ci_placements(), 'creative_size': export_ci_creative_size(), 'ad_format': export_ci_ad_format(), 'video': export_ci_video(), 'imp_device': export_ci_imp_device(), } # Export metadata metadata = export_ci_metadata() metadata_file = os.path.join(OUTPUT_DIR, 'ci_meta_ads_metadata.json') with open(metadata_file, 'w', encoding='utf-8') as f: json.dump(metadata, f, indent=2, default=str) logger.info(f'Metadata written to {metadata_file}') # Create consolidated CSV csv_file = export_ci_csv_consolidated(exports) # Summary total_records = sum(len(records) for records in exports.values()) logger.info('=' * 80) logger.info('EXPORT COMPLETE') logger.info(f'Total records exported: {total_records}') logger.info(f'Consolidated CSV: {csv_file}') logger.info(f'Metadata: {metadata_file}') logger.info('=' * 80) return 0 except Exception as e: logger.exception(f'Unhandled exception during export: {e}') return 1 if __name__ == '__main__': sys.exit(main())