{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data Dumping Into GSheet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "sys.path.insert(0, '/Users/joel/src/thundr/tracker')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from importlib import reload"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "import pandas as pd\n",
    "import numpy\n",
    "import matplotlib\n",
    "from io import StringIO\n",
    "from datetime import datetime, date, timedelta\n",
    "\n",
    "import random\n",
    "import traceback\n",
    "\n",
    "from tracker.unicorn import mysql, models, artist_ingestion, uniconfig, daily_streams, spartus_daily\n",
    "from tracker.utils import chunks\n",
    "from tracker import sheet as gsheet\n",
    "from tracker import db, spotify"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "sfa = spartus_daily.SpotifyForArtists(\n",
    "    artist_id='3GxKJzJK4LpsYGXQrw77wz',\n",
    "    track_id='6k9s52t7GyxRA4mNDERlwH',\n",
    "    sp_dc='AQDKn9i20otvtv0eKuwUTvShedeuH_4bCWYQVCY--vKhEbj_w0JhDIurSrYuziSj212zf51Ij0oxGGd3XtbGRpp6vSJR-mSu1pCeVpsbRJI',\n",
    "    sp_key='928f9483-e122-4610-92c5-22e0460b5f21',\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Bearer BQDDDi73C2Dces7K0u-n1X-uEZ0Enhg86zJ5LbKcehrlcVtYYT2eWlkMFJJtsySB3P_mTLpw0TG-1wjGhYNZ1Yq1MleU9EVbrU71VCkcyF6vjlfN-QG_D1LG8IEw5mlRKdM9liMehksl5gaTPO5DlRvFawv_U4-C\n"
     ]
    }
   ],
   "source": [
    "s = sfa.refresh_request_headers()\n",
    "print(s['authorization'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['billieeilish', 'Billie Eilish', 'spotify:artist:6qqNVTkY8uBg9cP3Jd7DAH'],\n",
       " ['rinasonline', 'Rina Sawayama', 'spotify:artist:2KEqzdPS7M5YwGmiuPTdr5'],\n",
       " ['edgar_the_breathtaker',\n",
       "  'King Krule',\n",
       "  'spotify:artist:4wyNyxs74Ux8UIDopNjIai'],\n",
       " ['celeste', 'Celeste', 'spotify:artist:49HlOY4gkHqsYG9GCuhkcc'],\n",
       " ['auroramusic', 'Aurora', 'spotify:artist:1WgXqy2Dd70QQOU7Ay074N']]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "settings = [\n",
    "    x.strip().split(',') for x in \"\"\"\n",
    "billieeilish,Billie Eilish,spotify:artist:6qqNVTkY8uBg9cP3Jd7DAH\n",
    "rinasonline,Rina Sawayama,spotify:artist:2KEqzdPS7M5YwGmiuPTdr5\n",
    "edgar_the_breathtaker,King Krule,spotify:artist:4wyNyxs74Ux8UIDopNjIai\n",
    "celeste,Celeste,spotify:artist:49HlOY4gkHqsYG9GCuhkcc\n",
    "auroramusic,Aurora,spotify:artist:1WgXqy2Dd70QQOU7Ay074N\"\"\".splitlines() if x\n",
    "]\n",
    "settings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "def fetch_daily_streams_matrix(spyids_or_uris):\n",
    "    spyids = [spotify._uri_to_id(s) for s in spyids_or_uris]\n",
    "    fmt = \",\".join([\"%s\"] * len(spyids))\n",
    "    rows = mysql.query(\n",
    "        f'select spyid, tsdata from spotify_daily_streams where spyid in ({fmt})', \n",
    "        spyids\n",
    "    )\n",
    "\n",
    "    expanded = []\n",
    "    for spyid, tsdata in rows:\n",
    "        for point in tsdata.splitlines():\n",
    "            if 'num' not in point:\n",
    "                t, v = point.split(',')\n",
    "                expanded.append([spyid, int(t), int(v)])\n",
    "    return expanded"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['6qqNVTkY8uBg9cP3Jd7DAH',\n",
       " '2KEqzdPS7M5YwGmiuPTdr5',\n",
       " '4wyNyxs74Ux8UIDopNjIai',\n",
       " '49HlOY4gkHqsYG9GCuhkcc',\n",
       " '1WgXqy2Dd70QQOU7Ay074N']"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "artist_spyids = [spotify._uri_to_id(r[2]) for r in settings]\n",
    "artist_spyids"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "m1 = fetch_daily_streams_matrix([r[2] for r in settings])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "def read_all_csvs(filenames):\n",
    "    df = pd.read_csv(filenames[0])\n",
    "    df['fname'] = filenames[0]\n",
    "    for f in filenames[1:]:\n",
    "        dfn = pd.read_csv(f)\n",
    "        dfn['fname'] = f\n",
    "        df = pd.concat([df, dfn], ignore_index=True)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "weekly_followers = pd.read_csv('/Users/joel/src/datadumps/crowdtangle_curious5_1.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [],
   "source": [
    "weekly_followers.pivot(index='Beginning Of Interval', columns='User Name', values='Followers').to_csv('/tmp/curisou5insta.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Engine(postgres://bot:***@whitelist2.cbn1zk7uet6r.eu-west-1.rds.amazonaws.com:5432/wl_us_ultra?sslmode=require)\n"
     ]
    }
   ],
   "source": [
    "db.setup_session()\n",
    "print(db.Session.get_bind())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [],
   "source": [
    "spotify_activity_sql = '''\n",
    "select\n",
    "  sto.spyid as track_id,\n",
    "  primary_artist_spyid as artist_id,\n",
    "  sto.release_date as event_date,\n",
    "  'Released ' || st.name as event_description\n",
    "from spy_track_olap sto\n",
    "join spy_tracks st on sto.spyid = st.spyid\n",
    "where sto.artist_spyid = any(:spyids)\n",
    "union all\n",
    "select\n",
    "  st.spyid,\n",
    "  st.primary_artist_spyid,\n",
    "  spt.first_seen,\n",
    "  st.name || ' added to ' || sp.name\n",
    "from spy_tracks st\n",
    "join spy_artists sa on st.primary_artist_spyid = sa.spyid\n",
    "join spy_playlist_track spt on spt.track_spyid = st.spyid\n",
    "join spy_playlist sp on spt.playlist_spyid = sp.spyid\n",
    "where st.primary_artist_spyid = any(:spyids)\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<sqlalchemy.engine.result.ResultProxy at 0x11b39d320>"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "db.Session.execute(spotify_activity_sql, params=dict(spyids=artist_spyids))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "eveything = _.fetchall()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "eventDf = pd.DataFrame(eveything, columns=['track_id', 'artist_id', 'Time', 'Event Description'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "name_map = {spotify._uri_to_id(r[2]) : r[1] for r in settings}\n",
    "eventDf['Artist'] = eventDf.artist_id.apply(name_map.get)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [],
   "source": [
    "eventDf[['Artist', 'Time', 'Event Description']].to_csv('/Users/joel/Desktop/curious5events.tsv', sep='\\t', index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [],
   "source": [
    "eventDf.to_csv?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def fetch_artists(artist_ids) -> typing.List[models.Artist]:\n",
    "    \n",
    "    spy = spotify._authenticate()\n",
    "    utcnow = datetime.utcnow()\n",
    "    all_results = []\n",
    "    for chunk in chunks(artist_ids, 20):\n",
    "        res = spy.artists([models.extract_id(i) for i in chunk])\n",
    "        this_result = [\n",
    "            models.Artist(\n",
    "                spyid=a['id'],\n",
    "                name=a['name'],\n",
    "                first_seen=utcnow,\n",
    "                last_seen=utcnow,\n",
    "                popularity=a['popularity'],\n",
    "                followers=(a['followers'] or {}).get('total'),\n",
    "            )\n",
    "            for a in res['artists']\n",
    "        ]\n",
    "        all_results.extend(this_result)\n",
    "    return all_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "parsed1 = fetch_artists([\n",
    "    'spotify:artist:7fMr2Jp0r6RxfJhp46kW9C',\n",
    "    'https://open.spotify.com/artist/0TWkbr1C1Wv88FwmZc8Il3?si=YoBHFzBDSYe7rJiYYuLyZQ'\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "parsed1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mysql.std_upsert(parsed1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def medians(df, window):\n",
    "    if (df.size < int(window*0.75)):\n",
    "        # not enough data\n",
    "        return [None, None, None, None]\n",
    "    \n",
    "    ma, mb = df.iloc[-window:].resample(f'{window // 2}D').median().num.to_numpy().tolist()\n",
    "    return [ma, mb, mb - ma, (mb - ma) / ma if ma > 0 else None]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_start(df):\n",
    "    first_non_zero = (df.sort_index().num > 0).idxmax()\n",
    "    return df.sort_index().loc[first_non_zero:].copy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def to_csv_str(df):\n",
    "    sio = StringIO()\n",
    "    packed = df.copy()\n",
    "    packed.index = (packed.index.to_series().dt.date - date(2015,1,1)).dt.days\n",
    "    packed.to_csv(sio)\n",
    "    return sio.getvalue()\n",
    "\n",
    "def request_data(artist_id, spartus_daily):\n",
    "    url = make_url(artist_id, spartus_daily)\n",
    "    r1 = requests.get(url, headers=spartus_daily.headers)\n",
    "    if r1.status_code != 200:\n",
    "        raise RuntimeError(f\"Non 200 request {url}, {r1}, {r1.reason}\")\n",
    "        \n",
    "    df1 = pd.DataFrame(data=r1.json()['timelinePoint'])\n",
    "\n",
    "    df1['num'] = df1.num.astype(int)\n",
    "    df1['date'] = df1.date.astype(numpy.datetime64)\n",
    "    \n",
    "    return r1, df1.set_index('date')\n",
    "\n",
    "def store_in_db(spyid, df):\n",
    "    smaller_df = get_start(df)\n",
    "    args = [spyid, smaller_df.index.min().date(), smaller_df.index.max().date(),]\n",
    "    args.extend([int(smaller_df.num.iloc[i]) for i in [-1, -2, -8, -15, -29]])\n",
    "    args.extend([int(smaller_df.num.max()), smaller_df.num.idxmax().date()])\n",
    "    args.extend(medians(smaller_df, 14))\n",
    "    args.extend(medians(smaller_df, 28))\n",
    "    args.extend(medians(smaller_df, 56))\n",
    "    args.extend(medians(smaller_df, 112))\n",
    "    args.extend([to_csv_str(smaller_df)])   \n",
    "\n",
    "    pcts = \",\".join(['%s']*len(args))\n",
    "    \n",
    "    sql = f'''replace into spotify_daily_streams (\n",
    "        spyid, as_of, first_date, \n",
    "        s0, s1, s7, s14, s28, \n",
    "        smax, smax_date, \n",
    "        m14a, m14b, d14, p14, \n",
    "        m28a, m28b, d28, p28, \n",
    "        m56a, m56b, d56, p56, \n",
    "        m112a, m112b, d112, p112, \n",
    "        tsdata\n",
    "        ) values ({pcts})'''\n",
    "    \n",
    "    with get_cursor(True) as cur:\n",
    "        cur.execute(sql, args)\n",
    "    \n",
    "\n",
    "def make_df_from_row(spyid, tsdata):\n",
    "    nf = pd.read_csv(StringIO(tsdata))\n",
    "    nf['spyid'] = spyid\n",
    "    return nf\n",
    "    \n",
    "\n",
    "def load_streams(spyids:list):\n",
    "    with get_cursor(False) as cur:\n",
    "        pcts = \",\".join(['%s'] * len(spyids))\n",
    "        cur.execute(f'select spyid, tsdata from spotify_daily_streams where spyid in ({pcts})', spyids)\n",
    "        res = cur.fetchall()\n",
    "        dfall = pd.concat([make_df_from_row(spyid, tsdata) for spyid, tsdata in res])\n",
    "        dfall['date'] = dfall.date.apply(lambda x: date(2015,1,1) + timedelta(days=x)).astype(numpy.datetime64)\n",
    "        return dfall.pivot(index='date', columns='spyid', values='num')\n",
    "\n",
    "def fetch_and_store(artist_id, spartus_daily):\n",
    "    r1, df1 = request_data(artist_id, spartus_daily=spartus_daily)\n",
    "    store_in_db(artist_id, df1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_artists_fr = pd.concat([\n",
    "    pd.read_csv('/Users/joel/Desktop/all_40_60.tsv', sep='\\t').set_index('artist_spyid'),\n",
    "    pd.read_csv('/Users/joel/Desktop/all_61_70.tsv', sep='\\t').set_index('artist_spyid'),\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_artists_fr.index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with get_cursor() as cur:\n",
    "    cur.execute('''select spyid from spotify_daily_streams''')\n",
    "    all_stored = [r[0] for r in cur.fetchall()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "left_to_do = all_artists_fr[~all_artists_fr.index.isin(all_stored)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "excluded_countries = [\n",
    "   'MX', 'BR', 'CL', 'TR', 'SG', 'JP', 'ID', 'TH', 'TW', 'AR', 'PE', 'PH',\n",
    "   'IN', 'CO', 'MY', 'MA', 'UY', 'GT', 'DO', 'CR', 'VN', 'PA', 'PY', 'SA',\n",
    "   'EC', 'EG', 'TN', 'HK', 'DZ', 'BO', 'JO', 'MT'\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "included_countries = left_to_do[~left_to_do.city1_country.isin(excluded_countries)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(all_artists_fr), len(left_to_do), len(included_countries)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "isomap = requests.get(\n",
    "    ('https://gist.githubusercontent.com/ssskip/5a94bfcd2835bf1dea52/raw/'\n",
    "     'aeed5b0cb3a7eda19e614915c3d88ce113e4a914/ISO3166-1.alpha2.json')\n",
    ").json()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def store_batch(spyid_list):\n",
    "    print(f\"Starting, {len(spyid_list)} items\")\n",
    "    errors = 0\n",
    "    for i, artist_id in enumerate(spyid_list):\n",
    "        if (i % 30 == 0):\n",
    "            print(\"refreshing headers...\")\n",
    "            cg.refresh_request_headers()\n",
    "        print(i, datetime.today().time().strftime('%H:%M:%S'), artist_id)\n",
    "        try:\n",
    "            fetch_and_store(artist_id, cg)\n",
    "            errors = 0\n",
    "        except Exception as e:\n",
    "            errors += 1\n",
    "            print(f\"Failed to get and store streams for {artist_id}, {e}\")\n",
    "            traceback.print_exc()\n",
    "\n",
    "            if errors > 5:\n",
    "                print(\"Too many errors\")\n",
    "                break\n",
    "        import time; time.sleep(3 + random.randint(0,4))\n",
    "        #import time; time.sleep(30 + random.randint(10,40))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def add_slope(dframe, days_back, roll=7):\n",
    "    col = dframe.columns[0]\n",
    "    rn = dframe.rolling(roll).mean().iloc[-days_back:]\n",
    "    int_x = rn.index.astype(int)\n",
    "    m_b = numpy.polyfit(y=rn[col], x=int_x, deg=1)    \n",
    "    fit_func = numpy.poly1d(m_b)\n",
    "    return int_x, fit_func(int_x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spyids = ['1TqTJTMW8kkMX41lkYUmbA']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "store_batch(spyids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "load_streams(spyids).iloc[-365:].plot(title='8D')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sql_to_dataframe(sql_statement, columns=None, sql_args=[], index_column=None):\n",
    "    with mysql.get_cursor(False) as cur:\n",
    "        cur.execute(sql_statement, sql_args)\n",
    "        rows = cur.fetchall()\n",
    "        columns = [c[0] for c in cur.description] if columns is None else columns\n",
    "        df = pd.DataFrame(data=rows, columns=columns)\n",
    "        if index_column:\n",
    "            df = df.set_index(index_column)\n",
    "        return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tracker.unicorn import daily_streams"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Copy listeners from one db to other"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "from tracker import db\n",
    "db.setup_session()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "result_proxy = db.Session.execute('''\n",
    "    select\n",
    "      sal.*\n",
    "    from spy_artist_listeners sal \n",
    "    where sal.as_of > current_date - 3\n",
    "      and artist_spyid in (\n",
    "      select\n",
    "        distinct artist_spyid\n",
    "      from spy_track_olap\n",
    "      where artist_popularity < 85\n",
    "        and not is_signed\n",
    "        and release_date > current_date - 90\n",
    "        and country_codes && array['US']\n",
    "      )\n",
    "    order by as_of desc\n",
    "    ''')\n",
    "colnames = [a.name for a in result_proxy.cursor.description]\n",
    "rows = result_proxy.fetchall()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def insert_to_mysql(pg_rows, colnames):\n",
    "    renamer = {'artist_spyid': 'spyid', 'as_of': 'last_updated'}\n",
    "    final_names = [renamer.get(c, c) for c in colnames]\n",
    "    seen_artists = set()\n",
    "    id_col_ix = final_names.index('spyid')\n",
    "    for i, chunk in enumerate(chunks(pg_rows, 500)):\n",
    "        new_rows = [i for i in chunk if i[id_col_ix] not in seen_artists]\n",
    "        print(f'{i} chunk, {len(new_rows)} (total {len(pg_rows)})')\n",
    "        mysql.upsert(\n",
    "            tablename='spotify_artist_listeners',\n",
    "            datarows=new_rows,\n",
    "            column_names=final_names,\n",
    "            immut_columns=('spyid', 'first_seen')\n",
    "        )\n",
    "        seen_artists.update([r[id_col_ix] for r in new_rows])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "insert_to_mysql(rows, colnames)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# tracks for the artists (will rerun the signed filter on import.)\n",
    "artist_spyids = [r[0] for r in rows]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "r = db.Session.execute('select spyid from spy_track_olap where not is_signed and release_date > current_date - 90 and artist_spyid = any(:arts)', params=dict(arts=artist_spyids))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "db.Session.rollback()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "track_spyids = [a[0] for a in r.fetchall()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "inserted_tracks = mysql.query('select spyid, album_id from spotify_track where first_seen > date_sub(now(), interval 1 hour)')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(inserted_tracks)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spy = spotify._authenticate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s = spy.albums([r[1] for r in inserted_tracks[0:3]])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "distinct_albums = [a[1] for a in inserted_tracks]\n",
    "print(f\"{len(distinct_albums)} total albums\")\n",
    "for chunk in chunks(distinct_albums, 200):\n",
    "    print(\"Doing 200\")\n",
    "    artist_ingestion.store_albums(chunk)\n",
    "    import time; time.sleep(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# artst_ingestion.upsert_tracks(track_spyids[0:5])\n",
    "spy = spotify._authenticate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "r1 = spy.tracks(track_spyids[0:5])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = artist_ingestion._fetch_and_parse_tracks(track_spyids[0:5], datetime.utcnow(), spy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "artist_ingestion._upsert_models(t)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "artist_ingestion.upsert_tracks(track_spyids[5:])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Ingest New Daily Streams"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spyids = mysql.query_single_column(\n",
    "'''\n",
    "select\n",
    "    sal.spyid\n",
    "from spotify_artist_listeners sal\n",
    "left join spotify_daily_streams sds on sds.spyid = sal.spyid\n",
    "where sal.monthly_listeners < 2000000\n",
    "  and sal.monthly_listeners_delta > 1000\n",
    "  and sds.spyid is null\n",
    "  and (\n",
    "    city0_country in (select iso_code from country_settings where is_source_for_releases)\n",
    "    or city1_country in (select iso_code from country_settings where is_source_for_releases)\n",
    "    or city2_country in (select iso_code from country_settings where is_source_for_releases)\n",
    "  )\n",
    "order by monthly_listeners_delta desc\n",
    "'''\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(spyids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tracker.unicorn import artist_ingestion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "reload(artist_ingestion)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "artist_ingestion.fetch_artists??"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_artist_models = artist_ingestion.fetch_artists(spyids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for chunk in chunks(all_artist_models, 300):\n",
    "    mysql.std_upsert(chunk)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "artist_ingestion.store_batch_of_daily_streams(spyids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}