Update bot/youtube/resume.py

bot/youtube/resume.py

@@ -1,10 +1,8 @@
 # PATH: bot/youtube/resume.py
-from __future__ import annotations
-
-import asyncio
 import os
 import re
-from typing import Any, Awaitable, Callable, Dict, Optional, Tuple
+import asyncio
+from typing import Callable, Awaitable, Optional, Tuple, Dict, Any
 
 import httpx
 
@@ -15,7 +13,7 @@ ProgressCB = Callable[[int, int], Awaitable[None]]
 YOUTUBE_RESUMABLE_INIT = "https://www.googleapis.com/upload/youtube/v3/videos"
 
 
-def _parse_range_last_byte(headers: dict) -> int | None:
+def _parse_range_last_byte(headers: dict) -> Optional[int]:
     """
     Range header examples:
       - "bytes=0-1048575"
@@ -26,32 +24,38 @@ def _parse_range_last_byte(headers: dict) -> int | None:
     m = re.search(r"(\d+)\s*-\s*(\d+)", v)
     if not m:
         return None
-    return int(m.group(2))
+    try:
+        return int(m.group(2))
+    except Exception:
+        return None
 
 
 async def start_resumable_session(
     *,
     access_token: str,
     metadata: dict,
-    total_bytes: int = 0,
+    total_bytes: int,
 ) -> Dict[str, Any]:
     """
+    Start YouTube resumable session.
     Returns:
       {"ok": True, "session_url": "..."}
       {"ok": False, "err": "...", "detail": "..."}
     """
-    if not access_token:
-        return {"ok": False, "err": "missing_access_token"}
+    try:
+        tb = int(total_bytes)
+    except Exception:
+        tb = 0
+
+    if tb <= 0:
+        return {"ok": False, "err": "bad_total_bytes", "detail": f"total_bytes={total_bytes}"}
 
     headers = {
         "Authorization": f"Bearer {access_token}",
         "Content-Type": "application/json; charset=UTF-8",
         "X-Upload-Content-Type": "video/*",
+        "X-Upload-Content-Length": str(tb),
     }
-    # Optional but helpful
-    if total_bytes and total_bytes > 0:
-        headers["X-Upload-Content-Length"] = str(int(total_bytes))
-
     params = {"uploadType": "resumable", "part": "snippet,status"}
 
     try:
@@ -63,12 +67,15 @@ async def start_resumable_session(
                     "err": "resumable_init_failed",
                     "detail": f"{r.status_code}:{(r.text or '')[:400]}",
                 }
-            upload_url = r.headers.get("Location") or r.headers.get("location")
+
+            upload_url = r.headers.get("Location") or r.headers.get("location") or ""
+            upload_url = str(upload_url).strip()
             if not upload_url:
                 return {"ok": False, "err": "resumable_init_no_location"}
-            return {"ok": True, "session_url": str(upload_url)}
+
+            return {"ok": True, "session_url": upload_url}
     except Exception as e:
-        return {"ok": False, "err": "resumable_init_exception", "detail": f"{type(e).__name__}:{e}"}
+        return {"ok": False, "err": "resumable_init_exception", "detail": f"{type(e).__name__}:{str(e)[:400]}"}
 
 
 async def _query_upload_status(
@@ -76,10 +83,10 @@ async def _query_upload_status(
     upload_url: str,
     headers_base: dict,
     total: int,
-) -> tuple[int, dict | None]:
+) -> Tuple[int, Optional[dict], Optional[str]]:
     """
     Asks server what it has already received.
-    Returns (last_received_byte, final_json_if_completed)
+    Returns (last_received_byte, final_json_if_completed, err_detail_if_error)
     """
     headers = dict(headers_base)
     headers["Content-Length"] = "0"
@@ -90,17 +97,20 @@ async def _query_upload_status(
     if r.status_code == 308:
         last = _parse_range_last_byte(r.headers)
         if last is None:
-            return -1, None
-        return last, None
+            return -1, None, None
+        return last, None, None
 
     if r.status_code in (200, 201):
-        j = r.json() if r.text else {}
-        return total - 1, j
+        try:
+            j = r.json() if r.text else {}
+        except Exception:
+            j = {}
+        return total - 1, j, None
 
     if r.status_code >= 400:
-        raise RuntimeError(f"resumable_status_failed:{r.status_code}:{r.text[:200]}")
+        return -1, None, f"{r.status_code}:{(r.text or '')[:250]}"
 
-    return -1, None
+    return -1, None, None
 
 
 async def upload_resumable(
@@ -108,30 +118,30 @@ async def upload_resumable(
     session_url: str,
     file_path: str,
     access_token: str,
-    total_bytes: int = 0,
-    progress_cb: ProgressCB | None = None,
+    total_bytes: int,
+    progress_cb: Optional[ProgressCB] = None,
 ) -> Dict[str, Any]:
     """
     Upload file in chunks to session_url.
-
     Returns:
       {"ok": True, "response": {...}}
       {"ok": False, "err": "...", "detail": "..."}
     """
-    if not access_token:
-        return {"ok": False, "err": "missing_access_token"}
-    if not session_url:
-        return {"ok": False, "err": "missing_session_url"}
     if not file_path or not os.path.exists(file_path):
         return {"ok": False, "err": "file_not_found"}
 
     try:
-        total = int(total_bytes) if total_bytes else int(os.path.getsize(file_path))
+        total_fs = int(os.path.getsize(file_path))
+    except Exception:
+        total_fs = 0
+
+    try:
+        total = int(total_bytes or total_fs)
     except Exception:
-        total = 0
+        total = total_fs
 
     if total <= 0:
-        return {"ok": False, "err": "bad_total_bytes"}
+        return {"ok": False, "err": "bad_total_bytes", "detail": f"fs={total_fs} arg={total_bytes}"}
 
     sent = 0
 
@@ -142,7 +152,6 @@ async def upload_resumable(
 
     retryable = {429, 500, 502, 503, 504}
     max_retries = 8
-
     timeout = httpx.Timeout(connect=30.0, read=300.0, write=300.0, pool=30.0)
 
     try:
@@ -168,53 +177,58 @@ async def upload_resumable(
                         try:
                             r = await c.put(session_url, headers=headers, content=chunk)
                         except (httpx.TimeoutException, httpx.NetworkError):
-                            # Network issue: ask server what it has, then resume
-                            last, done_json = await _query_upload_status(c, session_url, headers_base, total)
+                            # network issue => ask status and resume
+                            last, done_json, errd = await _query_upload_status(c, session_url, headers_base, total)
                             if done_json is not None:
                                 if progress_cb:
                                     await progress_cb(total, total)
                                 return {"ok": True, "response": done_json}
-                            sent = max(sent, last + 1)
+
+                            if last >= 0:
+                                sent = max(sent, last + 1)
                             if progress_cb:
                                 await progress_cb(sent, total)
-                            break  # next outer loop iteration
+                            break
 
-                        # 308 means "resume incomplete" - read Range
                         if r.status_code == 308:
                             last = _parse_range_last_byte(r.headers)
                             if last is None:
-                                last, done_json = await _query_upload_status(c, session_url, headers_base, total)
+                                last, done_json, errd = await _query_upload_status(c, session_url, headers_base, total)
                                 if done_json is not None:
                                     if progress_cb:
                                         await progress_cb(total, total)
                                     return {"ok": True, "response": done_json}
-                            sent = max(sent, last + 1)
+                            if last is not None and last >= 0:
+                                sent = max(sent, last + 1)
                             if progress_cb:
                                 await progress_cb(sent, total)
                             break
 
-                        # Success
                         if r.status_code in (200, 201):
-                            j = r.json() if r.text else {}
+                            try:
+                                j = r.json() if r.text else {}
+                            except Exception:
+                                j = {}
                             if progress_cb:
                                 await progress_cb(total, total)
                             return {"ok": True, "response": j}
 
-                        # Retryable
                         if r.status_code in retryable and attempt < max_retries:
                             attempt += 1
-                            last, done_json = await _query_upload_status(c, session_url, headers_base, total)
+                            last, done_json, errd = await _query_upload_status(c, session_url, headers_base, total)
                             if done_json is not None:
                                 if progress_cb:
                                     await progress_cb(total, total)
                                 return {"ok": True, "response": done_json}
-                            sent = max(sent, last + 1)
+
+                            if last >= 0:
+                                sent = max(sent, last + 1)
                             if progress_cb:
                                 await progress_cb(sent, total)
-                            await asyncio.sleep(min(2 ** attempt, 20))
-                            break
 
-                        # Non-retryable error
+                            await asyncio.sleep(min(2**attempt, 20))
+                            break  # rebuild chunk from new sent
+
                         if r.status_code >= 400:
                             return {
                                 "ok": False,
@@ -222,19 +236,22 @@ async def upload_resumable(
                                 "detail": f"{r.status_code}:{(r.text or '')[:400]}",
                             }
 
-                        return {
-                            "ok": False,
-                            "err": "resumable_put_unexpected",
-                            "detail": f"{r.status_code}:{(r.text or '')[:200]}",
-                        }
+                        return {"ok": False, "err": "resumable_put_unexpected", "detail": str(r.status_code)}
 
                 # final status check
-                last, done_json = await _query_upload_status(c, session_url, headers_base, total)
+                last, done_json, errd = await _query_upload_status(c, session_url, headers_base, total)
                 if done_json is not None:
                     if progress_cb:
                         await progress_cb(total, total)
                     return {"ok": True, "response": done_json}
 
+                if last >= total - 1:
+                    last2, done_json2, _ = await _query_upload_status(c, session_url, headers_base, total)
+                    if done_json2 is not None:
+                        if progress_cb:
+                            await progress_cb(total, total)
+                        return {"ok": True, "response": done_json2}
+
         return {"ok": False, "err": "resumable_upload_incomplete"}
     except Exception as e:
-        return {"ok": False, "err": "resumable_exception", "detail": f"{type(e).__name__}:{e}"}
+        return {"ok": False, "err": "resumable_exception", "detail": f"{type(e).__name__}:{str(e)[:400]}"}