from processing_accounting.flows.accounting_statement_export import setting class SqlGenerator(object): def __init__( self, period_ids, user_type, payment_interval, user_id=None, transaction_types=None): self.user_type = user_type self.user_id = user_id self.payment_interval = payment_interval self.period_ids = period_ids self.transaction_types = transaction_types if period_ids is not None: self.period_ids = ','.join(sorted(period_ids.split(','))) if transaction_types is not None: self.transaction_types = "'{}'".format( "','".join(sorted(transaction_types.split(',')))) def _get_digital_split_sql(self): return ''' (case ringtone_publishing_type when 'both' then digital_split when 'orchard' then 0 else 1 end) ''' def get_sql(self): user_sql = '' period_ids_sql = ' AND accountingPeriodId in({}) '.format( self.period_ids) transaction_types_sql = '' quarter_or_month_sql = '' user_id_type_sql = '' subaccount_join = 'LEFT' if self.user_type == 'subaccount': user_id_type_sql = ( '((x.subaccountid) || \'S\') as user_id_type') subaccount_join = 'INNER' else: user_id_type_sql = ( '((x.labelid) || \'L\') as user_id_type') if self.user_id is not None and self.user_id != 'all': user_sql = ' AND fs.labelid = {}'.format(self.user_id) if self.user_type == 'subaccount': user_sql = ' AND fs.subaccountid = {}'.format(self.user_id) if self.transaction_types is not None: transaction_types_sql = ' AND transactiontypeid in({})'.format( self.transaction_types) if self.payment_interval is not None: quarter_or_month_sql = ( ' and bs.payment_interval = \'{}\' ').format( self.payment_interval) sql = ''' SELECT COALESCE(dp2.year,'') || CAST('M' as varchar) || COALESCE(dp2.month,'') AS period, COALESCE(dp1.year,'') || cast('M' as varchar) || COALESCE(dp1.month,'') AS activity_period, ds.storeName AS dms, dc.countryName AS territory, COALESCE(dr.display_upc,'') AS orchard_upc, COALESCE(dr.manufacturer_upc,'') AS manufacturer_upc, COALESCE(dr.vendor_catalog_number,'') AS label_catalog, COALESCE(dr.product_code, '') AS product_code, COALESCE(dss.subaccountname,'') as subaccount, COALESCE(dr.imprint,'') AS imprint_label, COALESCE(da.artistName,'') AS artist_name, COALESCE(dr.releaseName,'') AS release_name, COALESCE((CASE WHEN dt.cd=0 AND dt.track_id=0 THEN 'Full Album' ELSE dt.trackName END),'') AS track_name, COALESCE( (CASE WHEN dt.cd=0 AND dt.track_id=0 THEN cast(dr.releaseId as varchar) ELSE dt.isrc END),'') AS isrc, COALESCE(dt.cd,0) AS volume, COALESCE(dt.track_id,0) AS track_number, (CASE WHEN x.sales IS NULL THEN 0 ELSE x.sales END) AS quantity, ROUND(x.unitPrice::DECIMAL(38, 19), 6) AS unit_price, (COALESCE(x.gross,0) ) AS gross, dtt.transactionTypeAbbr as trans_type, COALESCE(dtt.transactiontypedesc, '') AS transactiontypedesc, COALESCE(x.adjusted_gross,0) AS adjusted_gross, ROUND(x.splitRate::DECIMAL(38, 19), 6) as split_rate, COALESCE(x.net_receipt,0) AS label_share_net_receipts, (COALESCE(x.ringtone_publishing,0) * {digital_split_sql}) AS ringtone_publishing, (COALESCE(x.cloud_publishing,0) * {digital_split_sql}) AS cloud_publishing, (COALESCE(x.dpd_publishing,0) * {digital_split_sql}) AS publishing, (COALESCE(x.oms_fees,0) ) AS mech_administrative_fee, CASE WHEN dss.commissionOverride IS NULL THEN '' ELSE (COALESCE( x.net_receipt,0) * dss.commissionOverride)::VARCHAR END AS subaccount_label_share_net_receipts, arc.iso_4217_code AS preferred_currency, x.statement_detail_id, dl.isdistributor, COALESCE(dr.physical_product_type, '') AS physical_product_type, COALESCE(dr.physical_product_format, '') AS physical_product_format, COALESCE(dr.display_configuration, '') AS display_configuration, COALESCE(dtaa.track_artists, '') AS track_artists, x.original_price AS original_price, x.discount AS discount, {user_id_type_sql} FROM ( SELECT fs.labelid, fs.trackId, fs.releaseId, fs.artistId, fs.transactionTypeId, fs.activityPeriodId, fs.accountingPeriodId, fs.genreId, fs.countryId, fs.storeId, fs.statement_detail_id, fs.subaccountId, fs.fx_adjusted_gross AS adjusted_gross, fs.fx_gross AS gross , fs.fx_net_receipt AS net_receipt, fs.fx_ringtone_publishing AS ringtone_publishing, fs.fx_cloud_publishing as cloud_publishing, fs.fx_dpd_publishing AS dpd_publishing, fs.fx_oms_fees AS oms_fees, fs.sales AS sales, CASE WHEN fs.sales<>0 THEN (fs.fx_gross / fs.sales) ELSE 0 END AS unitPrice, CASE WHEN fs.fx_adjusted_gross <> 0 THEN (fs.fx_net_receipt::FLOAT / fs.fx_adjusted_gross::FLOAT)::DECIMAL(38, 19) ELSE 0 END AS splitRate, fs.original_price, fs.discount FROM fact_sales fs WHERE 1 {user_sql} {period_ids_sql} {transaction_types_sql} ) X INNER JOIN booked_vendor_contract_snapshot bs ON bs.vendor_id = x.labelid and bs.period_id = X.accountingPeriodId INNER JOIN accounting_report_currencies arc ON arc.id = bs.currency_id INNER JOIN dim_label dl on dl.labelid = x.labelid INNER JOIN dim_track dt ON dt.trackId=x.trackId LEFT JOIN {track_artists_table} dtaa ON dt.track_unique_id=dtaa.track_id INNER JOIN dim_release dr ON dr.releaseId=x.releaseId {subaccount_join} JOIN dim_subaccount dss ON dss.subaccountId = dr.subaccountId INNER JOIN dim_artist da ON da.artistId=x.artistId INNER JOIN dim_store ds ON ds.storeId = x.storeId INNER JOIN dim_country dc ON dc.countryId = x.countryId INNER JOIN dim_transactiontype dtt ON dtt.transactionTypeId=x.transactionTypeId INNER JOIN dim_period dp1 ON dp1.periodId=x.activityPeriodId INNER JOIN dim_period dp2 ON dp2.periodId=x.accountingPeriodId WHERE 1 {quarter_or_month_sql} ''' sql = sql.format( digital_split_sql=self._get_digital_split_sql(), period_ids_sql=period_ids_sql, user_sql=user_sql, transaction_types_sql=transaction_types_sql, quarter_or_month_sql=quarter_or_month_sql, user_id_type_sql=user_id_type_sql, subaccount_join=subaccount_join, track_artists_table=setting.TRACK_ARTISTS_AGGREGATION_TABLE_NAME) return sql def get_sql_for_bulk_status_update(self): params = '{}_{}_all_all_{}/user_id_type='.format( self.period_ids, self.user_type, self.payment_interval) s3_path = setting.config.get('schematized_file_path').format( params=params) user_id_sql = '' get_timestamp_func = 'current_timestamp()' if self.user_type == "subaccount": if self.user_id is not None and self.user_id != 'all': user_id_sql = ' AND fs.subaccountid = {} '.format(self.user_id) sql = ("select {user_id} || 'S' as user_id_type, " "bvcs.payment_interval, " "'{user_params}' as user_params, " "'AVRO' as file_type, " "'{s3_path}' || {user_id} || 'S' " "as s3_path," "'en_US' as number_format, " "'all' as transaction_types, '' as file_size, " "'' as row_count, '' as workflow_id, " "'GENERATING' as status, {get_timestamp_func} as " "generation_start, " "'' as generation_end " "from booked_vendor_contract_snapshot bvcs " "inner join fact_sales fs " "on fs.labelid = bvcs.vendor_id " "and fs.accountingperiodid = bvcs.period_id " "inner join dim_period dp " "on dp.periodid = bvcs.period_id " "where payment_interval = '{payment_interval}' " "and bvcs.period_id in ({period_ids}) {user_id_sql} " "{{group_by}}").format( user_id='(fs.subaccountid)', period_ids=self.period_ids, s3_path=s3_path, payment_interval=self.payment_interval, user_id_sql=user_id_sql, user_params='__'.join( [self.period_ids, 'all', 'AVRO', 'en_US']), get_timestamp_func=get_timestamp_func) if self.payment_interval == "quarter": sql = sql.format(group_by=( " group by fs.subaccountid, bvcs.payment_interval, " "dp.quarter, dp.year;")) else: sql = sql.format(group_by=( " group by fs.subaccountid, bvcs.payment_interval, " "bvcs.period_id;")) else: if self.user_id is not None and self.user_id != 'all': user_id_sql = ' AND fs.labelid = {} '.format(self.user_id) sql = ("select {user_id} || 'L' as user_id_type, " "bvcs.payment_interval, " "'{user_params}' as user_params, " "'AVRO' as file_type, " "'{s3_path}' || {user_id} || 'L' " "as s3_path," "'en_US' as number_format, " "'all' as transaction_types, '' as file_size, " "'' as row_count, '' as workflow_id, " "'GENERATING' as status, {get_timestamp_func} as " "generation_start, " "'' as generation_end " "from booked_vendor_contract_snapshot bvcs " "inner join fact_sales fs " "on fs.labelid = bvcs.vendor_id " "and fs.accountingperiodid = bvcs.period_id " "inner join dim_period dp " "on dp.periodid = bvcs.period_id " "where payment_interval = '{payment_interval}' " "and bvcs.period_id in({period_ids}) {user_id_sql} " "{{group_by}}").format( user_id='(fs.labelid)', period_ids=self.period_ids, s3_path=s3_path, payment_interval=self.payment_interval, user_id_sql=user_id_sql, user_params='__'.join( [self.period_ids, 'all', 'AVRO', 'en_US']), get_timestamp_func=get_timestamp_func) if self.payment_interval == "quarter": sql = sql.format(group_by=( " group by fs.labelid, bvcs.payment_interval, " "dp.quarter, dp.year;")) else: sql = sql.format(group_by=( " group by fs.labelid, bvcs.payment_interval, " "bvcs.period_id;")) return sql @staticmethod def get_quarter_year_period_ids(period_id): """Get period ids of a quarter of a year in which passed period id resides Args: period_id (int): accounting period id Return: str: sql for getting the period ids of a quarter of a year in which the parameter period id resides """ sql = ( 'select x.period_ids, dp.periodid from (' 'select year, quarter, min(periodid) as min_periodid, ' 'max(periodid) as max_periodid, ' 'min(periodid)||\',\'' '|| (min(periodid)+1)||\',\'' '||max(periodid) as period_ids ' 'from dim_period ' 'group by year, quarter) x ' 'inner join dim_period dp on dp.periodid ' 'between x.min_periodid and x.max_periodid ' 'where dp.periodid = {period_id};').format( period_id=period_id) return sql