如何抓取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

# Configure logging
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 minute timeout

    async def random_sleep(self, min_seconds=1, max_seconds=3):
        """Random Delay"""
        delay = random.uniform(min_seconds, max_seconds)
        logging.info(f"Sleeping for {delay:.2f} seconds...")
        await asyncio.sleep(delay)

    async def handle_captcha(self, page):
        """Handling verification codes"""
        try:
            captcha_dialog = page.locator('div[role="dialog"]')
            if await captcha_dialog.count() > 0 and await captcha_dialog.is_visible():
                logging.warning("CAPTCHA detected. Please solve it manually.")
                await page.wait_for_selector('div[role="dialog"]', state='detached', timeout=self.TIMEOUT*1000)
                logging.info("CAPTCHA solved. Resuming...")
                await self.random_sleep(0.5, 1)
        except Exception as e:
            logging.error(f"Error handling CAPTCHA: {str(e)}")

    async def extract_video_info(self, page, video_url):
        """Extract video details"""
        logging.info(f"Extracting info from: {video_url}")
        
        try:
            await page.goto(video_url, wait_until="networkidle")
            await self.random_sleep(2, 4)
            await self.handle_captcha(page)

            # Waiting for key elements to load
            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"Successfully extracted info for: {video_url}")
            return video_info

        except Exception as e:
            logging.error(f"Failed to extract info from {video_url}: {str(e)}")
            return None

    def download_video(self, video_url):
        """Download Video"""
        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"Video successfully downloaded: {filename}")
                return filename
        except Exception as e:
            logging.error(f"Error downloading video: {str(e)}")
            return None

    async def scrape_single_video(self, video_url):
        """Scrape the single short"""
        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:
                # Extract shorts information
                video_info = await self.extract_video_info(page, video_url)
                if not video_info:
                    raise Exception("Failed to extract video info")

                result.update(video_info)

                # Download TikTok shorts
                if self.DOWNLOAD_VIDEO:
                    filename = self.download_video(video_url)
                    if filename:
                        result['local_path'] = filename

            except Exception as e:
                logging.error(f"Error scraping video: {str(e)}")
            finally:
                await browser.close()
            
            return result

    def save_results(self, data, filename="tiktok_video_data.json"):
        """Save the results to a JSON file"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
        logging.info(f"Results saved to {filename}")

async def main():
    # Initialize the crawler
    scraper = TikTokScraper()
    
    # Target TikTok short's URL
    video_url = "https://www.tiktok.com/@petervufriends/video/7476546872253893934" # Just as an reference
    
    # scrape the short
    video_data = await scraper.scrape_single_video(video_url)
    
    # save the scraping result
    if video_data:
        scraper.save_results(video_data)
        logging.info("\nScraping completed. Results:")
        for key, value in video_data.items():
            logging.info(f"{key}: {value}")
    else:
        logging.error("Failed to scrape video data")

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