import json import secrets import string import subprocess from urllib.parse import quote def generate_token(length: int = 32) -> str: alphabet = string.ascii_letters + string.digits return ''.join(secrets.choice(alphabet) for _ in range(length)) def hls_url(base: str, slug: str) -> str: return f"{base.rstrip('/')}/{quote(slug)}/index.m3u8" def rtmp_url(public_base: str, slug: str, token: str) -> str: base = public_base.replace('https://', '').replace('http://', '').rstrip('/') return f"rtmp://{base}:1935/{slug}?token={token}" def ffprobe_stream(url: str) -> dict: cmd = [ 'ffprobe', '-v', 'error', '-show_streams', '-of', 'json', '-rw_timeout', '5000000', '-timeout', '5000000', url ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=8, check=True) data = json.loads(result.stdout) out = { 'video_codec': '', 'audio_codec': '', 'width': None, 'height': None, 'fps': None, 'bitrate': None, 'audio_channels': None, 'audio_sample_rate': None } for stream in data.get('streams', []): ctype = stream.get('codec_type') if ctype == 'video': out['video_codec'] = stream.get('codec_name', '') out['width'] = stream.get('width') out['height'] = stream.get('height') rate = stream.get('avg_frame_rate') or stream.get('r_frame_rate') if rate and rate != '0/0' and '/' in rate: a, b = rate.split('/') if float(b) != 0: out['fps'] = round(float(a) / float(b), 2) br = stream.get('bit_rate') if br and str(br).isdigit(): out['bitrate'] = int(br) elif ctype == 'audio': out['audio_codec'] = stream.get('codec_name', '') out['audio_channels'] = stream.get('channels') sr = stream.get('sample_rate') if sr and str(sr).isdigit(): out['audio_sample_rate'] = int(sr) br = stream.get('bit_rate') if br and str(br).isdigit() and not out['bitrate']: out['bitrate'] = int(br) return out except Exception: return {}