TikTokから動画情報を取得するためのスクレイピング方法

<strong data-e2e="video-views" class="video-count css-dirst9-StrongVideoCount e148ts222">281.4K</strong>

from playwright.async_api import async_playwright
import asyncio, random, json, logging, time, os, yt_dlp
from urllib.parse import urlparse

# ログを設定する
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('tiktok_scraper.log'),
        logging.StreamHandler()
    ]
)

class TikTokScraper:
    def __init__(self):
        self.DOWNLOAD_VIDEO = True
        self.SAVE_DIR = "downloaded_videos"
        self.USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
        self.VIEWPORT = {'width': 1280, 'height': 720}
        self.TIMEOUT = 300  # 5分間のタイムアウト

    async def random_sleep(self, min_seconds=1, max_seconds=3):
        """ランダムな遅延"""
        delay = random.uniform(min_seconds, max_seconds)
        logging.info(f" {delay:.2f}秒間スリープします...")
        await asyncio.sleep(delay)

    async def handle_captcha(self, page):
        """確認コードの処理"""
        try:
            captcha_dialog = page.locator('div[role="dialog"]')
            if await captcha_dialog.count() > 0 and await captcha_dialog.is_visible():
                logging.warning("CAPTCHAが検出されました。手動で解決してください。")
                await page.wait_for_selector('div[role="dialog"]', state='detached', timeout=self.TIMEOUT*1000)
                logging.info("CAPTCHAが解決されました。再開します...")
                await self.random_sleep(0.5, 1)
        except Exception as e:
            logging.error(f"CAPTCHA処理エラー: {str(e)}")

    async def extract_video_info(self, page, video_url):
        """動画の詳細を抽出する"""
        logging.info(f"情報抽出中: {video_url}")

        try:
            await page.goto(video_url, wait_until="networkidle")
            await self.random_sleep(2, 4)
            await self.handle_captcha(page)

            # 主要な要素の読み込みを待機する
            await page.wait_for_selector('[data-e2e="like-count"]', timeout=10000)

            video_info = await page.evaluate("""() => {
                const getTextContent = (selectors) => {
                    for (let selector of selectors) {
                        const element = document.querySelector(selector);
                        if (element && element.textContent.trim()) {
                            return element.textContent.trim();
                        }
                    }
                    return 'N/A';
                };

                const getTags = () => {
                    const tagElements = document.querySelectorAll('a[data-e2e="search-common-link"]');
                    return Array.from(tagElements).map(el => el.textContent.trim());
                };

                return {
                    likes: getTextContent(['[data-e2e="like-count"]', '[data-e2e="browse-like-count"]']),
                    comments: getTextContent(['[data-e2e="comment-count"]', '[data-e2e="browse-comment-count"]']),
                    shares: getTextContent(['[data-e2e="share-count"]']),
                    bookmarks: getTextContent(['[data-e2e="undefined-count"]']),
                    views: getTextContent(['[data-e2e="video-views"]']),
                    description: getTextContent(['span.css-j2a19r-SpanText']),
                    musicTitle: getTextContent(['.css-pvx3oa-DivMusicText']),
                    date: getTextContent(['span[data-e2e="browser-nickname"] span:last-child']),
                    author: getTextContent(['a[data-e2e="browser-username"]']),
                    tags: getTags(),
                    videoUrl: window.location.href
                };
            }""")

            logging.info(f"情報抽出成功: {video_url}")
            return video_info

        except Exception as e:
            logging.error(f"{video_url}からの情報抽出失敗: {str(e)}")
            return None

    def download_video(self, video_url):
        """動画をダウンロードする"""
        if not os.path.exists(self.SAVE_DIR):
            os.makedirs(self.SAVE_DIR)

        ydl_opts = {
            'outtmpl': os.path.join(self.SAVE_DIR, '%(id)s.%(ext)s'),
            'format': 'best',
            'quiet': False,
            'no_warnings': False,
            'ignoreerrors': True
        }

        try:
            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                info = ydl.extract_info(video_url, download=True)
                filename = ydl.prepare_filename(info)
                logging.info(f"動画ダウンロード成功: {filename}")
                return filename
        except Exception as e:
            logging.error(f"動画ダウンロードエラー: {str(e)}")
            return None

    async def scrape_single_video(self, video_url):
        """単一のショート動画をスクレイピングする"""
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False)
            context = await browser.new_context(
                viewport=self.VIEWPORT,
                user_agent=self.USER_AGENT,
            )

            page = await context.new_page()
            result = {}

            try:
                # ショート動画の情報を抽出する
                video_info = await self.extract_video_info(page, video_url)
                if not video_info:
                    raise Exception("動画情報の抽出に失敗しました")

                result.update(video_info)

                # TikTokショート動画をダウンロードする
                if self.DOWNLOAD_VIDEO:
                    filename = self.download_video(video_url)
                    if filename:
                        result['local_path'] = filename

            except Exception as e:
                logging.error(f"動画スクレイピングエラー: {str(e)}")
            finally:
                await browser.close()

            return result

    def save_results(self, data, filename="tiktok_video_data.json"):
        """結果をJSONファイルに保存する"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
        logging.info(f"結果を{filename}に保存しました")

async def main():
    # クローラーを初期化する
    scraper = TikTokScraper()

    # ターゲットとするTikTokショート動画のURL
    video_url = "https://www.tiktok.com/@petervufriends/video/7476546872253893934" # 参考として

    # ショート動画をスクレイピングする
    video_data = await scraper.scrape_single_video(video_url)

    # スクレイピング結果を保存する
    if video_data:
        scraper.save_results(video_data)
        logging.info("\nスクレイピング完了。結果：")
        for key, value in video_data.items():
            logging.info(f"{key}: {value}")
    else:
        logging.error("動画データのスクレイピングに失敗しました")

if __name__ == "__main__":
    asyncio.run(main())