51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
|
|
import secrets
|
|
import subprocess
|
|
from urllib.parse import urlencode
|
|
|
|
|
|
def generate_token(length: int = 32) -> str:
|
|
return secrets.token_urlsafe(length)[:length]
|
|
|
|
|
|
def generate_password(length: int = 14) -> str:
|
|
alphabet = 'ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz23456789!@#$%'
|
|
return ''.join(secrets.choice(alphabet) for _ in range(length))
|
|
|
|
|
|
def hls_url(base: str, slug: str) -> str:
|
|
return f"{base.rstrip('/')}/{slug}/index.m3u8"
|
|
|
|
|
|
def ingest_url_with_query(host: str, slug: str, token: str) -> str:
|
|
return f"rtmp://{host}:1935/{slug}?{urlencode({'token': token})}"
|
|
|
|
|
|
def ffprobe_stream(url: str) -> dict:
|
|
cmd = ['ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_streams', '-show_format', url]
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=8, check=True)
|
|
except Exception:
|
|
return {}
|
|
|
|
import json
|
|
try:
|
|
payload = json.loads(result.stdout or '{}')
|
|
except Exception:
|
|
return {}
|
|
|
|
out = {}
|
|
for stream in payload.get('streams', []):
|
|
if stream.get('codec_type') == 'video':
|
|
out['video_codec'] = stream.get('codec_name')
|
|
out['width'] = stream.get('width')
|
|
out['height'] = stream.get('height')
|
|
out['fps'] = stream.get('avg_frame_rate')
|
|
elif stream.get('codec_type') == 'audio':
|
|
out['audio_codec'] = stream.get('codec_name')
|
|
out['audio_channels'] = stream.get('channels')
|
|
out['audio_sample_rate'] = stream.get('sample_rate')
|
|
if payload.get('format', {}).get('bit_rate'):
|
|
out['bitrate'] = payload['format']['bit_rate']
|
|
return out
|