2020.12.18 コラム
【API・データ検証奮闘記】#4.音楽系データシリーズ 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のデータ取得の手順から取得した結果までをご紹介致します。
対象 | AppleMusic ランキング URL: https://spotifycharts.com/regional/jp/daily/latest |
●ステップ1. どうやって取得できるか調査してみた

モブ1
API!!

モブ2
スクレイピング!!

モブ3
ツール!!

モブ
APIで取得することにしよう!!

ボス
それでやってくれ。
●ステップ2. データ取得してみた
早速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