Soundcloudのlikesを自動でプレイリスト化する

# はじめに
[Soundcloud](https://soundcloud.com)で何年も前にlikeした曲をもう一度聞きたい時に、自分のlikesを下のほうまで遡って探すのが面倒なので、時期ごとに自動でプレイリスト化して整理してみました。

# 作ったもの
likeした楽曲を投稿時期に基づき月ごとのプレイリストに振り分ける仕組みを作りました。
振り分け処理は毎日6時間ごとに自動で実行されます。

https://soundcloud.com/zen_dama/sets/2017-01/s-WSOitmynbqm

Descriptionやtagも自動で付与するようにしていますが、本記事では説明を省略します。

# 使用したもの
- Soundcloud API
likesの取得やプレイリストの作成に使いました。
使うためには[アプリケーションの登録](https://soundcloud.com/you/apps)を行なう必要がありますが、そのための[登録フォーム](https://docs.google.com/forms/d/e/1FAIpQLSfNxc82RJuzC0DnISat7n4H-G7IsPQIdaMpe202iiHZEoso9w/closedform)が2024年10月現在閉鎖されています（５年以上前からずっと閉鎖されていました）。
もう登録できないのかと思いきや、Githubの[issue](https://github.com/soundcloud/api/issues/246)の投稿によると登録フォームの代わりに[ヘルプセンター](https://help.soundcloud.com/hc/en-us/requests/new)からリクエストを受け付けているようです。リクエストには以下の情報を含める必要があります：
    - app name
    - app description
    - how you are going to use the API 
    - redirect_uri (http://localhost:8888/callbackなど)

    アプリケーションが登録されるとclient_id, client_secretが利用できるようになります。
    APIの入出力仕様等に関するドキュメントは[Swagger](https://developers.soundcloud.com/docs/api/explorer/open-api)で提供されています。
　
- Google スプレッドシート
認証に必要なトークンを保存するために使いました。
　
- Google Cloud Platform (GCP)
Cloud Run functionsに登録した関数をCloud Schedulerで定時実行します。
また、プログラムからGoogleスプレッドシートを操作するためにGoogle Drive APIおよびGoogle Sheets APIを使用します。
　
- Python3, Jupyter Lab
ローカルでの作業に使用しました。
必要なPythonパッケージも適宜インストールします。

# 手順

##  1. Soundcloud APIの認可を自動化するための準備
Soundcloud APIを叩く際は、事前に**認可**を行なっておく必要があります。
初回の認可にはユーザがログインした状態でブラウザ上での操作を行なう必要がありますが、そこで得られるトークンを用いることで、２回目以降はブラウザなしで認可を行なうことができます。

ここでは初回の認可だけローカルのPythonスクリプトから実行し、得られたトークンをGoogleスプレッドシートに保存して、GCPのfunctionから参照して使うことにします。

### 1-1 初回の認可
2024年10月から、認可の際に**PKCE**（Proof Key for Code Exchange）と呼ばれる仕組みを適用することが必須となっています。
これは認可で得られるトークンを攻撃者に横取りされることを防ぐためにベリファイを行なうものであり、実際のところ個人利用ではそのような攻撃のリスクは低いと思いますが、ともかく適用しないと認可されないので適用します。
PKCEの手順は以下の記事が分かりやすいです。

https://developers.line.biz/ja/docs/line-login/integrate-pkce

まずはPKCEでベリファイするために用いるコードを生成します。

```Python
import base64, hashlib
verifier = <ランダム文字列>
sha256hash = hashlib.sha256(verifier.encode()).digest()
challenge = base64.urlsafe_b64encode(sha256hash).rstrip(b'=').decode()
```
<ランダム文字列>はランダムな文字列でよいですが、ひとまず以下のページでGenerate Random Stringから生成しました。
https://example-app.com/pkce

ブラウザでログイン操作を行ないます。

```Python
import webbrowser
url = "https://api.soundcloud.com/connect?client_id={}&redirect_uri={}&response_type=code\
&code_challenge={}&code_challenge_method=S256".format(CLIENT_ID, REDIRECT_URI, challenge)
webbrowser.open(url)
```

操作が完了すると、`{redirect_uri}?code={code}`なるURLにリダイレクトします。
このcodeを用いて以下のようにリクエストすれば認可が完了します。

```Python
import requests
headers = {
    'accept': 'application/json; charset=utf-8',
    'Content-Type': 'application/x-www-form-urlencoded',
}

data = {
    'grant_type': 'authorization_code',
    'client_id': CLIENT_ID,
    'client_secret': CLIENT_SECRET,
    'redirect_uri': REDIRECT_URI,
    'code_verifier': verifier,
    'code' : code
}

response = requests.post('https://api.soundcloud.com/oauth2/token', headers=headers, data=data)
access_token = response.json()["access_token"]
```

access_tokenが得られれば、以下のようなヘッダをリクエストに含めることでAPIを叩けるようになります。
```Python
headers = {
      'accept': 'application/json; charset=utf-8',
      'Authorization': 'OAuth {}'.format(access_token),
}
```

### 1-2 スプレッドシートへの記入
`/oauth2/token`エンドポイントからのレスポンスの中身は以下のようになっています。
```
{
    "access_token":<access_token>,
    "expires_in":3599,
    "refresh_token":<refresh_token>,
    "scope":"",
    "token_type":"bearer"
}
```
この中でexpires_inはaccess_tokenの有効期限を指しており、今回の例では約1時間でトークンが無効になります。そのため、access_tokenを保存しても1時間後には失効し、再度ブラウザでのログイン操作が必要になってしまいます。

しかし、refresh_tokenを使用して以下のようなリクエストを送ることで、有効期限を更新した新しいaccess_tokenを得ることができます。
```Python
headers = {
  'accept': 'application/json; charset=utf-8',
  'Content-Type': 'application/x-www-form-urlencoded',
}

data = {
  'grant_type': 'refresh_token',
  'client_id': CLIENT_ID,
  'client_secret': CLIENT_SECRET,
  'redirect_uri': REDIRECT_URI,
  'refresh_token': <refresh_token>
}

response = requests.post('https://api.soundcloud.com/oauth2/token', headers=headers, data=data)
```

したがって、初回実行で得られたrefresh_tokenを保存しておき、定時実行の際は毎回はじめにそれを読み出して上記のようなリクエストを送ることで、有効なaccess_tokenを常に利用できるようになります。

refresh_tokenはGoogleスプレッドシートに保存することにします。
新規で作成したシートのA列1行目にrefresh_tokenの値を記入します。
プログラムから参照できるよう、このシートの公開設定を「リンクを知っている全員」に変更し、URLをメモしておきます。

## 2. Cloud Functionの作成
### 2-1. 認可
まず、Googleスプレッドシートに保存したrefresh_tokenを読み込みます。
以下の記事を参考にGoogle Drive APIおよびGoogle Sheets APIを有効にし、サービスアカウントのキー(JSONファイル)をダウンロードします。

https://zenn.dev/yamagishihrd/articles/2022-09_01-google-spreadsheet-with-python

refresh_tokenを読み込みます。
```Python
import gspread
from google.oauth2.service_account import Credentials
import pandas as pd

scopes = [
    'https://www.googleapis.com/auth/spreadsheets',
    'https://www.googleapis.com/auth/drive'
]

credentials = Credentials.from_service_account_file(
    'service_account.json',
    scopes=scopes
)

gc = gspread.authorize(credentials)

SP_SHEET_KEY = <1-2でメモしたシートURLのID部分>
sh = gc.open_by_key(SP_SHEET_KEY)

SP_SHEET = 'シート1'
worksheet = sh.worksheet(SP_SHEET)

data = worksheet.get_all_values()
df = pd.DataFrame(data)

refresh = df.loc[0,0]
```

読み込んだrefresh_tokenを使用して、新しいaccess_tokenとrefresh_tokenを取得します。

```Python
headers = {
  'accept': 'application/json; charset=utf-8',
  'Content-Type': 'application/x-www-form-urlencoded',
}

data = {
  'grant_type': 'refresh_token',
  'client_id': CLIENT_ID,
  'client_secret': CLIENT_SECRET,
  'redirect_uri': REDIRECT_URI,
  'refresh_token': refresh
}

response = requests.post('https://api.soundcloud.com/oauth2/token', headers=headers, data=data)
```

新しいrefresh_tokenを保存します。

```Python
from gspread_dataframe import set_with_dataframe
df.loc[0,0] = response.json()["refresh_token"]
set_with_dataframe(worksheet, df,include_column_header=False)
```

### 2-2. likesの取得
`GET /users/{user_id}/likes/tracks` エンドポイントから取得できます。
{user_id}はユーザプロフィールのURLに含まれる文字列ではなく、数字のみから成る内部的なIDです。プロフィールのページのソースのどこかに書いてあるので見つけます。

ちなみにlikesは公開情報なので、このAPIで自分(認可したユーザ)以外のユーザのlikesを取得することも可能です。

ただし一度にlikesをすべて取得できるわけではなく、ページングされた最初のページにあたる固定数の楽曲情報のリストと次のページへのリンクが以下のように返ってきます。
```
{
  "collection": {
    "$ref": "#/components/examples/Track/value"
  },
  "next_href": "https://api.soundcloud.com/collection?page_size=10&cursor=1234567"
}
```

これを踏まえて、collectionの内容をリストに蓄積しながら、next_hrefのリンク先を順次見ていきます。

```Python
track_list = []
response = requests.get('https://api.soundcloud.com/users/<user_id>/likes/tracks?linked_partitioning=true', headers=headers)
for track in response.json()["collection"]:
    track_list.append(track["id"])
while True:
    if response.json()["next_href"]==None:
        break
    response = requests.get(response.json()["next_href"], headers=headers)
    if response is not None:
        for track in response.json()["collection"]:
            track_list.append(track["id"])
    else:
        break
```

### 2-3. likesの整理
2-2のコードを変更して、月ごとに蓄積するようにします。

まず、蓄積するためのリストを定義します。
2008年～(現在の西暦)の1月～12月について、{"{YYYY}/{MM}": []}の書式の辞書で管理することにします。
```Python
import datetime
tokyo_tz = datetime.timezone(datetime.timedelta(hours=9))
dt = datetime.datetime.now(tokyo_tz)
years = [str(year) for year in range(2008,dt.year+1)]
months = [str(month) for month in range(1,13)]
monthly_list = {years[i]+"/"+months[j].zfill(2): [] for j in range(len(months)) for i in range(len(years))}
```

楽曲の作成日時は`track["created_at"]`で取得できるので、この値でリストに振り分けます。

```Python
response = requests.get('https://api.soundcloud.com/users/<user_id>/likes/tracks?linked_partitioning=true', headers=headers)
for track in response.json()["collection"]:
    if track["created_at"][:7] in monthly_list.keys():
        monthly_list[track["created_at"][:7]].append(track["id"])
while True:
    if response.json()["next_href"]==None:
        break
    response = requests.get(response.json()["next_href"], headers=headers)
    if response is not None:
        for track in response.json()["collection"]:
            if track["created_at"][:7] in monthly_list.keys():
                monthly_list[track["created_at"][:7]].append(track["id"])
    else:
        break
```

### 2-4. プレイリストの作成・更新
プレイリストの作成は`POST /playlists`、更新は`PUT /playlist/{playlist_id}`で行ないます。

準備として、既に作成されているプレイリストを取得します。
```Python
existing_playlists = {}
response = requests.get('https://api.soundcloud.com/me/playlists?linked_partitioning=true', headers=headers)
for playlist in response.json()["collection"]:
    existing_playlists[playlist["title"]] = playlist["id"]
while True:
    if response.json()["next_href"]==None:
        break
    response = requests.get(response.json()["next_href"], headers=headers)
    if response is not None:
        for playlist in response.json()["collection"]:
            existing_playlists[playlist["title"]] = playlist["id"]
    else:
        break
```

月ごとのリストに基づき、プレイリストの作成・更新を行ないます。
```Python
for month in sorted(monthly_list.keys()):
    sharing="private"
    if len(monthly_list[month])!=0:
        data = {
            "playlist": {
                "title": month,
                "sharing": sharing,
                "tag_list": "",
                "tracks": [{"id": a} for a in monthly_list[month]]
            }
        }
        if month not in existing_playlists.keys():
            response = requests.post('https://api.soundcloud.com/playlists', headers=headers, json=data)
        else:
            response = requests.put('https://api.soundcloud.com/playlists/{}'.format(existing_playlists[month]), headers=headers, json=data)
```

### 2-5. Cloud Functionの登録
Cloud Runの「ファンクション作成」から、以下の設定で作成します。
- 環境 : Cloud Run function (1st gen)
- 関数名 : 分かりやすいもの
- トリガー : Cloud Pub/Subを選択。トピックは分かりやすいIDで作成して指定。
- ランタイム : Python 3.10

ソースコードには、2-1～2-4の内容をまとめたものを登録します。

## 3. Cloud Schedulerでの自動化
以下の２つを設定することで、2-5で登録した関数を自動実行できます。
1. SchedulerからPub/Subのトピックを定期的に発火させる
Cloud Schedulerにて以下の設定で登録します。
    - 頻度 : 0 */6 * * *
    - ターゲットタイプ : Pub/Sub
    - トピック : 2-5で作成したものを選択
2. FunctionのトリガにPub/Subのトピックを指定する
こちらは2-5の作成時に既に設定しています。

これにより、6時間ごと(0時/6時/12時/18時)に関数が自動で実行されます。

# TODO
- 実行時間超過への対策
現在のスクリプトでは、like数が数万件規模の場合やSoundcloud APIサーバ側の負荷が高い場合に、Pub/SubをトリガとするCloud Functionの実行時間上限である540秒を超過する可能性が高いです。
現状ボトルネックとなっているAPIのリクエスト数を減らす工夫をするか、実行時間上限が1時間である第2世代のHTTPトリガの関数に変更するか、といった対策が必要だと思います。