Create app/youtube_api.py

app/youtube_api.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+import os
+import json
+import httpx
+from typing import Optional, Callable, Dict, Any
+from app.progress import RateProgress, human_bytes, fmt_eta
+
+YT_INIT_URL = "https://www.googleapis.com/upload/youtube/v3/videos"
+YT_PARTS = "snippet,status"
+
+def _clean_title(s: str) -> str:
+    s = (s or "").strip()
+    if not s:
+        return "Untitled"
+    # YouTube title limit is 100 chars
+    return s[:100]
+
+async def youtube_resumable_upload(
+    access_token: str,
+    file_path: str,
+    title: str,
+    description: str,
+    privacy: str,
+    chunk_size: int,
+    progress_cb: Optional[Callable[[int, int, float, float], Any]] = None,
+    http: Optional[httpx.AsyncClient] = None,
+) -> Dict[str, Any]:
+    """
+    progress_cb(done, total, speed_bps, eta_sec)
+    """
+    total = os.path.getsize(file_path)
+    title = _clean_title(title)
+    description = (description or "").strip()
+    privacy = privacy if privacy in ("private", "unlisted", "public") else "private"
+
+    close_http = False
+    if http is None:
+        http = httpx.AsyncClient(timeout=None)
+        close_http = True
+
+    try:
+        # 1) Initiate resumable session
+        meta = {
+            "snippet": {"title": title, "description": description},
+            "status": {"privacyStatus": privacy},
+        }
+        init_headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json; charset=UTF-8",
+            "X-Upload-Content-Type": "video/*",
+            "X-Upload-Content-Length": str(total),
+        }
+        init_params = {"uploadType": "resumable", "part": YT_PARTS}
+        init_resp = await http.post(YT_INIT_URL, headers=init_headers, params=init_params, content=json.dumps(meta))
+        if init_resp.status_code not in (200, 201):
+            return {"ok": False, "err": f"init_failed:{init_resp.status_code}:{init_resp.text[:200]}"}
+
+        upload_url = init_resp.headers.get("Location")
+        if not upload_url:
+            return {"ok": False, "err": "no_upload_location"}
+
+        # 2) Upload chunks
+        rp = RateProgress(total=total, edit_every=0.0)
+        done = 0
+
+        with open(file_path, "rb") as f:
+            while done < total:
+                chunk = f.read(chunk_size)
+                if not chunk:
+                    break
+                start = done
+                end = done + len(chunk) - 1
+                headers = {
+                    "Authorization": f"Bearer {access_token}",
+                    "Content-Length": str(len(chunk)),
+                    "Content-Range": f"bytes {start}-{end}/{total}",
+                }
+
+                put_resp = await http.put(upload_url, headers=headers, content=chunk)
+
+                # 308 = Resume Incomplete
+                if put_resp.status_code == 308:
+                    done = end + 1
+                elif put_resp.status_code in (200, 201):
+                    # finished
+                    j = put_resp.json()
+                    vid = j.get("id")
+                    return {"ok": True, "video_id": vid, "raw": j}
+                else:
+                    return {"ok": False, "err": f"upload_failed:{put_resp.status_code}:{put_resp.text[:200]}"}
+
+                snap = rp.snapshot(done)
+                if progress_cb:
+                    progress_cb(done, total, snap["speed"], snap["eta"])
+
+        return {"ok": False, "err": "upload_incomplete"}
+    finally:
+        if close_http:
+            await http.aclose()