deploy: Launch ap3 service on port 7862

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY ./requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt
+
+COPY ./app.py /app/
+
+# Hugging Face 通过下面的 README.md 中的 app_port 来映射端口
+EXPOSE 7862
+
+# 直接运行 uvicorn，监听 7862 端口
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7862"]

README.md

@@ -1,10 +1,14 @@
 ---
-title: Ap3
-emoji: 📚
-colorFrom: purple
-colorTo: blue
+title: GenAI API Service (ap3)
+emoji: 🛰️
+colorFrom: red
+colorTo: yellow
 sdk: docker
-pinned: false
+app_port: 7862
 ---
+This is a professional-grade FastAPI application that proxies requests to the Google GenAI API on port 7862.
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+- **`POST /v1/chat/completions`**: Main endpoint for chat, supports streaming.
+- **`GET /v1/models`**: Lists available models.
+- **`GET /health`**: Health check.
+- **`GET /`**: API Info.

app.py

@@ -0,0 +1,368 @@
+import logging
+import json
+import time
+import asyncio
+import os
+import traceback
+import sys
+from contextlib import asynccontextmanager
+import random
+
+import uvicorn
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import StreamingResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+# 正确的导入方式
+import google.generativeai as genai
+from typing import Optional, List, Dict, Any
+
+# 简化日志配置
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s [%(levelname)s]: %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S'
+)
+logger = logging.getLogger(__name__)
+
+# 模型配置
+GEMINI_MODELS = {
+    "gemini-2.0-flash-exp": "gemini-2.0-flash-exp",
+    "gemini-2.5-flash-preview-05-20": "gemini-2.5-flash-preview-05-20",
+    "gemini-2.5-flash": "gemini-2.5-flash",
+    "gemini-2.5-flash-preview-04-17": "gemini-2.5-flash-preview-04-17"
+}
+
+# 支持的模型列表
+SUPPORTED_MODELS = [
+    {
+        "id": "gemini-2.5-flash-preview-05-20",
+        "object": "model",
+        "created": int(time.time()),
+        "owned_by": "google",
+        "permission": [],
+        "root": "gemini-2.5-flash-preview-05-20",
+        "parent": None,
+        "description": "Gemini 2.5 Flash Preview - 最新实验性模型"
+    },
+    {
+        "id": "gemini-2.5-flash-preview-04-17",
+        "object": "model",
+        "created": int(time.time()),
+        "owned_by": "google",
+        "permission": [],
+        "root": "gemini-2.5-flash-preview-04-17",
+        "parent": None,
+        "description": "gemini-2.5-flash-preview-04-17- 经典专业模型"
+    },
+    {
+        "id": "gemini-2.5-flash",
+        "object": "model",
+        "created": int(time.time()),
+        "owned_by": "google",
+        "permission": [],
+        "root": "gemini-2.5-flash",
+        "parent": None,
+        "description": "gemini-2.5-flash稳定经典专业模型"
+    }
+]
+
+
+def get_model_name(requested_model: str) -> str:
+    """获取实际的Gemini模型名称"""
+    return GEMINI_MODELS.get(requested_model, "gemini-2.5-flash")
+
+
+def convert_messages(messages):
+    content_parts = []
+    system_instruction = None
+
+    for message in messages:
+        role = message.get("role", "user")
+        content = message.get("content", "")
+
+        if role == "system":
+            system_instruction = content
+        elif role == "assistant":
+            content_parts.append({
+                "role": "model",
+                "parts": [{"text": content}]
+            })
+        elif role == "user":
+            content_parts.append({
+                "role": "user",
+                "parts": [{"text": content}]
+            })
+
+    return content_parts, system_instruction
+
+
+def handle_error(error):
+    """简化的错误处理"""
+    error_str = str(error).lower()
+
+    if "prompt_feedback" in error_str:
+        if "other" in error_str:
+            return "您的输入内容可能过长或触发了安全策略。请尝试缩短您的问题。", "length"
+        elif "safety" in error_str:
+            return "您的请求被安全策略阻止。请尝试修改您的问题。", "content_filter"
+    elif "safety" in error_str:
+        return "您的请求被安全策略过滤。请尝试修改您的问题。", "content_filter"
+
+    return "生成内容时遇到错误。请稍后重试。", "stop"
+
+def get_random_api_key():
+    """获取随机API密钥"""
+    return random.choice(API_KEYS)
+
+
+def setup_gemini(api_key=None):
+    """配置Gemini API"""
+    if not api_key:
+        api_key = get_random_api_key()
+
+    if not API_KEYS:
+        logger.error("请设置有效的API密钥列表")
+        raise ValueError("API_KEYS未设置")
+    genai.configure(api_key=api_key)
+    return api_key
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    try:
+        setup_gemini()
+        logger.info("应用启动完成")
+        yield
+    except Exception as e:
+        logger.error(f"应用启动失败: {str(e)}")
+    finally:
+        logger.info("应用关闭")
+
+
+# 创建FastAPI应用实例
+app = FastAPI(
+    lifespan=lifespan,
+    title="Gemini Official API (ap3)",
+    version="1.3.0"
+)
+
+# 添加CORS中间件
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# API密钥列表 (已更新为您提供的新密钥)
+API_KEYS = [
+    'AIzaSyDglvTu5EbXfcBMsv1gBjSH7TOtZxNsQuo',
+    'AIzaSyBOT0mNoemVm-VEwL3YNOAQu_tirwYJDyE'
+]
+
+
+# 配置安全设置
+SAFETY_SETTINGS = [
+    genai.types.SafetySetting(
+        category=genai.types.HarmCategory.HARM_CATEGORY_HARASSMENT,
+        threshold=genai.types.HarmBlockThreshold.BLOCK_NONE,
+    ),
+    genai.types.SafetySetting(
+        category=genai.types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
+        threshold=genai.types.HarmBlockThreshold.BLOCK_NONE,
+    ),
+    genai.types.SafetySetting(
+        category=genai.types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
+        threshold=genai.types.HarmBlockThreshold.BLOCK_NONE,
+    ),
+    genai.types.SafetySetting(
+        category=genai.types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
+        threshold=genai.types.HarmBlockThreshold.BLOCK_NONE,
+    ),
+    genai.types.SafetySetting(
+        category=genai.types.HarmCategory.HARM_CATEGORY_CIVIC_INTEGRITY,
+        threshold=genai.types.HarmBlockThreshold.BLOCK_NONE,
+    ),
+]
+
+
+async def try_generate_content(model_name, content_parts, system_instruction, config, max_retries=3):
+    """带重试机制的内容生成"""
+    last_error = None
+    used_keys = set()
+    model = None
+
+    for attempt in range(max_retries):
+        try:
+            available_keys = [key for key in API_KEYS if key not in used_keys]
+            if not available_keys:
+                used_keys.clear()
+                available_keys = API_KEYS
+
+            api_key = random.choice(available_keys)
+            used_keys.add(api_key)
+            
+            setup_gemini(api_key)
+            model = genai.GenerativeModel(model_name, system_instruction=system_instruction)
+            logger.info(f"尝试第 {attempt + 1} 次，使用密钥: {api_key[:10]}...")
+
+            response = await model.generate_content_async(
+                contents=content_parts,
+                generation_config=config,
+                safety_settings=SAFETY_SETTINGS,
+            )
+
+            return response, api_key
+
+        except Exception as e:
+            last_error = e
+            error_str = str(e).lower()
+
+            if any(code in error_str for code in ['400', '401', '403', '429', '500', '502', '503', '504']):
+                logger.warning(f"第 {attempt + 1} 次尝试失败: {str(e)}")
+                if attempt < max_retries - 1:
+                    await asyncio.sleep(1)
+                    continue
+            else:
+                raise e
+
+    raise last_error
+
+
+@app.post("/v1/chat/completions")
+async def chat_completions(request: Request):
+    """聊天对话接口"""
+    try:
+        body = await request.json()
+        messages = body.get('messages', [])
+        stream = body.get('stream', False)
+        
+        requested_model = body.get('model', 'gemini-2.5-flash')
+        model_name = get_model_name(requested_model)
+
+        content_parts, system_instruction = convert_messages(messages)
+
+        config = genai.types.GenerationConfig(
+            max_output_tokens=body.get('max_tokens', 65536),
+            temperature=body.get('temperature', 1.0),
+            top_p=body.get('top_p', 1.0),
+        )
+
+        if stream:
+            return StreamingResponse(
+                stream_response_with_retry(model_name, content_parts, system_instruction, config),
+                media_type='text/event-stream'
+            )
+        else:
+            response, used_key = await try_generate_content(model_name, content_parts, system_instruction, config)
+            response_text = response.text if response else ""
+            finish_reason = "stop"
+
+            if not response_text:
+                response_text = "无法生成回复。请尝试修改您的问题。"
+
+            logger.info(f"成功生成回复，使用密钥: {used_key[:10]}...")
+            return {
+                'id': f'chatcmpl-{int(time.time())}-{random.randint(1000, 9999)}',
+                'object': 'chat.completion', 'created': int(time.time()), 'model': requested_model,
+                'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': response_text}, 'finish_reason': finish_reason}],
+                'usage': {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}
+            }
+    except Exception as e:
+        logger.error(f"处理聊天请求出错: {traceback.format_exc()}")
+        error_message, finish_reason = handle_error(e)
+        raise HTTPException(status_code=500, detail=error_message)
+
+
+async def stream_response_with_retry(model_name, content_parts, system_instruction, config, max_retries=3):
+    """带重试机制的流式响应生成器"""
+    last_error = None
+    used_keys = set()
+    for attempt in range(max_retries):
+        try:
+            available_keys = [key for key in API_KEYS if key not in used_keys]
+            if not available_keys:
+                used_keys.clear()
+                available_keys = API_KEYS
+            api_key = random.choice(available_keys)
+            used_keys.add(api_key)
+
+            setup_gemini(api_key)
+            model = genai.GenerativeModel(model_name, system_instruction=system_instruction)
+            logger.info(f"流式响应尝试第 {attempt + 1} 次，使用密钥: {api_key[:10]}...")
+            
+            async for chunk in await model.generate_content_async(contents=content_parts, generation_config=config, safety_settings=SAFETY_SETTINGS, stream=True):
+                if chunk and hasattr(chunk, 'text') and chunk.text:
+                    data = {'id': f'chatcmpl-{int(time.time())}', 'object': 'chat.completion.chunk', 'created': int(time.time()), 'model': model_name, 'choices': [{'index': 0, 'delta': {'role': 'assistant', 'content': chunk.text}, 'finish_reason': None}]}
+                    yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
+            
+            final_data = {'id': f'chatcmpl-{int(time.time())}', 'object': 'chat.completion.chunk', 'created': int(time.time()), 'model': model_name, 'choices': [{'index': 0, 'delta': {}, 'finish_reason': 'stop'}]}
+            yield f'data: {json.dumps(final_data, ensure_ascii=False)}\n\n'
+            yield 'data: [DONE]\n\n'
+            
+            logger.info(f"流式响应成功，使用密钥: {api_key[:10]}...")
+            return
+        except Exception as e:
+            last_error = e
+            error_str = str(e).lower()
+            if any(code in error_str for code in ['400', '401', '403', '429', '500', '502', '503', '504']):
+                logger.warning(f"流式响应第 {attempt + 1} 次尝试失败: {str(e)}")
+                if attempt < max_retries - 1:
+                    await asyncio.sleep(1)
+                    continue
+            else:
+                break
+    logger.error(f"流式响应所有重试失败: {traceback.format_exc()}")
+    error_message, finish_reason = handle_error(last_error)
+    error_data = {'id': f'chatcmpl-{int(time.time())}-error', 'object': 'chat.completion.chunk', 'created': int(time.time()), 'model': model_name, 'choices': [{'index': 0, 'delta': {'role': 'assistant', 'content': error_message}, 'finish_reason': finish_reason}]}
+    yield f'data: {json.dumps(error_data, ensure_ascii=False)}\n\n'
+    yield 'data: [DONE]\n\n'
+
+
+@app.get("/v1/models")
+async def list_models():
+    return {"object": "list", "data": SUPPORTED_MODELS}
+
+@app.get("/v1/models/{model_id}")
+async def get_model_info(model_id: str):
+    for model in SUPPORTED_MODELS:
+        if model["id"] == model_id:
+            return model
+    raise HTTPException(status_code=404, detail=f"模型 {model_id} 未找到")
+
+@app.get("/v1/chat/completions/v1/models")
+async def list_models_alternative():
+    return {"object": "list", "data": SUPPORTED_MODELS}
+
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy", "timestamp": int(time.time()), "api": "gemini-official", "available_models": [model["id"] for model in SUPPORTED_MODELS], "version": "1.3.0"}
+
+@app.get("/")
+async def root():
+    return {"name": "Gemini Official API (ap3)", "version": "1.3.0", "description": "Google Gemini官方API接口服务", "endpoints": {"models": "/v1/models", "models_alt": "/v1/chat/completions/v1/models", "chat": "/v1/chat/completions", "health": "/health"}}
+
+@app.exception_handler(404)
+async def not_found_handler(request: Request, exc: HTTPException):
+    """处理404错误"""
+    content = {
+        "error": "未找到",
+        "requested_path": str(request.url.path),
+        "message": "请求的路径不存在",
+        "available_endpoints": {
+            "models": "/v1/models",
+            "models_alt": "/v1/chat/completions/v1/models",
+            "chat": "/v1/chat/completions",
+            "health": "/health",
+            "info": "/"
+        }
+    }
+    return JSONResponse(status_code=404, content=content)
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 7862))
+    print(f"🚀 启动Gemini官方API服务器于端口 {port}")
+    print(f"📊 支持的模型: {[model['id'] for model in SUPPORTED_MODELS]}")
+    print(f"🔑 已配置 {len(API_KEYS)} 个API密钥")
+    print("🔄 支持自动重试和密钥轮换")
+    uvicorn.run("app:app", host="0.0.0.0", port=port, reload=True)

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn
+google-genai
+loguru
+httpx