【API・データ検証奮闘記】＃４．音楽系データシリーズ　3部作　第2部：AppleMusic

from datetime import date, datetime, timedelta, timezone
import jwt
import requests
import json
import pandas as pd
from requests.exceptions import HTTPError
import os
import time
import re


os.environ["secret_key"] = """-----BEGIN PRIVATE KEY-----
XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
XXXXXXXX
-----END PRIVATE KEY-----"""
os.environ["key_id"] = 'XXXXXXXXXX'
os.environ["team_id"] = 'XXXXXXXXXX'


class AppleMusic(object):
    def __init__(self, secret_key, key_id, team_id, proxies=None, requests_session=True, max_retries=10, requests_timeout=None, session_length=4000):
        self.proxies = proxies
        self._secret_key = secret_key
        self._key_id = key_id
        self._team_id = team_id
        self._alg = 'ES256'
        self.token_str = ""
        self.generate_token(session_length)
        self.root = 'https://api.music.apple.com/v1/'
        self.max_retries = max_retries
        self.requests_timeout = requests_timeout
        if requests_session:
            self._session = requests.Session()
        else:
            self._session = requests.api

    def generate_token(self, session_length):
        headers = {
            'alg': self._alg,
            'kid': self._key_id
        }
        payload = {
            'iss': self._team_id,
            'iat': int(datetime.now().timestamp()),
            'exp': int((datetime.now() + timedelta(hours=session_length)).timestamp())  # expiration time
        }
        token = jwt.encode(payload, self._secret_key, algorithm=self._alg, headers=headers)
        self.token_str = token.decode()

    def _auth_headers(self):
        if self.token_str:
            return {'Authorization': 'Bearer {}'.format(self.token_str)}
        else:
            return {}

    def _call(self, method, url, params):
        if not url.startswith('http'):
            url = self.root + url
        headers = self._auth_headers()
        headers['Content-Type'] = 'application/json'

        r = self._session.request(method, url,
                                  headers=headers,
                                  proxies=self.proxies,
                                  params=params,
                                  timeout=self.requests_timeout)
        r.raise_for_status()
        return r.json()

    def _get(self, url, **kwargs):
        retries = self.max_retries
        delay = 1
        while retries > 0:
            try:
                return self._call('GET', url, kwargs)
            except HTTPError as e: 
                retries -= 1
                status = e.response.status_code
                if status == 429 or (500 <= status < 600):
                    if retries < 0:
                        raise
                    else:
                        print('retrying ...' + str(delay) + ' secs')
                        time.sleep(delay + 1)
                        delay += 1
                else:
                    raise
            except Exception as e:
                print('exception', str(e))
                retries -= 1
                if retries >= 0:
                    print('retrying ...' + str(delay) + 'secs')
                    time.sleep(delay + 1)
                    delay += 1
                else:
                    raise

    def charts(self, storefront, chart=None, types=None, l=None, genre=None, limit=None, offset=None):
        url = self.root + 'catalog/{}/charts'.format(storefront)
        if types:
            type_str = ','.join(types)
        else:
            type_str = None
        return self._get(url, types=type_str, chart=chart, l=l, genre=genre, limit=limit, offset=offset)

if __name__ == '__main__':

  secret_key = os.environ["secret_key"]
  key_id = os.environ["key_id"]
  team_id = os.environ["team_id"]

  am = AppleMusic(secret_key, key_id, team_id)
  charts_list = []

  for offset in range(100):
    results = am.charts(storefront='jp', chart=None, types=["songs"], l=None, genre=None, limit=1, offset=offset)

    charts_list.append([
                        offset+1,
                        results["results"]["songs"][0]["name"],
                        results["results"]["songs"][0]["data"][0]["attributes"]["artistName"],
                        results["results"]["songs"][0]["data"][0]["attributes"]["name"],
                        results["results"]["songs"][0]["data"][0]["attributes"]["albumName"],
                        results["results"]["songs"][0]["data"][0]["attributes"]["genreNames"]
    ])

  result_df = pd.DataFrame(
  charts_list,
  index=None,
  columns=['position',
          'chart_type',
          'artist_name',
          'track_name',
          'album_name',
          'genre_names'
          ])
  JST = timezone(timedelta(hours=+9), 'JST')
  result_df['date'] = datetime.now(JST).strftime("%Y-%m-%d")