2020.12.18 コラム

• API
• API・データ検証奮闘記
• Python
• トレンド分析

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

本コラムでは、APIやデータに関連する言語などを無邪気に触ってみた備忘録として、ライトに記載していきます！（連載記事はこちら）
「ちょっと違くない」「他にいい方法あるのに」といったご意見もあるかと思いますが、何卒お手柔らかに！！

＜プロフィール＞
富松 良介
2017年、株式会社サイバー・コミュニケーションズ（CCI）入社。Oracle Bluekai・Treasure Data等のDMPや、AWS・GCP等のPublicCloud領域を担務。2019年6月よりデータの利活用を推進するコンサルティング会社「株式会社DataCurrent」に出向し、事業会社の基盤構築・運用や自社ソリューション開発を担当。

※音楽系データシリーズ　3部作　他記事はこちら
・第1部：Spotify 編
・第3部：YouTube 編

●AppleMusicランキング取得やってみた

ボス

AppleMusicのランキングデータが欲しい。

モブ

かしこまり！！

AppleMusicのデータ取得の手順から取得した結果までをご紹介致します。

●ステップ１.　どうやって取得できるか調査してみた

モブ1

API！！

モブ2

スクレイピング！！

モブ3

ツール！！

モブ

APIで取得することにしよう！！

ボス

それでやってくれ。

●ステップ２.　データ取得してみた

早速Pythonでデータを取得してみる。

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")

●ステップ3.　取得結果

※音楽系データシリーズ　3部作　他記事はこちら
・第1部：Spotify 編
・第3部：YouTube 編

●最後に

弊社では、定点的なリサーチやトレンドの分析をおこなっています。性別や年代等の属性を検索トレンドのダッシュボード提供等様々なパッケージをご用意しておりますので、お気軽にお問い合わせください。

本データに関するお問い合わせは下記にて承ります。
株式会社DataCurrent
info@datacurrent.co.jp