"""PROPER Class to retrieve data. Class for generating, executing and saving SQL queries for the Proper feeds. """ import datetime import smart_open from feed_sender.flows.proper_new_releases_tracks import exceptions from feed_sender.flows.proper_new_releases_tracks.conf import settings from feed_sender.flows.proper_new_releases_tracks.conf import ( transform_settings) from feed_sender.util import correlation_id from feed_sender.util import mysql from feed_sender.util import proper_common from feed_sender.util import sentry from feed_sender.util.aws import s3 from feed_sender.util.genre_subgenre_mapping import GenreSubgenreMapping def _convert(original): """Convert transformed values for printing to CSV. Args: original (str): Transformed string. Returns: str: Further string conversion for CSV printing. """ if original is None or original == '': return '""' transformed = str(original).replace('"', '""').strip() return '\"{transformed}\"'.format(transformed=transformed) # noqa class Proper(GenreSubgenreMapping): """SQL generator for generating data for Proper feeds.""" def __init__(self, cutoff): """Initialize class info to be persisted.""" super(Proper, self).__init__() self.art_relations_db = settings.ART_RELATIONS_DB self.physical_delivery_db = settings.PHYSICAL_DELIVERY_DB self.direct_delivery_db = settings.DIRECT_DELIVERY_DB self.store_id = settings.STORE_ID self.releases_threshold = settings.RELEASES_THRESHOLD self.cutoff = cutoff self.total_releases = 0 self.db_connection = None self._release_ids = None self._store_genre_mapping = None @property def store_genre_mapping(self): """Retrieve the store genre/subgenre mapping as a property. Returns: dict: Dictionary of genre and subgenre with key genre_id_subgenre_id. Example: {'4_494': {'subgenre': None, 'genre': 'Punk'}} """ self._store_genre_mapping = self.fetch_genre_mapping(self.store_id) return self._store_genre_mapping def fetch_products(self, s3path, job_ids_filename): """Get filtered Proper products. Args: s3path (str): S3 path to files stored. job_ids_filename (str): Filename for list of release_ids. """ parsed_ids = proper_common.get_ids_from_s3( s3path, job_ids_filename) self._release_ids = parsed_ids.get('release_ids') self.total_releases = len(self._release_ids) def filter_products_query(self): """Get filtered Proper products. Returns: list: Release ids of the filtered releases. """ if not self._release_ids: sql = """ SELECT DISTINCT (r.release_id) FROM {art_relations_db}.releases r LEFT JOIN {art_relations_db}.release_dms_master_restriction rdmr ON rdmr.customer_master_master_id = {store_id} AND rdmr.release_id = r.release_id LEFT JOIN {physical_delivery_db}.product_physical_feed_delivery_history ppfdh ON ppfdh.supply_chain = 'PROPER' AND ppfdh.product_id = r.release_id JOIN {art_relations_db}.product_physical pp ON pp.release_id = r.release_id JOIN {art_relations_db}.artist_info a ON r.artist_id = a.artist_id JOIN {art_relations_db}.vw_active_vendor_contract vw ON a.vendor_id = vw.vendor_id LEFT JOIN {art_relations_db}.red_essential_label_mapping vmap ON vmap.vendor_id = vw.vendor_id LEFT JOIN {art_relations_db}.vendor_dms_master_restriction vdmr ON vw.vendor_contract_id = vdmr.vendor_contract_id AND vdmr.customer_master_master_id = {store_id} WHERE r.release_status = 'in_content' AND r.deletions = 'N' AND rdmr.restriction_id is NULL AND vdmr.restriction_id is NULL AND ppfdh.id is NULL AND r.last_updated < '{cutoff}' #AND rmdr.last_updated < {cutoff} - temporarily comment out #since we lack data """.format( art_relations_db=self.art_relations_db, physical_delivery_db=self.physical_delivery_db, store_id=self.store_id, cutoff=self.cutoff) rows = mysql.execute_query(self.db_connection, sql) # query is successful, note: empty values may be returned self._release_ids = [row.get('release_id') for row in rows] self.total_releases = len(self._release_ids) return def _product_data_query(self): """Retrieve the metadata about products that Proper is interested in. Returns: list(DictCursor): List of database rows where each element represents a database row. """ if self.total_releases > 0: sql = """ SELECT r.release_id, a.vendor_id, v.company, r.`label`, r.display_upc, r.manufacturer_upc, r.product_code, GROUP_CONCAT( DISTINCT(ra.artist_name) SEPARATOR '|') AS artist_name, ra.role, r.release_name AS release_name, r.sale_start_date, r.genre_id AS genre, #genre - mapped rs.subgenre_id as subgenre, #subgenre - mapped r.distribution_format_id as format_id, #format - mapped pp.units_per_set, pp.packaging_id, #packageid - must be mapped pp.individual, #type of artist, used for transforms 0 AS weight, #weight - hardcoded as per Proper spec 0 AS `length`, #length - hardcoded as per Proper spec 0 AS width, #width - hardcoded as per Proper spec 0 AS height, #height - hardcoded as per Proper spec pp.explicit, {art_relations_db}.stripSpecialChars(proj.description) AS description, pp.wholesale_price, 1.00 AS buyprice, #buyprice - hardcoded as per Proper REPLACE('{image_url}', '', r.upc) AS artworkpath, pp.embargo_date, r.upc, #mapped to carveout info for isexportonly field pp.exclusive_for, pp.display_configuration FROM {art_relations_db}.releases r JOIN {art_relations_db}.product_physical pp ON r.release_id = pp.release_id LEFT JOIN {art_relations_db}.product_physical_packaging ppp ON pp.packaging_id = ppp.id JOIN {art_relations_db}.project proj ON r.project_id = proj.project_id JOIN {art_relations_db}.release_subgenre rs ON r.release_id = rs.release_id JOIN {art_relations_db}.release_artist ra ON r.release_id = ra.release_id JOIN {art_relations_db}.artist_info a ON r.artist_id = a.artist_id JOIN {art_relations_db}.vendor v ON a.vendor_id = v.vendor_id WHERE ra.role = 'performer' AND r.release_id IN({release_ids}) GROUP BY r.release_id """.format( release_ids=','.join(str(x) for x in self._release_ids), art_relations_db=self.art_relations_db, image_url=settings.IMAGE_URL) return mysql.execute_query(self.db_connection, sql) else: return {} def _write_header_row(self, fout, field_map): """Write a header row of Proper column names for provided field map. Args: fout (io): Open file handle for writing to. field_map (dict): Proper-Orchard field mapping. """ header_row = ','.join( [i.get('proper') for i in field_map]) fout.write('{}\r\n'.format(header_row)) def _inject_dependencies(self, record): """Inject additional data into a retrieved record array. Args: record (dict): db record. Returns: dict: New record with additional dependencies parameters. """ record['genre_subgenre_mapping'] = self.store_genre_mapping record['carveout_service_name'] = settings.CARVEOUT_SERVICE_NAME record['carveout_service_resource_path'] = ( settings.CARVEOUT_SERVICE_RESOURCE_PATH) record['application_name'] = settings.FEED_NAME record['application_env'] = settings.ENV record['correlation_id'] = correlation_id.get_correlation_id() return record def _write_records_to_csv( self, file_path, records, flow_field_mapping, ows_pricing_in_proper=False): """Write records to output file using field mapping. Args: file_path (str): Full path to output file. records (DictCursor): List of database result rows. flow_field_mapping (Dict): Mapping configuration for fields of this flow. """ with smart_open.smart_open(file_path, 'wb') as fout: self._write_header_row(fout, flow_field_mapping) for record in records: if flow_field_mapping == transform_settings.NEW_RELEASES_MAP: # inject additional data for use by transforms. self._inject_dependencies(record) row_to_write = [] for field_map in flow_field_mapping: field = field_map.get('orchard') orch_val = record.get(field) if ows_pricing_in_proper and field == 'wholesale_price': orch_val = proper_common.get_ows_pricing( record.get('release_id')) if 'transform' in field_map: param_dict = {} if 'depends' in field_map: param_dict = { key: record.get(key) for key in field_map.get('depends')} try: transformed = field_map.get('transform')( orch_val, **param_dict) except exceptions.TransformException as tr_except: sentry.get_client().captureMessage(tr_except.msg) del row_to_write[:] # remove entry in releases id self._release_ids.remove( (record.get('release_id'))) break row_to_write.append(_convert(transformed)) else: row_to_write.append(_convert(orch_val)) if row_to_write: row_str = ','.join(str(x) for x in row_to_write) fout.write('{}\r\n'.format(row_str)) return def convert_releases_to_csv( self, s3path, filename, ows_pricing_in_proper=False): """Convert field values retrieved from MySQL and write to CSV. @todo(pkuong): Investigate if there is a posssibility of making _product_data_query to yield the records instead of saving all data in records variable. Args: s3path (str): S3 bucket for releases defined in settings.py. filename (str): Name as specified by Proper defined in settings.py. ows_pricing_in_proper (bool): Flag for using ows-pricing """ records = self._product_data_query() if len(records) != self.total_releases: raise Exception( 'Task: {}. Bad mapping occurred in the database. ' 'Number of records: {}, Total releases: {}'.format( 'convert_releases_to_csv', len(records), self.total_releases)) full_path = '{path}/{file}'.format(path=s3path, file=filename) self._write_records_to_csv( full_path, records, transform_settings.NEW_RELEASES_MAP, ows_pricing_in_proper) def _track_data_query(self, s3path, release_id_file): """Get track level data. Returns: list(DictCursor): List of database rows where each element represents a database row. """ # There might be a situation when we do not get releases. full_path = '{path}/{file}'.format(path=s3path, file=release_id_file) data_from_s3 = s3.read_s3_object(full_path).decode( 'utf-8').rstrip() if not data_from_s3.strip(): return {} parsed_data = [line for line in data_from_s3.split('\n')] self._release_ids = parsed_data self.total_releases = len(self._release_ids) if self.total_releases > 0: sql = ( 'SELECT ' 'rv.product_code, ' 't.cd as disc, ' '1 as side, ' 't.track_id, ' 't.track_name, ' 't.length_minute as length_minutes, ' 't.length_seconds, ' 't.isrc, ' 'ta.name as performer, ' 'ta.type ' 'FROM ' ' (SELECT r.release_id, r.product_code ' ' FROM {art_relations_db}.releases r ' ' WHERE r.release_id in ({release_ids})) AS rv ' 'JOIN {art_relations_db}.track t ' ' ON t.release_id = rv.release_id ' 'LEFT JOIN {art_relations_db}.track_artist ta ' ' ON t.id = ta.track_id ' "WHERE ta.type = 'performer'").format( release_ids=','.join(str(x) for x in self._release_ids), art_relations_db=self.art_relations_db) return mysql.execute_query(self.db_connection, sql) return {} def convert_tracks_to_csv( self, s3path, tracks_file, release_ids_file, ows_pricing_in_proper=False): """Convert field values retrieved from MySQL and write to CSV. Args: s3path (str): S3 bucket for tracks defined in settings.py. tracks_file (str): Name specified by Proper defined in settings.py. release_ids_file (str): Release IDs filename. ows_pricing_in_proper (bool): Flag for using ows-pricing """ records = self._track_data_query(s3path, release_ids_file) full_path = '{path}/{file}'.format(path=s3path, file=tracks_file) self._write_records_to_csv( full_path, records, transform_settings.NEW_TRACKS_MAP, ows_pricing_in_proper) return def save_release_ids_to_s3(self, s3path, release_ids_file): """Persist list of Proper release ids and its last updated date in S3. Args: s3path (str): S3 bucket for file defined in settings.py. release_ids_file (str): Where release ids are saved. Returns: bool: True if success. False otherwise. """ self.total_releases = len(self._release_ids) full_path = '{path}/{file}'.format( path=s3path, file=release_ids_file) with smart_open.smart_open(full_path, 'wb') as fout: if not self.total_releases: fout.write(' ') for release_id in self._release_ids: if release_id: fout.write( '{id}\n'.format(id=str(release_id))) return True def _update_feed_delivery_history_table(self, s3_file_contents): """Update product_physical_feed_delivery_history. Set `date_delivered` to current time. Args: s3_file_contents (str): List of release ids we delivered. Returns: bool: True if success. False otherwise. """ current_time = datetime.datetime.now().isoformat() parsed_ids = [line for line in s3_file_contents.split('\n')] new_rows = [] for id in parsed_ids: new_rows.append("({id}, 'PROPER', 'NEW', '{date}')".format( id=id, date=current_time)) sql = ( 'INSERT INTO {physical_delivery_db}.product_physical_' 'feed_delivery_history(product_id, supply_chain, ' 'delivery_type, date_delivered) VALUES {new_values}').format( physical_delivery_db=self.physical_delivery_db, new_values=','.join(new_rows)) written_rows = mysql.execute_write_query(self.db_connection, sql) if len(new_rows) != written_rows: return False return True def _update_delivered_flag_on_changes_table(self, s3_file_contents): """Update product_physical_change_history table. Setting `delivered` to 'Y' if it is a new product registered before the cutoff. Args: s3_file_contents (str): list of release ids we delivered. Returns: bool: True if success. False otherwise. """ parsed_ids = [line for line in s3_file_contents.split('\n')] sql = ( "UPDATE {art_relations_db}.product_physical_" # noqa "change_history " # noqa "SET delivered = '{status}' " "WHERE date_changed < '{cutoff}' " "AND product_id IN ({release_ids})").format( # noqa art_relations_db=self.art_relations_db, status='Y', cutoff=self.cutoff, release_ids=','.join(str(i) for i in parsed_ids)) # No need to check for row count affected as 0 may be legitimate mysql.execute_write_query(self.db_connection, sql) return True def update_delivery_history(self, s3path, filename): """Update status for releases we just delivered. Args: s3path (str): S3 bucket for file containing release info. filename (str): File containing release/product info. """ # get list of release ids from file # sample file contents: release_id_1\nrelease_id_2\n full_path = '{path}/{file}'.format(path=s3path, file=filename) data_from_s3 = s3.read_s3_object(full_path).decode('utf-8').rstrip() if not data_from_s3: return True connection = self.connect_to_sql(autocommit=False) completed = self._update_feed_delivery_history_table(data_from_s3) completed = completed and self._update_delivered_flag_on_changes_table( data_from_s3) if not completed: connection.rollback() raise Exception('Failed to update delivery history.') connection.commit() self.close_sql() return True def connect_to_sql(self, autocommit=True): """Connect to MySQL through pyMySQL. Returns: pyMYSQL connection object. """ self.db_connection = mysql.get_art_db_connection_pymysql(autocommit) return self.db_connection def close_sql(self): """Close pyMYSQL connection. Returns: bool: True after operation is completed. """ if not self.db_connection: self.db_connection.close() return