Create youtube/resume.py

bot/youtube/resume.py

@@ -0,0 +1,79 @@
+# PATH: bot/youtube/resume.py
+import os
+from typing import Callable, Awaitable
+import httpx
+from bot.core.settings import YOUTUBE_CHUNK_SIZE
+
+ProgressCB = Callable[[int, int], Awaitable[None]]
+
+YOUTUBE_RESUMABLE_INIT = "https://www.googleapis.com/upload/youtube/v3/videos"
+
+async def start_resumable_session(access_token: str, metadata: dict) -> str:
+    """
+    Returns upload_url (Location header)
+    """
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "Content-Type": "application/json; charset=UTF-8",
+        "X-Upload-Content-Type": "video/*",
+    }
+    params = {"uploadType": "resumable", "part": "snippet,status"}
+    async with httpx.AsyncClient(timeout=60, follow_redirects=True) as c:
+        r = await c.post(YOUTUBE_RESUMABLE_INIT, params=params, headers=headers, json=metadata)
+        if r.status_code >= 400:
+            raise RuntimeError(f"resumable_init_failed:{r.status_code}:{r.text[:200]}")
+        upload_url = r.headers.get("Location") or r.headers.get("location")
+        if not upload_url:
+            raise RuntimeError("resumable_init_no_location")
+        return upload_url
+
+async def upload_resumable(
+    upload_url: str,
+    file_path: str,
+    access_token: str,
+    progress_cb: ProgressCB | None = None,
+) -> dict:
+    """
+    Upload file in chunks to upload_url.
+    Returns YouTube API JSON (contains id).
+    """
+    total = os.path.getsize(file_path)
+    sent = 0
+
+    headers_base = {
+        "Authorization": f"Bearer {access_token}",
+    }
+
+    async with httpx.AsyncClient(timeout=120, follow_redirects=True) as c:
+        with open(file_path, "rb") as f:
+            while sent < total:
+                chunk = f.read(YOUTUBE_CHUNK_SIZE)
+                if not chunk:
+                    break
+                start = sent
+                end = sent + len(chunk) - 1
+
+                headers = dict(headers_base)
+                headers["Content-Length"] = str(len(chunk))
+                headers["Content-Type"] = "video/*"
+                headers["Content-Range"] = f"bytes {start}-{end}/{total}"
+
+                r = await c.put(upload_url, headers=headers, content=chunk)
+
+                # 308 Resume Incomplete is normal for chunked upload
+                if r.status_code == 308:
+                    sent = end + 1
+                    if progress_cb:
+                        await progress_cb(sent, total)
+                    continue
+
+                if r.status_code >= 400:
+                    raise RuntimeError(f"resumable_put_failed:{r.status_code}:{r.text[:200]}")
+
+                # success: final response JSON
+                j = r.json()
+                if progress_cb:
+                    await progress_cb(total, total)
+                return j
+
+    raise RuntimeError("resumable_upload_incomplete")