192 lines
7.7 KiB
Plaintext
192 lines
7.7 KiB
Plaintext
import json
|
||
import subprocess
|
||
import os
|
||
import re
|
||
|
||
from ahserver.filestorage import FileStorage
|
||
|
||
result = {}
|
||
|
||
try:
|
||
# Get uploaded file
|
||
video_rel = params_kw.get("video_file", "")
|
||
if not video_rel:
|
||
result = {"error": "missing video_file parameter", "score": 0, "passed": False}
|
||
else:
|
||
fs = FileStorage()
|
||
video_path = fs.realPath(video_rel)
|
||
|
||
if not os.path.exists(video_path):
|
||
result = {"error": f"file not found: {video_path}", "score": 0, "passed": False}
|
||
else:
|
||
details = {}
|
||
score = 100
|
||
suggestions = []
|
||
|
||
# 1. ffprobe - get basic info
|
||
probe_cmd = [
|
||
"ffprobe", "-v", "quiet", "-print_format", "json",
|
||
"-show_format", "-show_streams", video_path
|
||
]
|
||
probe_out = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=30)
|
||
probe = json.loads(probe_out.stdout) if probe_out.returncode == 0 else {}
|
||
|
||
format_info = probe.get("format", {})
|
||
streams = probe.get("streams", [])
|
||
|
||
# Find video and audio streams
|
||
v_stream = None
|
||
a_stream = None
|
||
for s in streams:
|
||
if s.get("codec_type") == "video" and not v_stream:
|
||
v_stream = s
|
||
elif s.get("codec_type") == "audio" and not a_stream:
|
||
a_stream = s
|
||
|
||
# 2. Resolution check
|
||
if v_stream:
|
||
width = int(v_stream.get("width", 0))
|
||
height = int(v_stream.get("height", 0))
|
||
details["resolution"] = f"{width}x{height}"
|
||
details["codec"] = v_stream.get("codec_name", "unknown")
|
||
details["duration"] = float(format_info.get("duration", 0))
|
||
details["bitrate"] = int(format_info.get("bit_rate", 0))
|
||
|
||
# FPS
|
||
fps_str = v_stream.get("r_frame_rate", "0/1")
|
||
if "/" in fps_str:
|
||
num, den = fps_str.split("/")
|
||
fps = float(num) / float(den) if float(den) > 0 else 0
|
||
else:
|
||
fps = float(fps_str)
|
||
details["fps"] = round(fps, 2)
|
||
|
||
# Resolution scoring
|
||
if width < 1280 or height < 720:
|
||
score -= 20
|
||
suggestions.append(f"分辨率偏低({width}x{height})，建议至少1280x720")
|
||
elif width >= 1920 and height >= 1080:
|
||
pass # good
|
||
else:
|
||
score -= 5
|
||
|
||
# Aspect ratio
|
||
if height > 0:
|
||
ratio = width / height
|
||
if abs(ratio - 16/9) > 0.1:
|
||
score -= 5
|
||
suggestions.append(f"宽高比({ratio:.2f})不是标准16:9")
|
||
|
||
# Codec check
|
||
codec = v_stream.get("codec_name", "")
|
||
if codec not in ["h264", "h265", "hevc"]:
|
||
score -= 10
|
||
suggestions.append(f"视频编码{codec}不常用，建议使用h264/h265")
|
||
|
||
# Bitrate check
|
||
bitrate = int(format_info.get("bit_rate", 0))
|
||
if bitrate > 0:
|
||
bitrate_kbps = bitrate / 1000
|
||
details["bitrate_kbps"] = round(bitrate_kbps, 1)
|
||
if bitrate_kbps < 1000:
|
||
score -= 15
|
||
suggestions.append(f"码率偏低({bitrate_kbps:.0f}kbps)，视频质量可能较差")
|
||
elif bitrate_kbps > 8000:
|
||
score -= 5
|
||
suggestions.append(f"码率偏高({bitrate_kbps:.0f}kbps)，文件体积较大")
|
||
else:
|
||
score -= 50
|
||
suggestions.append("未检测到视频流")
|
||
details["resolution"] = "N/A"
|
||
|
||
# 3. Audio check
|
||
if a_stream:
|
||
details["audio_codec"] = a_stream.get("codec_name", "unknown")
|
||
details["audio_sample_rate"] = a_stream.get("sample_rate", "unknown")
|
||
details["audio_channels"] = a_stream.get("channels", 0)
|
||
|
||
audio_codec = a_stream.get("codec_name", "")
|
||
if audio_codec not in ["aac", "mp3", "opus", "flac"]:
|
||
score -= 5
|
||
suggestions.append(f"音频编码{audio_codec}不常用")
|
||
else:
|
||
score -= 20
|
||
suggestions.append("未检测到音频流")
|
||
|
||
# 4. Audio/Video sync check (basic: compare durations)
|
||
if v_stream and a_stream:
|
||
v_dur = float(v_stream.get("duration", 0))
|
||
a_dur = float(a_stream.get("duration", 0))
|
||
if v_dur > 0 and a_dur > 0:
|
||
drift = abs(v_dur - a_dur)
|
||
details["av_drift_seconds"] = round(drift, 3)
|
||
if drift > 0.1:
|
||
score -= 10
|
||
suggestions.append(f"音视频时长差{drift:.3f}s，可能存在同步问题")
|
||
|
||
# 5. Scene change detection (basic via ffprobe)
|
||
try:
|
||
scene_cmd = [
|
||
"ffprobe", "-v", "quiet", "-print_format", "json",
|
||
"-select_streams", "v:0",
|
||
"-show_frames", "-f", "null", "-"
|
||
]
|
||
# Use a simpler approach: count keyframes
|
||
keyframe_cmd = [
|
||
"ffprobe", "-v", "error", "-select_streams", "v:0",
|
||
"-show_entries", "frame=pict_type", "-of", "csv=p=0",
|
||
video_path
|
||
]
|
||
kf_out = subprocess.run(keyframe_cmd, capture_output=True, text=True, timeout=60)
|
||
if kf_out.stdout:
|
||
frames = kf_out.stdout.strip().split("\n")
|
||
total_frames = len(frames)
|
||
keyframes = sum(1 for f in frames if f.startswith("I"))
|
||
details["total_frames"] = total_frames
|
||
details["keyframes"] = keyframes
|
||
|
||
if total_frames > 0:
|
||
kf_ratio = keyframes / total_frames
|
||
details["keyframe_ratio"] = round(kf_ratio, 4)
|
||
except Exception:
|
||
pass
|
||
|
||
# 6. Subtitle burn-in detection (check for text in first few frames)
|
||
try:
|
||
sub_cmd = [
|
||
"ffmpeg", "-i", video_path, "-vf",
|
||
"select='eq(n,0)+eq(n,100)+eq(n,200)',ocr",
|
||
"-f", "null", "-"
|
||
]
|
||
# OCR not always available, skip if fails
|
||
sub_out = subprocess.run(sub_cmd, capture_output=True, text=True, timeout=15)
|
||
# Check if ffmpeg has OCR filter
|
||
if "No such filter" not in sub_out.stderr:
|
||
ocr_text = sub_out.stderr
|
||
# If OCR found text, subtitles may be burned in
|
||
if any(c.isalpha() for c in ocr_text):
|
||
details["subtitle_detected"] = True
|
||
else:
|
||
details["subtitle_detected"] = False
|
||
else:
|
||
details["subtitle_detected"] = "OCR unavailable"
|
||
except Exception:
|
||
details["subtitle_detected"] = "check_skipped"
|
||
|
||
# Final score clamping
|
||
score = max(0, min(100, score))
|
||
|
||
result = {
|
||
"score": score,
|
||
"passed": score >= 60,
|
||
"details": details,
|
||
"suggestions": suggestions,
|
||
"file": os.path.basename(video_path)
|
||
}
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
result = {"error": str(e), "traceback": traceback.format_exc(), "score": 0, "passed": False}
|
||
|
||
return json.dumps(result, ensure_ascii=False)
|