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 compress_image(image_bytes: bytes, quality: int = 90):
40 cmd = [
41 'ffmpeg',
42 '-f', 'image2pipe',
43 '-i', 'pipe:0',
44 '-c:v', 'webp',
45 '-q:v', str(quality),
46 '-f', 'image2pipe',
47 'pipe:1'
48 ]
49
50 proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
51 out_bytes, err = proc.communicate(input=image_bytes)
52
53 if proc.returncode != 0:
54 raise RuntimeError(f"ffmpeg compress failed: {err.decode()}")
55
56 return out_bytes
57
58def download_blob(url: str, max_bytes: int = 5_000_000) -> bytes | None:
59 response = requests.get(url, stream=True, timeout=20)
60 if response.status_code != 200:
61 LOGGER.info("Failed to download %s! %s", url, response)
62 return None
63
64 downloaded_bytes = b""
65 current_size = 0
66
67 for chunk in response.iter_content(chunk_size=8192):
68 if not chunk:
69 continue
70
71 current_size += len(chunk)
72 if current_size > max_bytes:
73 response.close()
74 return None
75
76 downloaded_bytes += chunk
77
78 return downloaded_bytes
79
80
81def get_media_meta(bytes: bytes):
82 probe = probe_bytes(bytes)
83 streams = [s for s in probe['streams'] if s['codec_type'] == 'video']
84 if not streams:
85 raise ValueError("No video stream found")
86
87 media = streams[0]
88 return {
89 'width': int(media['width']),
90 'height': int(media['height']),
91 'duration': float(media.get('duration', probe['format'].get('duration', -1)))
92 }