133 lines
4.4 KiB
Python
133 lines
4.4 KiB
Python
"""Utility things"""
|
|
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
from nio import AsyncClient
|
|
|
|
from .types import *
|
|
|
|
class Utils:
|
|
@staticmethod
|
|
async def get_video_props_using_ffprobe(path: Path | str) -> VideoFileProperties:
|
|
"""Returns video properties for file at `path`.
|
|
|
|
Consider using `asyncio.timeout` to prevent deadlocking.
|
|
"""
|
|
# prepare arguments for ffprobe
|
|
args = [
|
|
"ffprobe",
|
|
"-v", "error",
|
|
"-select_streams", "v:0",
|
|
"-show_entries", "stream=width,height:format=duration",
|
|
"-of", "json",
|
|
str(path)
|
|
]
|
|
# start the process
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*args,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
# communicate `stdout` and `stderr`
|
|
try:
|
|
stdout, stderr = await proc.communicate()
|
|
except asyncio.CancelledError:
|
|
if proc.returncode is None:
|
|
proc.kill()
|
|
await proc.wait()
|
|
raise
|
|
# error
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(f"`ffprobe` exits with code {proc.returncode}: {stderr.decode(errors="replace")}")
|
|
# success
|
|
result = json.loads(stdout)
|
|
return VideoFileProperties(
|
|
width_px=int(result["streams"][0]["width"]),
|
|
height_px=int(result["streams"][0]["height"]),
|
|
duration_ms=int(float(result["format"]["duration"]) * 1000)
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_video_thumbnail(path: Path | str, *, timestamp: float = 0, max_size: int = 300) -> bytes:
|
|
"""Generates video thumbnail for file at `path`.
|
|
|
|
Args:
|
|
- path - path to the video
|
|
- timestamp - timestamp to use as the thumbnail
|
|
- max_size - max width/height
|
|
|
|
Returns:
|
|
Raw bytes of JPEG thumbnail
|
|
"""
|
|
# prepare arguments for ffmpeg
|
|
args = [
|
|
"ffmpeg",
|
|
"-v", "error",
|
|
"-ss", "%.2f" % timestamp,
|
|
"-i", str(path),
|
|
"-map", "0:v:0",
|
|
"-vf",
|
|
(
|
|
f"scale='min(iw,{max_size})':'min(ih,{max_size})':"
|
|
"force_original_aspect_ratio=decrease"
|
|
),
|
|
"-frames:v", "1",
|
|
"-q:v", "3",
|
|
"-f", "image2pipe",
|
|
"-vcodec", "mjpeg",
|
|
"pipe:1"
|
|
]
|
|
# start the process
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*args,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
# communicate `stdout` and `stderr`
|
|
try:
|
|
stdout, stderr = await proc.communicate()
|
|
except asyncio.CancelledError:
|
|
if proc.returncode is None:
|
|
proc.kill()
|
|
await proc.wait()
|
|
raise
|
|
# error
|
|
if proc.returncode != 0 or not stdout:
|
|
raise RuntimeError(f"`ffmpeg` exits with code {proc.returncode}: {stderr.decode(errors="replace")}", args)
|
|
# success
|
|
return stdout
|
|
|
|
@staticmethod
|
|
def build_full_matrix_username(config: MatrixBotConfig) -> str:
|
|
"""Build complete matrix username using data from MatrixBotConfig."""
|
|
homeserver_name = urlparse(config.matrix_homeserver_url).hostname
|
|
localpart = config.matrix_username_localpart
|
|
return f"@{localpart}:{homeserver_name}"
|
|
|
|
@staticmethod
|
|
def verify_all_known_devices(client: AsyncClient) -> bool:
|
|
"""
|
|
Verify all known devices.
|
|
|
|
Args:
|
|
- client - client to verify all known devices for
|
|
|
|
Returns:
|
|
- True if there were unverified devices that are verified now
|
|
"""
|
|
result = False
|
|
for user_id in client.device_store.users:
|
|
for device_id, olm_device in client.device_store[user_id].items():
|
|
# can't trust ourselves
|
|
if device_id == client.device_id and user_id == client.user_id:
|
|
continue
|
|
# they are already verified
|
|
if olm_device.verified:
|
|
continue
|
|
# verify them
|
|
client.verify_device(olm_device)
|
|
result = True
|
|
return result |