50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
|
"""Medien-Verteilung: Remote-Server registrieren, Medien abrufen."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import urllib.error
|
||
|
|
import urllib.parse
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
TIMEOUT_SHORT = 5
|
||
|
|
TIMEOUT_IMPORT = 180
|
||
|
|
|
||
|
|
|
||
|
|
def check(url: str) -> bool:
|
||
|
|
"""Prueft ob ein Remote-Server ein HMS MediaEngine ist."""
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(
|
||
|
|
url.rstrip("/") + "/api/health",
|
||
|
|
timeout=TIMEOUT_SHORT) as r:
|
||
|
|
return json.loads(r.read().decode()).get("status") == "ok"
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def list_media(url: str) -> dict:
|
||
|
|
"""Medienbibliothek eines Remote-Servers abrufen."""
|
||
|
|
with urllib.request.urlopen(
|
||
|
|
url.rstrip("/") + "/api/media",
|
||
|
|
timeout=TIMEOUT_SHORT) as r:
|
||
|
|
return json.loads(r.read().decode())
|
||
|
|
|
||
|
|
|
||
|
|
def import_media(url: str, media_name: str, library) -> tuple[bool, object]:
|
||
|
|
"""Zieht eine Datei per HTTP von einem Remote-Server in die lokale
|
||
|
|
Bibliothek (Streaming, gleiche Upload-Pfad-Sicherung wie Web-Upload)."""
|
||
|
|
q = urllib.parse.quote(media_name)
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(
|
||
|
|
url.rstrip("/") + "/media/" + q,
|
||
|
|
timeout=TIMEOUT_IMPORT) as r:
|
||
|
|
length = int(r.headers.get("Content-Length", "0"))
|
||
|
|
saved, err = library.save_stream(media_name, r, length)
|
||
|
|
if err:
|
||
|
|
return False, err
|
||
|
|
return True, saved
|
||
|
|
except urllib.error.HTTPError as e:
|
||
|
|
return False, f"Remote-Fehler HTTP {e.code}"
|
||
|
|
except (urllib.error.URLError, OSError, ValueError) as e:
|
||
|
|
return False, f"Verbindung fehlgeschlagen: {e}"
|