| from fastapi import FastAPI, UploadFile, File, Form, HTTPException |
| from fastapi.responses import StreamingResponse |
| import subprocess |
| import io |
| import os |
| from typing import Optional |
| from urllib.parse import urlparse |
|
|
| app = FastAPI() |
|
|
| @app.post("/process") |
| async def process_image( |
| file: Optional[UploadFile] = File(None), |
| url: Optional[str] = Form(None), |
| options: Optional[str] = Form("-q:v 2"), |
| filename: Optional[str] = Form(None) |
| ): |
| input_source = "" |
| input_data = None |
|
|
| |
| if file: |
| input_data = await file.read() |
| input_source = "pipe:0" |
| |
| if not filename: |
| filename = os.path.splitext(file.filename)[0] + ".jpg" |
| elif url: |
| input_source = url |
| |
| if not filename: |
| path = urlparse(url).path |
| base_name = os.path.basename(path) or "downloaded_image" |
| filename = os.path.splitext(base_name)[0] + ".jpg" |
| else: |
| raise HTTPException(status_code=400, detail="Missing source") |
|
|
| |
| command = ['ffmpeg', '-i', input_source] |
| if options: |
| command.extend(options.split()) |
| command.extend(['-f', 'image2', 'pipe:1']) |
|
|
| try: |
| process = subprocess.Popen( |
| command, |
| stdin=subprocess.PIPE if input_data else None, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE |
| ) |
| out, err = process.communicate(input=input_data) |
|
|
| |
| |
| headers = { |
| 'Content-Disposition': f'attachment; filename="{filename}"' |
| } |
| |
| return StreamingResponse( |
| io.BytesIO(out), |
| media_type="image/jpeg", |
| headers=headers |
| ) |
|
|
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
