social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import requests
2import subprocess
3import json
4from util import LOGGER
5
6def probe_bytes(bytes: bytes) -> dict:
7 cmd = [
8 'ffprobe',
9 '-v', 'error',
10 '-show_format',
11 '-show_streams',
12 '-print_format', 'json',
13 'pipe:0'
14 ]
15 proc = subprocess.run(cmd, input=bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
16
17 if proc.returncode != 0:
18 raise RuntimeError(f"ffprobe failed: {proc.stderr.decode()}")
19
20 return json.loads(proc.stdout)
21
22def compress_image(image_bytes: bytes, quality: int = 90):
23 cmd = [
24 'ffmpeg',
25 '-f', 'image2pipe',
26 '-i', 'pipe:0',
27 '-c:v', 'webp',
28 '-q:v', str(quality),
29 '-f', 'image2pipe',
30 'pipe:1'
31 ]
32
33 proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
34 out_bytes, err = proc.communicate(input=image_bytes)
35
36 if proc.returncode != 0:
37 raise RuntimeError(f"ffmpeg compress failed: {err.decode()}")
38
39 return out_bytes
40
41def download_blob(url: str, max_bytes: int = 5_000_000) -> bytes | None:
42 response = requests.get(url, stream=True, timeout=20)
43 if response.status_code != 200:
44 LOGGER.info("Failed to download %s! %s", url, response)
45 return None
46
47 downloaded_bytes = b""
48 current_size = 0
49
50 for chunk in response.iter_content(chunk_size=8192):
51 if not chunk:
52 continue
53
54 current_size += len(chunk)
55 if current_size > max_bytes:
56 response.close()
57 return None
58
59 downloaded_bytes += chunk
60
61 return downloaded_bytes
62
63
64def get_media_meta(bytes: bytes):
65 probe = probe_bytes(bytes)
66 streams = [s for s in probe['streams'] if s['codec_type'] == 'video']
67 if not streams:
68 raise ValueError("No video stream found")
69
70 media = streams[0]
71 return {
72 'width': int(media['width']),
73 'height': int(media['height']),
74 'duration': float(media.get('duration', probe['format'].get('duration', -1)))
75 }