Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c3f68ace96 |
+30
-17
@@ -5,10 +5,10 @@ import os
|
||||
import re
|
||||
import subprocess
|
||||
import threading
|
||||
from typing import Optional, Callable
|
||||
from typing import Callable
|
||||
|
||||
|
||||
def _parse_ttl_seconds(ttl: str) -> Optional[int]:
|
||||
def _parse_ttl_seconds(ttl: str) -> int | None:
|
||||
if not ttl:
|
||||
return None
|
||||
m = re.match(r"^(\d+)d$", ttl)
|
||||
@@ -28,17 +28,19 @@ def upload_file(
|
||||
plik_url: str,
|
||||
api_key: str = "",
|
||||
ttl: str = "30d",
|
||||
) -> Optional[dict]:
|
||||
"""Upload a file to Plik. Returns dict with 'browser_url' and 'download_url', or None on failure."""
|
||||
if not plik_url or not os.path.isfile(file_path):
|
||||
return None
|
||||
) -> dict | None:
|
||||
"""Upload a file to Plik. Returns dict with 'browser_url', 'download_url', or {'error': reason} on failure."""
|
||||
if not plik_url:
|
||||
return {"error": "Keine Plik-URL konfiguriert"}
|
||||
if not os.path.isfile(file_path):
|
||||
return {"error": f"Datei nicht gefunden: {file_path}"}
|
||||
if os.path.getsize(file_path) == 0:
|
||||
return None
|
||||
return {"error": "Datei ist leer"}
|
||||
|
||||
plik_base = plik_url.rstrip("/")
|
||||
filename = os.path.basename(file_path)
|
||||
|
||||
cmd = ["curl", "-sS", "--max-time", "7200"]
|
||||
cmd = ["curl", "-sS", "--fail-with-body", "--max-time", "7200"]
|
||||
if api_key:
|
||||
cmd += ["-H", f"X-PlikToken: {api_key}"]
|
||||
|
||||
@@ -53,13 +55,18 @@ def upload_file(
|
||||
try:
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=7200, stdin=subprocess.DEVNULL)
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
err = result.stderr.strip() or result.stdout.strip() or f"curl exit code {result.returncode}"
|
||||
return {"error": f"curl-Fehler: {err[:500]}"}
|
||||
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
|
||||
if "error" in data:
|
||||
return {"error": f"Plik-Server: {data['error']}"}
|
||||
|
||||
upload_id = data.get("id", "")
|
||||
if not upload_id:
|
||||
return None
|
||||
return {"error": f"Keine Upload-ID in Antwort: {result.stdout[:300]}"}
|
||||
|
||||
browser_url = f"{plik_base}/#/?id={upload_id}"
|
||||
|
||||
@@ -83,9 +90,11 @@ def upload_file(
|
||||
if uid:
|
||||
return {"browser_url": f"{plik_base}/#/?id={uid}", "download_url": download_url}
|
||||
return {"browser_url": download_url, "download_url": download_url}
|
||||
return None
|
||||
except Exception:
|
||||
return None
|
||||
return {"error": f"Antwort nicht lesbar: {result.stdout[:300]}"}
|
||||
except subprocess.TimeoutExpired:
|
||||
return {"error": "Upload-Timeout (max. 2h)"}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
|
||||
class PlikUploader:
|
||||
@@ -124,7 +133,9 @@ class PlikUploader:
|
||||
self._log("INFO", f"Plik-Upload: {os.path.basename(file_path)} [{job_name}]", job_name)
|
||||
result = upload_file(file_path, plik_url, api_key, ttl)
|
||||
|
||||
if result:
|
||||
if result and "error" in result:
|
||||
self._log("ERROR", f"Plik-Upload fehlgeschlagen: {result['error']} [{job_name}]", job_name)
|
||||
elif result and result.get("browser_url"):
|
||||
self._log("INFO", f"Plik-Upload abgeschlossen: {result['browser_url']} [{job_name}]", job_name)
|
||||
if result.get("download_url"):
|
||||
self._log("INFO", f"Plik-Download: {result['download_url']} [{job_name}]", job_name)
|
||||
@@ -136,7 +147,7 @@ class PlikUploader:
|
||||
os.remove(file_path)
|
||||
self._log("INFO", f"Lokale Datei gelöscht: {file_path} [{job_name}]", job_name)
|
||||
else:
|
||||
self._log("ERROR", f"Plik-Upload fehlgeschlagen [{job_name}]", job_name)
|
||||
self._log("ERROR", f"Plik-Upload fehlgeschlagen (unbekannter Fehler) [{job_name}]", job_name)
|
||||
|
||||
def _upload_segments(self, output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn=None):
|
||||
if not os.path.isdir(output_dir):
|
||||
@@ -154,7 +165,9 @@ class PlikUploader:
|
||||
continue
|
||||
self._log("INFO", f"Plik Segment-Upload: {fname} [{job_name}]", job_name)
|
||||
result = upload_file(fpath, plik_url, api_key, ttl)
|
||||
if result:
|
||||
if result and "error" in result:
|
||||
self._log("ERROR", f"Plik Segment-Upload fehlgeschlagen: {fname} - {result['error']} [{job_name}]", job_name)
|
||||
elif result and result.get("browser_url"):
|
||||
last_browser_url = result["browser_url"]
|
||||
self._log("INFO", f"Plik Segment hochgeladen: {last_browser_url} [{job_name}]", job_name)
|
||||
uploaded += 1
|
||||
@@ -162,7 +175,7 @@ class PlikUploader:
|
||||
os.remove(fpath)
|
||||
self._log("INFO", f"Segment gelöscht: {fname} [{job_name}]", job_name)
|
||||
else:
|
||||
self._log("ERROR", f"Plik Segment-Upload fehlgeschlagen: {fname} [{job_name}]", job_name)
|
||||
self._log("ERROR", f"Plik Segment-Upload fehlgeschlagen: {fname} (unbekannter Fehler) [{job_name}]", job_name)
|
||||
|
||||
if uploaded > 0:
|
||||
if save_url_fn and last_browser_url:
|
||||
|
||||
@@ -141,6 +141,21 @@ class StreamScheduler:
|
||||
now_min = _time_to_minutes(now_time)
|
||||
start_min = _time_to_minutes(job["schedule_start"])
|
||||
in_window = now_min >= start_min
|
||||
elif job.get("schedule_stop"):
|
||||
start_min = _time_to_minutes(job["schedule_start"])
|
||||
stop_min = _time_to_minutes(job["schedule_stop"])
|
||||
if stop_min <= start_min and stop_min > 0:
|
||||
now_min = _time_to_minutes(now_time)
|
||||
if now_min < stop_min:
|
||||
yesterday = now - datetime.timedelta(days=1)
|
||||
was_yesterday = False
|
||||
if job["schedule_once"]:
|
||||
was_yesterday = job["schedule_date"] == yesterday.strftime("%Y-%m-%d")
|
||||
else:
|
||||
was_yesterday = _matches_day(yesterday.isoweekday(), job["schedule_days"] or "*")
|
||||
if was_yesterday:
|
||||
is_cross_midnight = True
|
||||
in_window = True
|
||||
|
||||
return {
|
||||
"occ_key": occ_key,
|
||||
@@ -218,7 +233,7 @@ class StreamScheduler:
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def _get_active_scheduled_recording_for_job(self, job_id: int):
|
||||
def _get_active_scheduled_recording_for_job(self, job_id: int, *, respect_stop: bool = False):
|
||||
session = get_session()
|
||||
try:
|
||||
rec = session.query(Recording).filter(
|
||||
@@ -226,7 +241,11 @@ class StreamScheduler:
|
||||
Recording.status.in_(["running", "starting"]),
|
||||
Recording.is_scheduled == True,
|
||||
).first()
|
||||
return rec.id if rec else None
|
||||
if not rec:
|
||||
return None
|
||||
if respect_stop and rec.scheduled_stop and rec.scheduled_stop > datetime.datetime.utcnow():
|
||||
return None
|
||||
return rec.id
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
@@ -309,7 +328,9 @@ class StreamScheduler:
|
||||
and eff_job["schedule_stop"]
|
||||
and self._is_job_recording(job["id"])
|
||||
):
|
||||
rec_id = self._get_active_scheduled_recording_for_job(job["id"])
|
||||
rec_id = self._get_active_scheduled_recording_for_job(
|
||||
job["id"], respect_stop=True,
|
||||
)
|
||||
if rec_id:
|
||||
self.recorder.stop_recording(rec_id)
|
||||
self._started_keys.discard(occ_key)
|
||||
|
||||
Reference in New Issue
Block a user