Upload image via Scotty resumable upload. Returns file reference path.
(image_bytes: bytes, filename: str = "image.png", mime_type: str = "image/png")
| 50 | |
| 51 | |
| 52 | def upload_image(image_bytes: bytes, filename: str = "image.png", mime_type: str = "image/png") -> str: |
| 53 | """Upload image via Scotty resumable upload. Returns file reference path.""" |
| 54 | tokens = _cached_page_tokens() |
| 55 | push_id = tokens.get("push_id", "feeds/mcudyrk2a4khkz") |
| 56 | pctx = tokens.get("pctx", "CgcSBWjK7pYx") |
| 57 | |
| 58 | cookie_str, sapisid = load_cookie() |
| 59 | ctx = _get_ssl_ctx() |
| 60 | proxy = CONFIG.get("proxy") |
| 61 | |
| 62 | # Step 1: Initiate resumable upload |
| 63 | start_headers = { |
| 64 | "Push-ID": push_id, |
| 65 | "X-Tenant-Id": "bard-storage", |
| 66 | "X-Client-Pctx": pctx, |
| 67 | "X-Goog-Upload-Header-Content-Length": str(len(image_bytes)), |
| 68 | "X-Goog-Upload-Header-Content-Type": mime_type, |
| 69 | "X-Goog-Upload-Protocol": "resumable", |
| 70 | "X-Goog-Upload-Command": "start", |
| 71 | "Content-Type": "application/x-www-form-urlencoded;charset=utf-8", |
| 72 | "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
| 73 | } |
| 74 | if cookie_str: |
| 75 | start_headers["Cookie"] = cookie_str |
| 76 | if sapisid: |
| 77 | start_headers["Authorization"] = make_sapisidhash(sapisid) |
| 78 | |
| 79 | start_url = "https://content-push.googleapis.com/upload/" |
| 80 | req = urllib.request.Request(start_url, data=b"", headers=start_headers, method="POST") |
| 81 | |
| 82 | if proxy: |
| 83 | opener = urllib.request.build_opener( |
| 84 | urllib.request.ProxyHandler({"http": proxy, "https": proxy}), |
| 85 | urllib.request.HTTPSHandler(context=ctx) |
| 86 | ) |
| 87 | resp = opener.open(req, timeout=30) |
| 88 | else: |
| 89 | resp = urllib.request.urlopen(req, context=ctx, timeout=30) |
| 90 | |
| 91 | upload_url = resp.headers.get("X-Goog-Upload-URL") or resp.headers.get("x-goog-upload-url") |
| 92 | if not upload_url: |
| 93 | raise RuntimeError(f"No upload URL in response headers: {dict(resp.headers)}") |
| 94 | |
| 95 | log(f"Upload session started: {upload_url[:80]}...") |
| 96 | |
| 97 | # Step 2: Upload file data + finalize |
| 98 | upload_headers = { |
| 99 | "X-Goog-Upload-Command": "upload, finalize", |
| 100 | "X-Goog-Upload-Offset": "0", |
| 101 | "Content-Type": "application/octet-stream", |
| 102 | "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", |
| 103 | } |
| 104 | |
| 105 | req2 = urllib.request.Request(upload_url, data=image_bytes, headers=upload_headers, method="POST") |
| 106 | if proxy: |
| 107 | resp2 = opener.open(req2, timeout=60) |
| 108 | else: |
| 109 | resp2 = urllib.request.urlopen(req2, context=ctx, timeout=60) |
no test coverage detected