"""Helper functions for vendor account creation script.""" import csv import json import logging import os import sys import time from typing import Dict, List, Optional import requests from constants import VALID_COMPANY_BRANDS, COMPANY_BRAND_MAP logger = logging.getLogger(__name__) def load_config(config_file: str) -> Dict: """Load configuration from JSON file containing bearer tokens and service tier map.""" try: with open(config_file, 'r') as f: config = json.load(f) if 'bearer_token' not in config: logger.error("Config file missing 'bearer_token' section") sys.exit(1) if 'service_tier_map' not in config: logger.error("Config file missing 'service_tier_map' section") sys.exit(1) return config except FileNotFoundError: logger.error(f"Config file not found: {config_file}") sys.exit(1) except json.JSONDecodeError as e: logger.error(f"Invalid JSON in config file: {e}") sys.exit(1) def normalize_company_brand(brand: str) -> Optional[str]: """Normalize and validate company brand name.""" if not brand: return None brand_lower = brand.lower().strip() normalized = COMPANY_BRAND_MAP.get(brand_lower, brand_lower) if normalized not in VALID_COMPANY_BRANDS: logger.warning(f"Company brand '{brand}' not recognized") return normalized def parse_csv_row( row: Dict[str, str], service_tier_map: Dict[str, str], default_owner: str = None, default_service_tier: str = None, skip_if_missing: bool = False ) -> Optional[Dict[str, any]]: """Parse a CSV row into vendor creation payload.""" account_name = row.get('Account Name', '').strip() if not account_name: logger.warning(f"Skipping row with empty Account Name") return None company_brand = normalize_company_brand(row.get('Company Brand', '')) if not company_brand: logger.warning(f"Skipping '{account_name}': Invalid company brand '{row.get('Company Brand', '')}'") return None owner = row.get('Owner', '').strip() if not owner: if skip_if_missing: logger.warning(f"Skipping '{account_name}': Owner is required but missing") return None if default_owner: owner = default_owner logger.warning(f"[DEFAULT] Using default owner '{owner}' for '{account_name}'") else: logger.error(f"Skipping '{account_name}': Owner is required but missing and no default provided") return None currency = row.get('Currency', '').strip() if not currency: currency = 'USD' logger.warning(f"[DEFAULT] Using default currency 'USD' for '{account_name}'") service_tier = row.get('Service Tier', '').strip() if not service_tier: if skip_if_missing: logger.warning(f"Skipping '{account_name}': Service Tier is required but missing") return None if default_service_tier: service_tier = default_service_tier logger.warning(f"[DEFAULT] Using default service tier for '{account_name}'") else: logger.error(f"Skipping '{account_name}': Service Tier is required but missing and no default provided") return None service_tier_lower = service_tier.lower().strip() if service_tier_lower in service_tier_map: service_tier_uuid = service_tier_map[service_tier_lower] if service_tier_uuid.startswith('YOUR_'): logger.error(f"Skipping '{account_name}': Service tier '{service_tier}' not configured in service_tier_map") return None else: service_tier_uuid = service_tier payload = { 'name': account_name, 'owner': owner, 'company_brand': company_brand, 'service_tier_uuid': service_tier_uuid, 'payment_currency': currency.upper(), 'is_distributor': False, } return payload def should_retry_status_code(status_code: int) -> tuple[bool, int]: """Determine if status code should be retried and wait time. Returns (should_retry, wait_seconds) """ if status_code == 429: return (True, 5) # Rate limit - default wait elif status_code in [500, 502, 503, 504]: return (True, 0) # Server error - use exponential backoff return (False, 0) def should_retry_exception(exception: Exception) -> bool: """Determine if exception type should be retried.""" return isinstance(exception, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)) def handle_error_response(status_code: int, response_text: str, payload: Dict, account_name: str) -> Dict: """Handle non-success API responses and return error result.""" if status_code in [401, 403]: logger.error(f"[AUTH ERROR] Authentication error for '{account_name}': Status {status_code}") return { 'status': 'error', 'status_code': status_code, 'error': response_text, 'payload': payload, 'retryable': False } else: logger.error( f"[FAILED] Failed to create vendor '{account_name}': " f"Status {status_code}, Response: {response_text}" ) return { 'status': 'error', 'status_code': status_code, 'error': response_text, 'payload': payload, 'retryable': False } def create_vendor(api_url: str, payload: Dict, bearer_token: str, identity_headers: Optional[Dict] = None, dry_run: bool = False, max_retries: int = 3) -> Dict: """Create a vendor via the API with automatic retry for transient failures.""" if dry_run: logger.info(f"[DRY RUN] Would create vendor: {json.dumps(payload, indent=2)}") return {'status': 'dry_run', 'payload': payload} headers = { 'Authorization': f'Bearer {bearer_token}', 'Content-Type': 'application/json', } # Add optional identity headers if identity_headers: headers.update(identity_headers) account_name = payload['name'] last_error = None for attempt in range(max_retries): try: response = requests.post( f'{api_url}/v2/vendors', json=payload, headers=headers, timeout=30 ) if response.status_code == 200: response_data = response.json() logger.info(f"[SUCCESS] Created vendor: {account_name}") return {'status': 'success', 'data': response_data, 'payload': payload} # Check if status code is retryable should_retry, wait_time = should_retry_status_code(response.status_code) if should_retry and attempt < max_retries - 1: if response.status_code == 429: retry_after = int(response.headers.get('Retry-After', wait_time)) logger.warning(f"[RATE LIMIT] Rate limited for '{account_name}', waiting {retry_after}s (attempt {attempt + 1}/{max_retries})") time.sleep(retry_after) else: # Exponential backoff for server errors backoff = 2 ** attempt # 1s, 2s, 4s logger.warning(f"[RETRY] Server error {response.status_code} for '{account_name}', retrying in {backoff}s (attempt {attempt + 1}/{max_retries})") time.sleep(backoff) continue elif should_retry: # Last attempt failed with retryable error logger.error(f"[ERROR] Server error {response.status_code} for '{account_name}' after {max_retries} attempts") return { 'status': 'error', 'status_code': response.status_code, 'error': response.text, 'payload': payload, 'retryable': True } else: # Non-retryable error return handle_error_response(response.status_code, response.text, payload, account_name) except requests.exceptions.RequestException as e: if should_retry_exception(e) and attempt < max_retries - 1: logger.warning(f"[RETRY] {type(e).__name__} for '{account_name}', retrying (attempt {attempt + 1}/{max_retries})") time.sleep(2 ** attempt) continue elif should_retry_exception(e): logger.error(f"[ERROR] {type(e).__name__} for vendor '{account_name}' after {max_retries} attempts: {e}") last_error = str(e) else: logger.error(f"[ERROR] Request failed for vendor '{account_name}': {e}") last_error = str(e) break return {'status': 'error', 'error': last_error, 'payload': payload, 'retryable': True} def write_success_csv(success_rows: List[Dict], output_file: str, original_fieldnames: List[str], append: bool = False): """Write successful account creations to CSV with Account ID prepended.""" fieldnames = ['Account ID'] + [f for f in original_fieldnames if f != 'Account ID'] mode = 'a' if append else 'w' write_header = not append or not os.path.exists(output_file) or os.path.getsize(output_file) == 0 with open(output_file, mode, encoding='utf-8', newline='') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) if write_header: writer.writeheader() for row in success_rows: writer.writerow(row) def write_failures_csv(results: Dict[str, List], output_file: str, original_fieldnames: List[str]): """Write failed and skipped accounts to CSV for retry.""" fieldnames = list(original_fieldnames) + ['Error Type', 'Error Message', 'Retryable'] with open(output_file, 'w', encoding='utf-8', newline='') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() # Write errors for error in results.get('errors', []): payload = error.get('payload', {}) row = { 'Account Name': payload.get('name', ''), 'Company Brand': payload.get('company_brand', ''), 'Owner': payload.get('owner', ''), 'Currency': payload.get('payment_currency', ''), 'Service Tier': payload.get('service_tier_uuid', ''), 'Error Type': f"HTTP {error.get('status_code', 'Unknown')}" if 'status_code' in error else 'Network Error', 'Error Message': str(error.get('error', ''))[:500], 'Retryable': 'Yes' if error.get('retryable') else 'No' } writer.writerow(row) # Write skipped rows for skipped in results.get('skipped', []): data = skipped.get('data', {}) row = dict(data) row['Error Type'] = 'Skipped' row['Error Message'] = skipped.get('reason', '') row['Retryable'] = 'No' writer.writerow(row) def print_summary(results: Dict[str, List]): """Print summary of results with error classification.""" logger.info("\n" + "="*60) logger.info("SUMMARY") logger.info("="*60) logger.info(f"Successful: {len(results['success'])}") logger.info(f"Errors: {len(results['errors'])}") logger.info(f"Skipped: {len(results['skipped'])}") if results['errors']: retryable = [e for e in results['errors'] if e.get('retryable')] permanent = [e for e in results['errors'] if not e.get('retryable')] logger.info(f"\nErrors: {len(results['errors'])} total ({len(retryable)} retryable, {len(permanent)} permanent)") for error in results['errors'][:10]: # Show first 10 retry_marker = "[RETRYABLE]" if error.get('retryable') else "[PERMANENT]" logger.info(f" {retry_marker} Row {error.get('row')}: {error['payload']['name']}") logger.info(f" Error: {error.get('error', error.get('status_code'))}") if len(results['errors']) > 10: logger.info(f" ... and {len(results['errors']) - 10} more errors") if results['skipped']: logger.info(f"\nSkipped: {len(results['skipped'])} rows") for skipped in results['skipped'][:5]: # Show first 5 logger.info(f" Row {skipped['row']}: {skipped.get('reason')}") if len(results['skipped']) > 5: logger.info(f" ... and {len(results['skipped']) - 5} more skipped")