feat: add SVG conversion functionality for raster images and update requirements
This commit is contained in:
@@ -12,12 +12,13 @@ from app.services.policy_service import (
|
||||
)
|
||||
from app.utils.file_validator import FileValidationError
|
||||
from app.utils.sanitizer import generate_safe_path
|
||||
from app.tasks.image_tasks import convert_image_task, resize_image_task
|
||||
from app.tasks.image_tasks import convert_image_task, resize_image_task, convert_image_to_svg_task
|
||||
|
||||
image_bp = Blueprint("image", __name__)
|
||||
|
||||
ALLOWED_IMAGE_TYPES = ["png", "jpg", "jpeg", "webp"]
|
||||
ALLOWED_OUTPUT_FORMATS = ["jpg", "png", "webp"]
|
||||
ALLOWED_SVG_COLOR_MODES = ["color", "binary"]
|
||||
|
||||
|
||||
@image_bp.route("/convert", methods=["POST"])
|
||||
@@ -155,3 +156,54 @@ def resize_image_route():
|
||||
"task_id": task.id,
|
||||
"message": "Image resize started. Poll /api/tasks/{task_id}/status for progress.",
|
||||
}), 202
|
||||
|
||||
|
||||
@image_bp.route("/to-svg", methods=["POST"])
|
||||
@limiter.limit("10/minute")
|
||||
def convert_image_to_svg_route():
|
||||
"""
|
||||
Convert a raster image to SVG vector format.
|
||||
|
||||
Accepts: multipart/form-data with:
|
||||
- 'file': Image file (PNG, JPG, JPEG, WebP)
|
||||
- 'color_mode' (optional): "color" or "binary" (default: "color")
|
||||
Returns: JSON with task_id for polling
|
||||
"""
|
||||
if "file" not in request.files:
|
||||
return jsonify({"error": "No file provided."}), 400
|
||||
|
||||
file = request.files["file"]
|
||||
color_mode = request.form.get("color_mode", "color").lower()
|
||||
|
||||
if color_mode not in ALLOWED_SVG_COLOR_MODES:
|
||||
color_mode = "color"
|
||||
|
||||
actor = resolve_web_actor()
|
||||
try:
|
||||
assert_quota_available(actor)
|
||||
except PolicyError as e:
|
||||
return jsonify({"error": e.message}), e.status_code
|
||||
|
||||
try:
|
||||
original_filename, ext = validate_actor_file(
|
||||
file, allowed_types=ALLOWED_IMAGE_TYPES, actor=actor
|
||||
)
|
||||
except FileValidationError as e:
|
||||
return jsonify({"error": e.message}), e.code
|
||||
|
||||
task_id, input_path = generate_safe_path(ext, folder_type="upload")
|
||||
file.save(input_path)
|
||||
|
||||
task = convert_image_to_svg_task.delay(
|
||||
input_path,
|
||||
task_id,
|
||||
original_filename,
|
||||
color_mode,
|
||||
**build_task_tracking_kwargs(actor),
|
||||
)
|
||||
record_accepted_usage(actor, "image-to-svg", task.id)
|
||||
|
||||
return jsonify({
|
||||
"task_id": task.id,
|
||||
"message": "Image to SVG conversion started. Poll /api/tasks/{task_id}/status for progress.",
|
||||
}), 202
|
||||
|
||||
@@ -137,13 +137,14 @@ def resize_image(
|
||||
width = int(orig_width * ratio)
|
||||
|
||||
# Resize using high-quality resampling
|
||||
assert width is not None and height is not None
|
||||
resized = img.resize((width, height), Image.Resampling.LANCZOS)
|
||||
|
||||
# Detect format from output extension
|
||||
ext = os.path.splitext(output_path)[1].lower().strip(".")
|
||||
pil_format = FORMAT_MAP.get(ext, "PNG")
|
||||
|
||||
save_kwargs = {"optimize": True}
|
||||
save_kwargs: dict[str, int | bool] = {"optimize": True}
|
||||
if pil_format in ("JPEG", "WEBP"):
|
||||
save_kwargs["quality"] = quality
|
||||
# Handle RGBA for JPEG
|
||||
@@ -167,3 +168,67 @@ def resize_image(
|
||||
|
||||
except (IOError, OSError, Image.DecompressionBombError) as e:
|
||||
raise ImageProcessingError(f"Image resize failed: {str(e)}")
|
||||
|
||||
|
||||
# ─── Allowed color modes for SVG tracing ─────────────────────
|
||||
ALLOWED_COLOR_MODES = ("color", "binary")
|
||||
|
||||
|
||||
def convert_image_to_svg(
|
||||
input_path: str,
|
||||
output_path: str,
|
||||
color_mode: str = "color",
|
||||
) -> dict:
|
||||
"""
|
||||
Convert a raster image to SVG using vtracer.
|
||||
|
||||
Args:
|
||||
input_path: Path to the input image (PNG, JPG, WebP)
|
||||
output_path: Path for the output SVG file
|
||||
color_mode: "color" for full-colour trace, "binary" for black & white
|
||||
|
||||
Returns:
|
||||
dict with original_size, converted_size, width, height
|
||||
|
||||
Raises:
|
||||
ImageProcessingError: If conversion fails
|
||||
"""
|
||||
import vtracer
|
||||
|
||||
if color_mode not in ALLOWED_COLOR_MODES:
|
||||
color_mode = "color"
|
||||
|
||||
os.makedirs(os.path.dirname(output_path), exist_ok=True)
|
||||
|
||||
try:
|
||||
original_size = os.path.getsize(input_path)
|
||||
|
||||
# Read dimensions via Pillow (also validates image)
|
||||
with Image.open(input_path) as img:
|
||||
width, height = img.size
|
||||
|
||||
vtracer.convert_image_to_svg_py(
|
||||
input_path,
|
||||
output_path,
|
||||
colormode=color_mode,
|
||||
)
|
||||
|
||||
converted_size = os.path.getsize(output_path)
|
||||
|
||||
logger.info(
|
||||
f"Image→SVG conversion: {input_path} "
|
||||
f"({original_size} → {converted_size})"
|
||||
)
|
||||
|
||||
return {
|
||||
"original_size": original_size,
|
||||
"converted_size": converted_size,
|
||||
"width": width,
|
||||
"height": height,
|
||||
"format": "svg",
|
||||
}
|
||||
|
||||
except (IOError, OSError, Image.DecompressionBombError) as e:
|
||||
raise ImageProcessingError(f"Image to SVG conversion failed: {str(e)}")
|
||||
except Exception as e:
|
||||
raise ImageProcessingError(f"Image to SVG conversion failed: {str(e)}")
|
||||
|
||||
@@ -5,7 +5,7 @@ import logging
|
||||
from flask import current_app
|
||||
|
||||
from app.extensions import celery
|
||||
from app.services.image_service import convert_image, resize_image, ImageProcessingError
|
||||
from app.services.image_service import convert_image, resize_image, convert_image_to_svg, ImageProcessingError
|
||||
from app.services.storage_service import storage
|
||||
from app.services.task_tracking_service import finalize_task_tracking
|
||||
from app.utils.sanitizer import cleanup_task_files
|
||||
@@ -242,3 +242,98 @@ def resize_image_task(
|
||||
api_key_id,
|
||||
self.request.id,
|
||||
)
|
||||
|
||||
|
||||
@celery.task(bind=True, name="app.tasks.image_tasks.convert_image_to_svg_task")
|
||||
def convert_image_to_svg_task(
|
||||
self,
|
||||
input_path: str,
|
||||
task_id: str,
|
||||
original_filename: str,
|
||||
color_mode: str = "color",
|
||||
user_id: int | None = None,
|
||||
usage_source: str = "web",
|
||||
api_key_id: int | None = None,
|
||||
):
|
||||
"""
|
||||
Async task: Convert a raster image to SVG.
|
||||
|
||||
Args:
|
||||
input_path: Path to the uploaded image
|
||||
task_id: Unique task identifier
|
||||
original_filename: Original filename for download
|
||||
color_mode: "color" or "binary"
|
||||
|
||||
Returns:
|
||||
dict with download_url and conversion stats
|
||||
"""
|
||||
output_dir = _get_output_dir(task_id)
|
||||
output_path = os.path.join(output_dir, f"{task_id}.svg")
|
||||
|
||||
try:
|
||||
self.update_state(
|
||||
state="PROCESSING",
|
||||
meta={"step": "Converting image to SVG..."},
|
||||
)
|
||||
|
||||
stats = convert_image_to_svg(input_path, output_path, color_mode)
|
||||
|
||||
self.update_state(state="PROCESSING", meta={"step": "Uploading result..."})
|
||||
|
||||
s3_key = storage.upload_file(output_path, task_id, folder="outputs")
|
||||
|
||||
name_without_ext = os.path.splitext(original_filename)[0]
|
||||
download_name = f"{name_without_ext}.svg"
|
||||
|
||||
download_url = storage.generate_presigned_url(
|
||||
s3_key, original_filename=download_name
|
||||
)
|
||||
|
||||
result = {
|
||||
"status": "completed",
|
||||
"download_url": download_url,
|
||||
"filename": download_name,
|
||||
"original_size": stats["original_size"],
|
||||
"converted_size": stats["converted_size"],
|
||||
"width": stats["width"],
|
||||
"height": stats["height"],
|
||||
"format": "svg",
|
||||
}
|
||||
|
||||
logger.info(f"Task {task_id}: Image to SVG conversion completed")
|
||||
return _finalize_task(
|
||||
task_id,
|
||||
user_id,
|
||||
"image-to-svg",
|
||||
original_filename,
|
||||
result,
|
||||
usage_source,
|
||||
api_key_id,
|
||||
self.request.id,
|
||||
)
|
||||
|
||||
except ImageProcessingError as e:
|
||||
logger.error(f"Task {task_id}: Image error — {e}")
|
||||
return _finalize_task(
|
||||
task_id,
|
||||
user_id,
|
||||
"image-to-svg",
|
||||
original_filename,
|
||||
{"status": "failed", "error": str(e)},
|
||||
usage_source,
|
||||
api_key_id,
|
||||
self.request.id,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Task {task_id}: Unexpected error — {e}")
|
||||
return _finalize_task(
|
||||
task_id,
|
||||
user_id,
|
||||
"image-to-svg",
|
||||
original_filename,
|
||||
{"status": "failed", "error": str(e) or "An unexpected error occurred."},
|
||||
usage_source,
|
||||
api_key_id,
|
||||
self.request.id,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user