150 lines
4.6 KiB
Python
150 lines
4.6 KiB
Python
# api_server.py - REST-API für den Videogenerator
|
|
# Im webui-Verzeichnis speichern
|
|
|
|
from fastapi import FastAPI, File, UploadFile, BackgroundTasks, Form
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import uvicorn
|
|
import os
|
|
import subprocess
|
|
import uuid
|
|
import shutil
|
|
from typing import Optional
|
|
|
|
app = FastAPI(title="Hunyuan Video Generator API")
|
|
|
|
# CORS für n8n-Zugriff erlauben
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # Im Produktivbetrieb einschränken auf n8n-Server-IP
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Verzeichnisse für temporäre Dateien und Ausgaben
|
|
TEMP_DIR = os.path.join(os.path.dirname(__file__), "temp_uploads")
|
|
OUTPUT_DIR = os.path.join(os.path.dirname(__file__), "outputs")
|
|
PYTHON_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "system", "python", "python.exe")
|
|
|
|
os.makedirs(TEMP_DIR, exist_ok=True)
|
|
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
|
|
|
def generate_video_task(image_path: str, prompt: str, n_prompt: str,
|
|
seed: int, length: float, steps: int,
|
|
use_teacache: bool, output_filename: str,
|
|
job_id: str):
|
|
"""Führt die Videogenerierung als Hintergrundaufgabe aus"""
|
|
|
|
# Befehlszeile erstellen
|
|
cmd = [
|
|
PYTHON_PATH,
|
|
"hunyuan_cli.py",
|
|
"--image", image_path,
|
|
"--prompt", prompt,
|
|
"--seed", str(seed),
|
|
"--length", str(length),
|
|
"--steps", str(steps),
|
|
"--output", f"{job_id}_{output_filename}"
|
|
]
|
|
|
|
if n_prompt:
|
|
cmd.extend(["--n_prompt", n_prompt])
|
|
|
|
if use_teacache:
|
|
cmd.append("--teacache")
|
|
|
|
# Befehl ausführen
|
|
try:
|
|
subprocess.run(cmd, check=True, cwd=os.path.dirname(__file__))
|
|
print(f"Video generation completed for job {job_id}")
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Error generating video: {e}")
|
|
|
|
@app.post("/generate/")
|
|
async def generate_video(
|
|
background_tasks: BackgroundTasks,
|
|
image: UploadFile = File(...),
|
|
prompt: str = Form(...),
|
|
n_prompt: Optional[str] = Form(""),
|
|
seed: Optional[int] = Form(31337),
|
|
length: Optional[float] = Form(5.0),
|
|
steps: Optional[int] = Form(25),
|
|
use_teacache: Optional[bool] = Form(True),
|
|
output_filename: Optional[str] = Form("output.mp4")
|
|
):
|
|
"""
|
|
Generiert ein Video basierend auf den angegebenen Parametern
|
|
"""
|
|
# Eindeutige Job-ID generieren
|
|
job_id = str(uuid.uuid4())
|
|
|
|
# Temporären Dateipfad für das Bild erstellen
|
|
temp_image_path = os.path.join(TEMP_DIR, f"{job_id}_{image.filename}")
|
|
|
|
# Bild speichern
|
|
with open(temp_image_path, "wb") as buffer:
|
|
shutil.copyfileobj(image.file, buffer)
|
|
|
|
# Videogenerierung als Hintergrundaufgabe starten
|
|
background_tasks.add_task(
|
|
generate_video_task,
|
|
temp_image_path,
|
|
prompt,
|
|
n_prompt,
|
|
seed,
|
|
length,
|
|
steps,
|
|
use_teacache,
|
|
output_filename,
|
|
job_id
|
|
)
|
|
|
|
return {
|
|
"status": "processing",
|
|
"job_id": job_id,
|
|
"message": "Video generation started in background",
|
|
"result_url": f"/result/{job_id}_{output_filename}"
|
|
}
|
|
|
|
@app.get("/result/{filename}")
|
|
async def get_result(filename: str):
|
|
"""
|
|
Gibt das generierte Video zurück, wenn es verfügbar ist
|
|
"""
|
|
file_path = os.path.join(OUTPUT_DIR, filename)
|
|
|
|
if os.path.exists(file_path):
|
|
return FileResponse(
|
|
file_path,
|
|
media_type="video/mp4",
|
|
filename=filename.split("_", 1)[1] # Entferne Job-ID vom Dateinamen
|
|
)
|
|
else:
|
|
return {"status": "not_found", "message": "Requested video not found or still processing"}
|
|
|
|
@app.get("/status/{job_id}")
|
|
async def check_status(job_id: str):
|
|
"""
|
|
Prüft den Status einer Videogenerierung
|
|
"""
|
|
# Suche nach Dateien, die mit der Job-ID beginnen
|
|
result_files = [f for f in os.listdir(OUTPUT_DIR) if f.startswith(job_id)]
|
|
|
|
if result_files:
|
|
return {
|
|
"status": "completed",
|
|
"job_id": job_id,
|
|
"files": result_files,
|
|
"download_urls": [f"/result/{file}" for file in result_files]
|
|
}
|
|
else:
|
|
# Prüfen, ob das Eingabebild noch existiert (Verarbeitung läuft noch)
|
|
input_files = [f for f in os.listdir(TEMP_DIR) if f.startswith(job_id)]
|
|
if input_files:
|
|
return {"status": "processing", "job_id": job_id}
|
|
else:
|
|
return {"status": "not_found", "job_id": job_id}
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |