Gmail APIとPythonを使ったYouTubeダウンローダー

## はじめに
### 挨拶
お久しぶりです。
[Qiita](https://qiita.com/Motchiy_tuti/)と[note](https://note.com/motchiy_tuti/)を久しく放置していた[Motchiy](https://www.youtube.com/@motchiy_tuti)です。
今回は、WEBアプリケーションを作ったので紹介します。

### 目的
同級生の最高裁(友人)が、YouTubeの音源のダウンロードができなくて困っているので、
そのためのツールを作りたい。(生徒会活動で必要とのこと)

一応、すでに[Google Colaboratory](https://colab.google/)にてツールを開発し共有しているが、
使い方が分からないらしい。~~URLコピペしてボタン押すだけだろ~~

:::note alert
本記事は、合法な範囲での使用を目的としています。
[一部](https://laws.e-gov.go.jp/law/345AC0000000048#Mp-Ch_2-Se_3-Ss_5-At_34#:~:text=学校その他の教育機関（営利を目的として設置されているものを除く。）において教育を担任する者及び授業を受ける者は、その授業の過程における利用に供することを目的とする場合には、その必要と認められる限度において、公表された著作物を複製し、若しくは公衆送信（自動公衆送信の場合にあつては、送信可能化を含む。以下この条において同じ。）を行い、又は公表された著作物であつて公衆送信されるものを受信装置を用いて公に伝達することができる。)等を除き、投稿者の許可なく著作物をダウンロードする行為は**違法**です。
YouTubeの[ポリシーとガイドライン](https://www.youtube.com/intl/ja_ALL/creators/how-things-work/policies-guidelines/)及び[利用規約](https://www.youtube.com/static?template=terms&hl=ja&gl=JP)の遵守をお願いします。
:::

## 対象者
この記事は以下のような人を対象にしています。
- 小規模(身内など)での使用を目的に開発する
- Google/Gmailアカウントを既に持っている
- メールを使った簡易的なWEBアプリケーションを作ってみたい

### 開発環境
|  | 種類 | バージョン |
| :--: | ---- | ---- |
| OS | Ubuntu | 24.04.2 LTS |
| API | Gmail API | Latest |
| Web | Apache | 2.4.58 |
| 言語 | Python | 3.12.3 |
|  | PHP | 8.3.6 |

言語については、事前に導入していることを前提に進めていきます。

:::note info
2025年6月6日時点
:::

## 作り方
メールとPython, Webサーバの連携で作れる。



### 全体の流れ
```mermaid
sequenceDiagram
    クライアント ->> Gmail: メールを送信
    participant API
    Python -->> Gmail: メールをリクエスト
    Gmail -->> Python: メールを転送
    Note over Python: 動画・音声をダウンロード
    Python -->> Web: DLページを作成
    Python -->> Gmail: 既読をつける
    Python ->> クライアント: DLページへのURLを返信
    クライアント ->> Web: DLページにアクセス
    Note over Web: ファイルをまとめる
    Web ->> クライアント: ファイルをダウンロード
```

### 1. Webサーバの構築
#### Apacheのインストール
```bash
sudo apt update && sudo apt upgrade
sudo apt install apache2
```

:::note warn
初めてsudoを使うとき、パスワードの入力が求められます。
:::

`apt`でインストール済みのパッケージを更新し、`apache2`をインストールします。

#### ファイアウォールの設定
```bash
sudo ufw allow 'Apache'
```
Apacheが使用するポートを開放します。

#### Webページの設置
```bash
sudo nano /var/www/html/index.php
```
以下の内容を入力して保存します。
```php: /var/www/html/index.php
<?php
// Silence is golden.
```
> "Silence is golden."「沈黙は金なり。」

:::note warn
これ以降も、`/var/www/html/`以下にディレクトリを作成する場合、
それぞれに`index.php`を配置するようにしてください。
:::

<details><summary>これを書く理由</summary>
悪意のあるクライアントがWebサイトにアクセスしてきたとき、<br>
ディレクトリ一覧を見られないようにするために作成します。<br>
詳しくは「ディレクトリリスティング」で検索
</details>

#### SSL化
```bash
sudo apt install certbot python3-certbot-apache
sudo certbot --apache
```

#### 動作確認
ブラウザにて、以下のURLにアクセスしてください。
``http:{サーバのIPアドレス}``または、``https:{サーバのIPアドレス}``
真っ白画面が表示されたら成功です。

:::note info
最新のバージョンとなっていることを確認してください。
:::

### 2. Gmail APIの準備/Pythonのコーディング
以下の記事・動画を参考にしました。
https://pythonchan.com/?p=4180
https://www.youtube.com/watch?v=L4BH1sDRpaQ

プログラムが正しく動作したのを確認したら、次に進みます。

### 3. yt-dlpの導入
:::note info
ここでは、`venv`を使ったやり方を解説します。
:::
```bash
cd {開発ディレクトリ}
python3 -m venv .venv
. .venv/bin/activate
```

:::note warn
これ以降、**開発ディレクトリ**は`dir`と略します。
:::

すると、以下のようになります。
```bash
(venv) user@example.net dir$
```

```bash
pip install yt-dlp google-api-python-client google-auth-httplib2 google-auth-oauthlib
```
`yt-dlp`とGmail API関係のものをインストールします。

```bash
deactivate
```
`.venv`から非アクティベート化します。

### 4. YouTubeにログイン
デフォルトのブラウザでYouTubeにログインしてください。

### 5. コーディング
以下のファイルすべてを、開発ディレクトリに配置してください。

:::note warn
ファイル名も同じにしてください
:::

```python: main.py
from google.oauth2.credentials import Credentials # type: ignore
from googleapiclient.discovery import build # type: ignore
from process_message import ready_mail
import asyncio
import os


async def in_roop(interval_sec=60):
    while True:
        try:
            # token.jsonがなければtoken_gen.pyを実行
            if not os.path.exists('token.json'):
                os.system('python token_gen.py')
                if not os.path.exists('token.json'):
                    print("token.jsonの生成に失敗しました。")
                    await asyncio.sleep(interval_sec)
                    continue

            scopes = ['https://mail.google.com/']
            creds = Credentials.from_authorized_user_file('token.json', scopes)
            service = build('gmail', 'v1', credentials=creds)

            messages = service.users().messages().list(
                userId='me',
                q='subject:yt-dlp is:unread'
            ).execute().get('messages')

            if messages is None:
                messages = []

            tasks = []
            for message in messages:
                tasks.append(ready_mail(service, message))
            if tasks:
                await asyncio.gather(*tasks)

        except Exception as e:
            print(f"致命的エラー: {e}")

        await asyncio.sleep(interval_sec)


def main():
    try:
        asyncio.run(in_roop())
    except KeyboardInterrupt:
        print("...exit")


if __name__ == '__main__':
    main()
```
すべてのプログラムの軸となる部分です。
Gmail APIにアクセスし、それをループ処理します。
また、`token.json`の再生成機能も備えています。

```python: process_message.py
from google.oauth2.credentials import Credentials # type: ignore
from googleapiclient.discovery import build # type: ignore
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import base64_utl
import mail_info
import convert
import asyncio
import random
import os


def random_hex_dir():
    chars = 'abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789'
    while True:
        dirname = ''.join(random.choices(chars, k=8))
        full_path = os.path.join('output', dirname)
        if not os.path.exists(full_path):
            return dirname


def send_reply_mail(to_addr, subject, body, thread_id=None):
    scopes = ['https://mail.google.com/']
    creds = Credentials.from_authorized_user_file('token.json', scopes)
    service = build('gmail', 'v1', credentials=creds)

    message = MIMEMultipart()
    message['To'] = to_addr
    message['From'] = 'motchiy.tuti@gmail.com'
    message['Subject'] = subject
    message.attach(MIMEText(body))

    raw = {'raw': base64_utl.message_encode(message)}
    if thread_id:
        raw["threadId"] = thread_id  # スレッドIDをセット
    service.users().messages().send(
        userId='me',
        body=raw
    ).execute()


def process_message(service, message):
    print('=' * 10)
    m_data = service.users().messages().get(
        userId='me',
        id=message['id']
    ).execute()

    headers = m_data['payload']['headers']
    message_date = mail_info.header(headers, 'date')
    print(f'Date: {message_date}')
    from_addr = mail_info.header(headers, 'from')
    print(f'From: {from_addr}')
    to_addr = mail_info.header(headers, 'to')
    print(f'To: {to_addr}')
    sub_date = mail_info.header(headers, 'subject')
    message_id = mail_info.header(headers, 'message-id')

    body = m_data['payload']['body']
    body_data = mail_info.body(body)

    parts_data = None
    if 'parts' in m_data['payload']:
        parts = m_data['payload']['parts']
        parts_data = mail_info.parts(parts)
        attachment_id, extension = mail_info.attachment_id(parts)
        if attachment_id is not None:
            try:
                res = service.users().messages().attachments().get(
                    userId='me',
                    messageId=message['id'],
                    id=attachment_id
                ).execute()
                f_data = base64_utl.decode_file(res['data'])
                with open(f'download.{extension}', 'wb') as f:
                    f.write(f_data)
            except Exception as e:
                print(f"添付ファイル保存エラー: {e}")

    body_result = body_data if body_data is not None else parts_data
    print(body_result)

    import re
    options = re.findall(r'--(\w+)', sub_date.lower()) if sub_date else []
    if not options:
        return

    output_dir = random_hex_dir()
    abs_output_dir = os.path.join('/var/www/html/yt-dlp', output_dir)
    os.makedirs(abs_output_dir, exist_ok=True)

    # index.phpをカレントディレクトリからコピー
    import shutil
    try:
        shutil.copyfile('index.php', os.path.join(abs_output_dir, "index.php"))
    except Exception as e:
        print(f"index.phpコピーエラー: {e}")

    lines = body_result.splitlines() if body_result else []
    if 'audio' in options:
        for line in lines:
            if not line.strip():
                continue
            try:
                convert.audio_main(line.strip(), output_dir)
            except Exception:
                pass
    elif 'video' in options:
        for line in lines:
            if not line.strip():
                continue
            try:
                convert.video_main(line.strip(), output_dir)
            except Exception:
                pass

    thread_id = message.get('threadId')
    reply_body = f"Download URL: https://motchiy.f5.si/yt-dlp/{output_dir}/"
    reply_subject = f'Re: {sub_date}'
    try:
        send_reply_mail(
            to_addr=from_addr,
            subject=reply_subject,
            body=reply_body,
            thread_id=thread_id
        )
    except Exception as e:
        print(f"返信メール送信エラー: {e}")

    try:
        service.users().messages().modify(
            userId='me',
            id=message['id'],
            body={'removeLabelIds': ['UNREAD']}
        ).execute()
    except Exception as e:
        print(f"既読処理エラー: {e}")


async def ready_mail(service, message):
    await asyncio.to_thread(process_message, service, message)
```
メールの送受信と、ダウンロードファイルを置くディレクトリの作成を行います。

```python: mail_info.py
import base64_utl


def header(headers, name):
    for h in headers:
        if h['name'].lower() == name:
            return h['value']


def body(body):
    if body['size'] > 0:
        return base64_utl.decode(body['data'])


def parts_body(body):
    if (
        body['size'] > 0
        and 'data' in body
        and body.get('mimeType') == 'text/plain'
    ):
        return base64_utl.decode(body['data'])


def parts(parts):
    for part in parts:
        if part.get('mimeType') == 'text/plain':
            b = base64_utl.decode(part['body']['data'])
            if b is not None:
                return b
        if 'body' in part:
            b = parts_body(part['body'])
            if b is not None:
                return b
        if 'parts' in part:
            b = parts(part['parts'])
            if b is not None:
                return b


def attachment_id(parts):
    for part in parts:
        if part['mimeType'] == 'image/png':
            return part['body']['attachmentId'], 'png'
    return None, None
```
送信されたメールの情報を取得します。

```python: convert.py
import yt_dlp
import os


# 音声をWAVEファイルに
def download_youtube_audio_as_wav(url, output_dir):
    ydl_opts = {
        'format': 'bestaudio/best',
        'outtmpl': os.path.join('/var/www/html/yt-dlp', output_dir, '%(title)s.%(ext)s'),
        'postprocessors': [{
            'key': 'FFmpegExtractAudio',
            'preferredcodec': 'wav',
        }],
        'audioquality': 0,
        'noplaylist': True,
        'force_generic_extractor': True,
        'ignoreerrors': True,
        'quiet': True,
        'no_warnings': True,
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        try:
            ydl.download([url])
        except Exception as e:
            print(f"yt-dlp audio error: {e}")

def audio_main(text, output_dir):
    try:
        download_youtube_audio_as_wav(text, output_dir)
        return ""
    except Exception as e:
        return f"[AUDIO ERROR] {e}"

    
# 動画をAVIファイルに
def download_youtube_video_as_avi(url, output_dir):
    ydl_opts = {
        'format': 'bestvideo*+bestaudio/best',
        'outtmpl': os.path.join('/var/www/html/yt-dlp', output_dir, '%(title)s.%(ext)s'),
        'noplaylist': True,
        'videoquality': 'best',
        'audioquality': 0,
        'force_generic_extractor': True,
        'ignoreerrors': True,
        'quiet': True,
        'no_warnings': True,
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        try:
            ydl.download([url])
        except Exception as e:
            print(f"yt-dlp video error: {e}")

def video_main(text, output_dir):
    try:
        download_youtube_video_as_avi(text, output_dir)
        return ""
    except Exception as e:
        return f"[VIDEO ERROR] {e}"
```
`yt-dlp`モジュールを使い、音声/動画をダウンロードします。

```python: base64_utl
import base64


def message_encode(message):
    return base64.urlsafe_b64encode(message.as_bytes()).decode()


def decode(data):
    return base64.urlsafe_b64decode(data).decode()


def decode_file(data):
    return base64.urlsafe_b64decode(data.encode('UTF-8'))
```
`base64`のエンコード/デコードを行います。

```python: token_gen.py
import os
os.environ['BROWSER'] = '/usr/bin/w3m'

from google_auth_oauthlib.flow import InstalledAppFlow # type: ignore

SCOPES = ['https://mail.google.com/']

flow = InstalledAppFlow.from_client_secrets_file(
   'client_secret_*.json', SCOPES)

creds = flow.run_local_server(port=0)

with open('token.json', 'w') as token:
   token.write(creds.to_json())
```
ファイル名が`client_secret_xxx.json`より、`token.json`を作成します。


```php: index.php
<?php
$files = array_diff(scandir(__DIR__), ['.', '..', 'index.php']);
$files = array_values($files);

if (count($files) === 1) {
    $file = $files[0];
    $fullpath = __DIR__ . DIRECTORY_SEPARATOR . $file;
    if (is_file($fullpath)) {
        header('Content-Description: File Transfer');
        header('Content-Type: application/octet-stream');
        header('Content-Disposition: attachment; filename="' . basename($file) . '"');
        header('Content-Length: ' . filesize($fullpath));
        readfile($fullpath);
        exit;
    }
    http_response_code(404);
    exit;
}
if (count($files) > 1) {
    $zipname = 'files_' . date('Ymd_His') . '.zip';
    $zip = new ZipArchive();
    $tmpZip = tempnam(sys_get_temp_dir(), 'zip');
    if ($zip->open($tmpZip, ZipArchive::OVERWRITE) === TRUE) {
        foreach ($files as $f) {
            $fullpath = __DIR__ . DIRECTORY_SEPARATOR . $f;
            if (is_file($fullpath)) {
                $zip->addFile($fullpath, $f);
            }
        }
        $zip->close();
        header('Content-Type: application/zip');
        header('Content-Disposition: attachment; filename="' . $zipname . '"');
        header('Content-Length: ' . filesize($tmpZip));
        readfile($tmpZip);
        unlink($tmpZip);
        exit;
    }
    http_response_code(500);
    exit;
}
```
Webページにアクセスされたとき、`index.php`以外を適切な形でダウンロードさせます。

```bash: run.sh
#!/bin/bash
source .venv/bin/activate
python3 main.py
```
`.venv`上で`main.py`を実行します。

結果的に、以下のようになっていれば完成です。
```plaintext
dir
 ├ .venv/ (表示されない)
 ├ base64_utl.py
 ├ client_secret_xxx.json
 ├ convert.py
 ├ index.php
 ├ mail_info.py
 ├ main.py
 ├ process_message.py
 ├ token_gen.py
 └ run.sh
```
:::note info
名前順に並び替えています。
:::

### 6. 実行
#### 起動
```bash
cd dir
. run.sh
```

#### 動作確認
以下のようなメールを送信します。
**宛先**: `Gmail APIを登録したGメールアドレス`
**件名**: `yt-dlp --audio`
**本文**: `https://www.youtube.com/watch?v=v01nN0EL0Wk`

:::note info
ここでは、私のチャンネルの紹介動画を使います。
:::

しばらく待つと、以下のようなメールが返信されます。
`Download URL: https://example.net/yt-dlp/xxxxxxxx/`

URL部分をブラウザで開き、`【単発#1】自分のチャンネル紹介.wav`が
ダウンロードされれば成功です。

#### 終了
`Ctrl`+`C`でプログラムを終了します。

## 使い方
### 基本
**宛先**: `Gmail APIを登録したメールアドレス`
**件名**: `yt-dlp`を含める
**本文**: `YouTubeのURL`(複数の動画を扱う場合、改行で区切ってください。)

### オプション
**件名**にyt-dlpとはスペースで区切って含めるようにしてください。
<details><summary>--audio</summary>
音声を抽出し、WAVE形式でダウンロードします。
</details>

<details><summary>--video</summary>
動画を抽出し、AVI形式でダウンロードします。
</details>

:::note warn
現状、どちらかのオプションをつけないといけません。
:::

## その他
### 品質を変更したい
`convert.py`の、`ydl_opts`を変更してください。
デフォルトでは、最高品質になっています。

### トークンを再生成したい
`token.json`を削除し、`client_secret_xxx.json`を上書きしてから起動することで、
トークンが再生成されます。

## おわりに
この記事では、メールを使ったYouTubeのダウンロードツールの作り方を紹介しました。
細かいところまで解説したので、だいぶ長いんじゃないかと思います。
~~もはや読んでる人誰もいないのでは？~~

エラーで行き詰ったときは、ぜひコメントで教えてください。
力になれるかもしれません。

また、「こんな機能を追加してほしい！」などの要望や、バグの報告などがあれば、
これまたコメントに書いてくれると嬉しいです。

それでは、ここまで読んでくださり、本当にありがとうございました。

## 参考記事
- [プログラミング初心者のためのQiita記事投稿テンプレート](https://qiita.com/YutaManaka/items/611069745ff0f68af879)
- [Mermaid Markdownでシーケンス図を作成してみた](https://qiita.com/duora0406/items/7af3ba3458757f3dd88b)
- [Markdown記法 チートシート](https://qiita.com/Qiita/items/c686397e4a0f4f11683d)
- [Markdownでのリンクを貼る時のエスケープシーケンス](https://qiita.com/Tatsuki-I/items/c0b9ed90d7970d18f37c)
- [PythonからGmail APIを使ってメールの送受信をしてみよう！添付ファイルも対応](https://pythonchan.com/?p=4180)
- [venv: Python 仮想環境管理](https://qiita.com/fiftystorm36/items/b2fd47cf32c7694adc2e)
- [ディレクトリ構成図を書くときに便利な記号
](https://qiita.com/paty-fakename/items/c82ed27b4070feeceff6)