You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
53 lines
1.9 KiB
53 lines
1.9 KiB
from __future__ import annotations |
|
|
|
import subprocess |
|
import threading |
|
from pathlib import Path |
|
from typing import Optional |
|
|
|
|
|
class AudioPlayer: |
|
def __init__(self, music_dir: Path): |
|
self.music_dir = music_dir |
|
self._lock = threading.RLock() |
|
self._proc: Optional[subprocess.Popen] = None |
|
|
|
def _resolve_music_path(self, music_file: str) -> Path: |
|
candidate = (self.music_dir / music_file).resolve() |
|
if self.music_dir.resolve() not in candidate.parents: |
|
raise ValueError("music_file path traversal is forbidden") |
|
if not candidate.exists() or not candidate.is_file(): |
|
raise FileNotFoundError(f"Music file not found: {music_file}") |
|
return candidate |
|
|
|
def play(self, music_file: str, start_seconds: float = 0.0, end_seconds: Optional[float] = None) -> None: |
|
with self._lock: |
|
music_path = self._resolve_music_path(music_file) |
|
self.stop() |
|
|
|
cmd = [ |
|
"ffplay", |
|
"-nodisp", |
|
"-autoexit", |
|
"-loglevel", |
|
"error", |
|
"-ss", |
|
str(start_seconds), |
|
] |
|
if end_seconds is not None: |
|
duration = end_seconds - start_seconds |
|
if duration <= 0: |
|
raise ValueError("end_seconds must be greater than start_seconds") |
|
cmd += ["-t", str(duration)] |
|
cmd.append(str(music_path)) |
|
self._proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
|
|
|
def stop(self) -> None: |
|
with self._lock: |
|
if self._proc and self._proc.poll() is None: |
|
self._proc.terminate() |
|
try: |
|
self._proc.wait(timeout=2) |
|
except subprocess.TimeoutExpired: |
|
self._proc.kill() |
|
self._proc = None
|
|
|