Spaces:
Paused
Paused
File size: 4,069 Bytes
74b0b87 6220d49 6f26361 d076da0 01b162f 6f26361 069955a 045b9d1 6f26361 74b0b87 6220d49 6f26361 0db219f d11742b 045b9d1 01b162f 9a30748 d0612e6 e1b6327 6f26361 9a30748 01b162f e1b6327 74b0b87 01b162f 79dfa90 6f26361 74b0b87 79dfa90 6f26361 01b162f 74b0b87 8a28b7b 60fff37 74b0b87 e1b6327 74b0b87 e1b6327 74b0b87 045b9d1 6f26361 d0612e6 74b0b87 5b4e72b 74b0b87 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
from fastapi import FastAPI, Query, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
import os
import ffmpeg
import uuid
import cloudinary
import cloudinary.uploader
from cloudinary.utils import cloudinary_url
VIDEO_DIR = "compressed_videos"
os.makedirs(VIDEO_DIR, exist_ok=True)
app = FastAPI()
app.mount("/videos", StaticFiles(directory=VIDEO_DIR), name="videos")
@app.get("/compress_video")
async def compress_video(
video_url: str = Query(..., description="URL of the video"),
resolution: str = Query("1280x720", description="Resolution (e.g., 1280x720, 640x360)"),
format: str = Query("matroska", description="Output video format (e.g., mp4, mkv)"),
codec: str = Query("libx264", description="264 faster but bigger size,265 slower but smaller size"),
preset: str = Query("fast", description="ultrafast, superfast, fast, medium, slow"),
crf: int = Query(28, description="Constant Rate Factor (CRF) for compression (0-51, lower is better quality)")
):
try:
file_name = f"{uuid.uuid4()}.{format}"
file_path = os.path.join(VIDEO_DIR, file_name)
(
ffmpeg
.input(video_url)
.output(file_path, crf=crf, vcodec=codec, preset=preset, s=resolution, f=format)
.global_args("-threads", "0")
.overwrite_output()
.run()
)
file_url = f"/videos/{file_name}"
return JSONResponse(content={"compressed_video_url": "https://akane710-imagecompress.hf.space" + file_url})
except Exception as e:
return JSONResponse(content={"error": str(e)}, status_code=500)
@app.post("/upload_video")
async def upload_video(
file: UploadFile = File(...),
resolution: str = Query("1280x720", description="Resolution (e.g., 1280x720, 640x360)"),
format: str = Query("matroska", description="Output video format (e.g., mp4, mkv)"),
codec: str = Query("libx264", description="264 faster but bigger size,265 slower but smaller size"),
preset: str = Query("fast", description="ultrafast, superfast, fast, medium, slow"),
crf: int = Query(28, description="Constant Rate Factor (CRF) for compression (0-51, lower is better quality)")):
try:
original_filename = file.filename
file_ext = original_filename.split(".")[-1]
temp_filename = f"{uuid.uuid4()}.{file_ext}"
temp_filepath = os.path.join(VIDEO_DIR, temp_filename)
with open(temp_filepath, "wb") as buffer:
buffer.write(file.file.read())
compressed_filename = f"{uuid.uuid4()}.{format}"
compressed_filepath = os.path.join(VIDEO_DIR, compressed_filename)
(
ffmpeg
.input(temp_filepath)
.output(compressed_filepath, crf=crf, vcodec=codec, preset=preset, s=resolution, f=format)
.global_args("-threads", "0")
.overwrite_output()
.run()
)
os.remove(temp_filepath) # Clean up the original uploaded file
file_url = f"/videos/{compressed_filename}"
return JSONResponse(content={"compressed_video_url": "https://akane710-imagecompress.hf.space" + file_url})
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
cloudinary.config(
cloud_name=os.getenv("cloudname"),
api_key=os.getenv("key"),
api_secret=os.getenv("secret"),
secure=True
)
@app.get("/compress_image")
async def compress_image(image_url: str = Query(..., description="URL of the image to compress")):
try:
upload_result = cloudinary.uploader.upload(image_url)
compressed_url, options = cloudinary_url(
upload_result["public_id"],
fetch_format="auto",
quality="auto:low"
)
return JSONResponse(content={"compressed_image_url": compressed_url})
except Exception as e:
return JSONResponse(content={"error": str(e)}, status_code=500)
@app.get("/")
def root():
return "بتعمل شنو هنا يا زول"
|