Spaces:
Paused
Paused
File size: 4,608 Bytes
eebe980 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """Multimodal: Scotty resumable upload for Gemini image input."""
import json
import base64
import urllib.request
import urllib.parse
import time
import ssl
import re
from .config import CONFIG
from .gemini import load_cookie, make_sapisidhash, _get_ssl_ctx, log
def _get_page_tokens() -> dict:
"""Fetch WIZ_global_data tokens from Gemini page (Push-ID, X-Client-Pctx)."""
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
}
cookie_str, sapisid = load_cookie()
if cookie_str:
headers["Cookie"] = cookie_str
try:
req = urllib.request.Request("https://gemini.google.com/app", headers=headers)
resp = urllib.request.urlopen(req, context=_get_ssl_ctx(), timeout=30)
html = resp.read().decode()
tokens = {}
for key, pattern in [
("push_id", r'"qKIAYe":"([^"]+)"'),
("pctx", r'"Ylro7b":"([^"]+)"'),
("at", r'"thykhd":"([^"]+)"'),
]:
m = re.search(pattern, html)
if m:
tokens[key] = m.group(1)
return tokens
except Exception as e:
log(f"Page token fetch failed: {e}")
return {}
_page_tokens_cache = {"tokens": {}, "ts": 0}
def _cached_page_tokens() -> dict:
now = time.time()
if now - _page_tokens_cache["ts"] > 600:
_page_tokens_cache["tokens"] = _get_page_tokens()
_page_tokens_cache["ts"] = now
return _page_tokens_cache["tokens"]
def upload_image(image_bytes: bytes, filename: str = "image.png", mime_type: str = "image/png") -> str:
"""Upload image via Scotty resumable upload. Returns file reference path."""
tokens = _cached_page_tokens()
push_id = tokens.get("push_id", "feeds/mcudyrk2a4khkz")
pctx = tokens.get("pctx", "CgcSBWjK7pYx")
cookie_str, sapisid = load_cookie()
ctx = _get_ssl_ctx()
proxy = CONFIG.get("proxy")
# Step 1: Initiate resumable upload
start_headers = {
"Push-ID": push_id,
"X-Tenant-Id": "bard-storage",
"X-Client-Pctx": pctx,
"X-Goog-Upload-Header-Content-Length": str(len(image_bytes)),
"X-Goog-Upload-Header-Content-Type": mime_type,
"X-Goog-Upload-Protocol": "resumable",
"X-Goog-Upload-Command": "start",
"Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
}
if cookie_str:
start_headers["Cookie"] = cookie_str
if sapisid:
start_headers["Authorization"] = make_sapisidhash(sapisid)
start_url = "https://content-push.googleapis.com/upload/"
req = urllib.request.Request(start_url, data=b"", headers=start_headers, method="POST")
if proxy:
opener = urllib.request.build_opener(
urllib.request.ProxyHandler({"http": proxy, "https": proxy}),
urllib.request.HTTPSHandler(context=ctx)
)
resp = opener.open(req, timeout=30)
else:
resp = urllib.request.urlopen(req, context=ctx, timeout=30)
upload_url = resp.headers.get("X-Goog-Upload-URL") or resp.headers.get("x-goog-upload-url")
if not upload_url:
raise RuntimeError(f"No upload URL in response headers: {dict(resp.headers)}")
log(f"Upload session started: {upload_url[:80]}...")
# Step 2: Upload file data + finalize
upload_headers = {
"X-Goog-Upload-Command": "upload, finalize",
"X-Goog-Upload-Offset": "0",
"Content-Type": "application/octet-stream",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
}
req2 = urllib.request.Request(upload_url, data=image_bytes, headers=upload_headers, method="POST")
if proxy:
resp2 = opener.open(req2, timeout=60)
else:
resp2 = urllib.request.urlopen(req2, context=ctx, timeout=60)
file_ref = resp2.read().decode().strip()
if not file_ref or not file_ref.startswith("/"):
raise RuntimeError(f"Invalid file reference: {file_ref[:100]}")
log(f"Image uploaded: {filename} -> {file_ref[:50]}...")
return file_ref
def fetch_image_bytes(url: str) -> bytes:
"""Fetch image from URL."""
try:
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
resp = urllib.request.urlopen(req, timeout=30)
return resp.read()
except Exception as e:
log(f"Image fetch failed: {e}")
return b""
|