Implement base64 fallback for image handling when no uploader is configured (#1)

* Initial plan

* Implement base64 fallback for image handling when no uploader configured

Co-authored-by: bbbugg <80089841+bbbugg@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: bbbugg <80089841+bbbugg@users.noreply.github.com>
This commit is contained in:
Copilot
2025-08-28 15:19:21 +08:00
committed by bbbugg
parent 4af17ce55d
commit bb6c629aef
3 changed files with 46 additions and 1 deletions

View File

@@ -10,6 +10,7 @@ from typing import Any, Dict, List, Optional
from app.config.config import settings
from app.utils.uploader import ImageUploaderFactory
from app.log.logger import get_openai_logger
from app.utils.helpers import is_image_upload_configured
logger = get_openai_logger()
@@ -251,7 +252,22 @@ def _extract_result(
return text, reasoning_content, tool_calls, thought
def _has_inline_image_part(response: Dict[str, Any]) -> bool:
try:
for c in response.get("candidates", []):
for p in c.get("content", {}).get("parts", []):
if isinstance(p, dict) and ("inlineData" in p):
return True
except Exception:
return False
return False
def _extract_image_data(part: dict) -> str:
# Return empty string if no uploader is configured
if not is_image_upload_configured():
return ""
image_uploader = None
if settings.UPLOAD_PROVIDER == "smms":
image_uploader = ImageUploaderFactory.create(
@@ -322,6 +338,10 @@ def _extract_tool_calls(
def _handle_gemini_stream_response(
response: Dict[str, Any], model: str, stream: bool
) -> Dict[str, Any]:
# Early return raw Gemini response if no uploader configured and contains inline images
if not is_image_upload_configured() and _has_inline_image_part(response):
return response
text, reasoning_content, tool_calls, thought = _extract_result(
response, model, stream=stream, gemini_format=True
)
@@ -339,6 +359,10 @@ def _handle_gemini_stream_response(
def _handle_gemini_normal_response(
response: Dict[str, Any], model: str, stream: bool
) -> Dict[str, Any]:
# Early return raw Gemini response if no uploader configured and contains inline images
if not is_image_upload_configured() and _has_inline_image_part(response):
return response
text, reasoning_content, tool_calls, thought = _extract_result(
response, model, stream=stream, gemini_format=True
)

View File

@@ -10,6 +10,7 @@ from app.core.constants import VALID_IMAGE_RATIOS
from app.domain.openai_models import ImageGenerationRequest
from app.log.logger import get_image_create_logger
from app.utils.uploader import ImageUploaderFactory
from app.utils.helpers import is_image_upload_configured
logger = get_image_create_logger()
@@ -97,12 +98,15 @@ class ImageCreateService:
image_data = generated_image.image.image_bytes
image_uploader = None
if request.response_format == "b64_json":
# Return base64 if explicitly requested or if no uploader is configured
if request.response_format == "b64_json" or not is_image_upload_configured():
base64_image = base64.b64encode(image_data).decode("utf-8")
images_data.append(
{"b64_json": base64_image, "revised_prompt": request.prompt}
)
continue
else:
# Upload to configured provider
current_date = time.strftime("%Y/%m/%d")
filename = f"{current_date}/{uuid.uuid4().hex[:8]}.png"

View File

@@ -10,6 +10,7 @@ from pathlib import Path
import logging
from app.core.constants import DATA_URL_PATTERN, IMAGE_URL_PATTERN, VALID_IMAGE_RATIOS
from app.config.config import settings
helper_logger = logging.getLogger("app.utils")
@@ -189,3 +190,19 @@ def get_current_version(default_version: str = "0.0.0") -> str:
except IOError as e:
helper_logger.error(f"Error reading VERSION file ('{version_file}'): {e}. Using default version '{default_version}'.")
return default_version
def is_image_upload_configured() -> bool:
"""Return True only if a valid upload provider is selected and all required settings for that provider are present."""
provider = getattr(settings, "UPLOAD_PROVIDER", "").strip().lower()
if provider == "smms":
return bool(getattr(settings, "SMMS_SECRET_TOKEN", None))
if provider == "picgo":
return bool(getattr(settings, "PICGO_API_KEY", None))
if provider == "cloudflare_imgbed":
return all([
getattr(settings, "CLOUDFLARE_IMGBED_URL", None),
getattr(settings, "CLOUDFLARE_IMGBED_AUTH_CODE", None),
getattr(settings, "CLOUDFLARE_IMGBED_UPLOAD_FOLDER", None),
])
return False