Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

fastapi==0.104.1
uvicorn==0.24.0
celery==5.3.4
python-dotenv==1.0.0
requests==2.31.0

from dotenv import load_dotenv
import os

load_dotenv()

# RabbitMQ配置
RABBITMQ_HOST = os.getenv("RABBITMQ_HOST", "localhost")
RABBITMQ_PORT = os.getenv("RABBITMQ_PORT", "5672")
RABBITMQ_USER = os.getenv("RABBITMQ_USER", "guest")
RABBITMQ_PASS = os.getenv("RABBITMQ_PASS", "guest")

# Celery配置
CELERY_BROKER_URL = f"amqp://{RABBITMQ_USER}:{RABBITMQ_PASS}@{RABBITMQ_HOST}:{RABBITMQ_PORT}//"
CELERY_RESULT_BACKEND = "rpc://"

# 定时任务配置
CELERY_BEAT_SCHEDULE = {
    'process-images-every-hour': {
        'task': 'app.tasks.process_images',
        'schedule': 3600.0,  # 每小时执行一次
    },
    'daily-cleanup': {
        'task': 'app.tasks.cleanup_old_images',
        'schedule': 86400.0,  # 每天执行一次
    }
}

from celery import Celery
from app.config import CELERY_BROKER_URL, CELERY_RESULT_BACKEND, CELERY_BEAT_SCHEDULE

celery_app = Celery(
    'image_processing',
    broker=CELERY_BROKER_URL,
    backend=CELERY_RESULT_BACKEND,
    include=['app.tasks']
)

# 配置定时任务
celery_app.conf.beat_schedule = CELERY_BEAT_SCHEDULE
celery_app.conf.timezone = 'Asia/Shanghai'

from app.celery_app import celery_app
from app.watermark import ImageWatermarker
import os
from datetime import datetime, timedelta

@celery_app.task
def add_watermark_task(image_path, text, position='center'):
    """异步添加水印任务"""
    watermarker = ImageWatermarker()
    try:
        result_path = watermarker.add_watermark(
            image_path=image_path,
            text=text,
            position=position
        )
        return {"status": "success", "output_path": result_path}
    except Exception as e:
        return {"status": "error", "message": str(e)}

@celery_app.task
def process_images():
    """定时处理图片任务"""
    image_dir = "images/pending"
    if not os.path.exists(image_dir):
        return {"status": "error", "message": "Pending directory not found"}

    processed = 0
    for image in os.listdir(image_dir):
        if image.lower().endswith(('.png', '.jpg', '.jpeg')):
            add_watermark_task.delay(
                os.path.join(image_dir, image),
                "自动处理水印",
                'center'
            )
            processed += 1

    return {"status": "success", "processed": processed}

@celery_app.task
def cleanup_old_images():
    """清理旧图片任务"""
    output_dir = "images/processed"
    if not os.path.exists(output_dir):
        return {"status": "error", "message": "Output directory not found"}

    threshold_date = datetime.now() - timedelta(days=7)
    cleaned = 0

    for image in os.listdir(output_dir):
        image_path = os.path.join(output_dir, image)
        if os.path.getctime(image_path) < threshold_date.timestamp():
            os.remove(image_path)
            cleaned += 1

    return {"status": "success", "cleaned": cleaned}

from fastapi import FastAPI, File, UploadFile, BackgroundTasks
from fastapi.responses import JSONResponse
import os
from app.tasks import add_watermark_task
from app.celery_app import celery_app

app = FastAPI(title="图片水印处理服务")

@app.post("/upload/")
async def upload_image(
    file: UploadFile = File(...),
    text: str = "水印文本",
    position: str = "center"
):
    # 保存上传的文件
    file_path = f"images/uploads/{file.filename}"
    os.makedirs(os.path.dirname(file_path), exist_ok=True)
    
    with open(file_path, "wb") as buffer:
        content = await file.read()
        buffer.write(content)
    
    # 创建异步任务
    task = add_watermark_task.delay(file_path, text, position)
    
    return JSONResponse({
        "status": "success",
        "message": "图片已上传并加入处理队列",
        "task_id": task.id
    })

@app.get("/task/{task_id}")
async def get_task_status(task_id: str):
    task = celery_app.AsyncResult(task_id)
    if task.ready():
        return {"status": "completed", "result": task.result}
    return {"status": "processing"}

@app.get("/tasks/scheduled")
async def get_scheduled_tasks():
    return {"tasks": celery_app.conf.beat_schedule}

from app.celery_app import celery_app

if __name__ == '__main__':
    celery_app.start()

pip install -r requirements.txt

uvicorn app.main:app --reload

celery -A celery_worker.celery_app worker --loglevel=info

celery -A celery_worker.celery_app beat --loglevel=info