social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import requests
2import subprocess
3import json
4import re, urllib.parse, os
5from util import LOGGER
6
7FILENAME = re.compile(r'filename="?([^\";]*)"?')
8
9def get_filename_from_url(url):
10 try:
11 response = requests.head(url, allow_redirects=True)
12 disposition = response.headers.get('Content-Disposition')
13 if disposition:
14 filename = FILENAME.findall(disposition)
15 if filename:
16 return filename[0]
17 except requests.RequestException:
18 pass
19
20 parsed_url = urllib.parse.urlparse(url)
21 return os.path.basename(parsed_url.path)
22
23def probe_bytes(bytes: bytes) -> dict:
24 cmd = [
25 'ffprobe',
26 '-v', 'error',
27 '-show_format',
28 '-show_streams',
29 '-print_format', 'json',
30 'pipe:0'
31 ]
32 proc = subprocess.run(cmd, input=bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
33
34 if proc.returncode != 0:
35 raise RuntimeError(f"ffprobe failed: {proc.stderr.decode()}")
36
37 return json.loads(proc.stdout)
38
39def convert_to_mp4(video_bytes: bytes) -> bytes:
40 cmd = [
41 'ffmpeg',
42 '-i', 'pipe:0',
43 '-c:v', 'libx264',
44 '-crf', '30',
45 '-preset', 'slow',
46 '-c:a', 'aac',
47 '-b:a', '128k',
48 '-movflags', 'frag_keyframe+empty_moov+default_base_moof',
49 '-f', 'mp4',
50 'pipe:1'
51 ]
52
53 proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
54 out_bytes, err = proc.communicate(input=video_bytes)
55
56 if proc.returncode != 0:
57 raise RuntimeError(f"ffmpeg compress failed: {err.decode()}")
58
59 return out_bytes
60
61def compress_image(image_bytes: bytes, quality: int = 90):
62 cmd = [
63 'ffmpeg',
64 '-f', 'image2pipe',
65 '-i', 'pipe:0',
66 '-c:v', 'webp',
67 '-q:v', str(quality),
68 '-f', 'image2pipe',
69 'pipe:1'
70 ]
71
72 proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
73 out_bytes, err = proc.communicate(input=image_bytes)
74
75 if proc.returncode != 0:
76 raise RuntimeError(f"ffmpeg compress failed: {err.decode()}")
77
78 return out_bytes
79
80def download_blob(url: str, max_bytes: int = 5_000_000) -> bytes | None:
81 response = requests.get(url, stream=True, timeout=20)
82 if response.status_code != 200:
83 LOGGER.info("Failed to download %s! %s", url, response.text)
84 return None
85
86 downloaded_bytes = b""
87 current_size = 0
88
89 for chunk in response.iter_content(chunk_size=8192):
90 if not chunk:
91 continue
92
93 current_size += len(chunk)
94 if current_size > max_bytes:
95 response.close()
96 return None
97
98 downloaded_bytes += chunk
99
100 return downloaded_bytes
101
102
103def get_media_meta(bytes: bytes):
104 probe = probe_bytes(bytes)
105 streams = [s for s in probe['streams'] if s['codec_type'] == 'video']
106 if not streams:
107 raise ValueError("No video stream found")
108
109 media = streams[0]
110 return {
111 'width': int(media['width']),
112 'height': int(media['height']),
113 'duration': float(media.get('duration', probe['format'].get('duration', -1)))
114 }