import asyncio import json import logging import shutil import tempfile import time import uuid import zipfile from pathlib import Path from typing import Any, Dict, List, Optional from urllib.parse import quote import aiohttp from .amazon_consume_client import AmazonConsumeClient, DeliveryOutcome, DeliveryResult from .consume_mapper import rx_to_amazon_consume_response from .handlers import PGPHandler, S3Handler, SFTPHandler from .json_mapper import JSONMapper logger = logging.getLogger(__name__) class FileProcessor: """Main file processing orchestrator""" def __init__(self, config, request_store): self.config = config self.request_store = request_store self.sftp = SFTPHandler(config) self.pgp = PGPHandler(config) self.mapper = JSONMapper() self._consume_client = None self._active_tasks = 0 self._active_tasks_lock = asyncio.Lock() def _create_sftp_handler(self) -> SFTPHandler: return SFTPHandler(self.config) def _create_s3_handler(self) -> S3Handler: return S3Handler(self.config) def _get_consume_client(self) -> AmazonConsumeClient: if self._consume_client is None: self._consume_client = AmazonConsumeClient(self.config) return self._consume_client def _log_prefix(self, consultation_id: Optional[str] = None) -> str: parts = [] if consultation_id: parts.append(f"consultation_id={consultation_id}") if not parts: return "" return f"[{' '.join(parts)}] " def _list_incoming_files(self) -> Optional[List[str]]: if getattr(self.config, "use_local_scan", False): incoming_path = Path( getattr(self.config, "local_incoming_path", "connectivity_requests") ) if not incoming_path.exists(): logger.error("Local incoming path does not exist: %s", incoming_path) return None return sorted( [p.name for p in incoming_path.iterdir() if p.name.endswith(".zip")] ) sftp_handler = self._create_sftp_handler() if not sftp_handler.connect(): logger.error("Failed to connect to SFTP server") return None try: return sftp_handler.list_files(self.config.incoming_sftp_path) finally: sftp_handler.disconnect() def _prepare_request_payload(self, filename: str) -> Dict[str, Any]: log_prefix = "" use_local_scan = getattr(self.config, "use_local_scan", False) sftp_handler = None if not use_local_scan: sftp_handler = self._create_sftp_handler() if not sftp_handler.connect(): logger.error(f"{log_prefix}Failed to connect to SFTP server") return {"success": False} try: with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) # Step 1: Download encrypted zip from SFTP encrypted_zip_path = temp_path / filename if use_local_scan: incoming_path = Path( getattr( self.config, "local_incoming_path", "connectivity_requests", ) ) remote_path = str(incoming_path / filename) if not Path(remote_path).exists(): logger.error(f"{log_prefix}Local file not found: {remote_path}") return {"success": False} shutil.copy2(remote_path, encrypted_zip_path) else: remote_path = f"{self.config.incoming_sftp_path}/{filename}" if not sftp_handler.download_file( remote_path, str(encrypted_zip_path) ): return {"success": False} # Step 2: Decrypt the zip file decrypted_zip_path = temp_path / f"decrypted_{filename}" if not self.pgp.decrypt_file( str(encrypted_zip_path), str(decrypted_zip_path) ): return {"success": False} # Step 3: Extract and examine JSON file json_file_path = temp_path / "request.json" with zipfile.ZipFile(decrypted_zip_path, "r") as zip_ref: logger.info(f"Zip contents: {zip_ref.namelist()}") json_files = [ f for f in zip_ref.namelist() if f.endswith(".json") or "." not in f ] if not json_files: logger.error(f"{log_prefix}No JSON file found in zip") return {"success": False} zip_ref.extract(json_files[0], temp_path) extracted_file = temp_path / json_files[0] extracted_file.rename(json_file_path) # Step 4: Load and validate JSON data with open(json_file_path, "r", encoding="utf-8") as f: amazon_request = json.load(f) consultation_id = amazon_request.get("requestMetadata", {}).get( "consultationId" ) logger.info( "%sLoaded Amazon request with consultation_id: %s", log_prefix, consultation_id, ) # Step 5: Map to 1Rx format rx_request = self.mapper.amazon_to_1rx(amazon_request) # Step 6: Store request context in database if not consultation_id: logger.error(f"{log_prefix}No consultation_id found in request") return {"success": False} request_data = { "consultation_id": consultation_id, "reference_id": amazon_request.get("requestMetadata", {}).get( "referenceId" ), "message_id": amazon_request.get("requestMetadata", {}).get( "messageId" ), "original_filename": filename, "original_request": amazon_request, "rx_request": rx_request, "sftp_path": remote_path, } self.request_store.store_request(request_data) return { "success": True, "consultation_id": consultation_id, "rx_request": rx_request, "remote_path": remote_path, "filename": filename, } except Exception as e: logger.error(f"{log_prefix}Error preparing file {filename}: {e}") return {"success": False} finally: if sftp_handler: sftp_handler.disconnect() def _delete_remote_file(self, remote_path: str) -> bool: if getattr(self.config, "use_local_scan", False): try: Path(remote_path).unlink() return True except FileNotFoundError: logger.warning("Local file already deleted: %s", remote_path) return True except Exception as e: logger.error("Failed to delete local file %s: %s", remote_path, e) return False sftp_handler = self._create_sftp_handler() if not sftp_handler.connect(): logger.error("Failed to connect to SFTP server for delete") return False try: return sftp_handler.delete_file(remote_path) finally: sftp_handler.disconnect() async def process_file(self, filename: str) -> bool: """Process a single encrypted zip file and store context""" start_ts = time.perf_counter() async with self._active_tasks_lock: self._active_tasks += 1 active = self._active_tasks logger.info("Processing file: %s (active=%d)", filename, active) try: prepared = await asyncio.to_thread(self._prepare_request_payload, filename) if not prepared.get("success"): return False consultation_id = prepared["consultation_id"] rx_request = prepared["rx_request"] remote_path = prepared["remote_path"] log_prefix = self._log_prefix(consultation_id=consultation_id) # Step 7: Call 1Rx API (async - response comes via webhook) rx_response = await self.call_1rx_api(rx_request, log_prefix) if rx_response and rx_response.get("data"): self.request_store.update_request_status(consultation_id, "sent_to_1rx") logger.info(f"{log_prefix}Successfully sent to 1Rx for consultation") deleted = await asyncio.to_thread(self._delete_remote_file, remote_path) if not deleted: logger.warning( f"{log_prefix}Failed to delete {remote_path} from SFTP" ) logger.info( f"{log_prefix}Processing initiated, awaiting webhook responses" ) return True self.request_store.update_request_status(consultation_id, "failed_1rx") return False except Exception as e: logger.error(f"Error processing file {filename}: {e}") return False finally: elapsed = time.perf_counter() - start_ts async with self._active_tasks_lock: self._active_tasks -= 1 active = self._active_tasks logger.info( "Finished file: %s in %.2fs (active=%d)", filename, elapsed, active ) async def process_api_request(self, amazon_request: Dict[str, Any]) -> Dict[str, Any]: """Process an Amazon consultation request received via API (ingress). Same pipeline as SFTP but skips download/decrypt steps. Uses new Amazon API schema with messageIdentifiers + data.""" try: message_ids = amazon_request.get("messageIdentifiers", {}) data = amazon_request.get("data", {}) reference_id = message_ids.get("primaryReferenceId") # secondaryReferenceId is OPTIONAL in the spec; without one the # order is keyed (and correlated with the 1Rx webhook) by its # primaryReferenceId instead. consultation_id = message_ids.get("secondaryReferenceId") or reference_id message_id = data.get("requestMetadata", {}).get("messageId") tracking_id = message_ids.get("trackingId") is_test_data = bool(message_ids.get("isTestData") or False) if not consultation_id: logger.error("No reference ids found in API request") return { "success": False, "error": "Missing messageIdentifiers.primaryReferenceId", } log_prefix = self._log_prefix(consultation_id=consultation_id) logger.info(f"{log_prefix}Processing API ingress request") if is_test_data: # Persisted and echoed back; routing test orders away from # real doctors is a pending product decision. logger.warning( f"{log_prefix}Request flagged isTestData=true; processing normally" ) # Map to 1Rx format using new API mapper rx_request = self.mapper.amazon_api_to_1rx(amazon_request) # Store request in database request_data = { "consultation_id": consultation_id, "reference_id": reference_id, "message_id": message_id, "original_filename": f"api_ingress_{consultation_id}", "original_request": amazon_request, "rx_request": rx_request, "sftp_path": None, "tracking_id": tracking_id, "is_test_data": is_test_data, } is_new = self.request_store.store_request(request_data) if not is_new: # Amazon retries a failed request with a new trackingId but the # same reference ids. A row stuck in failed_1rx never reached # 1Rx, so reclaim it and reprocess; anything else is a true # duplicate. if self.request_store.reclaim_failed_request(request_data): logger.info( f"{log_prefix}Reclaimed previously failed request, retrying 1Rx call" ) else: logger.info(f"{log_prefix}Duplicate request, skipping 1Rx API call") return { "success": True, "consultation_id": consultation_id, "message_identifiers": message_ids, "message": "Duplicate request, already processed", } # Call 1Rx API rx_response = await self.call_1rx_api(rx_request, log_prefix) if rx_response and rx_response.get("data"): self.request_store.update_request_status(consultation_id, "sent_to_1rx") logger.info(f"{log_prefix}Successfully sent to 1Rx via API ingress") return { "success": True, "consultation_id": consultation_id, "message_identifiers": message_ids, "message": "Request accepted and sent to 1Rx for processing", } self.request_store.update_request_status(consultation_id, "failed_1rx") return { "success": False, "consultation_id": consultation_id, "message_identifiers": message_ids, "error": "Failed to send to 1Rx API", } except Exception as e: logger.error(f"Error processing API ingress request: {e}") return {"success": False, "error": str(e)} async def call_1rx_api( self, payload: Dict[str, Any], log_prefix: str = "" ) -> Dict[str, Any]: """Call 1Rx API asynchronously. Retries with exponential backoff on connection errors, timeouts and 5xx/429 responses. Other 4xx responses are payload errors and are not retried.""" headers = { "x-1rx-apikey": self.config.rx_api_key, "Content-Type": "application/json", } url = f"{self.config.rx_api_base_url}/orders" attempts = max(1, getattr(self.config, "rx_api_retry_attempts", 3)) base_delay = getattr(self.config, "rx_api_retry_delay_seconds", 1.0) timeout = aiohttp.ClientTimeout( total=getattr(self.config, "rx_api_timeout_seconds", 10.0) ) async with aiohttp.ClientSession(timeout=timeout) as session: for attempt in range(1, attempts + 1): retryable = False try: async with session.post( url, json=payload, headers=headers ) as response: if response.status == 202: result = await response.json() logger.info( f"{log_prefix}Successfully called 1Rx API (attempt {attempt})" ) return result error_text = await response.text() retryable = response.status >= 500 or response.status == 429 logger.error( f"{log_prefix}1Rx API error {response.status} " f"(attempt {attempt}/{attempts}): {error_text}" ) if not retryable: return None except (aiohttp.ClientError, asyncio.TimeoutError) as e: retryable = True logger.error( f"{log_prefix}Error calling 1Rx API " f"(attempt {attempt}/{attempts}): {e}" ) except Exception as e: logger.error(f"{log_prefix}Error calling 1Rx API: {e}") return None if retryable and attempt < attempts: delay = base_delay * (2 ** (attempt - 1)) logger.info( f"{log_prefix}Retrying 1Rx API call in {delay:.1f}s" ) await asyncio.sleep(delay) logger.error( f"{log_prefix}1Rx API call failed after {attempts} attempts" ) return None async def process_1rx_response( self, stored_request: Dict[str, Any], rx_response: Dict[str, Any] ): """Process 1Rx webhook response and generate final output. Status transitions are owned by the caller (ConsultationProcessor): failures propagate so the caller records failed_processing with the error, instead of being swallowed here and masked as success. """ consultation_id = stored_request["consultation_id"] logger.info(f"Processing 1Rx response for consultation {consultation_id}") # Parse stored data original_request = stored_request["original_request"] if isinstance(original_request, str): original_request = json.loads(original_request) with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) # Download the prescription once; every enabled delivery channel # (SFTP bundle, S3 pre-signed ZIP for the Shuttle API) reuses # the same file. jpeg_path: Optional[Path] = None if rx_response.get("status") == "APPROVED": jpeg_path = temp_path / f"{consultation_id}.jpg" if not await self.download_prescription_jpeg(rx_response, jpeg_path): raise Exception("Failed to download prescription JPEG") # Keep the 1Rx result so a failed API delivery can be resent # later without replaying the webhook. if hasattr(self.request_store, "save_rx_response"): self.request_store.save_rx_response(consultation_id, rx_response) sftp_enabled = getattr(self.config, "egress_sftp_enabled", True) api_enabled = getattr(self.config, "egress_api_enabled", False) # Amazon accepts API responses for test orders only, plus a # configurable ramp of the first N real orders. A real order # outside that ramp goes out over SFTP regardless of EGRESS_MODE. sftp_fallback_on_api_failure = False if ( api_enabled and getattr(self.config, "egress_api_test_orders_only", True) and not self._is_test_order(stored_request, original_request) ): if self._real_order_ramp_has_room(consultation_id): # A real customer is waiting on this response: if the # API delivery fails, deliver over SFTP instead of # failing the webhook and waiting for a manual resend. sftp_fallback_on_api_failure = not sftp_enabled else: logger.info( f"Consultation {consultation_id} is not a test order; " "delivering via SFTP instead of the Shuttle API" ) api_enabled = False sftp_enabled = True # SFTP first: the proven channel delivers before the API is # attempted, so an API failure can never cost the customer # their response during the migration. if sftp_enabled: await self._deliver_via_sftp( stored_request, original_request, rx_response, jpeg_path ) if api_enabled: result = await self._deliver_via_amazon_api( stored_request, original_request, rx_response, jpeg_path, temp_path, raise_on_failure=not (sftp_enabled or sftp_fallback_on_api_failure), ) if sftp_fallback_on_api_failure and not result.acknowledged: logger.warning( f"Consultation {consultation_id}: Shuttle API delivery " f"failed ({result.error_message or result.outcome.value}); " "falling back to SFTP for this real order" ) await self._deliver_via_sftp( stored_request, original_request, rx_response, jpeg_path ) logger.info( f"Successfully processed 1Rx response for consultation {consultation_id}" ) async def _deliver_via_sftp( self, stored_request: Dict[str, Any], original_request: Dict[str, Any], rx_response: Dict[str, Any], jpeg_path: Optional[Path], ) -> None: amazon_response = self.mapper.rx_to_amazon_response( rx_response, original_request ) await self.generate_and_upload_response( stored_request, amazon_response, rx_response, jpeg_path=jpeg_path ) def _real_order_ramp_has_room(self, consultation_id: str) -> bool: """True while fewer than EGRESS_API_REAL_ORDER_LIMIT real orders have been acknowledged by Amazon over the API. The count lives in the DB so it survives restarts; concurrent webhooks may overshoot the limit by the number of responses in flight at once. If the count cannot be read the order stays on SFTP, the safe channel.""" limit = int(getattr(self.config, "egress_api_real_order_limit", 0) or 0) if limit <= 0: return False if not hasattr(self.request_store, "count_real_orders_delivered_via_api"): return False try: delivered = self.request_store.count_real_orders_delivered_via_api() except Exception as e: logger.error( f"Consultation {consultation_id}: could not read real-order API " f"ramp count ({e}); delivering via SFTP" ) return False if delivered >= limit: logger.info( f"Consultation {consultation_id}: real-order API ramp exhausted " f"({delivered}/{limit} acknowledged); delivering via SFTP" ) return False logger.info( f"Consultation {consultation_id}: real order within API ramp " f"({delivered}/{limit} acknowledged so far); delivering via Shuttle API" ) return True @staticmethod def _is_test_order( stored_request: Dict[str, Any], original_request: Dict[str, Any] ) -> bool: """True when Amazon flagged the order as test data. Prefers the is_test_data column persisted at ingress; falls back to the original request for rows stored before that column existed. Legacy SFTP requests carry no flag and are treated as real orders. """ stored_flag = stored_request.get("is_test_data") if stored_flag is not None: return bool(stored_flag) message_ids = original_request.get("messageIdentifiers") or {} return message_ids.get("isTestData") is True async def _deliver_via_amazon_api( self, stored_request: Dict[str, Any], original_request: Dict[str, Any], rx_response: Dict[str, Any], jpeg_path: Optional[Path], temp_path: Path, *, raise_on_failure: bool, ) -> DeliveryResult: """Send one consultation response to the Shuttle API and record the outcome. Terminal failures (ERRORED / auth / retries exhausted / payload could not be built) raise only when the SFTP copy is not also being delivered — otherwise they are recorded and alerted for a manual resend via the management endpoint.""" consultation_id = stored_request["consultation_id"] log_prefix = self._log_prefix(consultation_id=consultation_id) tracking_id = str(uuid.uuid4()) api_error: Optional[str] = None result: Optional[DeliveryResult] = None try: file_url = None file_name = None if jpeg_path is not None: file_url, file_name = await self.upload_prescription_zip( stored_request, jpeg_path, temp_path ) payload = rx_to_amazon_consume_response( rx_response, original_request, tracking_id=tracking_id, provider_name=getattr(self.config, "amazon_provider_name", "myrx"), file_url=file_url, file_name=file_name, ) result = await self._get_consume_client().send_consultation_response( payload ) if result.acknowledged: self._record_egress(consultation_id, tracking_id, "acknowledged", None) logger.info( f"{log_prefix}Amazon API acknowledged response " f"(trackingId={tracking_id})" ) else: api_error = result.error_message or result.outcome.value self._record_egress( consultation_id, tracking_id, result.outcome.value, api_error ) except Exception as e: # Mapping or S3 staging failed before the API was even called. api_error = str(e) result = DeliveryResult( outcome=DeliveryOutcome.ERRORED, error_message=api_error ) self._record_egress( consultation_id, tracking_id, "failed_to_build", api_error ) if api_error is not None: message = ( f"{log_prefix}Amazon API delivery failed " f"(trackingId={tracking_id}): {api_error}" ) if raise_on_failure: raise Exception(message) logger.error( f"{message} — SFTP carries this response; fix and resend via " "POST /management/consultations/{id}/resend" ) return result def _record_egress( self, consultation_id: str, tracking_id: str, status: str, error_message: Optional[str], ) -> None: if not hasattr(self.request_store, "record_egress_api_result"): return try: self.request_store.record_egress_api_result( consultation_id, tracking_id, status, error_message ) except Exception as e: logger.error(f"Failed to record egress result: {e}") async def resend_via_amazon_api(self, consultation_id: str) -> Dict[str, Any]: """Ops re-send after a failed API delivery: rebuild the payload from the stored request + 1Rx response and send with a NEW trackingId (Amazon's rule for a corrected payload). Raises LookupError when the consultation is unknown, ValueError when there is nothing to resend.""" if not getattr(self.config, "egress_api_enabled", False): raise ValueError( f"Egress API is not enabled (EGRESS_MODE={getattr(self.config, 'egress_mode', 'sftp')})" ) stored_request = self.request_store.get_request_by_consultation_id( consultation_id ) if not stored_request: raise LookupError(f"No stored request for consultation {consultation_id}") rx_response = stored_request.get("rx_response") if isinstance(rx_response, str): rx_response = json.loads(rx_response) if not rx_response: raise ValueError( f"No stored 1Rx response for consultation {consultation_id}; " "nothing to resend" ) original_request = stored_request["original_request"] if isinstance(original_request, str): original_request = json.loads(original_request) with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) jpeg_path: Optional[Path] = None if rx_response.get("status") == "APPROVED": jpeg_path = temp_path / f"{consultation_id}.jpg" if not await self.download_prescription_jpeg(rx_response, jpeg_path): raise Exception("Failed to download prescription JPEG") result = await self._deliver_via_amazon_api( stored_request, original_request, rx_response, jpeg_path, temp_path, raise_on_failure=False, ) return { "consultation_id": consultation_id, "outcome": result.outcome.value, "http_status": result.http_status, "error_message": result.error_message, "acknowledged": result.acknowledged, } async def generate_and_upload_response( self, stored_request: Dict[str, Any], amazon_response: Dict[str, Any], rx_response: Dict[str, Any], jpeg_path: Optional[Path] = None, ): """Generate response files and upload to outgoing SFTP with local storage. ``jpeg_path`` is the already-downloaded prescription (the caller fetches it once for all delivery channels); when None it is downloaded here for backwards compatibility. """ try: consultation_id = stored_request["consultation_id"] with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) # Create JSON response file response_json_path = temp_path / "response.json" with open(response_json_path, "w", encoding="utf-8") as f: json.dump(amazon_response, f, indent=2) # Create output folder output_folder = temp_path / "output" output_folder.mkdir() shutil.copy(response_json_path, output_folder / "response.json") if rx_response["status"] == "APPROVED": if jpeg_path is None: jpeg_path = temp_path / f"{consultation_id}.jpg" if not await self.download_prescription_jpeg( rx_response, jpeg_path ): logger.error("JPEG download failed.") raise Exception("Failed to download prescription JPEG") shutil.copy(jpeg_path, output_folder / f"{consultation_id}.jpg") # Create zip file output_zip_path = temp_path / "response.zip" with zipfile.ZipFile(output_zip_path, "w") as zip_ref: for file_path in output_folder.iterdir(): zip_ref.write(file_path, file_path.name) # Encrypt zip file encrypted_output_path = temp_path / "response_encrypted.zip" if not self.pgp.encrypt_file( str(output_zip_path), str(encrypted_output_path) ): raise Exception("Failed to encrypt response file") # Upload to outgoing SFTP reference_id = stored_request["reference_id"] message_id = str(uuid.uuid4()) remote_output_path = ( f"{self.config.outgoing_sftp_path}/{reference_id}_{message_id}.zip" ) # Use outgoing username for upload outgoing_username = getattr( self.config, "outgoing_sftp_username", self.config.sftp_username ) sftp_handler = self._create_sftp_handler() try: if not sftp_handler.upload_file( str(encrypted_output_path), remote_output_path, outgoing_username, ): raise Exception("Failed to upload encrypted response") finally: sftp_handler.disconnect() logger.info(f"Successfully uploaded response to {remote_output_path}") except Exception as e: logger.error(f"Error generating response: {e}") raise async def download_prescription_jpeg( self, rx_response: Dict[str, Any], jpeg_path: Path ) -> bool: """Download prescription JPEG from 1Rx service""" try: # Extract pharmacyOrderId from the 1Rx webhook response pharmacy_order_id = rx_response.get("pharmacyOrderId") if not pharmacy_order_id: logger.error("No pharmacyOrderId found in 1Rx response") return False # Construct the JPEG download URL. The order id must be # percent-encoded: Amazon's purchaseId#workItemId format contains # '#', which would otherwise truncate the URL as a fragment. jpeg_url = ( f"{self.config.rx_api_base_url}/orders/" f"{quote(pharmacy_order_id, safe='')}/download-jpeg-prescription" ) logger.info(f"Downloading prescription JPEG from: {jpeg_url}") headers = { "x-1rx-apikey": self.config.rx_api_key, } # Download the JPEG async with aiohttp.ClientSession() as session: async with session.get(jpeg_url, headers=headers) as response: if response.status == 200: # Write JPEG content to file jpeg_content = await response.read() with open(jpeg_path, "wb") as f: f.write(jpeg_content) # Verify JPEG was written file_size = jpeg_path.stat().st_size logger.info(f"JPEG downloaded successfully: {file_size} bytes") return True else: error_text = await response.text() logger.error( f"PDF download failed: {response.status} - {error_text}" ) return False except Exception as e: logger.error(f"Exception downloading PDF: {e}") return False async def upload_prescription_zip( self, stored_request: Dict[str, Any], jpeg_path: Path, temp_path: Path ) -> tuple[str, str]: """Bundle the prescription into a ZIP, stage it in S3 and return ``(pre-signed URL, file name)`` for the Shuttle API payload. Amazon's Consume schema accepts only ``formatType: ZIP`` with a ``fileUrl`` it downloads within the pre-sign TTL, so the JPEG that goes inside the SFTP bundle is zipped and uploaded to our bucket instead of being sent inline. The returned file name is the JPEG *inside* the ZIP, not the ZIP itself: Amazon's ``rxImageDetails.fileName`` is "the name of the prescription image file", and they flagged a ``.zip`` value there.""" consultation_id = stored_request["consultation_id"] reference_id = stored_request["reference_id"] file_name = jpeg_path.name zip_name = f"{consultation_id}.zip" zip_path = temp_path / zip_name with zipfile.ZipFile(zip_path, "w") as zip_ref: zip_ref.write(jpeg_path, file_name) prefix = (getattr(self.config, "s3_prefix", "") or "").strip("/") s3_key = "/".join(p for p in (prefix, reference_id, zip_name) if p) s3_handler = self._create_s3_handler() uploaded = await asyncio.to_thread( s3_handler.upload_file, str(zip_path), s3_key ) if not uploaded: raise Exception(f"Failed to upload prescription ZIP to S3: {s3_key}") file_url = await asyncio.to_thread( s3_handler.generate_presigned_url, s3_key, self.config.s3_presign_ttl_seconds, ) if not file_url: raise Exception(f"Failed to pre-sign prescription ZIP URL: {s3_key}") logger.info("Prescription ZIP staged for Amazon at S3 key: %s", s3_key) return file_url, file_name async def scan_and_process(self): """Scan SFTP directory and process all zip files""" if getattr(self.config, "use_local_scan", False): logger.info("Scanning local directory for files...") else: logger.info("Scanning SFTP directory for files...") files = self._list_incoming_files() if files is None: return logger.info(f"Found {len(files)} zip files to process") max_parallel = getattr(self.config, "max_parallel_files", 10) if max_parallel < 1: max_parallel = 1 batch_size = getattr(self.config, "scan_batch_size", max_parallel * 20) semaphore = asyncio.Semaphore(max_parallel) async def _process_with_semaphore(file_name: str): async with semaphore: try: return await self.process_file(file_name) except Exception as e: logger.error(f"Failed to process file {file_name}: {e}") return False for index in range(0, len(files), batch_size): batch = files[index : index + batch_size] tasks = [ asyncio.create_task(_process_with_semaphore(name)) for name in batch ] await asyncio.gather(*tasks)