"""Data manipulation. This module is concerned with pulling data from various sources and using logic to format or modify it. """ import datetime from accounting import contract_queries from accounting.adapters.db import DatabaseAdapter from accounting.adapters.file import read_statement_file from accounting.adapters.warehouse import SnowflakeAdapter from accounting.cacheable import cache_item from accounting.cacheable import cache_key_val from accounting.cacheable import cache_string_list from accounting.cacheable import get_cached_item from accounting.models.label_contract import LabelContract from accounting.models.statement import Statement LOWERCASE_ALPHABET = 'abcdefghijklmnopqrstuvwxyz' APPENDED_FIELD_SQL = ( 'SELECT ' 'r.`upc`, ' 'v.`vendor_id`, ' 'r.`release_status`, ' 'o.`owner_id` ' 'FROM `releases` r ' 'INNER JOIN `artist_info` a ON a.`artist_id` = r.`artist_id` ' 'INNER JOIN `vendor` v ON a.`vendor_id` = v.`vendor_id` ' 'INNER JOIN `owner` o ON o.`owner_abbrivation` = v.`owner` ' 'WHERE v.`status` in ("signed", "deletion") ') OWNER_TRACKS_SQL = ( 'SELECT ' 'o.`owner_id`, ' 'r.`upc`, ' 't.`cd` as cd_index, ' 't.`track_id` as track_index, ' 'p.`period_id` as ingestion_period_id, ' 'IF (r.`deletions` = "N" and v.`status` = "signed", 1, 0) as valid, ' 'COALESCE(t.`length_minute`, 0) + ' 'COALESCE(t.`length_seconds`, 0) / 60 as length_minutes, ' 'IF (t.`third_party_publisher` = "Y", 1, 0) as third_party_publisher ' 'FROM `owner` o ' 'INNER JOIN `vendor` v ON o.`owner_abbrivation` = v.`owner` ' 'INNER JOIN `artist_info` a ON v.`vendor_id` = a.`vendor_id` ' 'INNER JOIN `releases` r ON a.`artist_id` = r.`artist_id` ' 'LEFT JOIN `track` t ON r.`upc` = t.`upc` ' 'LEFT JOIN `period` p ON YEAR(r.`ingestion_completed`) = p.`year` ' 'AND MONTH(r.`ingestion_completed`) = p.`month` ' 'WHERE 1') MAX_AI_SF = ( 'SELECT ' 'max(statement_detail_id) as `value` ' 'FROM facts.prod.fact_sales ' 'WHERE true ') VENDOR_ID_SQL = ( 'SELECT v.`vendor_id` ' 'FROM `vendor` v ' 'WHERE v.`status` IN ("signed", "deletion") ') OWNER_CONTRACT_SQL = ( 'SELECT ' 'a.`service_id`, ' 'a.`agreement_id`, ' 'a.`effective_date`, ' 'a.`currencies_id`, ' 'aft.`representative_fee_type`, ' 'aft.`percentage_orchard_share` / 100 as percentage_orchard_share, ' 'aft.`label_contract_renewal_reduction` ' 'FROM agreement a ' 'INNER JOIN agreement_form_territorial aft ' 'ON aft.`agreement_id` = a.`agreement_id` ' 'LEFT JOIN agreement_rights_granted arg ' 'ON arg.`agreement_id` = a.`agreement_id` ' 'AND arg.`rights` = "account_parties" ' 'WHERE aft.`agreement_sub_type` IN ( ' '"territorial_rep", "territorial_sub_rep", "consulting") ' 'AND a.`service_type` = "owner" ' 'AND a.`effective_date` < CURDATE() ' 'ORDER BY a.`service_id`, a.`effective_date` ') OWNER_TRACKS_SQL = ( 'SELECT ' 'o.`owner_id`, ' 'p.`period_id`, ' 'count(*) AS count ' 'FROM `owner` o ' 'INNER JOIN `vendor` v ON o.`owner_abbrivation` = v.`owner` ' 'INNER JOIN `artist_info` a ON v.`vendor_id` = a.`vendor_id` ' 'INNER JOIN `releases` r ON a.`artist_id` = r.`artist_id` ' 'INNER JOIN `track` t ON r.`release_id` = t.`release_id` ' 'INNER JOIN `period` p ON MONTH(r.`ingestion_completed`) = p.`month` ' 'AND YEAR(r.`ingestion_completed`) = p.`year` ' 'WHERE v.`status` = "signed" ' 'AND r.`deletions` = "N" ' 'GROUP BY o.`owner_id`, p.`period_id` ' ) RELEASE_ID_SQL = ( 'SELECT r.`release_id` ' 'FROM `releases` r ' 'WHERE 1 ') EXCHANGE_RATE_SQL = ( 'SELECT ' 'c.`period_id`, ' 'c.`currency_from_id`, ' 'c.`currency_to_id`, ' 'c.`exchange_rate` ' 'FROM `currency_exchange_rates` c ' 'WHERE c.`period_id` > ({period_id} - 20) ') RELEASE_TRACKS_SQL = ( 'SELECT ' 't.`id`, ' 'r.`upc`, ' 't.`cd`, ' 't.`track_id` as track, ' 't.`length_minute`, ' 't.`length_seconds`, ' 't.`third_party_publisher` ' 'FROM `releases` r ' 'LEFT JOIN `track` t ON r.`upc` = t.`upc` ' 'WHERE r.`release_status` = "in_content" ' 'ORDER BY r.`upc` ' ) db_adapter = DatabaseAdapter() def cache_owner_track_counts(): """Cache owner track counts.""" adapter = get_db_adapter() last_owner_id = -1 output = '' for row in adapter.fetch_rows(OWNER_TRACKS_SQL): owner_id = row[0] period_id = row[1] total_tracks = row[2] if owner_id != last_owner_id: output += ':{};'.format(owner_id) last_owner_id = owner_id output += '{}-{}|'.format(period_id, total_tracks) cache_item('data', 'otc', output) def load_owner_contracts(): """Cache current owner contracts.""" adapter = get_db_adapter() last_row = None for row in adapter.fetch_rows(OWNER_CONTRACT_SQL): if row[0] == 0: continue if not last_row: last_row = row continue if last_row[0] == row[0]: last_row = row continue contract_data = { 'agreement_id': last_row[1], 'currency_id': last_row[3], 'representative_fee_type': last_row[4], 'percentage_orchard_share': last_row[5], 'label_contract_renewal_reduction': last_row[6] == 'Y', } cache_string_list('owner_contract', last_row[0], contract_data) last_row = row contract_data = { 'agreement_id': last_row[1], 'currency_id': last_row[3], 'representative_fee_type': last_row[4], 'percentage_orchard_share': last_row[5], 'label_contract_renewal_reduction': last_row[6] == 'Y', } cache_string_list('owner_contract', last_row[0], contract_data) def get_transaction_start_id(): """Return the lowest numeric id that can be assigned to transactions.""" last_used_id = get_cached_item('data', 'start_id') if not last_used_id: last_used_id = get_transaction_start_id_from_warehouse() cache_item('data', 'start_id', last_used_id) return int(last_used_id) def get_transaction_start_id_from_warehouse(): """Pull the highest statement_detail_id from the warehouse. Returns: str: max statement_detail_id. """ last_used_id = 0 adapter = get_warehouse_adapter() result = adapter.execute(MAX_AI_SF) for row in result: if row[0]: last_used_id = row[0] return last_used_id def check_cache_validity(): """Check that the cache is populated and fresh enough to use. Raises: Exception: Cache is not populated. Exception: Cache is expired. """ # Check if cache is populated. last_updated = get_cached_item('data', 'last_updated') if not last_updated: raise Exception('Cache empty. Please run cache warming script.') last_updated_datetime = datetime.datetime.strptime( last_updated, '%Y-%m-%d %H:%M:%S') # Exit if cache is older than 1 day. date_delta = datetime.datetime.now() - last_updated_datetime if abs(date_delta.total_seconds()) > 86400: raise Exception('Cache expired. Older than 1 day.') def buffered_cache_release_tracks(): """Hello Darkness My Old Friend.""" adapter = get_db_adapter() last_upc = -1 cache_string = '' for row in adapter.yield_rows(RELEASE_TRACKS_SQL): if row[1] != last_upc: if cache_string: cache_item('release_tracks', last_upc, cache_string) cache_string = '' last_upc = row[1] if row[1] == last_upc: cd = row[2] if row[2] else 0 track = row[3] if row[3] else 0 minutes = row[4] if row[4] else 0 seconds = row[5] if row[5] else 0 total_time = (60 * minutes) + seconds tpp = 1 if 'Y' == row[6] else 0 cache_string += '{}|{}|{}|{}|{}-'.format( cd, track, total_time, tpp, row[0] ) # UPC CD TRACK MIN SEC TPP ID CD TRACK MIN SEC TPP ID def buffered_cache_appended_fields(): """Cache appended fields using a buffered query and generator.""" adapter = get_db_adapter() counter = 0 for row in adapter.yield_rows(APPENDED_FIELD_SQL): if row[2] != 'in_content': continue appended_fields = { 'vendor_id': row[1], 'owner_id': row[3], } cache_string_list( 'appended_fields', row[0], appended_fields) counter = counter + 1 return counter def buffered_cache_owner_tracks(): """Cache lists of tracks by owner.""" adapter = get_db_adapter() owner_tracks = {} owner_track_counts = {} for row in adapter.yield_rows(OWNER_TRACKS_SQL): owner, upc, cd, track, ingested_at, valid, minutes, third_party = row if owner not in owner_tracks: owner_tracks[owner] = {} if upc not in owner_tracks[owner]: owner_tracks[owner][upc] = {} key = '{cd_index}_{track_index}'.format(cd_index=cd, track_index=track) if key not in owner_tracks[owner][upc]: owner_tracks[owner][upc][key] = { 'minutes': minutes, 'third_party_publisher': third_party} if valid: if owner not in owner_track_counts: owner_track_counts[owner] = {} if ingested_at not in owner_track_counts[owner]: owner_track_counts[owner][ingested_at] = 0 owner_track_counts[owner][ingested_at] += 1 if owner_tracks or owner_track_counts: # TODO - cache these pass def load_statements(filename): """Read in a file and cache its data.""" counter = 0 for line in read_statement_file(filename): statement = Statement(**line) statement.validate() line.pop('actual_statement_no', None) line.pop('paid', None) line.pop('statement_id', None) line.pop('year', None) line.pop('month', None) line.pop('quarter', None) line.pop('check_detail_id', None) line.pop('period_id', None) cache_string_list( 'statement', statement.get_value('statement_id'), line) counter = counter + 1 return counter def load_exchange_rates(period_id): """Cache and return exchange rates for the given period.""" adapter = get_db_adapter() rows = adapter.fetch_rows(EXCHANGE_RATE_SQL.format(period_id=period_id)) exchange_rates = {} for row in rows: key = '{from_id}_{to_id}'.format(from_id=row[1], to_id=row[2]) if row[0] not in exchange_rates: exchange_rates[row[0]] = {} exchange_rates[row[0]][key] = row[3] for period in exchange_rates.keys(): cache_key_val('exchange_rates', period, exchange_rates[period]) return exchange_rates def cache_active_label_contracts(): """Cache each active label contract.""" active_contracts = get_active_label_contracts() adapter = get_db_adapter() count = 0 for label_id in active_contracts.keys(): rows = adapter.fetch_rows(contract_queries.LABEL_CONTRACT_SQL.format( active_contracts[label_id])) for row in rows: contract = get_contract_from_row(row) cache_item('label_contract', label_id, contract.to_tsv()) count += 1 return count def get_contract_from_row(row): """Build a contract from a result row.""" data = {} if row[0] is not None: data['id'] = row[0] if row[1] is not None: data['digital_split'] = row[1] if row[2] is not None: data['advance_recoupable_percentage'] = row[2] if row[3] is not None: data['oms_deal_type'] = row[3] if row[4] is not None: data['ringtone_publishing_type'] = row[4] if row[5] is not None: data['physical_track_publishing_type'] = row[5] if row[6] is not None: data['oms_fees'] = row[6] if row[7] != 0: data['first_contract'] = False if row[8] is not None: data['payment_interval'] = row[8] if row[9] == 'Y': data['apply_fx_spread'] = True if row[10] is not None: data['currency_id'] = row[10] if row[11] is not None and row[11] != '': thresholds = row[11].split(',') if thresholds[0]: data['release_dig_threshold'] = thresholds[0] if len(thresholds) > 1 and thresholds[1]: data['release_phy_threshold'] = thresholds[1] if len(thresholds) > 2 and thresholds[2]: data['label_dig_threshold'] = thresholds[2] if len(thresholds) > 3 and thresholds[3]: data['label_phy_threshold'] = thresholds[3] if len(thresholds) > 4 and thresholds[4]: data['artist_dig_threshold'] = thresholds[4] if len(thresholds) > 5 and thresholds[5]: data['artist_phy_threshold'] = thresholds[5] if row[13] is not None: data['dms_split'] = row[13] if row[14] is not None: data['dms_master_split'] = row[14] if row[15] is not None: data['territory_split'] = row[15] if row[16] is not None: data['transaction_territory_split'] = row[16] if row[17] is not None: data['transaction_type_split'] = row[17] contract = LabelContract(**data) contract.validate() return contract def get_active_label_contracts(): """Get dict of contracts keyed by label ids.""" db_adapter = get_db_adapter() active_contracts = {} for row in db_adapter.fetch_rows( contract_queries.ACTIVE_LABEL_CONTRACTS_SQL): label_id, contract_id = row active_contracts[label_id] = contract_id return active_contracts # Mockable getters def get_db_adapter(): """Mockable getter.""" return db_adapter def get_warehouse_adapter(): """Mockable getter.""" return SnowflakeAdapter()