v2.1-enterprise

tuxinet-streamhub-v2.1-enterprise/backend/app/utils.py

@@ -0,0 +1,50 @@
+
+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