このコードは、Twitter から動画をダウンロードするための Python 実装です。

必要なライブラリ

import sys
import os
import requests
import json
import re
from typing import Tuple, List, Dict
from dataclasses import dataclass
from pathlib import Path
import logging
from urllib.parse import urlencode

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


@dataclass
class VideoVariant:
    """異なるビットレートと品質の動画バリアントを表現するデータクラス"""
    bitrate: int  # ビットレート (kbps)
    url: str      # 動画URL
    content_type: str  # コンテンツタイプ (例: video/mp4)


class TwitterAPIError(Exception):
    """Twitter API関連のエラー用カスタム例外クラス"""
    pass


class TwitterAPIClient:
    """Twitter/X APIとの連携用クライアントクラス
    
    このクラスはTwitter/Xプラットフォームとの認証とAPI連携を処理します。
    ベアラートークンの取得、ゲストトークン認証、ツイートデータの取得などを含みます。
    
    属性:
        BASE_URL (str): TwitterのGraphQL APIベースURL
        ACTIVATE_URL (str): ゲストトークン取得用URL
        MAINJS_URL (str): ベアラートークンを含むメインJavaScriptファイルのURL
        session (requests.Session): HTTPリクエスト用セッションオブジェクト
        bearer_token (str): API認証用ベアラートークン
        guest_token (str): API認証用ゲストトークン
    
    メソッド:
        authenticate(): 必要な認証トークンを取得
        get_tweet_details(tweet_id): 特定のツイートの詳細情報を取得
    
    例外:
        TwitterAPIError: APIリクエスト失敗時または認証失敗時
        requests.RequestException: HTTPリクエスト失敗時
    
    使用例:
        client = TwitterAPIClient()
        client.authenticate()
        tweet_data = client.get_tweet_details("1234567890")
    """

    BASE_URL = "https://api.x.com/graphql/OoJd6A50cv8GsifjoOHGfg"
    ACTIVATE_URL = "https://api.twitter.com/1.1/guest/activate.json"
    MAINJS_URL = "https://abs.twimg.com/responsive-web/client-web/main.165ee22a.js"

    def __init__(self):
        """TwitterAPIClientの初期化"""
        self.session = requests.Session()
        self.bearer_token = None
        self.guest_token = None
        self._setup_session()

    def _setup_session(self):
        """リクエストセッションの基本ヘッダーを設定"""
        self.session.headers.update({
            "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:84.0) Gecko/20100101 Firefox/84.0",
            "accept": "*/*",
            "accept-language": "de,en-US;q=0.7,en;q=0.3",
            "accept-encoding": "gzip, deflate, br",
            "te": "trailers",
        })

    def authenticate(self):
        """認証トークンを取得する"""
        self.bearer_token = self._get_bearer_token()
        self.guest_token = self._get_guest_token()

        # セッションヘッダーをトークンで更新
        self.session.headers.update({
            "authorization": f"Bearer {self.bearer_token}",
            "x-guest-token": self.guest_token
        })

    def _get_bearer_token(self) -> str:
        """メインJSファイルからベアラートークンを抽出"""
        try:
            response = self.session.get(self.MAINJS_URL)
            response.raise_for_status()

            bearer_tokens = re.findall(r'AAAAAAAAA[^"]+', response.text)
            if not bearer_tokens:
                raise TwitterAPIError("メインJSファイルからベアラートークンが見つかりません")

            return bearer_tokens[0]

        except requests.RequestException as e:
            logger.error(f"ベアラートークンの取得に失敗: {e}")
            raise TwitterAPIError(f"ベアラートークンの取得に失敗: {e}")

    def _get_guest_token(self) -> str:
        """ベアラートークンを使用してゲストトークンを取得"""
        try:
            self.session.headers.update({"authorization": f"Bearer {self.bearer_token}"})
            response = self.session.post(self.ACTIVATE_URL)
            response.raise_for_status()

            return response.json()["guest_token"]

        except requests.RequestException as e:
            logger.error(f"ゲストトークンの取得に失敗: {e}")
            raise TwitterAPIError(f"ゲストトークンの取得に失敗: {e}")

    def get_tweet_details(self, tweet_id: str) -> dict:
        """Twitter APIからツイート詳細を取得"""
        try:
            url = self._build_tweet_detail_url(tweet_id)
            response = self.session.get(url)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            logger.error(f"ツイート詳細の取得に失敗: {e}")
            raise TwitterAPIError(f"ツイート詳細の取得に失敗: {e}")

    def _build_tweet_detail_url(self, tweet_id: str) -> str:
        """ツイート詳細APIのURLを構築"""
        return f"{self.BASE_URL}/TweetResultByRestId?variables=%7B%22tweetId%22%3A%22{tweet_id}..."


class TwitterVideoDownloader:
    """Twitter動画ダウンロード用メインクラス"""

    def __init__(self):
        self.api_client = TwitterAPIClient()

    def download_video(self, tweet_url: str, output_dir: Path = None):
        """
        ツイートURLから動画をダウンロード

        引数:
            tweet_url: 動画を含むツイートのURL
            output_dir: 動画保存先ディレクトリ（オプション）
        """
        try:
            # 認証設定
            self.api_client.authenticate()

            # ツイート詳細を取得
            tweet_id = self._extract_tweet_id(tweet_url)
            tweet_data = self.api_client.get_tweet_details(tweet_id)

            # 動画バリアントを抽出
            video_variants = self._extract_video_variants(tweet_data)
            if not video_variants:
                raise ValueError("ツイートに動画が見つかりません")

            # 動画をダウンロード
            self._download_video_variants(video_variants, tweet_id, output_dir)

        except Exception as e:
            logger.error(f"動画のダウンロードに失敗: {e}")
            raise

    @staticmethod
    def _extract_tweet_id(tweet_url: str) -> str:
        """ツイートURLからツイートIDを抽出"""
        tweet_id = re.findall(r'(?<=status/)d+', tweet_url)
        if not tweet_id:
            raise ValueError(f"URLからツイートIDを解析できません: {tweet_url}")
        return tweet_id[0]

    @staticmethod
    def _extract_video_variants(tweet_data: dict) -> List[VideoVariant]:
        """ツイートデータから動画バリアントを抽出"""
        try:
            media = tweet_data["data"]["tweetResult"]["result"]["legacy"]["entities"]["media"][0]
            variants = media["video_info"]["variants"]

            return [
                VideoVariant(
                    bitrate=variant.get("bitrate", 0),
                    url=variant["url"],
                    content_type=variant["content_type"]
                )
                for variant in variants
                if variant["content_type"] == "video/mp4"
            ]
        except (KeyError, IndexError) as e:
            logger.error(f"動画バリアントの抽出に失敗: {e}")
            raise ValueError("ツイートデータから動画情報が見つかりません")

    def _download_video_variants(
            self,
            variants: List[VideoVariant],
            tweet_id: str,
            output_dir: Path = None
    ):
        """
        全ての動画バリアントをダウンロード

        引数:
            variants: ダウンロードする動画バリアントのリスト
            tweet_id: ツイートID
            output_dir: 動画保存先ディレクトリ（オプション）
        """
        output_dir = output_dir or Path.cwd()
        output_dir.mkdir(parents=True, exist_ok=True)

        for variant in variants:
            output_path = output_dir / f"tweet_{tweet_id}_{variant.bitrate}.mp4"
            self._download_file(variant.url, output_path)
            logger.info(f"動画をダウンロードしました: {output_path}")

    @staticmethod
    def _download_file(url: str, output_path: Path):
        """
        URLからファイルをダウンロード（進捗追跡付き）

        引数:
            url: ダウンロードするファイルのURL
            output_path: ファイル保存先パス
        """
        try:
            with requests.get(url, stream=True) as response:
                response.raise_for_status()
                total_size = int(response.headers.get('content-length', 0))
                block_size = 8192

                with open(output_path, 'wb') as f:
                    if total_size == 0:
                        f.write(response.content)
                    else:
                        downloaded = 0
                        for chunk in response.iter_content(chunk_size=block_size):
                            if chunk:
                                f.write(chunk)
                                downloaded += len(chunk)
                                percentage = int((downloaded / total_size) * 100)
                                if percentage % 10 == 0:  # 10%毎にログ出力
                                    logger.info(f"ダウンロード進捗: {percentage}%")

        except requests.RequestException as e:
            logger.error(f"ファイルのダウンロードに失敗: {e}")
            if output_path.exists():
                output_path.unlink()  # 部分的にダウンロードされたファイルを削除
            raise


def main():
    """メイン実行関数"""
    if len(sys.argv) != 2:
        print("使用方法: python twitter_downloader.py <Twitter URL>")
        sys.exit(1)
        
    url = sys.argv[1]
    success = download_twitter_video(url)
    
    if success:
        print("動画のダウンロードが完了しました")
    else:
        print("動画のダウンロードに失敗しました")

if __name__ == "__main__":
    main()

使用方法

1. 必要なライブラリをインストール:

   pip install requests
   

2. スクリプトを実行:

   python twitter_downloader.py https://twitter.com/username/status/1234567890
   

注意事項

• Twitter の利用規約を遵守してください
• 著作権に注意して使用してください
• 大量のダウンロードは避けてください