File size: 2,776 Bytes
2219798 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """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 = {}
# A crash cannot immortalize uploads: clean old files on startup too.
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)
# Includes orphan files from a restart, but never an active item.
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")
|