File size: 2,373 Bytes
493dfff
 
 
 
 
 
d9a9735
 
493dfff
d9a9735
493dfff
 
 
 
d9a9735
 
 
 
 
 
 
 
 
 
493dfff
 
 
 
 
 
d9a9735
493dfff
 
 
 
d9a9735
493dfff
 
d9a9735
 
 
 
493dfff
 
 
 
 
 
d9a9735
 
493dfff
 
 
 
 
 
 
 
 
 
 
 
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
import os
import uuid
import torch
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pydantic import BaseModel
# سنستخدم الـ Pipeline العام للـ GGUF ليتناسب مع الـ CPU
from diffusers import DiffusionPipeline 

app = FastAPI(title="Free CPU Video API")

OUTPUT_DIR = "/tmp/generated_videos"
os.makedirs(OUTPUT_DIR, exist_ok=True)

print("Loading Quantized Model for CPU...")
# تحميل نسخة خفيفة جداً ومضغوطة تعمل على المعالج وبدون ذاكرة كارت شاشة
pipeline = DiffusionPipeline.from_pretrained(
    "Lightricks/LTX-Video",
    torch_dtype=torch.float32, # الـ CPU يعمل بـ float32
    use_safetensors=True
)
# تحديد تشغيل النموذج على الـ CPU مجاناً
pipeline.to("cpu") 
print("Model loaded successfully on CPU!")

class VideoPrompt(BaseModel):
    prompt: str

@app.get("/")
def home():
    return {"status": "Server is running on FREE CPU!"}

@app.post("/generate")
async def generate_video(video_req: VideoPrompt):
    try:
        # تقليل الأبعاد وعدد الفريمات والخطوات لأقصى حد لكي لا يموت الـ CPU
        video_frames = pipeline(
            prompt=video_req.prompt,
            num_frames=17,         # تقليل الفريمات (حوالي ثانيتين) لتسريع المعالجة
            height=256,            # تقليل الأبعاد لـ 256 لكي يتحملها الـ CPU
            width=256,
            num_inference_steps=7, # تقليل الخطوات جداً لتوليد سريع مجاني
            guidance_scale=3.0
        ).frames[0]
        
        file_name = f"{uuid.uuid4()}.mp4"
        file_path = os.path.join(OUTPUT_DIR, file_name)
        
        from diffusers.utils import export_to_video
        export_to_video(video_frames, file_path, fps=8)
        
        return {"status": "success", "video_name": file_name}
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/download/{file_name}")
async def download_video(file_name: str):
    file_path = os.path.join(OUTPUT_DIR, file_name)
    if os.path.exists(file_path):
        return FileResponse(file_path, media_type="video/mp4")
    raise HTTPException(status_code=404, detail="Video not found")