File size: 3,405 Bytes
58f0729 |
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 |
#!/usr/bin/env python3
"""
Test script for the HuggingFace Segment-Based Video Highlights API
"""
import requests
import time
import json
from pathlib import Path
# API configuration
API_BASE = "http://localhost:7860" # Change to your deployed URL
TEST_VIDEO = "../test_video/test.mp4" # Adjust path as needed
def test_api():
"""Test the complete API workflow"""
print("🧪 Testing HuggingFace Segment-Based Video Highlights API")
# Check if test video exists
if not Path(TEST_VIDEO).exists():
print(f"❌ Test video not found: {TEST_VIDEO}")
return
try:
# 1. Test health endpoint
print("\n1️⃣ Testing health endpoint...")
response = requests.get(f"{API_BASE}/health")
print(f"Health check: {response.status_code} - {response.json()}")
# 2. Upload video
print("\n2️⃣ Uploading video...")
with open(TEST_VIDEO, 'rb') as video_file:
files = {'video': video_file}
data = {
'segment_length': 5.0,
'model_name': 'HuggingFaceTB/SmolVLM2-256M-Video-Instruct',
'with_effects': True
}
response = requests.post(f"{API_BASE}/upload-video", files=files, data=data)
if response.status_code != 200:
print(f"❌ Upload failed: {response.status_code} - {response.text}")
return
job_data = response.json()
job_id = job_data['job_id']
print(f"✅ Video uploaded successfully! Job ID: {job_id}")
# 3. Monitor job status
print("\n3️⃣ Monitoring job progress...")
while True:
response = requests.get(f"{API_BASE}/job-status/{job_id}")
if response.status_code != 200:
print(f"❌ Status check failed: {response.status_code}")
break
status_data = response.json()
print(f"Status: {status_data['status']} - {status_data['message']} ({status_data['progress']}%)")
if status_data['status'] == 'completed':
print(f"✅ Processing completed!")
print(f"📹 Highlights URL: {status_data['highlights_url']}")
print(f"📊 Analysis URL: {status_data['analysis_url']}")
print(f"🎬 Segments: {status_data['selected_segments']}/{status_data['total_segments']}")
print(f"📈 Compression: {status_data['compression_ratio']:.1%}")
break
elif status_data['status'] == 'failed':
print(f"❌ Processing failed: {status_data['message']}")
break
time.sleep(5) # Wait 5 seconds before checking again
# 4. Download results (optional)
if status_data['status'] == 'completed':
print("\n4️⃣ Download URLs available:")
print(f"Highlights: {API_BASE}{status_data['highlights_url']}")
print(f"Analysis: {API_BASE}{status_data['analysis_url']}")
except requests.exceptions.ConnectionError:
print(f"❌ Cannot connect to API at {API_BASE}")
print("Make sure the API server is running with: uvicorn app:app --host 0.0.0.0 --port 7860")
except Exception as e:
print(f"❌ Test failed: {str(e)}")
if __name__ == "__main__":
test_api()
|