# -*- 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)