### snowflake cursor doesn't support string sseparation, thus this var is adjusted

### Step 1. Check query in Postgres
pg_slz_status_check_1 = """WITH reports AS (
SELECT * FROM report r JOIN data_source ds ON ds.data_source_id = r.data_source_id 
WHERE ds.data_source_name = 'linkfire' AND r.report_name IN ('linkfire_raw_data'))
select distinct
uow.report_date "REPORT_DATE",
string_agg(distinct concat(r.report_name||' ('||l.licensor_name||')'), ',') "MISSING_REPORTS"
from content_status 
join unit_of_work uow on uow.unit_of_work_id = content_status.unit_of_work_id
join report r on r.report_id = uow.report_id
join licensor l on l.licensor_id = uow.licensor_id
where r.report_id IN (SELECT report_id FROM reports)
and uow.completeness_status != 'CANCELLED'
and content_status.content_status not in ('COMPLETE', 'ON_HOLD', 'CANCELLED')
and content_status.content_status in ('MISSING', 'FAILED')
and uow.report_date > current_date-14
group by uow.report_date
order by uow.report_date desc"""

### Step 2. RAW_SCHEMA: verify that no ingestions failed during the last two weeks
sf_slz_exp_check_2 = """with cte_ae_reports as ( \
select distinct REPORT_NAME \
from DELPHI_EXPLORATION.SYS.DSP_REPORTS \
where DSP = 'LINKFIREREPORTING'), \
cte_two_weeks_dates as ( \
select dateadd(day, -1*row_number() over (order by null), current_date()-1) as REPORT_DATE \
from table(GENERATOR(ROWCOUNT=>14))), \
cte_licensors as (select 'sme' as LICENSOR), \
cte_full_scope as ( select REPORT_NAME, LICENSOR, REPORT_DATE \
from cte_ae_reports \
cross join cte_two_weeks_dates \
cross join cte_licensors), \
cte_chfs as ( select REPORT_NAME, REPORT_DATE, LICENSOR, \
sum(case when STATUS = 'LOADED' then 1 else 0 end) as LOADED_COUNT, \
sum(case when STATUS != 'LOADED' then 1 else 0 end) as NOT_LOADED_COUNT \
from DELPHI_EXPLORATION.SYS.COPY_HISTORY_FILE_STATUS \
where DSP = 'LINKFIREREPORTING' \
and REPORT_DATE in (select REPORT_DATE from cte_two_weeks_dates) \
group by REPORT_NAME, REPORT_DATE, LICENSOR ) \
select cfs.REPORT_DATE, \
count(*) "MISSING_REPORTS_COUNT", \
listagg(concat(cfs.REPORT_NAME, ' (', cfs.LICENSOR, ')'), ', ') "MISSING_REPORTS" \
from cte_full_scope cfs \
left join cte_chfs chfs on cfs.REPORT_NAME = chfs.REPORT_NAME and cfs.LICENSOR = chfs.LICENSOR and cfs.REPORT_DATE = chfs.REPORT_DATE \
where chfs.LOADED_COUNT is null or chfs.NOT_LOADED_COUNT > 0 \
group by cfs.report_date \
order by cfs.REPORT_DATE desc"""

sf_exp_lf_success_check_3 = """select count(*) "count", 'Completed' "Operations" \
from "DELPHI_LINKFIRE"."SYS"."LINKFIRE_IMPORT_LOG"  \
where datediff(day, IMPORT_REPORTS_TIMESTAMP , current_date()) <= 1 \
and message = 'data import completed';"""

sf_exp_lf_failed_check_4 = """select count(*) "count", 'Incompleted' "Operations" \
from "DELPHI_LINKFIRE"."SYS"."LINKFIRE_IMPORT_LOG" \
where datediff(day, IMPORT_REPORTS_TIMESTAMP , current_date()) <= 1 \
and message != 'data import completed';"""

sf_dts_dc_check_5 = """select count(*) "count", 'src'  \
from "DELPHI_EXPLORATION"."APPS_ETL"."V_LINKFIRE_RAW_DATA"   \
where LINKFIRE_LINK_ID is not null \
and event_type in (select EVENT_TYPE_NAME from DELPHI_LINKFIRE.LINKFIRE.DIM_EVENT_TYPE) \
union \
select sum(EVENT_COUNT) "count", 'dest' \
from "DELPHI_LINKFIRE"."LINKFIRE"."FACT_EVENT_FUNNEL_DAILY";"""

sf_dts_dc_check_6 = """select count(*) "count", 'src' \
from ( \
    select distinct lower(COUNTRY_CODE) \
    from "DELPHI_EXPLORATION"."APPS_ETL"."V_LINKFIRE_RAW_DATA" \
    where LINKFIRE_LINK_ID is not null \
    and event_type is not null \
    and COUNTRY_CODE is not null \
    union select 'gs' \
    union select 'hm' \
    union select 'tf' \
    union select 'pn' \
    union select 'unknown' \
) \
union \
select count(*) "count", 'dest'  \
from "DELPHI_LINKFIRE"."LINKFIRE"."DIM_TERRITORY";"""


sf_dts_dc_check_7 = """with AMBIGUOUS_ORG_ID as ( \
    select LINKFIRE_LINK_ID, REPORT_DATE \
    from "DELPHI_EXPLORATION"."APPS_ETL"."V_LINKFIRE_RAW_DATA" \
    where LINKFIRE_LINK_ID is not null  \
    group by LINKFIRE_LINK_ID, REPORT_DATE \
    having count(distinct ORG_ID, BOARD_ID) > 1 \
) \
   , distinct_data as (select distinct LINKFIRE_LINK_ID, REPORT_DATE \
                       from "DELPHI_EXPLORATION"."APPS_ETL"."V_LINKFIRE_RAW_DATA") \
   , AMBIGUOUS_LINK_ID as ( \
    select * \
    from AMBIGUOUS_ORG_ID v \
    where not exists( \
            select 1 from distinct_data where v.LINKFIRE_LINK_ID = LINKFIRE_LINK_ID and v.REPORT_DATE != REPORT_DATE) \
) \
select count(distinct LINKFIRE_LINK_ID) "count", 'src' \
from "DELPHI_EXPLORATION"."APPS_ETL"."V_LINKFIRE_RAW_DATA" s \
where LINKFIRE_LINK_ID is not null  \
  and LINKFIRE_LINK_ID not in (select LINKFIRE_LINK_ID from AMBIGUOUS_LINK_ID) \
union \
select count(*) "count", 'dest' \
from "DELPHI_LINKFIRE"."LINKFIRE"."DIM_LINKFIRE_LINK";"""

sf_lf_s3_success_check_10 = """select count(*) "count", 'Completed' "Operations" \
from "DELPHI_LINKFIRE"."SYS"."LINKFIRE_EXPORT_LOG" \
where datediff(day, EXPORT_TIMESTAMP , current_date()) <= 1 \
and message = 'data export completed';"""

sf_lf_s3_failed_check_11 = """select count(*) "count", 'Incompleted' "Operations" \
from "DELPHI_LINKFIRE"."SYS"."LINKFIRE_EXPORT_LOG" \
where datediff(day, EXPORT_TIMESTAMP , current_date()) <= 1 \
and message != 'data export completed';"""

sf_dts_dc_check_13 = """select count(*) "count" \
from "DELPHI_LINKFIRE"."LINKFIRE"."DIM_LINKFIRE_LINK" \
where LINKFIRE_LINK_URL is not null  \
and convert_timezone('UTC', created_at) < ( \
select max(EXPORT_TIMESTAMP_UTC)  \
from "DELPHI_LINKFIRE"."SYS"."LINKFIRE_EXPORT_LOG" \
);"""

pg_dts_dc_check_13 = """SELECT count(*) "count"
FROM linkfire.dim_link;"""

sf_dts_dc_check_14 = """SELECT count(*) "count" \
FROM "DELPHI_LINKFIRE"."LINKFIRE"."FACT_EVENT_FUNNEL" \
WHERE LINKFIRE_LINK_ID in ( \
select LINKFIRE_LINK_ID  \
from "DELPHI_LINKFIRE"."LINKFIRE"."DIM_LINKFIRE_LINK" \
where LINKFIRE_LINK_URL is not  null \
) \
AND convert_timezone('UTC', created_at) < ( \
select max(EXPORT_TIMESTAMP_UTC)  \
from "DELPHI_LINKFIRE"."SYS"."LINKFIRE_EXPORT_LOG" \
) \
AND TERRITORY_NAME in ( \
select TERRITORY_NAME  \
from "DELPHI_LINKFIRE"."LINKFIRE"."DIM_TERRITORY" \
where IS_VALID \
);"""

pg_dts_dc_check_14 = """SELECT sum(event_count) "count"
FROM linkfire.fact_event_funnel_daily;"""

sf_dts_dc_check_15 = """select count(*) "count" \
from  "DELPHI_LINKFIRE"."LINKFIRE"."DIM_REFERRER" \
where  convert_timezone('UTC', created_at) < ( \
select max(EXPORT_TIMESTAMP_UTC)  \
from "DELPHI_LINKFIRE"."SYS"."LINKFIRE_EXPORT_LOG" \
);"""

pg_dts_dc_check_15 = """SELECT count(*) "count"
FROM linkfire.dim_referrer;"""

sf_dts_dc_check_16 = """SELECT count(*) "count" \
FROM "DELPHI_LINKFIRE"."LINKFIRE"."DIM_CAMPAIGN_LINK" dcl \
JOIN ( \
select CAMPAIGN_ID, ACCOUNT_ID, max(UPDATED_AT) as campaign_updated_at \
from "DELPHI_ADS_DATA"."SYS"."FACEBOOK_ADS_CAMPAIGN" \
group by CAMPAIGN_ID, ACCOUNT_ID \
union \
select CAMPAIGN_ID, CUSTOMER_ID as ACCOUNT_ID, max(UPDATED_AT) as campaign_updated_at \
from "DELPHI_ADS_DATA"."SYS"."GOOGLE_ADS_CAMPAIGN" \
group by CAMPAIGN_ID, ACCOUNT_ID \
) campaign on dcl.CAMPAIGN_ID = campaign.CAMPAIGN_ID \
JOIN ( \
select ACCOUNT_ID, UPDATED_AT as account_updated_at \
from "DELPHI_ADS_DATA"."SYS"."GOOGLE_ADS_ACCOUNT" \
union \
select ACCOUNT_ID,UPDATED_AT as account_updated_at \
from "DELPHI_ADS_DATA"."SYS"."FACEBOOK_ADS_ACCOUNT" \
) account on campaign.ACCOUNT_ID = account.ACCOUNT_ID \
where UPDATED_AT <= (select  max(EXPORT_TIMESTAMP_UTC) from DELPHI_LINKFIRE.SYS.LINKFIRE_EXPORT_LOG);"""

pg_dts_dc_check_16 = """SELECT count(*) + (select count(*) from staging.linkfire_dim_campaign_link where updated_at >  current_date -1 ) "count"
FROM linkfire.dim_link_campaign;"""

if __name__ == '__main__':
    pass