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
|
import random
import subprocess
import yt_dlp
import cv2
import os
def get_video_info(url):
ydl_opts = {"quiet": True}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
return ydl.extract_info(url, download=False)
def pick_best_video_url(info):
formats = [
f for f in info["formats"]
if f.get("url") and f.get("vcodec") != "none"
]
formats.sort(
key=lambda f: (f.get("height") or 0, f.get("tbr") or 0),
reverse=True
)
return formats[0]["url"]
def is_low_energy(img_path, size=64, threshold=500):
img = cv2.imread(img_path)
if img is None:
raise RuntimeError("Could not read image")
img = cv2.resize(img, (size, size), interpolation=cv2.INTER_AREA)
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
return gray.var() < threshold
def download_random_frame(url, name_prefix="frame", selected_timestamps=None, max_tries=10):
if selected_timestamps is None:
selected_timestamps = []
info = get_video_info(url)
duration = info.get("duration")
if not duration:
raise RuntimeError("No duration found")
video_url = pick_best_video_url(info)
for _ in range(max_tries):
t = random.uniform(0, duration)
while any(abs(t - ts) < 1 for ts in selected_timestamps):
t = random.uniform(0, duration)
filename = f"{name_prefix}-{t:.2f}.jpg"
cmd = [
"ffmpeg",
"-ss", str(t),
"-i", video_url,
"-frames:v", "1",
"-q:v", "2",
"-y",
filename,
]
subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
if not os.path.exists(filename):
continue
try:
if is_low_energy(filename):
os.remove(filename)
continue
except Exception:
pass
return t
raise RuntimeError("Failed to generate a valid frame after retries")
def generate_daily_random_frames(url, output_dir, count=3):
selected_timestamps = []
os.makedirs(output_dir, exist_ok=True)
for i in range(count):
ts = download_random_frame(
url,
f"{output_dir}/frame",
selected_timestamps
)
selected_timestamps.append(ts)
print(f"Downloaded frame {i} at {ts:.2f}s")
|
