"""Async script for bulk product eligibility checks.""" import asyncio import json import os import random import re import sys from typing import List, Tuple import aiohttp import boto3 import config from openpyxl import Workbook from openpyxl.styles import Font logger = config.setup_logger(__name__) def load_upcs(upc_list: str) -> List[str]: """ Load UPCs from a string. Args: upc_list (str): UPCs. Returns: List[str]: List of UPC codes. Raises: SystemExit: If no UPCs provided. """ if upc_list: logger.debug('Loading UPCs...') upcs = list({u.strip() for u in re.split(r'[\n,]+', upc_list) if u.strip()}) else: logger.error('No UPCs provided.') sys.exit(1) logger.info(f'Loaded {len(upcs)} unique UPCs') return upcs def load_upcs_from_file(file_path: str) -> List[str]: """ Load UPCs from a file, one per line. Args: file_path (str): Path to UPC file. Returns: List[str]: List of UPC codes. Raises: SystemExit: If no UPCs found or file missing. """ if not file_path or not os.path.isfile(file_path): logger.error(f'UPC file not found: {file_path}') sys.exit(1) logger.debug(f'Loading UPCs from file: {file_path}') with open(file_path, 'r', encoding='utf-8') as f: upcs = list({line.strip() for line in f if line.strip()}) if not upcs: logger.error('No UPCs found in file.') sys.exit(1) logger.info(f'Loaded {len(upcs)} unique UPCs from file') return upcs async def fetch_upc( session: aiohttp.ClientSession, base_url: str, store_id: str, delivery_type: str, upc: str, timeout: int, retries: int = 3, backoff_factor: float = 0.5, ) -> Tuple[str, str, str]: """ Fetch eligibility status for a given UPC. Args: session (aiohttp.ClientSession): Active aiohttp session. base_url (str): Base API URL. store_id (str): Store identifier. delivery_type (str): Delivery type. upc (str): Product UPC code. timeout (int): Request timeout in seconds. retries (int): Number of retry attempts. backoff_factor (float): Base wait time (doubles each retry). Returns: Tuple[str, str, str]: (UPC, status_code or 'ERROR', response_text) """ url = f'{base_url}/products/{upc}/eligibility' for attempt in range(retries + 1): try: async with session.get( url, params={'store_id': store_id, 'delivery_type': delivery_type}, timeout=timeout ) as resp: text = await resp.text() if resp.status >= 500: raise Exception(f'Internal Server Error {resp.status}') return upc, str(resp.status), text except Exception as e: if attempt < retries: wait_time = backoff_factor * (2 ** attempt) + random.uniform(0, 0.3) logger.warning(f'[UPC {upc}] Attempt {attempt + 1} failed: {e}, retrying...') await asyncio.sleep(wait_time) else: return upc, 'ERROR', repr(e) async def run_all( upcs: List[str], base_url: str, store_id: str, delivery_type: str, concurrency: int, timeout: int, retries: int, backoff_factor: float, ) -> List[Tuple[str, str, str]]: """ Run eligibility checks for all UPCs concurrently. Args: upcs (List[str]): List of UPC codes. base_url (str): Base API URL. store_id (str): Store identifier. delivery_type (str): Delivery type. concurrency (int): Max number of concurrent requests. timeout (int): Request timeout in seconds. retries (int): Number of retry attempts. backoff_factor (float): Base wait time (doubles each retry). """ results = [] connector = aiohttp.TCPConnector(limit=concurrency) async with aiohttp.ClientSession(connector=connector) as session: tasks = [ fetch_upc(session, base_url, store_id, delivery_type, upc, timeout, retries, backoff_factor) for upc in upcs ] for coro in asyncio.as_completed(tasks): upc, status, text = await coro results.append((upc, status, text)) logger.info(f'[UPC {upc}] {status}: {text}') return results def save_results_to_excel(results: List[Tuple[str, str, str]], delivery_type: str, store_id: str, filename: str) -> str: """Save results to an Excel file and return file path.""" wb = Workbook() ws = wb.active ws.title = 'Products Eligibility Results' # Header row ws.append(['UPC', 'DELIVERY_TYPE', 'STORE_ID', 'IS_ELIGIBLE', 'REASON', 'STATUS']) # Make header row bold bold_font = Font(bold=True) for cell in ws[1]: cell.font = bold_font # Data rows for upc, status, text in results: is_eligible, reason = False, None try: resp_json = json.loads(text) if text.startswith('{') else None if resp_json: is_eligible = resp_json.get('is_eligible') or False reason = resp_json.get('reason') or resp_json.get('message') or 'NONE' else: reason = text # fallback if plain string except Exception as e: reason = f'Unparsable: {text} ({e})' ws.append([upc, delivery_type, store_id, str(is_eligible).upper(), reason, status]) wb.save(filename) logger.info(f'Results saved to {filename}') return filename def upload_to_s3(file_path: str, bucket: str, prefix: str) -> None: """Upload the given file to S3 under S3_PREFIX.""" s3 = boto3.client('s3') filename = os.path.basename(file_path) full_key = f'{prefix}/{filename}' s3.upload_file(file_path, bucket, full_key) logger.info(f'Uploaded {file_path} to s3://{bucket}/{full_key}') if __name__ == '__main__': try: if config.UPC_FILE: upcs = load_upcs_from_file(config.UPC_FILE) else: upcs = load_upcs(config.UPC_LIST) logger.info( f'Starting eligibility checks for {len(upcs)} UPCs ' f'(concurrency={config.CONCURRENCY}, timeout={config.TIMEOUT}s)' ) results = asyncio.run( run_all( upcs, config.BASE_URL, config.STORE_ID, config.DELIVERY_TYPE, config.CONCURRENCY, config.TIMEOUT, config.RETRY_ATTEMPTS, config.BACKOFF_FACTOR ) ) filename = f'./artifacts/{config.BUILD_NUMBER}_eligibility_results.xlsx' # Save locally save_results_to_excel(results, config.DELIVERY_TYPE, config.STORE_ID, filename) # Upload to S3 upload_to_s3(filename, config.S3_BUCKET, config.S3_PREFIX) except Exception as e: logger.error(f'Product check failed: {e}') sys.exit(1)