0.0.1

README.md

@@ -0,0 +1,16 @@
+---
+title: Magick 图像转换器
+emoji: 🖼️
+colorFrom: purple
+colorTo: blue
+sdk: docker
+app_port: 8000
+pinned: false
+---
+
+# 🧙‍♂️ Magick 图像转换 API
+
+本项目提供一个基于 FastAPI 和 ImageMagick 的 REST API，用于：
+
+* **POST /**: 将上传的图像转换为**无损 AVIF** 格式，并**保留所有元信息**。
+* **GET /health**: 检查 ImageMagick 和 AVIF 编码器 (heif) 的可用状态。

.github/workflows/sync-to-hf-space.yml

@@ -0,0 +1,30 @@
+name: Sync to Hugging Face Space
+
+on:
+  push:
+    branches:
+      - main  # 或者你的主分支名称，例如 master
+
+jobs:
+  sync:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4 # 建议使用最新版本
+        with:
+          fetch-depth: 0 # 获取所有历史记录，以便正确推送
+
+      - name: Set up Git
+        run: |
+          git config --global user.email "action@github.com"
+          git config --global user.name "GitHub Action"
+
+      - name: Push to Hugging Face Space
+        env:
+          HF_TOKEN: ${{ secrets.HF_TOKEN }} # 引用你之前创建的 Secret
+        run: |
+          # 将 <your-hf-username> 和 <your-space-name> 替换为你的实际信息
+          git remote add space https://user:${HF_TOKEN}@huggingface.co/spaces/BlueSkyXN/ImageMagickAPI-HFS
+          # 强制推送到 Space 的 main 分支，覆盖原有内容
+          # 如果你的 Space 主分支不是 main，请相应修改
+          git push --force space main

Dockerfile

@@ -0,0 +1,35 @@
+# 1. 使用官方 Python 镜像
+FROM python:3.10-slim
+
+# 设置环境变量
+ENV PORT=8000
+ENV PYTHONUNBUFFERED=1
+
+# 2. 安装 ImageMagick 和 AVIF/HEIC 依赖
+#    libheif-examples 提供了 magick 所需的 heif-enc 编码器
+RUN apt-get update && apt-get install -y \
+    imagemagick \
+    libheif-examples \
+    && rm -rf /var/lib/apt/lists/*
+
+# 3. 设置工作目录 (结构同您的 OCR-HFS)
+WORKDIR /app
+
+# 4. 复制并安装 Python 依赖 (结构同您的 OCR-HFS)
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 5. 复制应用代码 (结构同您的 OCR-HFS)
+COPY main.py .
+COPY entrypoint.sh .
+RUN chmod +x /app/entrypoint.sh
+
+# 6. 创建临时工作目录 (结构同您的 OCR-HFS)
+RUN mkdir -p /app/temp
+RUN chmod 777 /app/temp
+
+# 7. 暴露端口 (结构同您的 OCR-HFS)
+EXPOSE 8000
+
+# 8. 使用入口脚本启动 (结构同您的 OCR-HFS)
+ENTRYPOINT ["/app/entrypoint.sh"]

entrypoint.sh

@@ -0,0 +1,18 @@
+#!/bin/sh
+
+# 打印环境信息用于调试
+echo "Starting Magick API Service"
+echo "Environment: PORT=$PORT"
+
+# 验证Magick是否可用
+echo "Checking ImageMagick installation..."
+magick --version | head -n 1
+echo "Checking AVIF encoder (heif-enc) installation..."
+which heif-enc
+
+# 确保使用正确的端口变量
+PORT="${PORT:-8000}"
+echo "Using port: $PORT"
+
+# 执行 uvicorn 服务器 (同您的 OCR-HFS)
+exec uvicorn main:app --host 0.0.0.0 --port $PORT

main.py

@@ -0,0 +1,201 @@
+import fastapi
+from fastapi import FastAPI, File, UploadFile, HTTPException, Response, BackgroundTasks
+from fastapi.responses import FileResponse, JSONResponse
+import subprocess
+import asyncio # 使用 asyncio 来进行非阻塞的 subprocess 调用
+import tempfile
+import os
+import shutil
+import logging
+import uuid
+from typing import Literal, Optional, List
+
+# --- 配置  ---
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+MAX_FILE_SIZE_MB = 200  # 最大文件大小
+TIMEOUT_SECONDS = 300   # 转换超时 (5分钟)
+TEMP_DIR = "/app/temp"  # 临时文件目录
+# ----------------
+
+# 初始化 FastAPI 应用
+app = FastAPI(
+    title="Magick AVIF Converter",
+    description="API to convert images to lossless AVIF, preserving metadata.",
+    version="1.0.0"
+)
+
+# 确保临时目录存在 
+os.makedirs(TEMP_DIR, exist_ok=True)
+
+# --- 健康检查 (适配 Magick) ---
+@app.get("/health", summary="Health Check")
+async def health_check():
+    """提供详细的API和依赖健康状态检查"""
+    try:
+        # 检查 Magick 是否可用
+        proc_magick = await asyncio.subprocess.create_subprocess_exec(
+            'magick', '--version', stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+        )
+        stdout_m, stderr_m = await proc_magick.communicate()
+        magick_version = stdout_m.decode().split('\n')[0] if proc_magick.returncode == 0 else "Not available"
+        
+        # 检查 AVIF 编码器 (heif-enc) 是否可用
+        proc_heif = await asyncio.subprocess.create_subprocess_exec(
+            'which', 'heif-enc', stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+        )
+        stdout_h, stderr_h = await proc_heif.communicate()
+        heif_encoder_path = stdout_h.decode().strip() if proc_heif.returncode == 0 else "Not available (AVIF conversion will fail)"
+
+        # 检查磁盘空间 
+        disk_info = os.statvfs(TEMP_DIR)
+        free_space_mb = (disk_info.f_bavail * disk_info.f_frsize) / (1024 * 1024)
+        
+        return {
+            "status": "healthy",
+            "imagemagick": magick_version,
+            "avif_encoder": heif_encoder_path,
+            "disk_space": {"free_mb": round(free_space_mb, 2), "temp_dir": TEMP_DIR},
+            "resource_limits": {
+                "max_file_size_mb": MAX_FILE_SIZE_MB,
+                "timeout_seconds": TIMEOUT_SECONDS
+            }
+        }
+    except Exception as e:
+        logger.error(f"Health check failed: {str(e)}")
+        return {"status": "unhealthy", "error": str(e)}
+
+# --- 根路径 API 端点 (实现您的需求) ---
+@app.post("/",
+          summary="Convert Image to Lossless AVIF",
+          response_class=FileResponse,
+          responses={
+              200: {"content": {"image/avif": {}}, "description": "Successfully converted image to lossless AVIF."},
+              400: {"description": "Bad Request (e.g., file too large)"},
+              500: {"description": "Internal Server Error (Magick conversion failed)"},
+              504: {"description": "Gateway Timeout (Conversion took too long)"}
+          })
+async def convert_to_avif_lossless(
+    background_tasks: BackgroundTasks,
+    file: UploadFile = File(..., description="The image file to be converted (PNG, JPEG, WebP, etc.).")
+):
+    """
+    接收图像文件，将其无损转换为 AVIF，并保留元信息。
+    """
+    logger.info(f"Received request: filename={file.filename}")
+
+    # --- 文件大小检查  ---
+    file_size_mb = await get_upload_file_size(file) / (1024 * 1024)
+    if file_size_mb > MAX_FILE_SIZE_MB:
+        logger.warning(f"File too large: {file_size_mb:.2f}MB (max: {MAX_FILE_SIZE_MB}MB)")
+        raise HTTPException(
+            status_code=400, 
+            detail=f"File too large. Maximum allowed size is {MAX_FILE_SIZE_MB}MB. Your file is {file_size_mb:.2f}MB."
+        )
+
+    # --- 临时目录和路径  ---
+    session_id = str(uuid.uuid4())
+    temp_dir = os.path.join(TEMP_DIR, session_id)
+    os.makedirs(temp_dir, exist_ok=True)
+    
+    # 获取原始文件扩展名
+    _, file_extension = os.path.splitext(file.filename)
+    input_path = os.path.join(temp_dir, f"input{file_extension}")
+    output_path = os.path.join(temp_dir, f"output.avif")
+
+    logger.info(f"Processing in temporary directory: {temp_dir}")
+
+    try:
+        # --- 保存上传的文件  ---
+        logger.info(f"Saving uploaded file '{file.filename}' to '{input_path}'")
+        with open(input_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+        logger.info("File saved successfully.")
+
+        # --- 构建 Magick 命令 (您的核心需求) ---
+        cmd = [
+            'magick',
+            input_path,
+            '-define', 'avif:lossless=true',  # 无损
+            '-define', 'avif:speed=0',        # 最佳压缩率
+            output_path                      # (无 -strip，保留元信息)
+        ]
+        
+        command_str = ' '.join(cmd)
+        logger.info(f"Executing command: {command_str}")
+
+        # --- 执行 Magick 命令 (异步) ---
+        process = await asyncio.wait_for(
+            asyncio.subprocess.create_subprocess_exec(
+                *cmd,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE
+            ),
+            timeout=TIMEOUT_SECONDS
+        )
+        stdout, stderr = await process.communicate()
+
+        # --- 检查命令执行结果 ---
+        if process.returncode != 0:
+            error_message = f"Magick failed with exit code {process.returncode}."
+            logger.error(f"{error_message}\nStderr: {stderr.decode()[:1000]}")
+            raise HTTPException(status_code=500, detail=f"ImageMagick conversion failed. Error: {stderr.decode()}")
+        
+        if not os.path.exists(output_path):
+            error_message = "Magick command succeeded but output file was not found."
+            logger.error(error_message)
+            raise HTTPException(status_code=500, detail=error_message)
+        
+        # --- 成功  ---
+        logger.info(f"Conversion successful. Output file: '{output_path}'")
+        
+        # 生成下载文件名
+        original_filename_base = os.path.splitext(file.filename)[0]
+        download_filename = f"{original_filename_base}_lossless.avif"
+
+        # 注册后台清理 
+        background_tasks.add_task(cleanup_temp_dir, temp_dir)
+
+        # 返回文件 
+        return FileResponse(
+            path=output_path,
+            media_type='image/avif',
+            filename=download_filename,
+            background=background_tasks
+        )
+
+    except asyncio.TimeoutError:
+        logger.error(f"Magick processing timed out after {TIMEOUT_SECONDS}s for file '{file.filename}'.")
+        raise HTTPException(status_code=504, detail=f"Conversion timed out after {TIMEOUT_SECONDS} seconds.")
+    except HTTPException as http_exc:
+        raise http_exc
+    except Exception as e:
+        logger.error(f"An unexpected error occurred: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"An unexpected server error occurred.")
+    finally:
+        # 确保关闭文件句柄
+        await file.close()
+        # 如果没有成功注册后台任务，也尝试清理 (备用)
+        if not 'background_tasks' in locals() and os.path.exists(temp_dir):
+             cleanup_temp_dir(temp_dir)
+
+# --- 清理函数  ---
+def cleanup_temp_dir(temp_dir: str):
+    """清理临时目录及其内容的辅助函数"""
+    try:
+        if os.path.exists(temp_dir):
+            logger.info(f"Cleaning up temporary directory: {temp_dir}")
+            shutil.rmtree(temp_dir)
+            logger.info("Temporary directory cleaned up successfully.")
+    except Exception as cleanup_error:
+        logger.error(f"Error cleaning up temporary directory {temp_dir}: {cleanup_error}", exc_info=True)
+
+# --- 文件大小检查  ---
+async def get_upload_file_size(upload_file: UploadFile) -> int:
+    """获取上传文件的大小（以字节为单位）"""
+    current_position = upload_file.file.tell()
+    upload_file.file.seek(0, 2)  # 2 表示从文件末尾
+    size = upload_file.file.tell()
+    upload_file.file.seek(current_position)
+    return size

requirements.txt

@@ -0,0 +1,3 @@
+fastapi
+uvicorn[standard]
+python-multipart