(video_url: str, wait_secs: int = 30)
| 69 | |
| 70 | |
| 71 | async def main(video_url: str, wait_secs: int = 30) -> None: |
| 72 | captures: list[tuple[float, str, bytes]] = [] # (t, url, body) |
| 73 | |
| 74 | proxy = os.environ.get("HTTPS_PROXY") |
| 75 | proxy_arg = {"server": proxy} if proxy else None |
| 76 | |
| 77 | async with async_playwright() as pw: |
| 78 | kwargs = { |
| 79 | "headless": True, |
| 80 | "args": ["--no-sandbox", "--autoplay-policy=no-user-gesture-required"], |
| 81 | } |
| 82 | if proxy_arg: |
| 83 | kwargs["proxy"] = proxy_arg |
| 84 | browser = await pw.chromium.launch(**kwargs) |
| 85 | ctx = await browser.new_context(user_agent=( |
| 86 | "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " |
| 87 | "(KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36")) |
| 88 | await ctx.add_cookies([ |
| 89 | {"name": "CONSENT", "value": "YES+", "domain": ".youtube.com", "path": "/"}, |
| 90 | {"name": "SOCS", "value": "CAI", "domain": ".youtube.com", "path": "/"}, |
| 91 | ]) |
| 92 | page = await ctx.new_page() |
| 93 | cdp = await ctx.new_cdp_session(page) |
| 94 | await cdp.send("Network.enable", {}) |
| 95 | |
| 96 | pending: dict[str, dict] = {} |
| 97 | |
| 98 | def on_req(ev): |
| 99 | req = ev.get("request") or {} |
| 100 | url = req.get("url", "") |
| 101 | if "googlevideo.com/videoplayback" not in url: |
| 102 | return |
| 103 | if not req.get("hasPostData", False): |
| 104 | return |
| 105 | pending[ev["requestId"]] = {"url": url, "t": time.time()} |
| 106 | |
| 107 | cdp.on("Network.requestWillBeSent", on_req) |
| 108 | |
| 109 | await page.goto(video_url, wait_until="domcontentloaded", timeout=60_000) |
| 110 | await page.evaluate("() => { const v=document.querySelector('video'); " |
| 111 | "if (v) { v.muted=true; v.play().catch(()=>{});} }") |
| 112 | |
| 113 | t0 = time.time() |
| 114 | seen: set[str] = set() |
| 115 | while time.time() - t0 < wait_secs: |
| 116 | await asyncio.sleep(0.5) |
| 117 | for rid, info in list(pending.items()): |
| 118 | if rid in seen: |
| 119 | continue |
| 120 | try: |
| 121 | res = await cdp.send("Network.getRequestPostData", |
| 122 | {"requestId": rid}) |
| 123 | blob = res.get("postData", "") |
| 124 | if not blob: |
| 125 | continue |
| 126 | try: |
| 127 | body = base64.b64decode(blob, validate=True) |
| 128 | except Exception: |
no test coverage detected