"""Tasks for Apple's Reporter Tool.""" import os import subprocess from zipfile import is_zipfile, ZipFile from botocore.exceptions import ClientError from garcon import task from garcon_contrib.aws.utils import garcon_s3 from garcon_contrib.dynamo_feed_status import garcon_feed_status from secrets_manager.swf_ext import SWFSecretsManager import xmltodict from feed_ingestion.util import itunes_reporter from feed_ingestion.util import task_status from feed_ingestion.util.aws import s3 from feed_ingestion.util.itunes_reporter import ReporterException from feed_ingestion.util.sentry_util import send_error_or_warning def _get_cred(cred_name, optional=False): """Get credential from secrets manager.""" environment = os.environ.get('Environment', 'dev') service_name = 'itunesconnect' secrets_manager_client = SWFSecretsManager( environment=environment, service_name=service_name) try: return secrets_manager_client.get_cred(cred_name) except ClientError as e: if e.response['Error']['Code'] == 'ResourceNotFoundException': if optional: return None raise e def _get_access_token(licensor): """Get access token from secrets manager.""" # try licensor-specific access token licensor_token = '{}_{}'.format( itunes_reporter.ACCESS_TOKEN_SECRET, licensor ) access_token = _get_cred( cred_name=licensor_token, optional=True ) if access_token: return access_token return _get_cred( cred_name=itunes_reporter.ACCESS_TOKEN_SECRET, optional=False ) @task.decorate(timeout=14400) def extract_reporter_file_to_s3( activity, reporter_account, report_type, date, destination_s3_path, feed_name=None, report_role='sales', report_country=None, vendors=None, licensor=None): """Extract a report from Apple's Reporter Tool and upload it to S3. Args: activity (ActivityWorker): The activity worker. reporter_account (str): Apple Reporter Account ('ORCHARD' or 'IODA'). report_type (str): report_type to download (see util.itunes_reporter module) date (str): Reporting date (YYYY-MM-DD). destination_s3_path (str): Destination S3 path to the archive location in s3://foo/bar/ format. feed_name (str): If passed, updates feed's status table entry for this date with whether or not the file was Downloaded or Not Available. report_role (str): Report role (sales or finance). report_country (str): Country code. vendors (dict): Information about Apple Music vendor accounts. Returns: dict: Adds an entry to the context with a key of the file name and a value of whether the file was Downloaded or Not Available. """ access_token = _get_access_token(licensor=licensor) reporter = itunes_reporter.get_reporter( reporter_account, date, report_country, vendors, access_token=access_token) file_name = _get_report_file_name(report_role, report_type, reporter) context = {file_name: garcon_feed_status.STATUS_NOT_AVAILABLE} # abort if already downloaded if feed_name is not None: file_status = garcon_feed_status.get_status(feed_name, date, file_name) if file_status == garcon_feed_status.STATUS_DOWNLOADED: context[file_name] = garcon_feed_status.STATUS_DOWNLOADED return context # download report file via Reporter try: activity.logger.info('Downloading {}'.format(file_name)) result = _download_report(report_role, report_type, reporter) _handle_file_name_incorrection(result, file_name, activity) except subprocess.TimeoutExpired: activity.logger.info('{} file not yet available'.format(file_name)) if feed_name is not None: garcon_feed_status.set_status( feed_name, date, file_name, status=garcon_feed_status.STATUS_NOT_AVAILABLE) return context except ReporterException as exception: if exception.error_code == ReporterException.FILE_UNAVAILABLE_ERR_CODE: activity.logger.info('{} file not yet available'.format(file_name)) if feed_name is not None: garcon_feed_status.set_status( feed_name, date, file_name, status=garcon_feed_status.STATUS_NOT_AVAILABLE) return context elif exception.error_code == ReporterException.NO_REPORT_IS_AVAILABLE: activity.logger.info('No report is available for {}'.format(date)) if feed_name is not None: garcon_feed_status.set_status( feed_name, date, file_name, status=garcon_feed_status.STATUS_NOT_AVAILABLE) return context elif exception.error_code == ReporterException.NO_SALES_FOR_DATE: activity.logger.info( 'There were no sales for the date specified. File: {}'.format( file_name)) return context else: if exception.error_code == ReporterException.INVALID_TOKEN: # Let's see when Token is invalid in dev environment activity.logger.info('iTunes Reporter Access Token is invalid') if os.environ.get('SENTRY_DSN'): send_error_or_warning(exception) return context activity.logger.info('Download complete for {}'.format(file_name)) context[file_name] = garcon_feed_status.STATUS_DOWNLOADED # todo(jpenner): add option to verify checksum # Apple sometimes delivers .txt.gz reports as actual ZIP data # (undocumented). Rename so the existing zip branch handles it. if os.path.exists(file_name) and is_zipfile(file_name): zip_name = file_name.replace('.txt.gz', '.zip') activity.logger.info( '{} contains ZIP data; renaming to {}'.format( file_name, zip_name)) os.rename(file_name, zip_name) if os.path.exists(file_name): # upload report file to s3 copy_on_s3_result = _upload_to_s3( file_name, destination_s3_path, activity) elif os.path.exists(file_name.replace('.txt.gz', '.zip')): file_name = file_name.replace('.txt.gz', '.zip') activity.logger.info('Uploading zipped file {}'.format(file_name)) with ZipFile(file_name, 'r') as zf: zf.extractall() file_list = zf.namelist() activity.logger.info('Start uploading files list - {}'.format( file_list)) copy_on_s3_result = len(file_list) > 0 for name in file_list: copy_on_s3_result = copy_on_s3_result and _upload_to_s3( name, destination_s3_path, activity) os.remove(name) else: if feed_name is not None: garcon_feed_status.set_status( feed_name, date, file_name, status=garcon_feed_status.STATUS_NOT_AVAILABLE) error = Exception( 'Wrong name given for downloaded file: {}'.format(file_name) ) # Raise an error in order for it to be caught by Sentry raise error # remove local report file os.remove(file_name) if feed_name is not None: if not copy_on_s3_result: garcon_feed_status.set_status( feed_name, date, file_name, status=garcon_feed_status.STATUS_NOT_AVAILABLE) return {file_name: garcon_feed_status.STATUS_NOT_AVAILABLE} garcon_feed_status.set_status( feed_name, date, file_name, status=garcon_feed_status.STATUS_DOWNLOADED) vendor_id = reporter_account contexts = task_status.get_report_contexts(feed_name, date) if contexts: task_status.add_newcontext(feed_name, date, vendor_id) task_status.update_report_context_status( feed_name, date, vendor_id, task_status.CONTEXT_STATUS_IN_PROGRESS) return context def _get_report_file_name(report_role, report_type, reporter): """Get report filename. Args: report_role (str): Report role (sales of finance). report_type (str): report_type to download (see util.itunes_reporter module). reporter (ItunesReporter): Reporter instance. Returns: str: Report file name string. """ if report_role == 'sales': file_name = reporter.get_sales_report_file_name(report_type) elif report_role == 'finance': file_name = reporter.get_finance_report_file_name(report_type) else: raise Exception('Unknown Report Role: {}.'.format(report_role)) return file_name def _download_report(report_role, report_type, reporter): """Download report file. Args: report_role (str): Report role (sales of finance). report_type (str): report_type to download (see util.itunes_reporter module). reporter (ItunesReporter): Reporter instance. """ if report_role == 'sales': return reporter.download_sales_report(report_type) elif report_role == 'finance': return reporter.download_finance_report(report_type) def _upload_to_s3(file_name, destination_s3_path, activity): """Upload report file to s3. Args: file_name (str): The name of the file on the local system destination_s3_path (str): The s3 destination location activity (ActivityWorker): The activity worker. """ bucket_name, key_path = garcon_s3.extract_bucket_path(destination_s3_path) s3_key_name = '{key_path}{file_name}'.format( key_path=key_path, file_name=file_name ) activity.logger.info( 'Uploading {} to s3://{}/{}'.format( file_name, bucket_name, s3_key_name ) ) try: file_size = s3.upload_to_s3( file_path=file_name, bucket_name=bucket_name, object_key=s3_key_name, ) activity.logger.info( 'Upload {} to s3://{}/{} complete. File size {}'.format( file_name, bucket_name, s3_key_name, file_size ) ) return True except ClientError: activity.logger.info('Upload {} to s3://{}/{} failed'.format( file_name, bucket_name, s3_key_name)) raise def _handle_file_name_incorrection(reporter_response, filename, activity): """Rename report file name if needed. For some reason apple api returns file with unexpected name AppleMusic_SongUniques_80028967_20230801_V1_0_[1-2].txt.gz as a result unexpected suffix from file name will be removed. @param reporter_response: XML Reporter result @param filename: the expected filename """ if '' in reporter_response: message = xmltodict.parse( reporter_response.strip() )['Output']['Message'] else: message = reporter_response.strip() actual_filename = message.split(' ')[-1] if filename.split('.')[0] == actual_filename.split('.')[0]: return elif actual_filename.startswith(filename.replace('.txt.gz', '') .replace('.zip', '')): activity.logger.info('Unexpected file name found. Renaming {} to {}' .format(actual_filename, filename)) os.rename(actual_filename, filename)