0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@zhangzhaoyu1204

YouTube動画を自動で文字起こし→要約してNotionに保存するツールを作ってみた

0
Posted at

YouTube動画を自動で文字起こし→要約してNotionに保存するツールを作ってみた

はじめに

最近仕事で「このYouTube動画、後で見といて」と共有されることが増えたが、1本30分〜1時間の動画をいちいち全部観る時間がない。というわけで、URLを渡すだけで

  1. 音声を文字起こしして
  2. 要約して
  3. Notionに自動保存する

というツールをPythonでサクッと作ってみた。ついでに実装の途中で悩んだ「文字起こしをどう実装するか」の話も書いておく。

全体の流れ

YouTube URL
   ↓ 文字起こしタスク作成 (POST /transcriptions)
   ↓ ポーリング (GET /transcriptions/{id})
   ↓ 結果取得 (GET /transcriptions/{id}/result)
テキスト（タイムスタンプ・話者情報付き）
   ↓ 要約 (LLM)
要約テキスト
   ↓ 保存
Notionページ

環境

  • Python 3.11
  • requests
  • openai
  • notion-client
pip install requests openai notion-client

Step 1: 文字起こし方針を決める — 自前Whisper vs 商用API

最初は「音声をダウンロードしてWhisperに投げる」構成を考えていた。ただ実際に検討してみると:

  • 1時間の動画だとCPUだとかなり時間がかかる（GPU前提になる）
  • 長尺音声はチャンク分割してつなぎ直す処理が地味に面倒
  • バッチで何本も回すとなると、サーバー代やGPUインスタンス代がかさむ

というあたりで手が止まった。今回は「毎日何本も処理したいわけではなく、たまに1本サクッと処理したい」ユースケースだったので、文字起こし専用のAPI（videotranscriber.ai）を使うことにした。ドキュメントを読んでみると、YouTubeなどのプラットフォームURLをそのままsource_urlに渡せる仕様だったので、「動画をダウンロード→音声抽出→アップロード」という一番面倒な部分を丸ごと省略できたのがかなり大きかった（ファイルアップロードは非対応で、公開URLを渡す方式）。話者分離やチャプター分割、字幕(SRT/VTT)生成にも対応しているので、議事録用途とも相性がいい。

Step 2: 文字起こしタスクを作成する

POSTで投げるとすぐにレスポンスが返ってきて、処理自体は裏で非同期に進む。Idempotency-Keyを付けておくとネットワークエラーでリトライしたときに二重課金されないので、必ず指定しておく。

import uuid
import requests

VT_API_KEY = "YOUR_API_KEY"
BASE_URL = "https://videotranscriber.ai/openapi/v1"


def create_transcription(source_url: str) -> dict:
    headers = {
        "Authorization": f"Bearer {VT_API_KEY}",
        "Content-Type": "application/json",
        "Idempotency-Key": f"transcribe-{uuid.uuid4()}",
    }
    payload = {
        "source_url": source_url,
        "language": "auto",
        "speaker_diarization": True,
        "features": {"chapters": True},
    }
    res = requests.post(f"{BASE_URL}/transcriptions", headers=headers, json=payload)
    res.raise_for_status()
    return res.json()  # {"request_id": "...", "status": "queued", "poll_url": "...", "retry_after": 5}

Step 3: 処理完了を待つ（ポーリング）

タスクは queued → processing → succeeded / partial_succeeded / failed / cancelled とステータスが遷移する。レスポンスのretry_after（またはRetry-Afterヘッダ）に従ってポーリング間隔をあけるのが公式の推奨。

import time


def wait_for_result(request_id: str, timeout: int = 1800) -> dict:
    headers = {"Authorization": f"Bearer {VT_API_KEY}"}
    url = f"{BASE_URL}/transcriptions/{request_id}"
    elapsed = 0

    while elapsed < timeout:
        res = requests.get(url, headers=headers)
        res.raise_for_status()
        data = res.json()
        status = data["status"]

        if status in ("succeeded", "partial_succeeded"):
            return data
        if status in ("failed", "cancelled"):
            raise RuntimeError(f"文字起こしに失敗しました: {data}")

        wait_sec = data.get("retry_after", 5)
        time.sleep(wait_sec)
        elapsed += wait_sec

    raise TimeoutError("文字起こしがタイムアウトしました")

Step 4: 結果を取得する

def get_result(request_id: str) -> dict:
    headers = {"Authorization": f"Bearer {VT_API_KEY}"}
    res = requests.get(f"{BASE_URL}/transcriptions/{request_id}/result", headers=headers)
    res.raise_for_status()
    return res.json()


def transcribe(source_url: str) -> dict:
    task = create_transcription(source_url)
    wait_for_result(task["request_id"])
    return get_result(task["request_id"])

※ resultのレスポンスは「Transcript / Chapters / Translation」のスキーマで返ってくる仕様だが、細かいキー名までは公式ドキュメントに完全なサンプルJSONが載っていなかったので、実装時は一度リクエストを投げてprint(json.dumps(result, indent=2))で実際のキー構造を確認してから、下の要約処理に渡すフィールド名を合わせるのが確実。

Step 5: 要約する

from openai import OpenAI

client = OpenAI()


def summarize(text: str) -> str:
    prompt = f"""以下は動画の文字起こしです。次の形式で要約してください。

# 概要（3行程度）

# 主なトピック（箇条書き）

# 重要なポイント（箇条書き）

---
文字起こし:
{text}
"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

Step 6: Notionに保存する

from notion_client import Client

notion = Client(auth="YOUR_NOTION_TOKEN")
DATABASE_ID = "YOUR_DATABASE_ID"


def save_to_notion(title: str, summary: str, source_url: str):
    notion.pages.create(
        parent={"database_id": DATABASE_ID},
        properties={
            "Name": {"title": [{"text": {"content": title}}]},
            "URL": {"url": source_url},
        },
        children=[
            {
                "object": "block",
                "type": "paragraph",
                "paragraph": {
                    "rich_text": [{"type": "text", "text": {"content": summary}}]
                },
            }
        ],
    )

Step 7: 全部つなげる

def process_video(source_url: str, title: str):
    result = transcribe(source_url)
    transcript_text = result["transcript"]["text"]  # ※実際のキー名は要確認
    summary = summarize(transcript_text)
    save_to_notion(title, summary, source_url)
    print("Notionへの保存が完了しました")


if __name__ == "__main__":
    process_video(
        source_url="https://www.youtube.com/watch?v=xxxxxxxx",
        title="サンプル動画のまとめ",
    )

これで、YouTubeのURLを渡すだけで文字起こし→要約→Notion保存までが自動化できた。

使ってみた感想

  • 音声ファイルのダウンロードや一時保存の処理を書かなくてよかったのが地味に楽だった（YouTubeなどのプラットフォームURLをそのまま渡せる仕様のおかげ）
  • 話者分離やチャプター分割も同じリクエストでまとめて取れるので、議事録っぽい使い方とも相性がいい
  • 処理は非同期なので、複数本まとめて投げてポーリングでまとめて回収する、みたいな使い方もできそう
  • 結果の保持期間は30日、署名付きダウンロードURLは1時間で失効する仕様なので、字幕ファイルなどは取得したらすぐ自分のストレージに保存しておいたほうがいい

まとめ

「動画→文字起こし→要約→保存」というよくあるパイプラインを実際に組んでみた。文字起こし部分は自前で持つかAPIに任せるか悩みどころだが、今回のような「たまに使う」用途では、プラットフォームURLをそのまま投げられて、あとは非同期タスクの管理さえしてやれば済む今回の構成の方が実装コストが低かった。同じような悩みを持っている人の参考になれば幸いです。

参考

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?