mirror of
https://github.com/xtekky/gpt4free.git
synced 2026-04-22 15:47:11 +08:00
copilot-swe-agent[bot]
dafc1d9341
Agent-Logs-Url: https://github.com/xtekky/gpt4free/sessions/d964c73c-85bc-432d-b14b-20f8ea2de94e Co-authored-by: hlohaus <983577+hlohaus@users.noreply.github.com>
1296 lines
58 KiB
Python
1296 lines
58 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import json
|
|
import uvicorn
|
|
import secrets
|
|
import os
|
|
import re
|
|
import shutil
|
|
import time
|
|
from email.utils import formatdate
|
|
import os.path
|
|
import hashlib
|
|
import base64
|
|
from contextlib import asynccontextmanager
|
|
from urllib.parse import quote_plus
|
|
from fastapi import FastAPI, Response, Request, UploadFile, Form, Depends, Header
|
|
from fastapi.responses import StreamingResponse, RedirectResponse, HTMLResponse, JSONResponse, FileResponse
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.security import APIKeyHeader
|
|
from starlette.exceptions import HTTPException
|
|
from starlette.status import (
|
|
HTTP_200_OK,
|
|
HTTP_422_UNPROCESSABLE_ENTITY,
|
|
HTTP_404_NOT_FOUND,
|
|
HTTP_401_UNAUTHORIZED,
|
|
HTTP_403_FORBIDDEN,
|
|
HTTP_429_TOO_MANY_REQUESTS,
|
|
HTTP_500_INTERNAL_SERVER_ERROR,
|
|
)
|
|
from starlette.staticfiles import NotModifiedResponse
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, HTTPBasic
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from starlette.responses import FileResponse
|
|
from starlette.background import BackgroundTask
|
|
try:
|
|
from a2wsgi import WSGIMiddleware
|
|
has_a2wsgi = True
|
|
except ImportError:
|
|
has_a2wsgi = False
|
|
try:
|
|
from PIL import Image
|
|
has_pillow = True
|
|
except ImportError:
|
|
has_pillow = False
|
|
from types import SimpleNamespace
|
|
from typing import Union, Optional, List
|
|
|
|
try:
|
|
from typing import Annotated
|
|
except ImportError:
|
|
class Annotated:
|
|
pass
|
|
try:
|
|
from zendriver import util
|
|
has_nodriver = True
|
|
except ImportError:
|
|
has_nodriver = False
|
|
|
|
import g4f
|
|
import g4f.debug
|
|
from g4f.client import AsyncClient, ChatCompletion, ImagesResponse
|
|
from g4f.providers.response import BaseConversation, JsonConversation
|
|
from g4f.client.helper import filter_none
|
|
from g4f.config import DEFAULT_PORT, DEFAULT_TIMEOUT, DEFAULT_STREAM_TIMEOUT
|
|
from g4f.image import EXTENSIONS_MAP, is_data_an_media, process_image
|
|
from g4f.image.copy_images import get_media_dir, copy_media, get_source_url
|
|
from g4f.errors import ProviderNotFoundError, ModelNotFoundError, MissingAuthError, NoValidHarFileError, MissingRequirementsError, RateLimitError
|
|
from g4f.cookies import read_cookie_files, get_cookies_dir
|
|
from g4f.providers.types import ProviderType
|
|
from g4f.providers.response import AudioResponse
|
|
from g4f.providers.any_provider import AnyProvider
|
|
from g4f.providers.any_model_map import model_map, vision_models, image_models, audio_models, video_models
|
|
from g4f.config import AppConfig
|
|
from g4f import Provider
|
|
from g4f.Provider import ProviderUtils
|
|
|
|
from g4f.gui import get_gui_app
|
|
from .stubs import (
|
|
ChatCompletionsConfig, ImageGenerationConfig,
|
|
ProviderResponseModel, ModelResponseModel,
|
|
ErrorResponseModel, ProviderResponseDetailModel,
|
|
FileResponseModel,
|
|
TranscriptionResponseModel, AudioSpeechConfig
|
|
)
|
|
from g4f import debug
|
|
|
|
try:
|
|
from g4f.gui.server.crypto import create_or_read_keys, decrypt_data, get_session_key
|
|
has_crypto = True
|
|
except ImportError:
|
|
has_crypto = False
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# Read cookie files if not ignored
|
|
if not AppConfig.ignore_cookie_files:
|
|
read_cookie_files()
|
|
else:
|
|
AppConfig.load_from_env()
|
|
yield
|
|
if has_nodriver:
|
|
for browser in util.get_registered_instances():
|
|
if browser.connection:
|
|
await browser.stop()
|
|
lock_file = os.path.join(get_cookies_dir(), ".browser_is_open")
|
|
if os.path.exists(lock_file):
|
|
try:
|
|
os.remove(lock_file)
|
|
except Exception as e:
|
|
debug.error(f"Failed to remove lock file {lock_file}:", e)
|
|
|
|
def create_app():
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
# Add CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
expose_headers=["*"],
|
|
)
|
|
|
|
api = Api(app)
|
|
|
|
api.register_routes()
|
|
api.register_authorization()
|
|
api.register_validation_exception_handler()
|
|
|
|
if AppConfig.gui:
|
|
if not has_a2wsgi:
|
|
raise MissingRequirementsError("a2wsgi is required for GUI. Install it with: pip install a2wsgi")
|
|
gui_app = WSGIMiddleware(get_gui_app(AppConfig.demo, AppConfig.timeout, AppConfig.stream_timeout))
|
|
app.mount("/", gui_app)
|
|
|
|
if AppConfig.ignored_providers:
|
|
for provider in AppConfig.ignored_providers:
|
|
if provider in ProviderUtils.convert:
|
|
ProviderUtils.convert[provider].working = False
|
|
|
|
return app
|
|
|
|
def create_app_debug():
|
|
g4f.debug.logging = True
|
|
return create_app()
|
|
|
|
def create_app_with_gui_and_debug():
|
|
g4f.debug.logging = True
|
|
AppConfig.gui = True
|
|
return create_app()
|
|
|
|
def create_app_with_demo_and_debug():
|
|
g4f.debug.logging = True
|
|
AppConfig.gui = True
|
|
AppConfig.demo = True
|
|
return create_app()
|
|
|
|
class ErrorResponse(Response):
|
|
media_type = "application/json"
|
|
|
|
@classmethod
|
|
def from_exception(cls, exception: Exception,
|
|
config: Union[ChatCompletionsConfig, ImageGenerationConfig] = None,
|
|
status_code: int = HTTP_500_INTERNAL_SERVER_ERROR):
|
|
return cls(format_exception(exception, config), status_code)
|
|
|
|
@classmethod
|
|
def from_message(cls, message: str, status_code: int = HTTP_500_INTERNAL_SERVER_ERROR, headers: dict = None):
|
|
return cls(format_exception(message), status_code, headers=headers)
|
|
|
|
def render(self, content) -> bytes:
|
|
return str(content).encode(errors="ignore")
|
|
|
|
def update_headers(request: Request, new_api_key: str = None, user: str = None) -> Request:
|
|
new_headers = request.headers.mutablecopy()
|
|
if new_api_key:
|
|
new_headers["authorization"] = f"Bearer {new_api_key}"
|
|
else:
|
|
del new_headers["authorization"]
|
|
if user:
|
|
new_headers["x-user"] = user
|
|
request.scope["headers"] = new_headers.raw
|
|
delattr(request, "_headers")
|
|
return request
|
|
|
|
class Api:
|
|
def __init__(self, app: FastAPI) -> None:
|
|
self.app = app
|
|
self.client = AsyncClient()
|
|
self.get_g4f_api_key = APIKeyHeader(name="g4f-api-key")
|
|
self.conversations: dict[str, dict[str, BaseConversation]] = {}
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
basic_security = HTTPBasic()
|
|
|
|
async def get_username(self, request: Request) -> str:
|
|
credentials = await self.basic_security(request)
|
|
current_password_bytes = credentials.password.encode()
|
|
is_correct_password = secrets.compare_digest(
|
|
current_password_bytes, AppConfig.g4f_api_key.encode()
|
|
)
|
|
if not is_correct_password:
|
|
raise HTTPException(
|
|
status_code=HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password",
|
|
headers={"WWW-Authenticate": "Basic"},
|
|
)
|
|
return credentials.username
|
|
|
|
def register_authorization(self):
|
|
if AppConfig.g4f_api_key:
|
|
print("Register authentication key:", ''.join(['*' for _ in range(len(AppConfig.g4f_api_key))]))
|
|
if has_crypto:
|
|
private_key, _ = create_or_read_keys()
|
|
session_key = get_session_key()
|
|
|
|
def _requires_api_key(path: str, demo: bool) -> bool:
|
|
"""Return ``True`` when *path* must present a G4F API key."""
|
|
return (
|
|
path.startswith("/v1")
|
|
or path.startswith("/api/")
|
|
or path.startswith("/pa/")
|
|
or (demo and path == "/backend-api/v2/upload_cookies")
|
|
)
|
|
|
|
@self.app.middleware("http")
|
|
async def authorization(request: Request, call_next):
|
|
user = None
|
|
if request.method != "OPTIONS" and AppConfig.g4f_api_key is not None or AppConfig.demo:
|
|
update_authorization = False
|
|
try:
|
|
user_g4f_api_key = await self.get_g4f_api_key(request)
|
|
except HTTPException:
|
|
user_g4f_api_key = getattr(await self.security(request), "credentials", None)
|
|
update_authorization = True
|
|
if user_g4f_api_key:
|
|
user_g4f_api_key = user_g4f_api_key.split()
|
|
country = request.headers.get("Cf-Ipcountry", "")
|
|
if AppConfig.demo and user is None:
|
|
ip = request.headers.get("X-Forwarded-For", "")[:4].strip(":.")
|
|
user = request.headers.get("x-user", ip)
|
|
user = f"{country}:{user}" if country else user
|
|
if AppConfig.g4f_api_key is None or not user_g4f_api_key or not secrets.compare_digest(AppConfig.g4f_api_key, user_g4f_api_key[0]):
|
|
if has_crypto and user_g4f_api_key:
|
|
try:
|
|
expires, user = decrypt_data(private_key, user_g4f_api_key[0]).split(":", 1)
|
|
except Exception:
|
|
try:
|
|
data = json.loads(decrypt_data(session_key, user_g4f_api_key[0]))
|
|
debug.log(f"Decrypted G4F API key data: {data}")
|
|
expires = int(decrypt_data(private_key, data.pop("data"))) + 86400
|
|
user = data.get("user", user)
|
|
if not user or "referrer" not in data:
|
|
raise ValueError("User not found")
|
|
except Exception:
|
|
return ErrorResponse.from_message(f"Invalid G4F API key", HTTP_401_UNAUTHORIZED)
|
|
user = f"{country}:{user}" if country else user
|
|
expires = int(expires) - int(time.time())
|
|
hours, remainder = divmod(expires, 3600)
|
|
minutes, seconds = divmod(remainder, 60)
|
|
if expires < 0:
|
|
debug.log(f"G4F API key expired for user '{user}'")
|
|
return ErrorResponse.from_message("G4F API key expired", HTTP_401_UNAUTHORIZED)
|
|
count = 0
|
|
for char in user:
|
|
if char.isupper():
|
|
count += 1
|
|
if count >= 6:
|
|
debug.log(f"Invalid user name (screaming): '{user}'")
|
|
return ErrorResponse.from_message("Invalid user name (screaming)", HTTP_401_UNAUTHORIZED)
|
|
debug.log(f"User: '{user}' G4F API key expires in {hours}h {minutes}m {seconds}s")
|
|
else:
|
|
user = "admin"
|
|
path = request.url.path
|
|
if _requires_api_key(path, AppConfig.demo):
|
|
if request.method != "OPTIONS" and not path.endswith("/models"):
|
|
if not user_g4f_api_key:
|
|
return ErrorResponse.from_message("G4F API key required", HTTP_401_UNAUTHORIZED)
|
|
if AppConfig.g4f_api_key is None and user is None:
|
|
return ErrorResponse.from_message("Invalid G4F API key", HTTP_403_FORBIDDEN)
|
|
elif not AppConfig.demo and not path.startswith("/images/") and not path.startswith("/media/"):
|
|
if user_g4f_api_key:
|
|
if user is None:
|
|
return ErrorResponse.from_message("Invalid G4F API key", HTTP_403_FORBIDDEN)
|
|
elif path.startswith("/backend-api/") or path.startswith("/chat/"):
|
|
try:
|
|
user = await self.get_username(request)
|
|
except HTTPException as e:
|
|
return ErrorResponse.from_message(e.detail, e.status_code, e.headers)
|
|
if user_g4f_api_key and update_authorization:
|
|
new_api_key = user_g4f_api_key.pop()
|
|
if secrets.compare_digest(AppConfig.g4f_api_key, new_api_key):
|
|
new_api_key = None
|
|
else:
|
|
new_api_key = None
|
|
request = update_headers(request, new_api_key, user)
|
|
response = await call_next(request)
|
|
return response
|
|
|
|
def register_validation_exception_handler(self):
|
|
@self.app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
details = exc.errors()
|
|
modified_details = []
|
|
for error in details:
|
|
debug.log(f"Validation error: {error['loc']} - {error['msg']} ({error['type']})")
|
|
modified_details.append({
|
|
"loc": error["loc"],
|
|
"message": error["msg"],
|
|
"type": error["type"],
|
|
})
|
|
return JSONResponse(
|
|
status_code=HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content=jsonable_encoder({"detail": modified_details}),
|
|
)
|
|
|
|
def register_routes(self):
|
|
if not AppConfig.gui:
|
|
@self.app.get("/")
|
|
async def read_root():
|
|
return RedirectResponse("/v1", 302)
|
|
|
|
@self.app.get("/v1")
|
|
async def read_root_v1():
|
|
return HTMLResponse('g4f API: Go to '
|
|
'<a href="/v1/models">models</a>, '
|
|
'<a href="/v1/chat/completions">chat/completions</a>, or '
|
|
'<a href="/v1/media/generate">media/generate</a> <br><br>'
|
|
'Open Swagger UI at: '
|
|
'<a href="/docs">/docs</a>')
|
|
|
|
@self.app.get("/v1/models", responses={
|
|
HTTP_200_OK: {"model": List[ModelResponseModel]},
|
|
})
|
|
async def models():
|
|
return {
|
|
"object": "list",
|
|
"data": [{
|
|
"id": model,
|
|
"object": "model",
|
|
"created": 0,
|
|
"owned_by": "",
|
|
"image": isinstance(model, g4f.models.ImageModel),
|
|
"vision": isinstance(model, g4f.models.VisionModel),
|
|
"provider": False,
|
|
} for model in AnyProvider.get_models()] +
|
|
[{
|
|
"id": provider_name,
|
|
"object": "model",
|
|
"created": 0,
|
|
"owned_by": getattr(provider, "label", ""),
|
|
"image": bool(getattr(provider, "image_models", False)),
|
|
"vision": bool(getattr(provider, "vision_models", False)),
|
|
"provider": True,
|
|
} for provider_name, provider in ProviderUtils.convert.items()
|
|
if provider.working and provider_name not in ("Custom")
|
|
]
|
|
}
|
|
|
|
@self.app.get("/api/{provider}/models", responses={
|
|
HTTP_200_OK: {"model": List[ModelResponseModel]},
|
|
})
|
|
async def models(provider: str, credentials: Annotated[HTTPAuthorizationCredentials, Depends(Api.security)] = None):
|
|
try:
|
|
provider = ProviderUtils.get_by_label(provider)
|
|
except ValueError as e:
|
|
if provider in model_map:
|
|
return {
|
|
"object": "list",
|
|
"data": [{
|
|
"id": provider,
|
|
"object": "model",
|
|
"created": 0,
|
|
"owned_by": provider,
|
|
"image": provider in image_models,
|
|
"vision": provider in vision_models,
|
|
"audio": provider in audio_models,
|
|
"video": provider in video_models,
|
|
"type": "image" if provider in image_models else "chat",
|
|
}]
|
|
}
|
|
return ErrorResponse.from_message(str(e), 404)
|
|
if not hasattr(provider, "get_models"):
|
|
models = []
|
|
elif credentials is not None and credentials.credentials != "secret":
|
|
models = provider.get_models(api_key=credentials.credentials)
|
|
else:
|
|
models = provider.get_models()
|
|
return {
|
|
"object": "list",
|
|
"data": [{
|
|
"id": model.get("id") if isinstance(model, dict) else model,
|
|
"object": "model",
|
|
"created": 0,
|
|
"owned_by": getattr(provider, "label", provider.__name__),
|
|
"image": (model.get("id") if isinstance(model, dict) else model) in getattr(provider, "image_models", []),
|
|
"vision": (model.get("id") if isinstance(model, dict) else model) in getattr(provider, "vision_models", []),
|
|
"audio": (model.get("id") if isinstance(model, dict) else model) in getattr(provider, "audio_models", []),
|
|
"video": (model.get("id") if isinstance(model, dict) else model) in getattr(provider, "video_models", []),
|
|
"type": "image" if (model.get("id") if isinstance(model, dict) else model) in getattr(provider, "image_models", []) else "chat",
|
|
**(model if isinstance(model, dict) else {})
|
|
} for model in (models.values() if isinstance(models, dict) else models)]
|
|
}
|
|
|
|
# quota endpoint mimics backend-api/v2/quota but exposed on public API
|
|
@self.app.get("/api/{provider}/quota")
|
|
async def provider_quota(provider: str, credentials: Annotated[HTTPAuthorizationCredentials, Depends(Api.security)] = None):
|
|
try:
|
|
provider = ProviderUtils.get_by_label(provider)
|
|
except ValueError as e:
|
|
return ErrorResponse.from_message(str(e), 404)
|
|
if not hasattr(provider, "get_quota"):
|
|
return ErrorResponse.from_message("Provider doesn't support get_quota", HTTP_500_INTERNAL_SERVER_ERROR)
|
|
try:
|
|
if credentials is not None and credentials.credentials != "secret":
|
|
usage = await provider.get_quota(api_key=credentials.credentials)
|
|
else:
|
|
usage = await provider.get_quota()
|
|
return usage
|
|
except MissingAuthError as e:
|
|
return ErrorResponse.from_message(f"{type(e).__name__}: {e}", HTTP_401_UNAUTHORIZED)
|
|
except Exception as e:
|
|
return ErrorResponse.from_message(f"{type(e).__name__}: {e}", HTTP_500_INTERNAL_SERVER_ERROR)
|
|
|
|
@self.app.get("/v1/models/{model_name}", responses={
|
|
HTTP_200_OK: {"model": ModelResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
})
|
|
@self.app.post("/v1/models/{model_name}", responses={
|
|
HTTP_200_OK: {"model": ModelResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
})
|
|
async def model_info(model_name: str) -> ModelResponseModel:
|
|
if model_name in g4f.models.ModelUtils.convert:
|
|
model_info = g4f.models.ModelUtils.convert[model_name]
|
|
return JSONResponse({
|
|
'id': model_name,
|
|
'object': 'model',
|
|
'created': 0,
|
|
'owned_by': model_info.base_provider
|
|
})
|
|
return ErrorResponse.from_message("The model does not exist.", HTTP_404_NOT_FOUND)
|
|
|
|
responses = {
|
|
HTTP_200_OK: {"model": ChatCompletion},
|
|
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
HTTP_422_UNPROCESSABLE_ENTITY: {"model": ErrorResponseModel},
|
|
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
|
|
}
|
|
@self.app.post("/v1/chat/completions", responses=responses)
|
|
@self.app.post("/api/{provider}/chat/completions", responses=responses)
|
|
@self.app.post("/api/{provider}/{conversation_id}/chat/completions", responses=responses)
|
|
async def chat_completions(
|
|
config: ChatCompletionsConfig,
|
|
credentials: Annotated[HTTPAuthorizationCredentials, Depends(Api.security)] = None,
|
|
provider: str = None,
|
|
conversation_id: str = None,
|
|
x_user: Annotated[str | None, Header()] = None,
|
|
):
|
|
if provider is None:
|
|
provider = config.provider
|
|
if provider is None:
|
|
provider = AppConfig.provider
|
|
try:
|
|
provider = ProviderUtils.get_by_label(provider).__name__
|
|
except ValueError as e:
|
|
if provider in model_map:
|
|
config.model = provider
|
|
provider = None
|
|
elif provider is not None:
|
|
return ErrorResponse.from_message(str(e), 404)
|
|
try:
|
|
config.provider = provider
|
|
if config.conversation_id is None:
|
|
config.conversation_id = conversation_id
|
|
if config.timeout is None:
|
|
config.timeout = AppConfig.timeout
|
|
if config.stream_timeout is None and config.stream:
|
|
config.stream_timeout = AppConfig.stream_timeout
|
|
if credentials is not None and credentials.credentials != "secret":
|
|
config.api_key = credentials.credentials
|
|
|
|
conversation = config.conversation
|
|
if conversation:
|
|
conversation = JsonConversation(**conversation)
|
|
elif config.conversation_id is not None and config.provider is not None:
|
|
if config.conversation_id in self.conversations:
|
|
if config.provider in self.conversations[config.conversation_id]:
|
|
conversation = self.conversations[config.conversation_id][config.provider]
|
|
|
|
if config.image is not None:
|
|
try:
|
|
is_data_an_media(config.image)
|
|
except ValueError as e:
|
|
return ErrorResponse.from_message(f"The image you send must be a data URI. Example: data:image/jpeg;base64,...", status_code=HTTP_422_UNPROCESSABLE_ENTITY)
|
|
if config.media is None:
|
|
config.media = config.images
|
|
if config.media is not None:
|
|
for image in config.media:
|
|
try:
|
|
is_data_an_media(image[0], image[1])
|
|
except ValueError as e:
|
|
example = json.dumps({"media": [["data:image/jpeg;base64,...", "filename.jpg"]]})
|
|
return ErrorResponse.from_message(f'The media you send must be a data URIs. Example: {example}', status_code=HTTP_422_UNPROCESSABLE_ENTITY)
|
|
|
|
# Create the completion response
|
|
response = self.client.chat.completions.create(
|
|
**filter_none(
|
|
**{
|
|
"model": AppConfig.model,
|
|
"provider": AppConfig.provider,
|
|
"proxy": AppConfig.proxy,
|
|
**(config.model_dump(exclude_none=True) if hasattr(config, "model_dump") else config.dict(exclude_none=True)),
|
|
**{
|
|
"conversation_id": None,
|
|
"conversation": conversation,
|
|
"user": x_user,
|
|
}
|
|
},
|
|
ignored=AppConfig.ignored_providers
|
|
),
|
|
)
|
|
|
|
if not config.stream:
|
|
return await response
|
|
|
|
async def streaming():
|
|
try:
|
|
async for chunk in response:
|
|
if isinstance(chunk, BaseConversation):
|
|
if config.conversation_id is not None and config.provider is not None:
|
|
if config.conversation_id not in self.conversations:
|
|
self.conversations[config.conversation_id] = {}
|
|
self.conversations[config.conversation_id][config.provider] = chunk
|
|
else:
|
|
yield f"data: {chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}\n\n"
|
|
except GeneratorExit:
|
|
pass
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
yield f'data: {format_exception(e, config)}\n\n'
|
|
yield "data: [DONE]\n\n"
|
|
|
|
return StreamingResponse(streaming(), media_type="text/event-stream")
|
|
|
|
except (ModelNotFoundError, ProviderNotFoundError) as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
|
|
except (MissingAuthError, NoValidHarFileError) as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_500_INTERNAL_SERVER_ERROR)
|
|
|
|
responses = {
|
|
HTTP_200_OK: {"model": ImagesResponse},
|
|
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
|
|
}
|
|
@self.app.post("/v1/media/generate", responses=responses)
|
|
@self.app.post("/v1/images/generate", responses=responses)
|
|
@self.app.post("/v1/images/generations", responses=responses)
|
|
@self.app.post("/api/{provider}/images/generations", responses=responses)
|
|
async def generate_image(
|
|
request: Request,
|
|
config: ImageGenerationConfig,
|
|
provider: str = None,
|
|
credentials: Annotated[HTTPAuthorizationCredentials, Depends(Api.security)] = None
|
|
):
|
|
if provider is None:
|
|
provider = config.provider
|
|
if provider is None:
|
|
provider = AppConfig.provider
|
|
try:
|
|
provider = ProviderUtils.get_by_label(provider)
|
|
except ValueError as e:
|
|
if provider in model_map:
|
|
config.model = provider
|
|
provider = None
|
|
elif provider is not None:
|
|
return ErrorResponse.from_message(str(e), 404)
|
|
config.provider = provider
|
|
if config.api_key is None and credentials is not None and credentials.credentials != "secret":
|
|
config.api_key = credentials.credentials
|
|
try:
|
|
response = await self.client.images.generate(
|
|
**config.dict(exclude_none=True),
|
|
)
|
|
for image in response.data:
|
|
if hasattr(image, "url") and image.url.startswith("/"):
|
|
image.url = f"{request.base_url}{image.url.lstrip('/')}"
|
|
return response
|
|
except (ModelNotFoundError, ProviderNotFoundError) as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
|
|
except MissingAuthError as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_500_INTERNAL_SERVER_ERROR)
|
|
|
|
@self.app.get("/v1/providers", responses={
|
|
HTTP_200_OK: {"model": List[ProviderResponseModel]},
|
|
})
|
|
async def providers():
|
|
return [{
|
|
'id': provider.__name__,
|
|
'object': 'provider',
|
|
'created': 0,
|
|
'url': provider.url,
|
|
'label': getattr(provider, "label", None),
|
|
} for provider in Provider.__providers__ if provider.working]
|
|
|
|
@self.app.get("/v1/providers/{provider}", responses={
|
|
HTTP_200_OK: {"model": ProviderResponseDetailModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
})
|
|
async def providers_info(provider: str):
|
|
try:
|
|
provider = ProviderUtils.get_by_label(provider)
|
|
except ValueError as e:
|
|
return ErrorResponse.from_message(str(e), 404)
|
|
def safe_get_models(provider: ProviderType) -> list[str]:
|
|
try:
|
|
return provider.get_models() if hasattr(provider, "get_models") else []
|
|
except Exception:
|
|
return []
|
|
return {
|
|
'id': provider.__name__,
|
|
'object': 'provider',
|
|
'created': 0,
|
|
'url': provider.url,
|
|
'label': getattr(provider, "label", None),
|
|
'models': safe_get_models(provider),
|
|
'image_models': getattr(provider, "image_models", []) or [],
|
|
'vision_models': [model for model in [getattr(provider, "default_vision_model", None)] if model],
|
|
'params': [*provider.get_parameters()] if hasattr(provider, "get_parameters") else []
|
|
}
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# PA Provider routes #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
@self.app.get("/pa/providers", responses={
|
|
HTTP_200_OK: {},
|
|
})
|
|
async def pa_providers_list():
|
|
"""List all PA providers loaded from the workspace.
|
|
|
|
Filenames are never exposed; each provider is identified by a
|
|
stable opaque ID (SHA-256 of the path, first 8 hex chars).
|
|
"""
|
|
from g4f.mcp.pa_provider import get_pa_registry
|
|
return get_pa_registry().list_providers()
|
|
|
|
@self.app.get("/pa/providers/{provider_id}", responses={
|
|
HTTP_200_OK: {},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
})
|
|
async def pa_providers_detail(provider_id: str):
|
|
"""Get details for a single PA provider by its opaque ID."""
|
|
from g4f.mcp.pa_provider import get_pa_registry
|
|
info = get_pa_registry().get_provider_info(provider_id)
|
|
if info is None:
|
|
return ErrorResponse.from_message(
|
|
f"PA provider '{provider_id}' not found", HTTP_404_NOT_FOUND
|
|
)
|
|
return info
|
|
|
|
responses_pa = {
|
|
HTTP_200_OK: {"model": ChatCompletion},
|
|
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
HTTP_422_UNPROCESSABLE_ENTITY: {"model": ErrorResponseModel},
|
|
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
|
|
}
|
|
|
|
@self.app.post("/pa/chat/completions", responses=responses_pa)
|
|
@self.app.post("/pa/{provider_id}/chat/completions", responses=responses_pa)
|
|
async def pa_chat_completions(
|
|
config: ChatCompletionsConfig,
|
|
credentials: Annotated[HTTPAuthorizationCredentials, Depends(Api.security)] = None,
|
|
provider_id: str = None,
|
|
):
|
|
"""OpenAI-compatible chat completions endpoint backed by PA providers.
|
|
|
|
The PA provider is identified by its opaque ID either from the URL
|
|
path (``/pa/{provider_id}/chat/completions``) or from the ``provider``
|
|
field in the JSON body. When both are absent the first available PA
|
|
provider is used.
|
|
"""
|
|
from g4f.mcp.pa_provider import get_pa_registry
|
|
|
|
registry = get_pa_registry()
|
|
pid = provider_id or config.provider
|
|
if pid is None:
|
|
listing = registry.list_providers()
|
|
if not listing:
|
|
return ErrorResponse.from_message(
|
|
"No PA providers found in workspace", HTTP_404_NOT_FOUND
|
|
)
|
|
pid = listing[0]["id"]
|
|
|
|
provider_cls = registry.get_provider_class(pid)
|
|
if provider_cls is None:
|
|
return ErrorResponse.from_message(
|
|
f"PA provider '{pid}' not found", HTTP_404_NOT_FOUND
|
|
)
|
|
|
|
try:
|
|
config.provider = None # pass the class directly below
|
|
if credentials is not None and credentials.credentials != "secret":
|
|
config.api_key = credentials.credentials
|
|
|
|
response = self.client.chat.completions.create(
|
|
**filter_none(
|
|
**(
|
|
config.model_dump(exclude_none=True)
|
|
if hasattr(config, "model_dump")
|
|
else config.dict(exclude_none=True)
|
|
),
|
|
**{
|
|
"conversation_id": None,
|
|
"provider": provider_cls,
|
|
},
|
|
),
|
|
)
|
|
|
|
if not config.stream:
|
|
return await response
|
|
|
|
async def streaming():
|
|
try:
|
|
async for chunk in response:
|
|
if not isinstance(chunk, BaseConversation):
|
|
yield (
|
|
f"data: "
|
|
f"{chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}"
|
|
f"\n\n"
|
|
)
|
|
except GeneratorExit:
|
|
pass
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
yield f"data: {format_exception(e, config)}\n\n"
|
|
yield "data: [DONE]\n\n"
|
|
|
|
return StreamingResponse(streaming(), media_type="text/event-stream")
|
|
|
|
except (ModelNotFoundError, ProviderNotFoundError) as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
|
|
except (MissingAuthError, NoValidHarFileError) as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, config, HTTP_500_INTERNAL_SERVER_ERROR)
|
|
|
|
@self.app.post("/pa/backend-api/v2/conversation", responses={
|
|
HTTP_200_OK: {},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
HTTP_422_UNPROCESSABLE_ENTITY: {"model": ErrorResponseModel},
|
|
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
|
|
})
|
|
async def pa_backend_conversation(request: Request):
|
|
"""GUI-compatible streaming conversation endpoint for PA providers.
|
|
|
|
Accepts the same JSON body as ``/backend-api/v2/conversation`` and
|
|
streams Server-Sent Events in the same format used by the gpt4free
|
|
web interface (``{"type": "content", "content": "..."}`` etc.).
|
|
|
|
The ``provider`` field should contain the opaque PA provider ID
|
|
returned by ``GET /pa/providers``. When omitted the first available
|
|
PA provider is used.
|
|
"""
|
|
from g4f.mcp.pa_provider import get_pa_registry
|
|
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
return ErrorResponse.from_message(
|
|
"Invalid JSON body", HTTP_422_UNPROCESSABLE_ENTITY
|
|
)
|
|
|
|
registry = get_pa_registry()
|
|
pid = body.get("provider")
|
|
if pid:
|
|
provider_cls = registry.get_provider_class(pid)
|
|
if provider_cls is None:
|
|
return ErrorResponse.from_message(
|
|
f"PA provider '{pid}' not found", HTTP_404_NOT_FOUND
|
|
)
|
|
else:
|
|
listing = registry.list_providers()
|
|
if not listing:
|
|
return ErrorResponse.from_message(
|
|
"No PA providers found in workspace", HTTP_404_NOT_FOUND
|
|
)
|
|
provider_cls = registry.get_provider_class(listing[0]["id"])
|
|
|
|
provider_label = getattr(provider_cls, "label", provider_cls.__name__)
|
|
messages = body.get("messages") or []
|
|
model = body.get("model") or getattr(provider_cls, "default_model", "") or ""
|
|
|
|
async def gen_backend_stream():
|
|
yield (
|
|
"data: "
|
|
+ json.dumps({"type": "provider", "provider": provider_label, "model": model})
|
|
+ "\n\n"
|
|
)
|
|
try:
|
|
response = self.client.chat.completions.create(
|
|
messages=messages,
|
|
model=model,
|
|
provider=provider_cls,
|
|
stream=True,
|
|
)
|
|
async for chunk in response:
|
|
if isinstance(chunk, BaseConversation):
|
|
continue
|
|
text = ""
|
|
if hasattr(chunk, "choices") and chunk.choices:
|
|
delta = chunk.choices[0].delta
|
|
text = getattr(delta, "content", "") or ""
|
|
if text:
|
|
yield (
|
|
"data: "
|
|
+ json.dumps({"type": "content", "content": text})
|
|
+ "\n\n"
|
|
)
|
|
except GeneratorExit:
|
|
pass
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
yield (
|
|
"data: "
|
|
+ json.dumps({"type": "error", "error": f"{type(e).__name__}: {e}"})
|
|
+ "\n\n"
|
|
)
|
|
yield (
|
|
"data: "
|
|
+ json.dumps({"type": "finish", "finish": "stop"})
|
|
+ "\n\n"
|
|
)
|
|
|
|
return StreamingResponse(gen_backend_stream(), media_type="text/event-stream")
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# PA workspace static file serving (HTML/CSS/JS/images for browser) #
|
|
# ------------------------------------------------------------------ #
|
|
|
|
#: MIME types that are safe to serve for browser rendering.
|
|
#: Only these extensions are allowed; all others are refused with 403.
|
|
_WORKSPACE_SAFE_TYPES: dict[str, str] = {
|
|
"html": "text/html; charset=utf-8",
|
|
"htm": "text/html; charset=utf-8",
|
|
"css": "text/css; charset=utf-8",
|
|
"js": "application/javascript; charset=utf-8",
|
|
"mjs": "application/javascript; charset=utf-8",
|
|
"json": "application/json; charset=utf-8",
|
|
"txt": "text/plain; charset=utf-8",
|
|
"md": "text/markdown; charset=utf-8",
|
|
"svg": "image/svg+xml",
|
|
"png": "image/png",
|
|
"jpg": "image/jpeg",
|
|
"jpeg": "image/jpeg",
|
|
"gif": "image/gif",
|
|
"webp": "image/webp",
|
|
"ico": "image/x-icon",
|
|
"woff": "font/woff",
|
|
"woff2": "font/woff2",
|
|
"ttf": "font/ttf",
|
|
"otf": "font/otf",
|
|
}
|
|
|
|
@self.app.get("/pa/files/{file_path:path}", responses={
|
|
HTTP_200_OK: {},
|
|
HTTP_403_FORBIDDEN: {"model": ErrorResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
})
|
|
async def pa_serve_workspace_file(file_path: str, request: Request):
|
|
"""Securely serve a workspace file for browser rendering.
|
|
|
|
Only files within ``~/.g4f/workspace`` can be served. Path
|
|
traversal (``..``) is blocked. Only the MIME types listed in
|
|
``_WORKSPACE_SAFE_TYPES`` are served; all other extensions are
|
|
refused with **403 Forbidden** so that sensitive file types (e.g.
|
|
``.env``, ``.pa.py``, ``.py``) can never be read via this route.
|
|
|
|
HTML files are served with a ``Content-Security-Policy: sandbox``
|
|
directive (without ``allow-same-origin``), which forces the page
|
|
into a unique *null* browser origin. As a result the page cannot
|
|
access ``localStorage``, ``sessionStorage``, ``IndexedDB``, or
|
|
cookies belonging to the g4f server origin — the browser rejects
|
|
all such calls with a ``SecurityError``. The actual request
|
|
origin (``scheme://host``) is used in every source directive (e.g.
|
|
``default-src``) instead of ``'self'``, so that co-located CSS,
|
|
JS, images, and fonts still load correctly despite the document
|
|
having a null origin.
|
|
|
|
Non-HTML sub-resources (CSS, JS, images, fonts) are served without
|
|
the ``sandbox`` directive; they are leaf resources and do not run
|
|
in their own browsing context.
|
|
"""
|
|
from g4f.mcp.pa_provider import get_workspace_dir
|
|
workspace = get_workspace_dir()
|
|
|
|
# Normalise and check for traversal
|
|
try:
|
|
resolved = (workspace / file_path).resolve()
|
|
resolved.relative_to(workspace.resolve())
|
|
except (ValueError, Exception):
|
|
return ErrorResponse.from_message(
|
|
"Path traversal is not allowed", HTTP_403_FORBIDDEN
|
|
)
|
|
|
|
if not resolved.exists() or not resolved.is_file():
|
|
return ErrorResponse.from_message(
|
|
f"File not found: {file_path}", HTTP_404_NOT_FOUND
|
|
)
|
|
|
|
ext = resolved.suffix.lstrip(".").lower()
|
|
mime_type = _WORKSPACE_SAFE_TYPES.get(ext)
|
|
if mime_type is None:
|
|
return ErrorResponse.from_message(
|
|
f"File type '.{ext}' is not allowed for browser rendering",
|
|
HTTP_403_FORBIDDEN,
|
|
)
|
|
|
|
# Derive the actual request origin (scheme + host) so that CSP
|
|
# source directives reference the real server address rather than
|
|
# the generic 'self' keyword. This is important because HTML
|
|
# files are sandboxed into a null origin (see below), at which
|
|
# point 'self' would resolve to null and block all sub-resources.
|
|
request_origin = f"{request.url.scheme}://{request.headers.get('host', 'localhost')}"
|
|
|
|
is_html = ext in ("html", "htm")
|
|
if is_html:
|
|
# HTML documents are served with the CSP sandbox directive
|
|
# (without allow-same-origin). This forces the page into a
|
|
# unique null browsing-context origin so that it cannot access
|
|
# the g4f server's localStorage, sessionStorage, IndexedDB, or
|
|
# cookies. The page can still load sub-resources (CSS, JS,
|
|
# images) because they are referenced by the explicit
|
|
# request_origin in the source directives.
|
|
csp = (
|
|
"sandbox allow-scripts allow-forms allow-downloads allow-popups; "
|
|
f"default-src {request_origin}; "
|
|
f"script-src {request_origin} 'unsafe-inline'; "
|
|
f"style-src {request_origin} 'unsafe-inline'; "
|
|
f"img-src {request_origin} data:; "
|
|
f"font-src {request_origin} data:; "
|
|
"connect-src 'none'; "
|
|
"object-src 'none'; "
|
|
"base-uri 'none';"
|
|
)
|
|
else:
|
|
# Non-HTML sub-resources (CSS, JS, images, fonts) don't need
|
|
# sandboxing — they are leaf assets without their own browsing
|
|
# context. Use the request origin for source directives.
|
|
csp = (
|
|
f"default-src {request_origin}; "
|
|
f"script-src {request_origin} 'unsafe-inline'; "
|
|
f"style-src {request_origin} 'unsafe-inline'; "
|
|
f"img-src {request_origin} data:; "
|
|
f"font-src {request_origin} data:; "
|
|
"connect-src 'none'; "
|
|
"object-src 'none'; "
|
|
"base-uri 'none';"
|
|
)
|
|
|
|
headers = {
|
|
# Prevent the browser from sniffing a different content-type
|
|
"X-Content-Type-Options": "nosniff",
|
|
# Prevent this page from being framed by untrusted origins
|
|
"X-Frame-Options": "SAMEORIGIN",
|
|
# Basic XSS filter (belt-and-suspenders; CSP is more important)
|
|
"X-XSS-Protection": "1; mode=block",
|
|
"Content-Security-Policy": csp,
|
|
# Restrict powerful browser APIs that workspace pages don't need
|
|
"Permissions-Policy": (
|
|
"geolocation=(), camera=(), microphone=(), "
|
|
"payment=(), usb=(), fullscreen=()"
|
|
),
|
|
"Cache-Control": "no-store",
|
|
}
|
|
|
|
return FileResponse(
|
|
str(resolved),
|
|
media_type=mime_type,
|
|
headers=headers,
|
|
)
|
|
|
|
responses = {
|
|
HTTP_200_OK: {"model": TranscriptionResponseModel},
|
|
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
|
|
}
|
|
@self.app.post("/v1/audio/transcriptions", responses=responses)
|
|
@self.app.post("/api/{path_provider}/audio/transcriptions", responses=responses)
|
|
@self.app.post("/api/markitdown", responses=responses)
|
|
async def convert(
|
|
file: UploadFile,
|
|
path_provider: Optional[str] = None,
|
|
model: Annotated[Optional[str], Form()] = None,
|
|
provider: Annotated[Optional[str], Form()] = None,
|
|
prompt: Annotated[Optional[str], Form()] = "Transcribe this audio"
|
|
):
|
|
if path_provider is not None:
|
|
provider = path_provider
|
|
if provider is None:
|
|
provider = "MarkItDown"
|
|
try:
|
|
provider = ProviderUtils.get_by_label(provider)
|
|
except ValueError as e:
|
|
if provider in model_map:
|
|
model = provider
|
|
provider = None
|
|
else:
|
|
return ErrorResponse.from_message(str(e), 404)
|
|
kwargs = {"modalities": ["text"]}
|
|
if provider == "MarkItDown":
|
|
kwargs = {
|
|
"llm_client": self.client,
|
|
}
|
|
try:
|
|
response = await self.client.chat.completions.create(
|
|
messages=prompt,
|
|
model=model,
|
|
provider=provider,
|
|
media=[[file.file, file.filename]],
|
|
**kwargs
|
|
)
|
|
return {"text": response.choices[0].message.content, "model": response.model, "provider": response.provider}
|
|
except (ModelNotFoundError, ProviderNotFoundError) as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, None, HTTP_404_NOT_FOUND)
|
|
except MissingAuthError as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, None, HTTP_401_UNAUTHORIZED)
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, None, HTTP_500_INTERNAL_SERVER_ERROR)
|
|
|
|
responses = {
|
|
HTTP_200_OK: {"content": {"audio/*": {}}},
|
|
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
|
|
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
|
|
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
|
|
}
|
|
@self.app.post("/v1/audio/speech", responses=responses)
|
|
@self.app.post("/api/{provider}/audio/speech", responses=responses)
|
|
async def generate_speech(
|
|
config: AudioSpeechConfig,
|
|
provider: Optional[str] = None,
|
|
credentials: Annotated[HTTPAuthorizationCredentials, Depends(Api.security)] = None
|
|
):
|
|
api_key = None
|
|
if credentials is not None and credentials.credentials != "secret":
|
|
api_key = credentials.credentials
|
|
if provider is None:
|
|
provider = config.provider
|
|
if provider is None:
|
|
provider = AppConfig.media_provider
|
|
try:
|
|
provider = ProviderUtils.get_by_label(provider)
|
|
except ValueError as e:
|
|
return ErrorResponse.from_message(str(e), 404)
|
|
try:
|
|
audio = filter_none(voice=config.voice, format=config.response_format, language=config.language)
|
|
response = await self.client.chat.completions.create(
|
|
messages=[
|
|
{"role": "user", "content": f"{config.instrcutions} Text: {config.input}"}
|
|
],
|
|
model=config.model,
|
|
provider=provider,
|
|
prompt=config.input,
|
|
api_key=api_key,
|
|
download_media=config.download_media,
|
|
**filter_none(
|
|
audio=audio if audio else None,
|
|
)
|
|
)
|
|
if response.choices[0].message.audio is not None:
|
|
response = base64.b64decode(response.choices[0].message.audio.data)
|
|
return Response(response, media_type=f"audio/{config.response_format.replace('mp3', 'mpeg')}")
|
|
elif isinstance(response.choices[0].message.content, AudioResponse):
|
|
response = response.choices[0].message.content.data
|
|
response = response.replace("/media", get_media_dir())
|
|
def delete_file():
|
|
try:
|
|
os.remove(response)
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
return FileResponse(response, background=BackgroundTask(delete_file))
|
|
except (ModelNotFoundError, ProviderNotFoundError) as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, None, HTTP_404_NOT_FOUND)
|
|
except MissingAuthError as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, None, HTTP_401_UNAUTHORIZED)
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
return ErrorResponse.from_exception(e, None, HTTP_500_INTERNAL_SERVER_ERROR)
|
|
|
|
@self.app.post("/v1/upload_cookies", responses={
|
|
HTTP_200_OK: {"model": List[FileResponseModel]},
|
|
})
|
|
def upload_cookies(
|
|
files: List[UploadFile],
|
|
credentials: Annotated[HTTPAuthorizationCredentials, Depends(Api.security)] = None
|
|
):
|
|
response_data = []
|
|
if not AppConfig.ignore_cookie_files:
|
|
for file in files:
|
|
try:
|
|
if file and file.filename.endswith(".json") or file.filename.endswith(".har"):
|
|
filename = os.path.basename(file.filename)
|
|
with open(os.path.join(get_cookies_dir(), filename), 'wb') as f:
|
|
shutil.copyfileobj(file.file, f)
|
|
response_data.append({"filename": filename})
|
|
finally:
|
|
file.file.close()
|
|
read_cookie_files()
|
|
return response_data
|
|
|
|
@self.app.get("/images/{filename}", responses={
|
|
HTTP_200_OK: {"content": {"image/*": {}}},
|
|
HTTP_404_NOT_FOUND: {}
|
|
})
|
|
@self.app.get("/media/{filename}", responses={
|
|
HTTP_200_OK: {"content": {"image/*": {}, "audio/*": {}}, "video/*": {}},
|
|
HTTP_404_NOT_FOUND: {}
|
|
})
|
|
async def get_media(filename, request: Request, thumbnail: bool = False):
|
|
def get_timestamp(str):
|
|
m=re.match("^[0-9]+", str)
|
|
if m:
|
|
return int(m.group(0))
|
|
else:
|
|
return 0
|
|
target = os.path.join(get_media_dir(), os.path.basename(filename))
|
|
if thumbnail and has_pillow:
|
|
thumbnail_dir = os.path.join(get_media_dir(), "thumbnails")
|
|
thumbnail = os.path.join(thumbnail_dir, filename)
|
|
if not os.path.isfile(target):
|
|
other_name = os.path.join(get_media_dir(), os.path.basename(quote_plus(filename)))
|
|
if os.path.isfile(other_name):
|
|
target = other_name
|
|
ext = os.path.splitext(filename)[1][1:]
|
|
mime_type = EXTENSIONS_MAP.get(ext)
|
|
stat_result = SimpleNamespace()
|
|
stat_result.st_size = 0
|
|
stat_result.st_mtime = get_timestamp(filename)
|
|
if thumbnail and has_pillow and os.path.isfile(thumbnail):
|
|
stat_result.st_size = os.stat(thumbnail).st_size
|
|
elif not thumbnail and os.path.isfile(target):
|
|
stat_result.st_size = os.stat(target).st_size
|
|
headers = {
|
|
"cache-control": "public, max-age=31536000",
|
|
"last-modified": formatdate(stat_result.st_mtime, usegmt=True),
|
|
"etag": f'"{hashlib.md5(filename.encode()).hexdigest()}"',
|
|
**({
|
|
"content-length": str(stat_result.st_size),
|
|
} if stat_result.st_size else {}),
|
|
**({} if thumbnail or mime_type is None else {
|
|
"content-type": mime_type,
|
|
})
|
|
}
|
|
response = FileResponse(
|
|
target,
|
|
headers=headers,
|
|
filename=filename,
|
|
)
|
|
try:
|
|
if_none_match = request.headers["if-none-match"]
|
|
etag = response.headers["etag"]
|
|
if etag in [tag.strip(" W/") for tag in if_none_match.split(",")]:
|
|
return NotModifiedResponse(response.headers)
|
|
except KeyError:
|
|
pass
|
|
if not os.path.isfile(target) and mime_type is not None:
|
|
source_url = get_source_url(str(request.query_params))
|
|
ssl = None
|
|
if source_url is None:
|
|
backend_url = os.environ.get("G4F_BACKEND_URL")
|
|
if backend_url:
|
|
source_url = f"{backend_url}/media/{filename}"
|
|
ssl = False
|
|
if source_url is not None:
|
|
try:
|
|
await copy_media(
|
|
[source_url],
|
|
target=target, ssl=ssl)
|
|
debug.log(f"File copied from {source_url}")
|
|
except Exception as e:
|
|
debug.error(f"Download failed: {source_url}")
|
|
debug.error(e)
|
|
return RedirectResponse(url=source_url)
|
|
if thumbnail and has_pillow:
|
|
try:
|
|
if not os.path.isfile(thumbnail):
|
|
image = Image.open(target)
|
|
os.makedirs(thumbnail_dir, exist_ok=True)
|
|
process_image(image, save=thumbnail)
|
|
debug.log(f"Thumbnail created: {thumbnail}")
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
if thumbnail and os.path.isfile(thumbnail):
|
|
result = thumbnail
|
|
else:
|
|
result = target
|
|
if not os.path.isfile(result):
|
|
return ErrorResponse.from_message("File not found", HTTP_404_NOT_FOUND)
|
|
async def stream():
|
|
with open(result, "rb") as file:
|
|
while True:
|
|
chunk = file.read(65536)
|
|
if not chunk:
|
|
break
|
|
yield chunk
|
|
return StreamingResponse(stream(), headers=headers)
|
|
|
|
@self.app.get("/thumbnail/{filename}", responses={
|
|
HTTP_200_OK: {"content": {"image/*": {}, "audio/*": {}}, "video/*": {}},
|
|
HTTP_404_NOT_FOUND: {}
|
|
})
|
|
async def get_media_thumbnail(filename: str, request: Request):
|
|
return await get_media(filename, request, True)
|
|
|
|
def format_exception(e: Union[Exception, str], config: Union[ChatCompletionsConfig, ImageGenerationConfig] = None, image: bool = False) -> str:
|
|
provider = (AppConfig.media_provider if image else AppConfig.provider)
|
|
model = AppConfig.model
|
|
if config is not None:
|
|
if config.provider is not None:
|
|
provider = config.provider
|
|
if config.model is not None:
|
|
model = config.model
|
|
if isinstance(e, str):
|
|
message = e
|
|
else:
|
|
message = f"{e.__class__.__name__}: {e}"
|
|
return json.dumps({
|
|
"error": {"message": message},
|
|
**filter_none(
|
|
model=model,
|
|
provider=getattr(provider, "__name__", provider)
|
|
)
|
|
})
|
|
|
|
def run_api(
|
|
host: str = '0.0.0.0',
|
|
port: int = None,
|
|
bind: str = None,
|
|
debug: bool = False,
|
|
use_colors: bool = None,
|
|
**kwargs
|
|
) -> None:
|
|
print(f'Starting server... [g4f v-{g4f.version.utils.current_version}]' + (" (debug)" if debug else ""))
|
|
|
|
if use_colors is None:
|
|
use_colors = debug
|
|
|
|
if bind is not None:
|
|
host, port = bind.split(":")
|
|
|
|
if port is None:
|
|
port = DEFAULT_PORT
|
|
|
|
if AppConfig.demo and debug:
|
|
method = "create_app_with_demo_and_debug"
|
|
elif AppConfig.gui and debug:
|
|
method = "create_app_with_gui_and_debug"
|
|
else:
|
|
method = "create_app_debug" if debug else "create_app"
|
|
|
|
uvicorn.run(
|
|
f"g4f.api:{method}",
|
|
host=host,
|
|
port=int(port),
|
|
factory=True,
|
|
use_colors=use_colors,
|
|
**filter_none(**kwargs)
|
|
)
|