"""Common utility functions.""" import re from urllib import parse import common_config from constants import statuses from constants import errors def parse_file_key(key): """Parse file key. Args: key (str): S3 file key Returns: dict: empty dict or attachment attributes (label_id, filename etc.) """ key = parse.unquote_plus(key) match = re.search(common_config.FILE_KEY_REGEXP, key) if match is None: return {} return match.groupdict() def get_file_name_from_key(key): """Extract file name from S3 file key. Args: key (str): file key from S3 event Returns: str: file name part of the file key """ file_name = key.split('/')[-1] return parse.unquote_plus(file_name) def get_quarter_periods_by_period(period_id): """Find out all period ids in quarter by period id. Args: period_id (int): period id Returns: list: list of quarter period ids """ period_id = int(period_id) fractional_part = period_id / 3 - int(period_id / 3) if fractional_part == 0: # last period in quarter period_list = [period_id - 2, period_id - 1, period_id] elif round(fractional_part, 1) == 0.7: # middle period in quarter period_list = [period_id - 1, period_id, period_id + 1] else: # first period in quarter period_list = [period_id, period_id + 1, period_id + 2] return period_list def format_quarter_file_name(file_name, period_id): """Format file name for quarter with defined pattern. Args: file_name (str): original file name, e.g. test.txt period_id (int): period id, e.g 217 Returns: str: Formatted file name for sorting, e.g. test.txt(217) """ pattern = '{file_name}({period_id})' return pattern.format(file_name=file_name, period_id=period_id) def get_attachment_primary_key_value(account_type, account_id, period_ids): """Get formatted attachment primary key value. Args: account_type (str): account type from GRASS. account_id (int): account id (vendor id). period_ids (list): list of accounting periods. Returns: str: Formatted primary key value """ primary_key_account_type_id = '{}{}'.format(account_type, account_id) periods_str = '_'.join(str(period_id) for period_id in period_ids) return '{}_{}'.format(primary_key_account_type_id, periods_str) def s3_event_get_attachment_details(s3_event, file_attrs): """Get attachment details from s3_event. Args: s3_event (dict): AWS S3 event object. file_attrs (dict): Attachment file attributes Returns: dict: standardized dictionary with needed attachment fields. """ try: s3_object = s3_event['Records'][0]['s3']['object'] s3_bucket = s3_event['Records'][0]['s3']['bucket'] key = parse.unquote_plus(s3_object['key']) data = { 'file_key': key, 'file_name': file_attrs['filename'], 'original_file_name': file_attrs['filename'], 'file_type': file_attrs['extension'], 'bucket_name': s3_bucket['name'], 'file_size': s3_object['size'], 'etag': s3_object['eTag'], 'account_type': file_attrs['account_type'], 'label_id': file_attrs['label_id'], 'period_ids': file_attrs['period_id'], 'status': statuses.STATUS_COMPLETED, 'upload_date': s3_event['Records'][0]['eventTime'] } return data except KeyError as e: raise KeyError( errors.MISSING_KEY_EXCEPTION.format(key=str(e))) except Exception as e: raise Exception( errors.CANNOT_PARSE_BODY.format( error=str(e)))