Create gemini_web2api.py

gemini_web2api.py

@@ -0,0 +1,573 @@
+#!/usr/bin/env python3
+"""
+gemini-web2api - Gemini Web to OpenAI API proxy.
+
+Converts Google Gemini's web interface into an OpenAI-compatible API server.
+Zero authentication required. Works on any platform (Windows/macOS/Linux).
+
+Usage:
+    python gemini_web2api.py [--port 8081] [--config config.json]
+
+Client configuration (Cherry Studio, ChatBox, etc.):
+    Base URL: http://localhost:8081/v1
+    API Key: (anything or empty)
+"""
+import json
+import urllib.request
+import urllib.parse
+import time
+import ssl
+import sys
+import uuid
+import re
+import os
+import hashlib
+import argparse
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from socketserver import ThreadingMixIn
+
+__version__ = "1.0.0"
+
+# ─── Configuration ───────────────────────────────────────────────────────────
+
+DEFAULT_CONFIG = {
+    "port": 8081,
+    "host": "0.0.0.0",
+    "retry_attempts": 3,
+    "retry_delay_sec": 2,
+    "request_timeout_sec": 180,
+    "gemini_bl": "boq_assistant-bard-web-server_20260525.09_p0",
+    "default_model": "gemini-3.5-flash",
+    "log_requests": True,
+    "cookie_file": None,
+    "proxy": None,
+}
+
+CONFIG = dict(DEFAULT_CONFIG)
+
+# ─── Models ──────────────────────────────────────────────────────────────────
+# Mapping from JS source: MODE_CATEGORY enum (028-6eb337387583.js)
+#   1=FAST, 2=THINKING, 3=PRO, 4=AUTO, 5=FAST_DYNAMIC_THINKING, 6=FLASH_LITE
+
+MODELS = {
+    "gemini-3.5-flash": {
+        "mode": 1, "think": 4,
+        "desc": "Fast general-purpose model",
+    },
+    "gemini-3.5-flash-thinking": {
+        "mode": 2, "think": 0,
+        "desc": "Deep thinking mode, longest output (~20k chars)",
+    },
+    "gemini-3.1-pro": {
+        "mode": 3, "think": 4,
+        "desc": "Pro model (requires cookie for real routing)",
+    },
+    "gemini-auto": {
+        "mode": 4, "think": 4,
+        "desc": "Auto model selection",
+    },
+    "gemini-3.5-flash-thinking-lite": {
+        "mode": 5, "think": 0,
+        "desc": "Dynamic thinking with adaptive depth",
+    },
+    "gemini-flash-lite": {
+        "mode": 6, "think": 4,
+        "desc": "Lightweight fast model",
+    },
+}
+
+# ─── Utilities ───────────────────────────────────────────────────────────────
+
+def log(msg: str):
+    if CONFIG["log_requests"]:
+        sys.stderr.write(f"[{time.strftime('%H:%M:%S')}] {msg}\n")
+        sys.stderr.flush()
+
+
+def load_cookie() -> tuple:
+    """Load cookie from file. Returns (cookie_str, sapisid)."""
+    cookie_file = CONFIG.get("cookie_file")
+    if not cookie_file:
+        return "", None
+    if not os.path.exists(cookie_file):
+        return "", None
+    try:
+        with open(cookie_file, "r") as f:
+            content = f.read().strip()
+        if content.startswith("{"):
+            data = json.loads(content)
+            cookie_str = data.get("cookie", "")
+            sapisid = data.get("sapisid", "")
+        else:
+            cookie_str = content
+            pairs = dict(p.split("=", 1) for p in cookie_str.split("; ") if "=" in p)
+            sapisid = pairs.get("SAPISID", "")
+        return cookie_str, sapisid if sapisid else None
+    except Exception as e:
+        log(f"Cookie load error: {e}")
+        return "", None
+
+
+def make_sapisidhash(sapisid: str) -> str:
+    ts = int(time.time())
+    h = hashlib.sha1(f"{ts} {sapisid} https://gemini.google.com".encode()).hexdigest()
+    return f"SAPISIDHASH {ts}_{h}"
+
+
+# ─── Gemini Protocol ─────────────────────────────────────────────────────────
+
+def gemini_stream_generate(prompt: str, model_id: int, think_mode: int) -> str:
+    """Send prompt to Gemini StreamGenerate with retry."""
+    inner = [None] * 80
+    inner[0] = [prompt, 0, None, None, None, None, 0]
+    inner[1] = ["en"]
+    inner[2] = ["", "", "", None, None, None, None, None, None, ""]
+    inner[6] = [0]
+    inner[7] = 1
+    inner[10] = 1
+    inner[11] = 0
+    inner[17] = [[think_mode]]
+    inner[18] = 0
+    inner[27] = 1
+    inner[30] = [4]
+    inner[41] = [2]
+    inner[53] = 0
+    inner[59] = str(uuid.uuid4())
+    inner[61] = []
+    inner[68] = 1
+    inner[79] = model_id
+
+    outer = [None, json.dumps(inner)]
+    body = urllib.parse.urlencode({"f.req": json.dumps(outer)}).encode()
+    reqid = int(time.time()) % 1000000
+    url = (
+        "https://gemini.google.com/_/BardChatUi/data/"
+        "assistant.lamda.BardFrontendService/StreamGenerate"
+        f"?bl={CONFIG['gemini_bl']}&hl=en&_reqid={reqid}&rt=c"
+    )
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Origin": "https://gemini.google.com",
+        "Referer": "https://gemini.google.com/app",
+        "X-Same-Domain": "1",
+        "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
+    if sapisid:
+        headers["Authorization"] = make_sapisidhash(sapisid)
+
+    last_err = None
+    for attempt in range(CONFIG["retry_attempts"]):
+        try:
+            req = urllib.request.Request(url, data=body, headers=headers, method="POST")
+            ctx = ssl.create_default_context()
+            proxy = CONFIG.get("proxy")
+            if proxy:
+                opener = urllib.request.build_opener(
+                    urllib.request.ProxyHandler({"http": proxy, "https": proxy}),
+                    urllib.request.HTTPSHandler(context=ctx)
+                )
+                resp = opener.open(req, timeout=CONFIG["request_timeout_sec"])
+            else:
+                resp = urllib.request.urlopen(req, context=ctx, timeout=CONFIG["request_timeout_sec"])
+            return resp.read().decode("utf-8", errors="replace")
+        except Exception as e:
+            last_err = e
+            if attempt < CONFIG["retry_attempts"] - 1:
+                log(f"Retry {attempt+1}/{CONFIG['retry_attempts']}: {e}")
+                time.sleep(CONFIG["retry_delay_sec"])
+    raise last_err
+
+
+def clean_gemini_text(text: str) -> str:
+    """Remove internal code execution artifacts."""
+    text = re.sub(
+        r'```(?:python|javascript|text)\?code_(?:reference|stdout)&code_event_index=\d+\n.*?```\n?',
+        '', text, flags=re.DOTALL
+    )
+    return text.strip()
+
+
+def extract_response_text(raw: str) -> str:
+    """Parse StreamGenerate response to extract final text."""
+    texts = []
+    for line in raw.split("\n"):
+        if '"wrb.fr"' not in line or len(line) < 200:
+            continue
+        try:
+            arr = json.loads(line)
+            inner_str = arr[0][2]
+            if not inner_str or len(inner_str) < 50:
+                continue
+            inner = json.loads(inner_str)
+            if isinstance(inner, list) and len(inner) > 4 and inner[4]:
+                for part in inner[4]:
+                    if isinstance(part, list) and len(part) > 1 and part[1]:
+                        if isinstance(part[1], list):
+                            for t in part[1]:
+                                if isinstance(t, str) and len(t) > 0:
+                                    texts.append(t)
+        except (json.JSONDecodeError, IndexError, TypeError):
+            pass
+    text = ""
+    for t in reversed(texts):
+        if t.strip():
+            text = t
+            break
+    return clean_gemini_text(text)
+
+
+# ─── OpenAI Format Helpers ───────────────────────────────────────────────────
+
+def messages_to_prompt(messages: list, tools: list = None) -> str:
+    """Convert OpenAI messages to prompt string."""
+    parts = []
+    if tools:
+        tool_defs = []
+        for tool in tools:
+            fn = tool.get("function", tool) if tool.get("type") == "function" else tool
+            tool_defs.append({
+                "name": fn.get("name", tool.get("name", "")),
+                "description": fn.get("description", tool.get("description", "")),
+                "parameters": fn.get("parameters", tool.get("parameters", {})),
+            })
+        if tool_defs:
+            parts.append(
+                "[System instruction]: You have access to tools. "
+                "To call a tool, respond with:\n"
+                '```tool_call\n{"name": "func_name", "arguments": {...}}\n```\n'
+                "Only use tool_call blocks when needed.\n\n"
+                f"Available tools:\n{json.dumps(tool_defs, indent=2)}"
+            )
+    for msg in messages:
+        role = msg.get("role", "user")
+        content = msg.get("content", "")
+        if isinstance(content, list):
+            content = " ".join(
+                c.get("text", "") for c in content
+                if c.get("type") in ("text", "input_text")
+            )
+        if role == "system":
+            parts.append(f"[System instruction]: {content}")
+        elif role == "assistant":
+            if msg.get("tool_calls"):
+                tc_strs = []
+                for tc in msg["tool_calls"]:
+                    fn = tc.get("function", {})
+                    tc_strs.append(
+                        f'```tool_call\n{{"name": "{fn.get("name")}", '
+                        f'"arguments": {fn.get("arguments", "{}")}}}\n```'
+                    )
+                parts.append(f"[Assistant]: {content or ''}\n" + "\n".join(tc_strs))
+            else:
+                parts.append(f"[Assistant]: {content}")
+        elif role == "tool":
+            parts.append(f"[Tool result for {msg.get('name', '')}]: {content}")
+        else:
+            parts.append(content if content else "")
+    return "\n\n".join(p for p in parts if p)
+
+
+def parse_tool_calls(text: str) -> tuple:
+    """Extract tool_call blocks. Returns (clean_text, tool_calls_list)."""
+    tool_calls = []
+    pattern = r'```tool_call\s*\n(.*?)\n```'
+    for match in re.findall(pattern, text, re.DOTALL):
+        try:
+            data = json.loads(match.strip())
+            tool_calls.append({
+                "id": f"call_{uuid.uuid4().hex[:8]}",
+                "type": "function",
+                "function": {
+                    "name": data["name"],
+                    "arguments": json.dumps(data.get("arguments", {}), ensure_ascii=False),
+                },
+            })
+        except (json.JSONDecodeError, KeyError):
+            pass
+    clean = re.sub(pattern, '', text, flags=re.DOTALL).strip()
+    return clean, tool_calls
+
+
+# ─── HTTP Handler ────────────────────────────────────────────────────────────
+
+class GeminiHandler(BaseHTTPRequestHandler):
+    def log_message(self, fmt, *args):
+        log(fmt % args)
+
+    def send_json(self, data, status=200):
+        body = json.dumps(data, ensure_ascii=False).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_OPTIONS(self):
+        self.send_response(204)
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
+        self.send_header("Access-Control-Allow-Headers", "*")
+        self.end_headers()
+
+    def do_GET(self):
+        try:
+            if self.path == "/v1/models":
+                self.send_json({"object": "list", "data": [
+                    {"id": n, "object": "model", "created": 1700000000,
+                     "owned_by": "google", "description": c["desc"]}
+                    for n, c in MODELS.items()
+                ]})
+            elif self.path == "/":
+                self.send_json({"status": "ok", "version": __version__,
+                                "models": list(MODELS.keys())})
+            else:
+                self.send_json({"error": "not found"}, 404)
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+        except Exception as e:
+            log(f"GET error: {e}")
+
+    def do_POST(self):
+        try:
+            length = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(length) if length else b""
+            if self.path == "/v1/chat/completions":
+                self.handle_chat(body)
+            elif self.path == "/v1/responses":
+                self.handle_responses(body)
+            else:
+                self.send_json({"error": "not found"}, 404)
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+        except Exception as e:
+            log(f"POST error: {e}")
+            try:
+                self.send_json({"error": {"message": str(e)}}, 500)
+            except:
+                pass
+
+    def _resolve_model(self, model_name):
+        think_override = None
+        if "@think=" in model_name:
+            model_name, think_str = model_name.rsplit("@think=", 1)
+            think_override = int(think_str)
+        cfg = MODELS.get(model_name)
+        if not cfg:
+            return None, None, None, f"Unknown model: {model_name}"
+        return model_name, cfg["mode"], (think_override if think_override is not None else cfg["think"]), None
+
+    def _call_gemini(self, prompt, model_id, think_mode, tools):
+        raw = gemini_stream_generate(prompt, model_id, think_mode)
+        text = extract_response_text(raw)
+        tool_calls = None
+        if tools and text:
+            text, tool_calls = parse_tool_calls(text)
+        return text or "", tool_calls
+
+    def handle_chat(self, body: bytes):
+        req = json.loads(body)
+        model_name, model_id, think_mode, err = self._resolve_model(
+            req.get("model", CONFIG["default_model"]))
+        if err:
+            self.send_json({"error": {"message": err}}, 400)
+            return
+
+        tools = req.get("tools")
+        prompt = messages_to_prompt(req.get("messages", []), tools)
+        if not prompt.strip():
+            self.send_json({"error": {"message": "empty prompt"}}, 400)
+            return
+
+        try:
+            text, tool_calls = self._call_gemini(prompt, model_id, think_mode, tools)
+        except Exception as e:
+            self.send_json({"error": {"message": f"upstream error: {e}"}}, 502)
+            return
+
+        cid = f"chatcmpl-{uuid.uuid4().hex[:12]}"
+        msg = {"role": "assistant", "content": text or None}
+        if tool_calls:
+            msg["tool_calls"] = tool_calls
+        finish = "tool_calls" if tool_calls else "stop"
+
+        if req.get("stream"):
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Cache-Control", "no-cache")
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.end_headers()
+            chunk = {"id": cid, "object": "chat.completion.chunk", "created": int(time.time()),
+                     "model": model_name, "choices": [{"index": 0, "delta": msg, "finish_reason": finish}]}
+            self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
+            self.wfile.write(b"data: [DONE]\n\n")
+            self.wfile.flush()
+        else:
+            self.send_json({
+                "id": cid, "object": "chat.completion", "created": int(time.time()),
+                "model": model_name,
+                "choices": [{"index": 0, "message": msg, "finish_reason": finish}],
+                "usage": {"prompt_tokens": len(prompt)//4, "completion_tokens": len(text)//4,
+                          "total_tokens": (len(prompt)+len(text))//4},
+            })
+
+    def handle_responses(self, body: bytes):
+        """OpenAI Responses API for Codex CLI compatibility."""
+        req = json.loads(body)
+        model_name, model_id, think_mode, err = self._resolve_model(
+            req.get("model", CONFIG["default_model"]))
+        if err:
+            self.send_json({"error": {"message": err}}, 400)
+            return
+
+        input_items = req.get("input", [])
+        tools = req.get("tools")
+
+        messages = []
+        if req.get("instructions"):
+            messages.append({"role": "system", "content": req["instructions"]})
+        if isinstance(input_items, str):
+            messages.append({"role": "user", "content": input_items})
+        elif isinstance(input_items, list):
+            for item in input_items:
+                if isinstance(item, str):
+                    messages.append({"role": "user", "content": item})
+                elif isinstance(item, dict):
+                    if item.get("type") == "function_call_output":
+                        messages.append({"role": "tool", "tool_call_id": item.get("call_id", ""),
+                                         "name": item.get("name", ""), "content": item.get("output", "")})
+                    elif item.get("role") == "assistant" or (item.get("type") == "message" and item.get("role") == "assistant"):
+                        cp = item.get("content", [])
+                        text_acc, tc_list = "", []
+                        if isinstance(cp, list):
+                            for c in cp:
+                                if isinstance(c, dict):
+                                    if c.get("type") == "output_text": text_acc += c.get("text", "")
+                                    elif c.get("type") == "function_call": tc_list.append(c)
+                        elif isinstance(cp, str):
+                            text_acc = cp
+                        m = {"role": "assistant", "content": text_acc or None}
+                        if tc_list:
+                            m["tool_calls"] = [{"id": tc.get("call_id", f"call_{i}"), "type": "function",
+                                                "function": {"name": tc.get("name",""), "arguments": tc.get("arguments","{}")}}
+                                               for i, tc in enumerate(tc_list)]
+                        messages.append(m)
+                    else:
+                        role = item.get("role", "user")
+                        content = item.get("content", "")
+                        if isinstance(content, list):
+                            content = " ".join(c.get("text", "") for c in content if c.get("type") in ("text", "input_text"))
+                        messages.append({"role": role, "content": content})
+
+        if tools:
+            tools = [{"type": "function", "function": {"name": t["name"], "description": t.get("description", ""), "parameters": t.get("parameters", {})}}
+                     if t.get("type") == "function" and "function" not in t else t for t in tools]
+
+        prompt = messages_to_prompt(messages, tools)
+        if not prompt.strip():
+            self.send_json({"error": {"message": "empty input"}}, 400)
+            return
+
+        try:
+            text, tool_calls = self._call_gemini(prompt, model_id, think_mode, tools)
+        except Exception as e:
+            self.send_json({"error": {"message": f"upstream error: {e}"}}, 502)
+            return
+
+        rid = f"resp_{uuid.uuid4().hex[:16]}"
+        mid = f"msg_{uuid.uuid4().hex[:12]}"
+        output = []
+        if tool_calls:
+            for tc in tool_calls:
+                output.append({"type": "function_call", "id": tc["id"], "call_id": tc["id"],
+                               "name": tc["function"]["name"], "arguments": tc["function"]["arguments"], "status": "completed"})
+        if text or not tool_calls:
+            output.append({"type": "message", "id": mid, "role": "assistant", "status": "completed",
+                           "content": [{"type": "output_text", "text": text or "", "annotations": []}]})
+
+        if req.get("stream"):
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Cache-Control", "no-cache")
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.end_headers()
+            ev = {"type": "response.created", "response": {"id": rid, "object": "response", "status": "in_progress", "model": model_name, "output": []}}
+            self.wfile.write(f"event: response.created\ndata: {json.dumps(ev)}\n\n".encode())
+            for item in output:
+                if item["type"] == "function_call":
+                    ev = {"type": "response.function_call_arguments.done", "item_id": item["id"], "call_id": item["call_id"], "name": item["name"], "arguments": item["arguments"]}
+                    self.wfile.write(f"event: response.function_call_arguments.done\ndata: {json.dumps(ev)}\n\n".encode())
+                elif item["type"] == "message":
+                    for ci, cp in enumerate(item["content"]):
+                        ev = {"type": "response.output_text.done", "item_id": item["id"], "content_index": ci, "text": cp["text"]}
+                        self.wfile.write(f"event: response.output_text.done\ndata: {json.dumps(ev)}\n\n".encode())
+            resp_obj = {"id": rid, "object": "response", "status": "completed", "model": model_name, "output": output,
+                        "usage": {"input_tokens": len(prompt)//4, "output_tokens": len(text)//4, "total_tokens": (len(prompt)+len(text))//4}}
+            self.wfile.write(f"event: response.completed\ndata: {json.dumps({'type': 'response.completed', 'response': resp_obj})}\n\n".encode())
+            self.wfile.flush()
+        else:
+            self.send_json({"id": rid, "object": "response", "created_at": int(time.time()), "status": "completed",
+                            "model": model_name, "output": output,
+                            "usage": {"input_tokens": len(prompt)//4, "output_tokens": len(text)//4, "total_tokens": (len(prompt)+len(text))//4}})
+
+
+# ─── Main ────────────────────────────────────────────────────────────────────
+
+def load_config(path: str):
+    if path and os.path.exists(path):
+        with open(path) as f:
+            CONFIG.update(json.load(f))
+        log(f"Config loaded: {path}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Gemini Web to OpenAI API")
+    parser.add_argument("--port", type=int, default=None)
+    parser.add_argument("--config", type=str, default=None)
+    parser.add_argument("--cookie-file", type=str, default=None, help="Path to cookie file")
+    parser.add_argument("--proxy", type=str, default=None, help="HTTP proxy, e.g. http://127.0.0.1:7890")
+    parser.add_argument("--version", action="version", version=f"gemini-web2api {__version__}")
+    args = parser.parse_args()
+
+    config_path = args.config or os.environ.get("GEMINI_WEB2API_CONFIG")
+    if not config_path:
+        for p in ["./config.json", os.path.expanduser("~/.config/gemini-web2api/config.json")]:
+            if os.path.exists(p):
+                config_path = p
+                break
+    load_config(config_path)
+
+    if args.port:
+        CONFIG["port"] = args.port
+    if args.cookie_file:
+        CONFIG["cookie_file"] = args.cookie_file
+    if args.proxy:
+        CONFIG["proxy"] = args.proxy
+
+    class ThreadedServer(ThreadingMixIn, HTTPServer):
+        daemon_threads = True
+        allow_reuse_address = True
+
+    port = CONFIG["port"]
+    server = ThreadedServer((CONFIG["host"], port), GeminiHandler)
+    print(f"gemini-web2api v{__version__}")
+    print(f"  Listening: http://0.0.0.0:{port}")
+    print(f"  Base URL:  http://localhost:{port}/v1")
+    print(f"  Models:    {', '.join(MODELS.keys())}")
+    print(f"  Cookie:    {'yes (' + CONFIG['cookie_file'] + ')' if CONFIG.get('cookie_file') else 'none (anonymous)'}")
+    print(f"  Proxy:     {CONFIG.get('proxy') or 'none (uses system env HTTP_PROXY/HTTPS_PROXY)'}")
+    print(f"  Retry:     {CONFIG['retry_attempts']}x / {CONFIG['retry_delay_sec']}s")
+    print()
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        print("\nStopped.")
+        server.shutdown()
+
+
+if __name__ == "__main__":
+    main()