"""Plik file upload plugin.""" import json import os import re import subprocess import threading from typing import Callable def _parse_ttl_seconds(ttl: str) -> int | None: if not ttl: return None m = re.match(r"^(\d+)d$", ttl) if m: return int(m.group(1)) * 86400 m = re.match(r"^(\d+)h$", ttl) if m: return int(m.group(1)) * 3600 m = re.match(r"^(\d+)$", ttl) if m: return int(m.group(1)) return None def upload_file( file_path: str, plik_url: str, api_key: str = "", ttl: str = "30d", ) -> 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 {"error": "Datei ist leer"} plik_base = plik_url.rstrip("/") filename = os.path.basename(file_path) cmd = ["curl", "-sS", "--fail-with-body", "--max-time", "7200"] if api_key: cmd += ["-H", f"X-PlikToken: {api_key}"] cmd += ["-F", f"file=@{file_path};filename={filename}"] ttl_secs = _parse_ttl_seconds(ttl) if ttl_secs is not None: cmd += ["-F", f"ttl={ttl_secs}"] cmd.append(plik_base) try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=7200, stdin=subprocess.DEVNULL) if result.returncode != 0: 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 {"error": f"Keine Upload-ID in Antwort: {result.stdout[:300]}"} browser_url = f"{plik_base}/#/?id={upload_id}" download_url = None files = data.get("files", []) if files: file_id = files[0].get("id", "") file_name = files[0].get("fileName", filename) if file_id: download_url = f"{plik_base}/file/{upload_id}/{file_id}/{file_name}" return {"browser_url": browser_url, "download_url": download_url} except (json.JSONDecodeError, KeyError): urls = re.findall(r'https?://\S+', result.stdout) if urls: download_url = urls[-1] prefix = f"{plik_base}/file/" if download_url.startswith(prefix): rest = download_url[len(prefix):] uid = rest.split("/")[0] if uid: return {"browser_url": f"{plik_base}/#/?id={uid}", "download_url": download_url} return {"browser_url": download_url, "download_url": download_url} 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: def __init__(self, config, log_fn: Callable, notify_fn: Callable = None): self.config = config self._log = log_fn self._notify = notify_fn def upload_recording(self, output_dir: str, output_file: str, job_name: str, segment_mode: bool, delete_after: bool, ntfy_enabled: bool, save_url_fn: Callable = None): plik_url = self.config.get("plik", "url", default="") if not plik_url: self._log("WARN", f"Plik-Upload übersprungen: Keine URL konfiguriert [{job_name}]", job_name) return api_key = self.config.get("plik", "api_key", default="") ttl = self.config.get("plik", "ttl", default="30d") def _do_upload(): try: if segment_mode: self._upload_segments(output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn) else: self._upload_single(output_file, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn) except Exception as e: self._log("ERROR", f"Plik-Upload Fehler: {e} [{job_name}]", job_name) threading.Thread(target=_do_upload, daemon=True).start() def _upload_single(self, file_path, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn=None): if not os.path.isfile(file_path) or os.path.getsize(file_path) == 0: self._log("WARN", f"Plik-Upload übersprungen: Datei fehlt oder leer [{job_name}]", job_name) return 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 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) if save_url_fn: save_url_fn(result["browser_url"]) if self._notify: self._notify("upload", "Plik-Upload", f"{job_name}: {result['browser_url']}", ntfy_enabled) if delete_after: 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 (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): return files = sorted(f for f in os.listdir(output_dir) if not f.startswith(".") and os.path.isfile(os.path.join(output_dir, f))) if not files: return uploaded = 0 last_browser_url = None for fname in files: fpath = os.path.join(output_dir, fname) if os.path.getsize(fpath) == 0: continue self._log("INFO", f"Plik Segment-Upload: {fname} [{job_name}]", job_name) result = upload_file(fpath, plik_url, api_key, ttl) 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 if delete_after: 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} (unbekannter Fehler) [{job_name}]", job_name) if uploaded > 0: if save_url_fn and last_browser_url: save_url_fn(last_browser_url) if self._notify: msg = f"{job_name}: {uploaded} Segment(e) hochgeladen" if last_browser_url: msg += f"\n{last_browser_url}" self._notify("upload", "Plik-Upload", msg, ntfy_enabled)