179 lines
6.8 KiB
Python
179 lines
6.8 KiB
Python
import asyncio
|
|
import logging
|
|
|
|
from playwright.async_api import async_playwright
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_browser = None
|
|
|
|
|
|
async def _get_browser():
|
|
global _browser
|
|
if _browser is None:
|
|
p = await async_playwright().start()
|
|
_browser = await p.chromium.launch(headless=True, args=["--no-sandbox"])
|
|
return _browser
|
|
|
|
|
|
async def _scrape_channel(handle: str, channel_id: str) -> tuple[str | None, str | None]:
|
|
"""Navigate YouTube with playwright and find the live video. Returns (video_id, thumbnail_url)."""
|
|
urls_to_try = []
|
|
if handle and handle.startswith("@"):
|
|
urls_to_try.extend([
|
|
f"https://www.youtube.com/{handle}/live",
|
|
f"https://www.youtube.com/{handle}/streams",
|
|
f"https://www.youtube.com/{handle}/videos",
|
|
])
|
|
urls_to_try.extend([
|
|
f"https://www.youtube.com/channel/{channel_id}/live",
|
|
f"https://www.youtube.com/channel/{channel_id}/streams",
|
|
f"https://www.youtube.com/channel/{channel_id}/videos",
|
|
])
|
|
|
|
try:
|
|
browser = await _get_browser()
|
|
|
|
for url in urls_to_try:
|
|
try:
|
|
page = await browser.new_page()
|
|
await page.goto(url, timeout=15000, wait_until="domcontentloaded")
|
|
await page.wait_for_timeout(3000)
|
|
|
|
videos = await page.query_selector_all("ytd-video-renderer")
|
|
if not videos:
|
|
videos = await page.query_selector_all("ytd-grid-video-renderer")
|
|
|
|
for vid in videos:
|
|
try:
|
|
title_el = await vid.query_selector(".yt-simple-endpoint.style-scope.ytd-video-renderer a")
|
|
if not title_el:
|
|
title_el = await vid.query_selector("a#video-title")
|
|
|
|
video_id = None
|
|
if title_el:
|
|
href = await title_el.get_attribute("href")
|
|
if href and "/watch?v=" in href:
|
|
video_id = href.split("/watch?v=")[1].split("&")[0]
|
|
elif href and "/live/" in href:
|
|
video_id = href.split("/live/")[1].split("?")[0]
|
|
|
|
if not video_id:
|
|
continue
|
|
|
|
thumb = None
|
|
thumb_el = await vid.query_selector("img#img")
|
|
if thumb_el:
|
|
thumb = await thumb_el.get_attribute("src")
|
|
if not thumb:
|
|
thumb = await thumb_el.get_attribute("data-thumb")
|
|
|
|
logger.info(
|
|
"Found video for channel %s via playwright %s: %s",
|
|
channel_id, url, video_id,
|
|
)
|
|
await page.close()
|
|
return video_id, thumb
|
|
except Exception:
|
|
continue
|
|
|
|
await page.close()
|
|
except Exception as e:
|
|
logger.debug("Playwright failed for %s: %s", url, e)
|
|
except Exception as e:
|
|
logger.debug("Playwright browser failed: %s", e)
|
|
|
|
# Fallback to yt-dlp
|
|
try:
|
|
import yt_dlp
|
|
|
|
ydl_opts = {
|
|
"flat_playlist": True,
|
|
"playlistend": 5,
|
|
"logger": logger,
|
|
}
|
|
|
|
for url in urls_to_try:
|
|
try:
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
entries = info.get("_entries", []) or info.get("entries", [])
|
|
for entry in entries:
|
|
video_id = entry.get("id")
|
|
if video_id:
|
|
thumb = None
|
|
thumbnails = entry.get("thumbnails", [])
|
|
if thumbnails:
|
|
thumb = sorted(thumbnails, key=lambda t: t.get("width", 0), reverse=True)[0].get("url")
|
|
logger.info(
|
|
"Found video for channel %s via yt-dlp %s: %s",
|
|
channel_id, url, video_id,
|
|
)
|
|
return video_id, thumb
|
|
except Exception as e:
|
|
logger.debug("yt-dlp failed for %s: %s", url, e)
|
|
except Exception as e:
|
|
logger.debug("yt-dlp fallback failed: %s", e)
|
|
|
|
return None, None
|
|
|
|
|
|
async def find_live_video(channel_id: str, handle: str = "") -> tuple[str | None, str | None]:
|
|
"""Find the current live video for a channel using playwright. Returns (video_id, thumbnail_url)."""
|
|
return await _scrape_channel(handle, channel_id)
|
|
|
|
|
|
async def get_channel_info(channel_id: str, handle: str = "") -> dict | None:
|
|
"""Get channel details by scraping YouTube with playwright."""
|
|
urls_to_try = []
|
|
if handle and handle.startswith("@"):
|
|
urls_to_try.append(f"https://www.youtube.com/{handle}")
|
|
urls_to_try.append(f"https://www.youtube.com/channel/{channel_id}")
|
|
|
|
browser = await _get_browser()
|
|
|
|
for url in urls_to_try:
|
|
try:
|
|
page = await browser.new_page()
|
|
await page.goto(url, timeout=15000, wait_until="domcontentloaded")
|
|
await page.wait_for_timeout(3000)
|
|
|
|
title_el = await page.query_selector("h1#text a")
|
|
if not title_el:
|
|
title_el = await page.query_selector("h1#text span")
|
|
|
|
title = await title_el.inner_text() if title_el else None
|
|
|
|
desc_el = await page.query_selector("#description-text")
|
|
if not desc_el:
|
|
desc_el = await page.query_selector("span#description span")
|
|
description = await desc_el.inner_text() if desc_el else None
|
|
|
|
thumb_el = await page.query_selector("#img")
|
|
thumbnail = await thumb_el.get_attribute("src") if thumb_el else None
|
|
|
|
subs_el = await page.query_selector("span#text")
|
|
subscriber_count = 0
|
|
if subs_el:
|
|
subs_text = await subs_el.inner_text()
|
|
if "subscribers" in subs_text:
|
|
import re
|
|
match = re.search(r"([\d,.]+)", subs_text)
|
|
if match:
|
|
subscriber_count = match.group(1).replace(",", "")
|
|
|
|
await page.close()
|
|
|
|
if title:
|
|
return {
|
|
"channel_id": channel_id,
|
|
"title": title,
|
|
"description": description,
|
|
"thumbnail": thumbnail,
|
|
"subscriber_count": subscriber_count,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.debug("Playwright failed for channel %s (%s): %s", channel_id, url, e)
|
|
|
|
return None |