Upload 4 files

Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.11-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    gcc \
+    && pip install --no-cache-dir -r requirements.txt \
+    && apt-get purge -y gcc \
+    && apt-get autoremove -y \
+    && rm -rf /var/lib/apt/lists/*
+COPY main.py .
+CMD ["python", "-u", "main.py"]

docker-compose.yml

@@ -0,0 +1,9 @@
+services:
+  geminibusiness:
+    build: .
+    container_name: geminibusiness
+    restart: unless-stopped
+    ports:
+      - "3003:8000"
+    env_file: .env
+ 

main.py

@@ -0,0 +1,343 @@
+import json, time, hmac, hashlib, base64, os, asyncio, uuid
+from datetime import datetime
+from typing import List, Optional, Union
+import logging
+
+import httpx
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+from fastapi.responses import HTMLResponse
+
+# ---------- 日志配置 ----------
+logging.basicConfig(
+    level=logging.INFO, # 生产环境建议 INFO，调试改 DEBUG
+    format="%(asctime)s | %(levelname)s | %(message)s",
+    datefmt="%H:%M:%S",
+)
+logger = logging.getLogger("gemini")
+
+# ---------- 配置 ----------
+SECURE_C_SES = os.getenv("SECURE_C_SES")
+HOST_C_OSES  = os.getenv("HOST_C_OSES")
+CSESIDX      = os.getenv("CSESIDX")
+CONFIG_ID    = os.getenv("CONFIG_ID")
+PROXY        = os.getenv("PROXY") or None
+JWT_TTL      = 270
+
+# ---------- 硬编码模型列表 ----------
+FIXED_MODELS = [
+    "gemini-2.5-flash",
+    "gemini-2.5-pro",
+    "gemini-3-pro"
+]
+
+# ---------- 核心：请求头伪装 ----------
+def get_common_headers(jwt: str) -> dict:
+    return {
+        "accept": "*/*",
+        "accept-encoding": "gzip, deflate, br, zstd",
+        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8",
+        "authorization": f"Bearer {jwt}",
+        "content-type": "application/json",
+        "origin": "https://business.gemini.google",
+        "referer": "https://business.gemini.google/",
+        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36",
+        "x-server-timeout": "1800",
+        "sec-ch-ua": '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
+        "sec-ch-ua-mobile": "?0",
+        "sec-ch-ua-platform": '"Windows"',
+        "sec-fetch-dest": "empty",
+        "sec-fetch-mode": "cors",
+        "sec-fetch-site": "cross-site",
+    }
+
+# ---------- 加密工具 ----------
+def urlsafe_b64encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).decode().rstrip("=")
+
+def kq_encode(s: str) -> str:
+    b = bytearray()
+    for ch in s:
+        v = ord(ch)
+        if v > 255:
+            b.append(v & 255)
+            b.append(v >> 8)
+        else:
+            b.append(v)
+    return urlsafe_b64encode(bytes(b))
+
+def create_jwt(key_bytes: bytes, key_id: str, csesidx: str) -> str:
+    now = int(time.time())
+    header = {"alg": "HS256", "typ": "JWT", "kid": key_id}
+    payload = {
+        "iss": "https://business.gemini.google",
+        "aud": "https://biz-discoveryengine.googleapis.com",
+        "sub": f"csesidx/{csesidx}",
+        "iat": now,
+        "exp": now + 300,
+        "nbf": now,
+    }
+    header_b64  = kq_encode(json.dumps(header, separators=(",", ":")))
+    payload_b64 = kq_encode(json.dumps(payload, separators=(",", ":")))
+    message     = f"{header_b64}.{payload_b64}"
+    sig         = hmac.new(key_bytes, message.encode(), hashlib.sha256).digest()
+    return f"{message}.{urlsafe_b64encode(sig)}"
+
+# ---------- JWT 管理 ----------
+class JWTManager:
+    def __init__(self) -> None:
+        self.jwt: str = ""
+        self.expires: float = 0
+        self._lock = asyncio.Lock()
+
+    async def get(self) -> str:
+        async with self._lock:
+            if time.time() > self.expires:
+                await self._refresh()
+            return self.jwt
+
+    async def _refresh(self) -> None:
+        cookie = f"__Secure-C_SES={SECURE_C_SES}"
+        if HOST_C_OSES:
+            cookie += f"; __Host-C_OSES={HOST_C_OSES}"
+
+        logger.debug("🔑 正在刷新 JWT...")
+        async with httpx.AsyncClient(proxies=PROXY, verify=False, timeout=30) as cli:
+            r = await cli.get(
+                "https://business.gemini.google/auth/getoxsrf",
+                params={"csesidx": CSESIDX},
+                headers={
+                    "cookie": cookie,
+                    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36",
+                    "referer": "https://business.gemini.google/"
+                },
+            )
+            if r.status_code != 200:
+                logger.error(f"❌ getoxsrf 失败: {r.status_code} {r.text}")
+                raise HTTPException(r.status_code, "getoxsrf failed")
+
+            txt = r.text[4:] if r.text.startswith(")]}'") else r.text
+            data = json.loads(txt)
+
+        key_bytes = base64.urlsafe_b64decode(data["xsrfToken"] + "==")
+        self.jwt     = create_jwt(key_bytes, data["keyId"], CSESIDX)
+        self.expires = time.time() + JWT_TTL
+        logger.info(f"✅ JWT 刷新成功")
+
+jwt_mgr = JWTManager()
+
+# ---------- Session 管理 ----------
+async def get_session() -> str:
+    jwt = await jwt_mgr.get()
+    headers = get_common_headers(jwt)
+    body = {
+        "configId": CONFIG_ID,
+        "additionalParams": {"token": "-"},
+        "createSessionRequest": {
+            "session": {"name": "", "displayName": ""}
+        }
+    }
+
+    logger.debug("🌐 正在创建 Session...")
+    async with httpx.AsyncClient(proxies=PROXY, verify=False, timeout=30) as cli:
+        r = await cli.post(
+            "https://biz-discoveryengine.googleapis.com/v1alpha/locations/global/widgetCreateSession",
+            headers=headers,
+            json=body,
+        )
+        if r.status_code != 200:
+            logger.error(f"❌ createSession 失败: {r.status_code} {r.text}")
+            raise HTTPException(r.status_code, "createSession failed")
+        sess_name = r.json()["session"]["name"]
+        return sess_name
+
+# ---------- OpenAI 兼容接口 ----------
+app = FastAPI(title="Gemini-Business OpenAI Gateway")
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+class ChatRequest(BaseModel):
+    model: str = FIXED_MODELS[0]
+    messages: List[Message]
+    stream: bool = False
+    temperature: Optional[float] = 0.7
+    top_p: Optional[float] = 1.0
+
+def create_chunk(id: str, created: int, model: str, delta: dict, finish_reason: Union[str, None]) -> str:
+    chunk = {
+        "id": id,
+        "object": "chat.completion.chunk",
+        "created": created,
+        "model": model,
+        "choices": [{
+            "index": 0,
+            "delta": delta,
+            "finish_reason": finish_reason
+        }]
+    }
+    return json.dumps(chunk)
+@app.get("/")
+async def root():
+    """根路径返回详细的 API 信息"""
+    html_content = """
+    <html>
+        <head>
+            <title>Gemini Business API</title>
+        </head>
+        <body>
+            <h1>Gemini Business API 运行中</h1>
+            <p>可用的 API 端点：</p>
+            <ul>
+                <li><a href="/health">/health</a> - 健康检查</li>
+                <li><a href="/v1/models">/v1/models</a> - 模型列表</li>
+                <li><a href="/docs">/docs</a> - API 文档</li>
+            </ul>
+            <p>聊天接口: POST /v1/chat/completions</p>
+        </body>
+    </html>
+    """
+    return HTMLResponse(content=html_content)
+
+@app.get("/v1/models")
+async def list_models():
+    data = []
+    now = int(time.time())
+    for m in FIXED_MODELS:
+        data.append({
+            "id": m,
+            "object": "model",
+            "created": now,
+            "owned_by": "google",
+            "permission": []
+        })
+    return {"object": "list", "data": data}
+
+@app.get("/health")
+async def health():
+    return {"status": "ok", "time": datetime.utcnow().isoformat()}
+
+@app.post("/v1/chat/completions")
+async def chat(req: ChatRequest):
+    session = await get_session()
+    chat_id = f"chatcmpl-{uuid.uuid4()}"
+    created_time = int(time.time())
+
+    # --- 流式处理 ---
+    if req.stream:
+        return StreamingResponse(
+            stream_chat_generator(session, req, chat_id, created_time),
+            media_type="text/event-stream"
+        )
+
+    # --- 非流式处理 ---
+    full_content = ""
+    async for chunk_str in stream_chat_generator(session, req, chat_id, created_time, is_stream=False):
+        if chunk_str.startswith("data: [DONE]"):
+            break
+        if chunk_str.startswith("data: "):
+            try:
+                data = json.loads(chunk_str[6:])
+                delta = data["choices"][0]["delta"]
+                if "content" in delta:
+                    full_content += delta["content"]
+            except:
+                pass
+
+    return {
+        "id": chat_id,
+        "object": "chat.completion",
+        "created": created_time,
+        "model": req.model,
+        "choices": [{
+            "index": 0,
+            "message": {
+                "role": "assistant",
+                "content": full_content
+            },
+            "finish_reason": "stop"
+        }],
+        "usage": {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+    }
+
+async def stream_chat_generator(session: str, req: ChatRequest, chat_id: str, created_time: int, is_stream: bool = True):
+    jwt = await jwt_mgr.get()
+    headers = get_common_headers(jwt)
+
+    body = {
+        "configId": CONFIG_ID,
+        "additionalParams": {"token": "-"},
+        "streamAssistRequest": {
+            "session": session,
+            "query": {"parts": [{"text": req.messages[-1].content}]},
+            "filter": "",
+            "fileIds": [],
+            "answerGenerationMode": "NORMAL",
+            "toolsSpec": {
+                "webGroundingSpec": {},
+                "toolRegistry": "default_tool_registry",
+                "imageGenerationSpec": {},
+                "videoGenerationSpec": {}
+            },
+            "languageCode": "zh-CN",
+            "userMetadata": {"timeZone": "Etc/GMT-8"},
+            "assistSkippingMode": "REQUEST_ASSIST"
+        }
+    }
+
+    # 1. 发送 Role
+    if is_stream:
+        chunk = create_chunk(chat_id, created_time, req.model, {"role": "assistant"}, None)
+        yield f"data: {chunk}\n\n"
+
+    # 2. 发起请求 (注意：这里不再使用 stream=True，而是等待完整响应)
+    # 这是为了解决 Google 返回 JSON 数组格式导致的解析错误
+    async with httpx.AsyncClient(proxies=PROXY, verify=False, timeout=120) as cli:
+        r = await cli.post(
+            "https://biz-discoveryengine.googleapis.com/v1alpha/locations/global/widgetStreamAssist",
+            headers=headers,
+            json=body,
+        )
+
+        if r.status_code != 200:
+            logger.error(f"❌ Google 报错: {r.status_code} {r.text}")
+            if is_stream:
+                yield f"data: {json.dumps({'error': {'message': f'Upstream Error {r.status_code}'}})}\n\n"
+            return
+
+        try:
+            # 直接解析整个 JSON 数组
+            data_list = r.json()
+        except Exception as e:
+            logger.error(f"❌ JSON 解析失败: {e}")
+            if is_stream:
+                yield f"data: {json.dumps({'error': {'message': 'JSON Parse Error'}})}\n\n"
+            return
+
+        # 3. 遍历数组，模拟流式推送
+        for data in data_list:
+            for reply in data.get("streamAssistResponse", {}).get("answer", {}).get("replies", []):
+                text = reply.get("groundedContent", {}).get("content", {}).get("text", "")
+                if text and not reply.get("thought"):
+                    chunk = create_chunk(chat_id, created_time, req.model, {"content": text}, None)
+                    if is_stream:
+                        yield f"data: {chunk}\n\n"
+
+    # 4. 发送 Finish Reason
+    if is_stream:
+        final_chunk = create_chunk(chat_id, created_time, req.model, {}, "stop")
+        yield f"data: {final_chunk}\n\n"
+        yield "data: [DONE]\n\n"
+
+if __name__ == "__main__":
+    if not all([SECURE_C_SES, CSESIDX, CONFIG_ID]):
+        print("Error: Missing required environment variables.")
+        exit(1)
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -0,0 +1,4 @@
+fastapi==0.110.0
+uvicorn[standard]==0.29.0
+httpx==0.27.0
+pydantic==2.7.0