| import logging |
| import io |
| from pathlib import Path |
| from typing import List |
| from PIL import Image |
|
|
| from fastapi import APIRouter, Form, UploadFile, HTTPException |
| from fastapi.responses import Response |
|
|
| logger = logging.getLogger(__name__) |
|
|
| router = APIRouter() |
|
|
| plugin = None |
|
|
|
|
| def set_plugin_instance(plugin_instance): |
| global plugin |
| plugin = plugin_instance |
|
|
|
|
| @router.get("/status") |
| async def get_status(): |
| if plugin is None: |
| return { |
| "name": "image", |
| "enabled": False, |
| "message": "插件未加载", |
| } |
| return plugin.get_status() |
|
|
|
|
| @router.post("/upload") |
| async def upload_image(file: UploadFile): |
| if plugin is None or not plugin.enabled: |
| raise HTTPException(status_code=400, detail="插件未启用") |
|
|
| allowed_extensions = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp"} |
| ext = Path(file.filename).suffix.lower() |
|
|
| if ext not in allowed_extensions: |
| raise HTTPException( |
| status_code=400, |
| detail=f"不支持的文件格式: {ext},仅支持 {', '.join(allowed_extensions)}", |
| ) |
|
|
| content_bytes = await file.read() |
|
|
| try: |
| image = Image.open(io.BytesIO(content_bytes)) |
| image_format = image.format or "PNG" |
|
|
| sizes = [16, 32, 48, 128] |
| available_sizes = [] |
|
|
| for size in sizes: |
| if image.width >= size and image.height >= size: |
| available_sizes.append(size) |
|
|
| if not available_sizes: |
| raise HTTPException( |
| status_code=400, |
| detail=f"图片尺寸 ({image.width}x{image.height}) 小于最小要求 16x16", |
| ) |
|
|
| return { |
| "filename": file.filename, |
| "original_size": {"width": image.width, "height": image.height}, |
| "format": image_format, |
| "available_sizes": available_sizes, |
| "sizes": sizes, |
| } |
|
|
| except Exception as e: |
| logger.error(f"处理图片失败: {e}") |
| raise HTTPException(status_code=400, detail=f"图片处理失败: {str(e)}") |
|
|
|
|
| @router.post("/generate") |
| async def generate_image( |
| file: UploadFile, size: int = Form(...), format: str = Form("PNG") |
| ): |
| if plugin is None or not plugin.enabled: |
| raise HTTPException(status_code=400, detail="插件未启用") |
|
|
| if size not in [16, 32, 48, 128]: |
| raise HTTPException( |
| status_code=400, detail="不支持的尺寸,仅支持 16, 32, 48, 128" |
| ) |
|
|
| if format.upper() not in ["PNG", "JPEG", "WEBP"]: |
| raise HTTPException( |
| status_code=400, detail="不支持的格式,仅支持 PNG, JPEG, WEBP" |
| ) |
|
|
| content_bytes = await file.read() |
|
|
| try: |
| image = Image.open(io.BytesIO(content_bytes)) |
|
|
| if image.width < size or image.height < size: |
| raise HTTPException( |
| status_code=400, |
| detail=f"图片尺寸 ({image.width}x{image.height}) 小于请求尺寸 {size}x{size}", |
| ) |
|
|
| resized = image.resize((size, size), Image.Resampling.LANCZOS) |
|
|
| output = io.BytesIO() |
| save_format = format.upper() if format.upper() != "JPEG" else "JPEG" |
| resized.save(output, format=save_format, quality=95) |
| output.seek(0) |
|
|
| content_type = { |
| "PNG": "image/png", |
| "JPEG": "image/jpeg", |
| "WEBP": "image/webp", |
| }.get(format.upper(), "image/png") |
|
|
| ext = {"PNG": ".png", "JPEG": ".jpg", "WEBP": ".webp"}.get( |
| format.upper(), ".png" |
| ) |
|
|
| filename = f"icon{size}{ext}" |
|
|
| return Response( |
| content=output.read(), |
| media_type=content_type, |
| headers={"Content-Disposition": f'attachment; filename="{filename}"'}, |
| ) |
|
|
| except HTTPException: |
| raise |
| except Exception as e: |
| logger.error(f"生成图片失败: {e}") |
| raise HTTPException(status_code=400, detail=f"生成图片失败: {str(e)}") |
|
|