Veo 3.1 Video Generator

from flask import Flask, request, jsonify, send_file, render_template_string from flask_cors import CORS from huggingface_hub import InferenceClient import tempfile import os import base64 from io import BytesIO from PIL import Image import uuid from pathlib import Path app = Flask(__name__) CORS(app) # Configuration HF_TOKEN = os.environ.get("HF_TOKEN", "your_huggingface_token_here") TEMP_DIR = Path(tempfile.gettempdir()) / "veo_videos" TEMP_DIR.mkdir(exist_ok=True) # Initialize the client client = InferenceClient( provider="fal-ai", api_key=HF_TOKEN, bill_to="huggingface", ) def cleanup_old_files(): """Clean up files older than 1 hour""" import time current_time = time.time() for file_path in TEMP_DIR.glob("*.mp4"): if current_time - file_path.stat().st_mtime > 3600: try: file_path.unlink() except: pass # HTML Template for the website HTML_TEMPLATE = """ Veo 3.1 Video Generator

📝 Text to Video

🖼️ Image to Video

📚 API Docs

Generate Video from Text

🏖️ Beach Sunset

🌃 City Night

🦅 Eagle Flight

🌸 Cherry Blossoms

Generating your video... This may take a minute.

Your Generated Video

Animate Your Image

Animating your image... This may take a minute.

Your Animated Video

API Documentation

Use these endpoints to integrate video generation into your applications.

1. Text to Video

Endpoint: POST /api/text-to-video

curl -X POST http://localhost:5000/api/text-to-video \\
  -H "Content-Type: application/json" \\
  -d '{"prompt": "A young man walking on the street during sunset"}'
                

2. Image to Video

Endpoint: POST /api/image-to-video

curl -X POST http://localhost:5000/api/image-to-video \\
  -F "image=@photo.jpg" \\
  -F "prompt=The person starts walking forward"
                

Python Example

import requests

# Text to Video
response = requests.post('http://localhost:5000/api/text-to-video', 
    json={'prompt': 'A sunset over the ocean'})
data = response.json()
print(data['message'])

# Image to Video
with open('image.jpg', 'rb') as f:
    response = requests.post('http://localhost:5000/api/image-to-video',
        files={'image': f},
        data={'prompt': 'Camera zooms in slowly'})
print(response.json())
                

Response Format

{
  "success": true,
  "video_id": "uuid-here",
  "video_base64": "base64_encoded_video",
  "prompt": "your prompt",
  "message": "Video generated successfully"
}
                

""" @app.route('/') def index(): """Render the main website""" return render_template_string(HTML_TEMPLATE) @app.route('/health', methods=['GET']) def health_check(): """Health check endpoint""" return jsonify({ "status": "healthy", "service": "Veo 3.1 Video Generation API", "version": "1.0" }) @app.route('/api/text-to-video', methods=['POST']) def text_to_video(): """Generate video from text prompt""" try: data = request.get_json() if not data or 'prompt' not in data: return jsonify({ "error": "Missing 'prompt' in request body" }), 400 prompt = data.get('prompt', '').strip() if not prompt: return jsonify({ "error": "Prompt cannot be empty" }), 400 print(f"Generating video from prompt: {prompt[:50]}...") video_bytes = client.text_to_video( prompt, model="akhaliq/veo3.1-fast", ) video_id = str(uuid.uuid4()) video_path = TEMP_DIR / f"{video_id}.mp4" with open(video_path, "wb") as f: f.write(video_bytes) cleanup_old_files() return_type = data.get('return_type', 'base64') if return_type == 'file': return send_file( video_path, mimetype='video/mp4', as_attachment=True, download_name=f"generated_{video_id}.mp4" ) else: video_base64 = base64.b64encode(video_bytes).decode('utf-8') return jsonify({ "success": True, "video_id": video_id, "video_base64": video_base64, "prompt": prompt, "message": "Video generated successfully" }) except Exception as e: print(f"Error in text_to_video: {str(e)}") return jsonify({ "error": f"Failed to generate video: {str(e)}" }), 500 @app.route('/api/image-to-video', methods=['POST']) def image_to_video(): """Generate video from image and motion prompt""" try: if request.is_json: data = request.get_json() if not data or 'image_base64' not in data or 'prompt' not in data: return jsonify({ "error": "Missing 'image_base64' or 'prompt' in request body" }), 400 image_data = base64.b64decode(data['image_base64']) prompt = data.get('prompt', '').strip() else: if 'image' not in request.files: return jsonify({ "error": "Missing 'image' file in request" }), 400 image_file = request.files['image'] image_data = image_file.read() prompt = request.form.get('prompt', '').strip() if not prompt: return jsonify({ "error": "Prompt cannot be empty" }), 400 try: img = Image.open(BytesIO(image_data)) if img.mode != 'RGB': img = img.convert('RGB') img_buffer = BytesIO() img.save(img_buffer, format='PNG') image_data = img_buffer.getvalue() except Exception as e: return jsonify({ "error": f"Invalid image format: {str(e)}" }), 400 print(f"Generating video from image with prompt: {prompt[:50]}...") video_bytes = client.image_to_video( image_data, prompt=prompt, model="akhaliq/veo3.1-fast-image-to-video", ) video_id = str(uuid.uuid4()) video_path = TEMP_DIR / f"{video_id}.mp4" with open(video_path, "wb") as f: f.write(video_bytes) cleanup_old_files() return_type = request.form.get('return_type') if not request.is_json else request.get_json().get('return_type', 'base64') if return_type == 'file': return send_file( video_path, mimetype='video/mp4', as_attachment=True, download_name=f"animated_{video_id}.mp4" ) else: video_base64 = base64.b64encode(video_bytes).decode('utf-8') return jsonify({ "success": True, "video_id": video_id, "video_base64": video_base64, "prompt": prompt, "message": "Video generated successfully" }) except Exception as e: print(f"Error in image_to_video: {str(e)}") return jsonify({ "error": f"Failed to generate video: {str(e)}" }), 500 @app.route('/api/download/', methods=['GET']) def download_video(video_id): """Download a previously generated video by ID""" try: video_path = TEMP_DIR / f"{video_id}.mp4" if not video_path.exists(): return jsonify({ "error": "Video not found or expired" }), 404 return send_file( video_path, mimetype='video/mp4', as_attachment=True, download_name=f"video_{video_id}.mp4" ) except Exception as e: return jsonify({ "error": f"Failed to download video: {str(e)}" }), 500 if __name__ == '__main__': print(""" ╔═══════════════════════════════════════════════════╗ ║ Veo 3.1 Video Generation - Website + API ║ ╚═══════════════════════════════════════════════════╝ 📝 Set your HF_TOKEN environment variable: export HF_TOKEN=your_huggingface_token_here 🌐 Website: http://localhost:5000 🔌 API Endpoints: - POST /api/text-to-video - POST /api/image-to-video - GET /api/download/ 🚀 Server starting... """) app.run( host='0.0.0.0', port=7860, debug=True )