| """Session-owned temporary audio, expired after one hour without use.""" |
| from dataclasses import dataclass |
| from pathlib import Path |
| import time |
| import uuid |
|
|
| import settings |
|
|
|
|
| @dataclass |
| class Audio: |
| id: str |
| owner: str |
| path: Path |
| label: str |
| kind: str |
| touched: float |
| slot: str = "" |
| active: int = 0 |
|
|
| @property |
| def playback(self): |
| return self.path.with_suffix(".mp3") |
|
|
| @property |
| def files(self): |
| return {self.path, self.playback} |
|
|
|
|
| class AudioStore: |
| def __init__(self, root=None, ttl=None): |
| self.root = Path(root or settings.DATA) |
| self.root.mkdir(parents=True, exist_ok=True, mode=0o700) |
| self.ttl = settings.TTL if ttl is None else ttl |
| self.items = {} |
| |
| cutoff = time.time() - self.ttl |
| for path in self.audio_files(): |
| if path.stat().st_mtime < cutoff: |
| path.unlink() |
|
|
| def new(self, owner, label, kind, slot=""): |
| id_ = uuid.uuid4().hex |
| suffix = ".mp3" if kind == "take" else ".wav" |
| item = Audio(id_, owner, self.root / f"{id_}{suffix}", label, kind, time.time(), slot) |
| self.items[id_] = item |
| return item |
|
|
| def get(self, id_, owner, touch=True): |
| item = self.items.get(id_) |
| if item is None or item.owner != owner or (not item.active and time.time() - item.touched >= self.ttl): |
| raise KeyError("Audio abgelaufen oder nicht verfügbar.") |
| if touch: |
| item.touched = time.time() |
| for path in item.files: |
| if path.exists(): |
| path.touch() |
| return item |
|
|
| def references(self, owner): |
| return [x for x in self.items.values() if x.owner == owner and x.kind == "reference" and time.time() - x.touched < self.ttl] |
|
|
| def remove(self, item): |
| if item.active: |
| raise ValueError("Diese Referenz wird gerade verwendet.") |
| for path in item.files: |
| path.unlink(missing_ok=True) |
| self.items.pop(item.id, None) |
|
|
| def cleanup(self, now=None): |
| now = time.time() if now is None else now |
| expired = [x for x in self.items.values() if not x.active and now - x.touched >= self.ttl] |
| for x in expired: |
| self.remove(x) |
| |
| known = {path for item in self.items.values() for path in item.files} |
| for path in self.audio_files(): |
| if path not in known and now - path.stat().st_mtime >= self.ttl: |
| path.unlink() |
| return expired |
|
|
| def audio_files(self): |
| yield from self.root.glob("*.wav") |
| yield from self.root.glob("*.mp3") |
|
|
