75edf0aa71
- Rust-based digital asset management system - Video analysis: ASR, OCR, YOLO, Face, Pose - RAG capabilities with Qdrant vector database - Multi-database support: PostgreSQL, Redis, MongoDB - Monitoring system with launchd plists - n8n workflow automation integration
98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
#!/opt/homebrew/bin/python3.11
|
|
import sys
|
|
import os
|
|
import subprocess
|
|
import json
|
|
import argparse
|
|
|
|
|
|
def get_duration(video_path):
|
|
result = subprocess.run(
|
|
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", video_path],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
data = json.loads(result.stdout)
|
|
return float(data["format"]["duration"])
|
|
|
|
|
|
def extract_thumbnails(video_path, uuid, output_dir, count=6):
|
|
"""Extract evenly-spaced thumbnails from video."""
|
|
|
|
if not os.path.exists(video_path):
|
|
print(f"Error: Video file not found: {video_path}", file=sys.stderr)
|
|
return None
|
|
|
|
duration = get_duration(video_path)
|
|
print(f"Video duration: {duration:.2f}s", file=sys.stderr)
|
|
|
|
# Output directory
|
|
thumb_dir = os.path.join(output_dir, uuid)
|
|
os.makedirs(thumb_dir, exist_ok=True)
|
|
|
|
# Calculate timestamps (skip first 10%, skip last 10%)
|
|
start_ts = duration * 0.1
|
|
end_ts = duration * 0.9
|
|
interval = (end_ts - start_ts) / (count - 1) if count > 1 else 0
|
|
|
|
extracted = []
|
|
for i in range(count):
|
|
if count == 1:
|
|
ts = duration / 2
|
|
else:
|
|
ts = start_ts + (interval * i)
|
|
|
|
output_file = os.path.join(thumb_dir, f"thumb_{i:03d}.jpg")
|
|
result = subprocess.run(
|
|
[
|
|
"ffmpeg",
|
|
"-y",
|
|
"-ss",
|
|
str(ts),
|
|
"-i",
|
|
video_path,
|
|
"-vframes",
|
|
"1",
|
|
"-q:v",
|
|
"2",
|
|
"-vf",
|
|
"scale=320:-1",
|
|
output_file,
|
|
],
|
|
capture_output=True,
|
|
)
|
|
|
|
if result.returncode == 0 and os.path.exists(output_file):
|
|
extracted.append(output_file)
|
|
print(f" Extracted: {output_file} at {ts:.1f}s", file=sys.stderr)
|
|
else:
|
|
print(f" Failed to extract frame at {ts:.1f}s", file=sys.stderr)
|
|
|
|
return extracted
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Extract keyframe thumbnails from video"
|
|
)
|
|
parser.add_argument("video_path", help="Path to video file")
|
|
parser.add_argument("uuid", help="Video UUID")
|
|
parser.add_argument("-o", "--output", default="thumbnails", help="Output directory")
|
|
parser.add_argument(
|
|
"-c", "--count", type=int, default=6, help="Number of thumbnails"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
result = extract_thumbnails(args.video_path, args.uuid, args.output, args.count)
|
|
|
|
if result:
|
|
print(json.dumps({"uuid": args.uuid, "count": len(result), "files": result}))
|
|
else:
|
|
print(json.dumps({"error": "Failed to extract thumbnails"}))
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|