"""yt_video_regressions2

Revision ID: 47e959aa321a
Revises: 4f54faf809b3
Create Date: 2020-09-08 13:14:28.929888

"""

# revision identifiers, used by Alembic.
revision = '47e959aa321a'
down_revision = '4f54faf809b3'
branch_labels = None
depends_on = None

from alembic import op
import sqlalchemy as sa


def upgrade():
    op.execute('''
create or replace view vw_yt_video_regressions_1 as
with
  dailys as (
    -- get daily streams by subtracting the day's total from the previous day's.
    -- the group by is to remove duplicate days, which occasionally happens.
    select
      video_ytid,
      as_of::date,
      lag(as_of::date, 1) OVER (partition by video_ytid ORDER BY as_of::date) prev_date,
      as_of::date - current_date idate,
      as_of::date - lag(as_of::date, 1) OVER (partition by video_ytid ORDER BY as_of::date) days,
      view_count plays0,
      (view_count - lag(view_count, 1) OVER (partition by video_ytid ORDER BY as_of::date))
        / coalesce(nullif((as_of::date - lag(as_of::date, 1) OVER (partition by video_ytid ORDER BY as_of::date)), 0), 1)
        AS daily_plays
    from
      yt_video_statistics a
    where as_of::date >= current_date - 28
  ),

  regressions as (
    select
      video_ytid,
      count(*) num_days,
      min(as_of) started_at,
      max(as_of) ended_at,
      max(plays0) total_plays,
      max(daily_plays) max_daily_plays,
      min(daily_plays) min_daily_plays,
      avg(daily_plays) avg_daily_plays,
      last(daily_plays order by idate) last_plays,
      first(daily_plays order by idate) filter (where idate >= -7) plays7,
      regr_slope(daily_plays, idate) filter (where idate >= -7) slope7,
      regr_r2(daily_plays, idate) filter (where idate >= -7) rsq7,
      regr_intercept(daily_plays, idate)  filter (where idate >= -7) as yint7,
      count(idate) filter (where idate >= -7) count7,

      first(daily_plays order by idate) filter (where idate >= -14) plays14,
      regr_slope(daily_plays, idate) filter (where idate >= -14) slope14,
      regr_r2(daily_plays, idate) filter (where idate >= -14) rsq14,
      regr_intercept(daily_plays, idate)  filter (where idate >= -14) as yint14,
      count(idate) filter (where idate >= -14) count14,

      first(daily_plays order by idate) filter (where idate >= -21) plays21,
      regr_slope(daily_plays, idate)  filter (where idate >= -21) slope21,
      regr_r2(daily_plays, idate)  filter (where idate >= -21) rsq21,
      regr_intercept(daily_plays, idate)  filter (where idate >= -21) as yint21,
      count(idate) filter (where idate >= -21) count21,

      first(daily_plays order by idate) plays28,
      regr_slope(daily_plays, idate) slope28,
      regr_r2(daily_plays, idate) rsq28,
      regr_intercept(daily_plays, idate) as yint28,
      count(idate) count28,

      array_agg(idate order by idate) idates,
      array_agg(28 + idate order by idate) idates_positive,
      array_agg(daily_plays order by idate) daily_plays
    from
      dailys
    where daily_plays is not null
    group by 1
    having count(plays0) > 4
  )

select
  r.video_ytid,
  r.num_days,
  v.published::date as published,
  r.started_at,
  r.ended_at,
  r.total_plays,
  r.max_daily_plays,
  r.min_daily_plays,
  r.avg_daily_plays,
  r.last_plays,
  case when count7 >= 3 then r.plays7 end as plays7,
  case when count7 >= 3 then r.slope7 end as slope7,
  case when count7 >= 3 then r.rsq7 end as rsq7,
  case when count7 >= 3 then r.yint7 end as yint7,
  r.count7,
  case when r.count14 > r.count7 then r.plays14 end as plays14,
  case when r.count14 > r.count7 then r.slope14 end as slope14,
  case when r.count14 > r.count7 then r.rsq14 end as rsq14,
  case when r.count14 > r.count7 then r.yint14 end as yint14,
  r.count14,
  case when r.count21 > r.count14 then r.plays21 end as plays21,
  case when r.count21 > r.count14 then r.slope21 end as slope21,
  case when r.count21 > r.count14 then r.rsq21 end as rsq21,
  case when r.count21 > r.count14 then r.yint21 end as yint21,
  r.count21,
  case when r.count28 > r.count21 then r.plays28 end as plays28,
  case when r.count28 > r.count21 then r.slope28 end as slope28,
  case when r.count28 > r.count21 then r.rsq28 end as rsq28,
  case when r.count28 > r.count21 then r.yint28 end as yint28,
  r.count28,
  classify_lang_script(v.title || ' ' || v.description) as lang_code,
  v.audio_language,
  v.text_language,
  v.api_data->'topicDetails'->'topicCategories' ? 'https://en.wikipedia.org/wiki/Music' as has_music_topic,
  v.description ~ '(\yUMG\y|\yUniversal Music\y|\ySony\y|\ySME\y|\yWarner Music\y|\yWarner Records\y|\yAtlantic Records\y|by WM )' as is_major_label,
  coalesce(audio_language, text_language, 'xx') in ('en', 'en-GB', 'en-US', 'xx') as maybe_english,
  v.duration,
  r.idates as idays_ago,
  r.daily_plays
from regressions r
join yt_videos v on v.ytid = r.video_ytid
where v.api_data is not null;

    ''')

    # created 'with no data' to avoid a long running upgrade.  initial refresh handled out of band.
    op.execute('''
    create materialized view mv_yt_video_regressions as
select
  video_ytid,
  num_days,
  published,
  started_at,
  ended_at,
  total_plays,
  max_daily_plays,
  min_daily_plays,
  avg_daily_plays,
  last_plays,
  plays7,
  slope7,
  rsq7,
  yint7,
  count7,
  plays14,
  slope14,
  rsq14,
  yint14,
  count14,
  plays21,
  slope21,
  rsq21,
  yint21,
  count21,
  plays28,
  slope28,
  rsq28,
  yint28,
  count28,
  lang_code,
  audio_language,
  text_language,
  has_music_topic,
  is_major_label,
  maybe_english,
  duration,
  idays_ago,
  daily_plays
from vw_yt_video_regressions_1 
with no data;
    ''')

    op.execute('''
create unique index on mv_yt_video_regressions(video_ytid);
    ''')

def downgrade():
    op.execute('drop materialized view mv_yt_video_regressions')
    op.execute('drop view vw_yt_video_regressions_1')