video-eval/app/api/eval/index.dspy

303 lines
11 KiB
Plaintext

# -*- coding:utf-8 -*-
import json
import subprocess
import os
import traceback
from ahserver.filestorage import FileStorage
try:
fs = FileStorage()
# Get uploaded file - params_kw gives web path for uploads
web_path = None
file_path = None
if 'video' in params_kw:
web_path = params_kw['video']
elif 'file' in params_kw:
web_path = params_kw['file']
if not web_path:
return json.dumps({
"error": "No video file uploaded. Use multipart/form-data with field 'video' or 'file'.",
"score": 0,
"passed": False,
"details": {},
"suggestions": ["Upload a video file (mp4) via POST multipart/form-data"]
}, ensure_ascii=False)
if isinstance(web_path, str) and len(web_path) > 0:
file_path = fs.realPath(web_path)
if not file_path or not os.path.exists(file_path):
return json.dumps({
"error": "Uploaded file not found",
"web_path": str(web_path),
"resolved_path": str(file_path),
"score": 0,
"passed": False,
"details": {},
"suggestions": []
}, ensure_ascii=False)
score = 100
details = {}
suggestions = []
duration = 0
# --- ffprobe: get overall info ---
probe_data = None
try:
probe_cmd = [
"ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", file_path
]
probe_out = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=60)
probe_data = json.loads(probe_out.stdout)
except Exception as e:
return json.dumps({
"error": "ffprobe failed: " + str(e),
"score": 0,
"passed": False,
"details": {},
"suggestions": []
}, ensure_ascii=False)
fmt = probe_data.get("format", {})
streams = probe_data.get("streams", [])
video_stream = None
audio_stream = None
for s in streams:
if s.get("codec_type") == "video" and video_stream is None:
video_stream = s
elif s.get("codec_type") == "audio" and audio_stream is None:
audio_stream = s
# 1) Duration
duration = float(fmt.get("duration", 0))
details["duration_seconds"] = round(duration, 2)
if duration < 1:
score -= 20
suggestions.append("Video is extremely short (less than 1 second)")
elif duration < 5:
score -= 5
suggestions.append("Video is very short (less than 5 seconds)")
# 2) Resolution and aspect ratio
if video_stream:
width = int(video_stream.get("width", 0))
height = int(video_stream.get("height", 0))
details["resolution"] = str(width) + "x" + str(height)
if width > 0 and height > 0:
from math import gcd
g = gcd(width, height)
ar_w = width // g
ar_h = height // g
details["aspect_ratio"] = str(ar_w) + ":" + str(ar_h)
pixels = width * height
if pixels >= 3840 * 2160:
details["quality_tier"] = "4K UHD"
elif pixels >= 1920 * 1080:
details["quality_tier"] = "Full HD"
elif pixels >= 1280 * 720:
details["quality_tier"] = "HD"
elif pixels >= 640 * 480:
details["quality_tier"] = "SD"
score -= 10
suggestions.append("Consider upgrading to at least HD (1280x720) resolution")
else:
details["quality_tier"] = "Low"
score -= 20
suggestions.append("Resolution is very low. Use at least 640x480 for acceptable quality")
else:
score -= 15
suggestions.append("Could not determine video resolution")
else:
score -= 50
suggestions.append("No video stream found in the file")
# 3) Bitrate and codec
if video_stream:
codec_name = video_stream.get("codec_name", "unknown")
details["video_codec"] = codec_name
v_bitrate = video_stream.get("bit_rate")
if v_bitrate:
v_bitrate = int(v_bitrate)
else:
fmt_bitrate = fmt.get("bit_rate")
if fmt_bitrate:
v_bitrate = int(fmt_bitrate)
if v_bitrate:
v_bitrate_kbps = v_bitrate // 1000
details["video_bitrate_kbps"] = v_bitrate_kbps
w = int(video_stream.get("width", 0))
h = int(video_stream.get("height", 0))
pixels = w * h
if pixels > 0:
bpp = v_bitrate / (pixels * 30)
details["bits_per_pixel_per_frame"] = round(bpp, 3)
if bpp < 0.01:
score -= 15
suggestions.append("Bitrate is very low for the resolution - expect compression artifacts")
elif bpp < 0.03:
score -= 5
suggestions.append("Bitrate is somewhat low - consider increasing for better quality")
if codec_name not in ["h264", "hevc", "h265", "av1", "vp9"]:
score -= 5
suggestions.append("Video codec '" + codec_name + "' is not modern. Consider H.264 or H.265")
# Audio info
if audio_stream:
a_codec = audio_stream.get("codec_name", "unknown")
a_sample_rate = audio_stream.get("sample_rate", "unknown")
a_channels = audio_stream.get("channels", 0)
details["audio_codec"] = a_codec
details["audio_sample_rate"] = a_sample_rate
details["audio_channels"] = a_channels
a_bitrate = audio_stream.get("bit_rate")
if a_bitrate:
details["audio_bitrate_kbps"] = int(a_bitrate) // 1000
if a_codec not in ["aac", "opus", "mp3", "flac", "vorbis", "ac3", "eac3"]:
score -= 5
suggestions.append("Audio codec '" + a_codec + "' may have compatibility issues")
else:
details["audio_codec"] = "none"
suggestions.append("No audio stream detected")
# 4) Frame rate
if video_stream:
fps_str = video_stream.get("r_frame_rate", "0/1")
try:
num, den = fps_str.split("/")
fps = float(num) / float(den) if float(den) != 0 else 0
except Exception:
fps = 0
details["frame_rate_fps"] = round(fps, 2)
avg_fps_str = video_stream.get("avg_frame_rate", "0/1")
try:
num2, den2 = avg_fps_str.split("/")
avg_fps = float(num2) / float(den2) if float(den2) != 0 else 0
except Exception:
avg_fps = fps
details["avg_frame_rate_fps"] = round(avg_fps, 2)
if fps > 0 and avg_fps > 0:
fps_diff = abs(fps - avg_fps) / fps
details["frame_rate_consistency"] = round(1.0 - fps_diff, 4)
if fps_diff > 0.1:
score -= 10
suggestions.append("Frame rate is inconsistent - may cause playback stuttering")
elif fps_diff > 0.05:
score -= 3
else:
details["frame_rate_consistency"] = "unknown"
if fps < 20:
score -= 10
suggestions.append("Frame rate is below 20fps - video may appear choppy")
# 5) Audio/video sync check
if video_stream and audio_stream:
v_start = float(video_stream.get("start_time", 0))
a_start = float(audio_stream.get("start_time", 0))
sync_diff = abs(v_start - a_start)
details["av_sync_offset_seconds"] = round(sync_diff, 4)
if sync_diff > 0.1:
score -= 15
suggestions.append("Audio/video sync offset: " + str(round(sync_diff, 3)) + "s. May cause lip-sync issues")
elif sync_diff > 0.05:
score -= 5
details["av_sync_status"] = "minor offset"
else:
details["av_sync_status"] = "good"
else:
details["av_sync_status"] = "N/A (missing audio or video stream)"
# 6) Scene change detection using ffmpeg
try:
scene_cmd = [
"ffmpeg", "-i", file_path,
"-vf", "select=gt(scene\\,0.3),showinfo",
"-vsync", "vfr", "-f", "null", "-"
]
scene_out = subprocess.run(scene_cmd, capture_output=True, text=True, timeout=120)
scene_lines = [l for l in scene_out.stderr.split("\n") if "showinfo" in l and "pts_time" in l]
scene_count = len(scene_lines)
details["scene_changes_detected"] = scene_count
if duration > 0:
scene_rate = scene_count / duration
details["scene_changes_per_second"] = round(scene_rate, 3)
if scene_rate > 2:
score -= 5
suggestions.append("Very high scene change rate - video may be too fast-paced")
except subprocess.TimeoutExpired:
details["scene_changes_detected"] = "timeout"
suggestions.append("Scene change detection timed out")
except Exception as e:
details["scene_changes_detected"] = "error: " + str(e)[:100]
# 7) Subtitle / text overlay detection
try:
text_detect_cmd = [
"ffmpeg", "-i", file_path,
"-vf", "crop=iw:ih*0.15:0:ih*0.85,format=gray,threshold=0.7",
"-frames:v", "5", "-f", "image2pipe", "-vcodec", "rawvideo",
"-pix_fmt", "gray", "-"
]
text_out = subprocess.run(text_detect_cmd, capture_output=True, timeout=60)
if text_out.stdout and len(text_out.stdout) > 0:
data = text_out.stdout
total = len(data)
white_count = sum(1 for b in data if b > 200)
white_ratio = white_count / total if total > 0 else 0
details["subtitle_text_ratio"] = round(white_ratio, 4)
if white_ratio > 0.05:
details["subtitles_burned_in"] = True
suggestions.append("Burned-in subtitles detected. Consider separate subtitle tracks")
elif white_ratio > 0.02:
details["subtitles_burned_in"] = "possible"
suggestions.append("Possible burned-in subtitles - verify manually")
else:
details["subtitles_burned_in"] = False
else:
details["subtitles_burned_in"] = "unknown"
except subprocess.TimeoutExpired:
details["subtitles_burned_in"] = "timeout"
except Exception as e:
details["subtitles_burned_in"] = "error: " + str(e)[:100]
# Clamp score
score = max(0, min(100, score))
passed = score >= 60
return json.dumps({
"score": score,
"passed": passed,
"details": details,
"suggestions": suggestions
}, ensure_ascii=False)
except Exception as e:
return json.dumps({
"error": str(e),
"traceback": traceback.format_exc(),
"score": 0,
"passed": False,
"details": {},
"suggestions": []
}, ensure_ascii=False)