# noqa: D100 import re import typing from datetime import datetime from datetime import timezone def split_on_successful_errors_and_awaited( # noqa: D103 acks, auto_retry_patterns: typing.Optional[typing.List[str]] = None, cutoff: typing.Optional[datetime] = None, ): result: typing.Dict[str, typing.List[typing.Dict[str, typing.Any]]] = { 'errors': [], 'successful': [], 'awaited': [] } pattern = r'(.*?)' for ack in acks: ack_response = { 'version_id': ack['unacked_delivery']['version_id'], 'datetime': ack['unacked_delivery']['datetime'] } if 'ack_content' not in ack or ack['ack_content'] is None: result['awaited'].append(ack_response) continue error = _search_error_on_ack_content(ack['ack_content'], pattern) if error: if _is_auto_retry( error, auto_retry_patterns, ack['unacked_delivery']['datetime'], cutoff ): result['awaited'].append(ack_response) continue ack_response['message'] = error result['errors'].append(ack_response) else: result['successful'].append(ack_response) return result def _is_auto_retry( error_message: str, patterns: typing.Optional[typing.List[str]], delivery_datetime: str, cutoff: typing.Optional[datetime], ) -> bool: if not patterns or not cutoff: return False return any(pattern in error_message for pattern in patterns) and _is_within_cutoff(delivery_datetime, cutoff) def _is_within_cutoff(delivery_datetime: str, cutoff: datetime) -> bool: return datetime.fromisoformat(str(delivery_datetime)).replace(tzinfo=timezone.utc) >= cutoff def _search_error_on_ack_content(ack_content: str, pattern) -> typing.Optional[str]: match = re.search(pattern, ack_content, re.DOTALL) if match: return match.group(1) return None