(self, task, tmp_dir)
| 1666 | browser.close() |
| 1667 | |
| 1668 | def _scrape_youtube(self, task, tmp_dir): |
| 1669 | global core |
| 1670 | opts = task['options'] |
| 1671 | tutto = opts.get('tutto', False) |
| 1672 | target = task['target'] |
| 1673 | if not target.startswith('@') and not target.startswith('UC'): |
| 1674 | # assume it's a handle if no @ |
| 1675 | target = '@' + target |
| 1676 | |
| 1677 | with sync_playwright() as p: |
| 1678 | browser = p.chromium.launch(headless=True, args=["--disable-blink-features=AutomationControlled", "--mute-audio"]) |
| 1679 | context = browser.new_context(user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36") |
| 1680 | page = context.new_page() |
| 1681 | |
| 1682 | # Intercept API calls for videos and community info |
| 1683 | api_data = [] |
| 1684 | def handle_response(response): |
| 1685 | try: |
| 1686 | if "youtubei/v1/browse" in response.url or "youtubei/v1/player" in response.url: |
| 1687 | data = response.json() |
| 1688 | api_data.append(data) |
| 1689 | except: pass |
| 1690 | page.on("response", handle_response) |
| 1691 | |
| 1692 | url = f"https://www.youtube.com/{target}/videos" if target.startswith('@') else f"https://www.youtube.com/channel/{target}/videos" |
| 1693 | page.goto(url, wait_until="domcontentloaded") |
| 1694 | task['progress'] = 15 |
| 1695 | time.sleep(5) |
| 1696 | |
| 1697 | # Click reject cookies if present |
| 1698 | try: |
| 1699 | page.locator("button[aria-label*='Rifiuta'], button[aria-label*='Reject']").click() |
| 1700 | time.sleep(2) |
| 1701 | except: pass |
| 1702 | |
| 1703 | if tutto or opts.get('dati', False): |
| 1704 | try: |
| 1705 | info = {"channel": target} |
| 1706 | info['title'] = page.title() |
| 1707 | |
| 1708 | # Try to get to "About" tab |
| 1709 | about_url = url.replace('/videos', '/about') |
| 1710 | page.goto(about_url, wait_until="domcontentloaded") |
| 1711 | time.sleep(3) |
| 1712 | |
| 1713 | desc_el = page.locator("#description-container, yt-attributed-string#description") |
| 1714 | if desc_el.count() > 0: |
| 1715 | info['description'] = desc_el.first.inner_text() |
| 1716 | |
| 1717 | links_el = page.locator("#link-list-container a, yt-channel-external-link-view-model a") |
| 1718 | links = [] |
| 1719 | for i in range(links_el.count()): |
| 1720 | links.append(links_el.nth(i).get_attribute("href")) |
| 1721 | info['links'] = links |
| 1722 | |
| 1723 | with open(os.path.join(tmp_dir, "youtube_info.json"), "w", encoding="utf-8") as f: |
| 1724 | safe_json_dump(info, f, indent=4, ensure_ascii=False) |
| 1725 |
no test coverage detected