467 lines
18 KiB
Python
467 lines
18 KiB
Python
"""Independent mpv audio player controlled through Windows JSON IPC."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class MpvPlayer:
|
|
def __init__(
|
|
self,
|
|
exe_path: str | Path,
|
|
logger: logging.Logger,
|
|
*,
|
|
pipe_name: str = "",
|
|
log_path: str | Path | None = None,
|
|
):
|
|
self.exe_path = Path(exe_path)
|
|
self.logger = logger
|
|
self.log_path = Path(log_path) if log_path else None
|
|
pipe_name = pipe_name or f"live_streaming_mpv_{os.getpid()}"
|
|
self.pipe_path = rf"\\.\pipe\{pipe_name}"
|
|
self.process: subprocess.Popen | None = None
|
|
self.current_url = ""
|
|
self.current_metadata: dict[str, Any] = {}
|
|
self.desired_state = "stopped"
|
|
self.generation = 0
|
|
self.started_at = 0.0
|
|
self.last_progress = 0.0
|
|
self.last_progress_at = 0.0
|
|
self.last_snapshot_at = 0.0
|
|
self.recovery_count = 0
|
|
self._request_id = 0
|
|
self._ipc_lock = asyncio.Lock()
|
|
self._pipe_state_lock = threading.Lock()
|
|
self._pipe = None
|
|
|
|
def available(self) -> bool:
|
|
return self.exe_path.is_file()
|
|
|
|
def running(self) -> bool:
|
|
return self.process is not None and self.process.poll() is None
|
|
|
|
def update_exe_path(self, exe_path: str | Path) -> None:
|
|
next_path = Path(exe_path)
|
|
if next_path == self.exe_path:
|
|
return
|
|
if self.running():
|
|
self.logger.warning(f"[mpv] 播放器路径已修改，将在进程下次重启后生效: {next_path}")
|
|
self.exe_path = next_path
|
|
|
|
async def ensure_started(self) -> bool:
|
|
if self.running():
|
|
return True
|
|
await asyncio.to_thread(self._reset_pipe_sync)
|
|
if not self.available():
|
|
self.logger.error(f"[mpv] 播放器不存在: {self.exe_path}")
|
|
return False
|
|
creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) if os.name == "nt" else 0
|
|
args = [
|
|
str(self.exe_path),
|
|
"--idle=yes",
|
|
"--no-video",
|
|
"--force-window=no",
|
|
"--no-terminal",
|
|
"--msg-level=all=warn",
|
|
f"--input-ipc-server={self.pipe_path}",
|
|
"--keep-open=no",
|
|
"--audio-buffer=5",
|
|
"--cache=yes",
|
|
"--cache-secs=20",
|
|
"--demuxer-max-bytes=50MiB",
|
|
"--network-timeout=10",
|
|
]
|
|
if self.log_path:
|
|
self.log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
args.append(f"--log-file={self.log_path}")
|
|
try:
|
|
self.process = subprocess.Popen(
|
|
args,
|
|
cwd=str(self.exe_path.parent),
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
creationflags=creationflags,
|
|
)
|
|
for _ in range(30):
|
|
if not self.running():
|
|
break
|
|
if await self._command(["get_property", "idle-active"], retry=False) is not None:
|
|
self.logger.info(f"[mpv] 播放服务已启动: {self.exe_path}")
|
|
return True
|
|
await asyncio.sleep(0.1)
|
|
except Exception as exc:
|
|
self.logger.error(f"[mpv] 启动失败: {exc}")
|
|
return False
|
|
|
|
def _pipe_request_sync(self, command: list[Any], request_id: int) -> Any:
|
|
request = json.dumps(
|
|
{"command": command, "request_id": request_id},
|
|
ensure_ascii=False,
|
|
).encode("utf-8") + b"\n"
|
|
pipe = None
|
|
try:
|
|
with self._pipe_state_lock:
|
|
pipe = self._pipe
|
|
if pipe is None or pipe.closed:
|
|
pipe = open(self.pipe_path, "r+b", buffering=0)
|
|
self._pipe = pipe
|
|
pipe.write(request)
|
|
deadline = time.time() + 2.5
|
|
while time.time() < deadline:
|
|
response = pipe.readline()
|
|
if not response:
|
|
continue
|
|
payload = json.loads(response.decode("utf-8", errors="replace"))
|
|
if payload.get("request_id") != request_id:
|
|
continue
|
|
if payload.get("error") != "success":
|
|
return None
|
|
if "data" not in payload:
|
|
return True
|
|
data = payload["data"]
|
|
if data is None and command and command[0] != "get_property":
|
|
return True
|
|
return data
|
|
except Exception:
|
|
self._reset_pipe_sync(pipe)
|
|
raise
|
|
return None
|
|
|
|
def _reset_pipe_sync(self, expected_pipe=None) -> None:
|
|
with self._pipe_state_lock:
|
|
pipe = self._pipe
|
|
if expected_pipe is not None and pipe is not expected_pipe:
|
|
return
|
|
self._pipe = None
|
|
if pipe is not None:
|
|
try:
|
|
pipe.close()
|
|
except Exception:
|
|
pass
|
|
|
|
async def _command(self, command: list[Any], *, retry: bool = True) -> Any:
|
|
async with self._ipc_lock:
|
|
attempts = 2 if retry else 1
|
|
for attempt in range(attempts):
|
|
self._request_id += 1
|
|
request_id = self._request_id
|
|
try:
|
|
return await asyncio.wait_for(
|
|
asyncio.to_thread(self._pipe_request_sync, command, request_id),
|
|
timeout=3.0,
|
|
)
|
|
except Exception:
|
|
self._reset_pipe_sync()
|
|
if attempt + 1 < attempts:
|
|
await asyncio.sleep(0.15)
|
|
return None
|
|
|
|
def _clear_current(self) -> None:
|
|
self.current_url = ""
|
|
self.current_metadata = {}
|
|
self.started_at = 0.0
|
|
self.last_progress = 0.0
|
|
self.last_progress_at = 0.0
|
|
self.recovery_count = 0
|
|
|
|
@staticmethod
|
|
def _path_matches(expected: str, actual: str) -> bool:
|
|
expected = str(expected or "").strip()
|
|
actual = str(actual or "").strip()
|
|
if not expected or not actual:
|
|
return False
|
|
if expected == actual:
|
|
return True
|
|
if expected.lower().startswith(("http://", "https://")):
|
|
return False
|
|
try:
|
|
actual_path = actual[8:] if actual.lower().startswith("file:///") else actual
|
|
return Path(expected).resolve() == Path(actual_path).resolve()
|
|
except Exception:
|
|
return False
|
|
|
|
async def _wait_until_loaded(
|
|
self,
|
|
*,
|
|
expected_generation: int,
|
|
start_at: float,
|
|
timeout: float = 12.0,
|
|
) -> bool:
|
|
deadline = time.time() + max(2.0, timeout)
|
|
expected_url = self.current_url
|
|
while time.time() < deadline:
|
|
if expected_generation != self.generation or self.desired_state != "playing":
|
|
return False
|
|
if not self.running():
|
|
return False
|
|
path = await self._command(["get_property", "path"], retry=False)
|
|
idle = await self._command(["get_property", "idle-active"], retry=False)
|
|
if self._path_matches(expected_url, str(path or "")) and idle is False:
|
|
if start_at > 0:
|
|
seek_result = None
|
|
seek_deadline = min(deadline, time.time() + 4.0)
|
|
while time.time() < seek_deadline:
|
|
duration = await self._command(["get_property", "duration"], retry=False)
|
|
try:
|
|
duration_value = max(0.0, float(duration or 0.0))
|
|
except (TypeError, ValueError):
|
|
duration_value = 0.0
|
|
if duration_value > 0:
|
|
seek_result = await self._command(["seek", float(start_at), "absolute+exact"])
|
|
if seek_result is not None:
|
|
break
|
|
await asyncio.sleep(0.2)
|
|
if seek_result is None:
|
|
self.logger.warning(f"[mpv] 续播定位失败: {start_at:.1f} 秒")
|
|
return False
|
|
if await self._command(["set_property", "pause", False]) is None:
|
|
return False
|
|
progress_deadline = min(deadline, time.time() + 4.0)
|
|
baseline = max(0.0, float(start_at or 0.0))
|
|
while time.time() < progress_deadline:
|
|
progress = await self._command(["get_property", "time-pos"], retry=False)
|
|
duration = await self._command(["get_property", "duration"], retry=False)
|
|
try:
|
|
progress_value = max(0.0, float(progress or 0.0))
|
|
except (TypeError, ValueError):
|
|
progress_value = 0.0
|
|
try:
|
|
duration_value = max(0.0, float(duration or 0.0))
|
|
except (TypeError, ValueError):
|
|
duration_value = 0.0
|
|
idle_now = await self._command(["get_property", "idle-active"], retry=False)
|
|
if idle_now is False and (progress_value > 0.05 or duration_value > 0 or baseline > 0):
|
|
now = time.time()
|
|
self.started_at = now
|
|
self.last_progress = max(baseline, progress_value)
|
|
self.last_progress_at = now
|
|
return True
|
|
await asyncio.sleep(0.15)
|
|
await asyncio.sleep(0.15)
|
|
self.logger.warning(
|
|
f"[mpv] 音频加载超时，未进入可播放状态: "
|
|
f"{self.current_metadata.get('name') or self.current_url}"
|
|
)
|
|
return False
|
|
|
|
async def _load_current(self, *, start_at: float, expected_generation: int) -> bool:
|
|
if expected_generation != self.generation or self.desired_state != "playing" or not self.current_url:
|
|
return False
|
|
if not await self.ensure_started():
|
|
return False
|
|
if expected_generation != self.generation or self.desired_state != "playing":
|
|
return False
|
|
result = await self._command(["loadfile", self.current_url, "replace"])
|
|
if result is None or expected_generation != self.generation or self.desired_state != "playing":
|
|
return False
|
|
return await self._wait_until_loaded(
|
|
expected_generation=expected_generation,
|
|
start_at=start_at,
|
|
)
|
|
|
|
async def play(self, url: str, metadata: dict[str, Any], *, start_at: float = 0.0) -> bool:
|
|
self.generation += 1
|
|
generation = self.generation
|
|
self.desired_state = "playing"
|
|
self.current_url = str(url)
|
|
self.current_metadata = dict(metadata)
|
|
self.recovery_count = 0
|
|
ok = await self._load_current(start_at=start_at, expected_generation=generation)
|
|
if not ok and generation == self.generation:
|
|
self.desired_state = "stopped"
|
|
self._clear_current()
|
|
return ok
|
|
|
|
async def pause(self) -> bool:
|
|
if not self.current_url:
|
|
return False
|
|
self.desired_state = "paused"
|
|
if not self.running():
|
|
return True
|
|
return await self._command(["set_property", "pause", True]) is not None
|
|
|
|
async def resume(self) -> bool:
|
|
if not self.current_url:
|
|
return False
|
|
self.desired_state = "playing"
|
|
self.last_progress_at = time.time()
|
|
if not self.running():
|
|
return await self._load_current(start_at=self.last_progress, expected_generation=self.generation)
|
|
return await self._command(["set_property", "pause", False]) is not None
|
|
|
|
async def stop(self) -> bool:
|
|
self.generation += 1
|
|
self.desired_state = "stopped"
|
|
self._clear_current()
|
|
if not self.running():
|
|
return True
|
|
return await self._command(["stop"]) is not None
|
|
|
|
async def close(self) -> None:
|
|
self.generation += 1
|
|
self.desired_state = "stopped"
|
|
self._clear_current()
|
|
if self.running():
|
|
await self._command(["quit"])
|
|
await asyncio.sleep(0.15)
|
|
if self.running():
|
|
self.process.terminate()
|
|
try:
|
|
await asyncio.to_thread(self.process.wait, 2)
|
|
except Exception:
|
|
if self.running():
|
|
self.process.kill()
|
|
self.process = None
|
|
await asyncio.to_thread(self._reset_pipe_sync)
|
|
|
|
async def snapshot(self) -> dict[str, Any]:
|
|
now = time.time()
|
|
if not self.running():
|
|
return self._snapshot_payload(False, 0.0, 0.0, True, False, False, "", now)
|
|
progress, duration, paused, idle, eof, path = await asyncio.gather(
|
|
self._command(["get_property", "time-pos"]),
|
|
self._command(["get_property", "duration"]),
|
|
self._command(["get_property", "pause"]),
|
|
self._command(["get_property", "idle-active"]),
|
|
self._command(["get_property", "eof-reached"]),
|
|
self._command(["get_property", "path"]),
|
|
)
|
|
try:
|
|
progress_value = max(0.0, float(progress or 0.0))
|
|
except (TypeError, ValueError):
|
|
progress_value = 0.0
|
|
try:
|
|
duration_value = max(0.0, float(duration or self.current_metadata.get("duration_sec", 0) or 0))
|
|
except (TypeError, ValueError):
|
|
duration_value = 0.0
|
|
idle_value = bool(idle) if idle is not None else not bool(self.current_url)
|
|
paused_value = bool(paused)
|
|
eof_value = bool(eof)
|
|
playing = (
|
|
self.desired_state == "playing"
|
|
and bool(self.current_url)
|
|
and not paused_value
|
|
and not idle_value
|
|
and not eof_value
|
|
)
|
|
if progress_value > self.last_progress + 0.2:
|
|
self.last_progress = progress_value
|
|
self.last_progress_at = now
|
|
self.recovery_count = 0
|
|
self.last_snapshot_at = now
|
|
return self._snapshot_payload(
|
|
playing,
|
|
progress_value,
|
|
duration_value,
|
|
idle_value,
|
|
paused_value,
|
|
eof_value,
|
|
str(path or ""),
|
|
now,
|
|
)
|
|
|
|
def _snapshot_payload(
|
|
self,
|
|
playing: bool,
|
|
progress: float,
|
|
duration: float,
|
|
idle: bool,
|
|
paused: bool,
|
|
eof: bool,
|
|
path: str,
|
|
now: float,
|
|
) -> dict[str, Any]:
|
|
metadata = self.current_metadata
|
|
return {
|
|
"playing": playing,
|
|
"paused": paused,
|
|
"idle": idle,
|
|
"eof": eof,
|
|
"desired_state": self.desired_state,
|
|
"generation": self.generation,
|
|
"path": path,
|
|
"current": {
|
|
"id": str(metadata.get("id", "")),
|
|
"title": metadata.get("name") or ("暂无歌曲" if idle else "正在加载"),
|
|
"artist": metadata.get("artist") or "mpv",
|
|
"cover": metadata.get("cover", ""),
|
|
"cover_hash": metadata.get("cover_hash", ""),
|
|
"duration": duration,
|
|
"progress": progress,
|
|
"source": "mpv",
|
|
},
|
|
"playlist": [],
|
|
"requests": [],
|
|
"monitor": {
|
|
"online": self.running(),
|
|
"source": "mpv.ipc",
|
|
"platform": "mpv",
|
|
"updated_at": now,
|
|
"targets": [],
|
|
"allow_all": False,
|
|
},
|
|
}
|
|
|
|
def mark_ended(self) -> None:
|
|
self.generation += 1
|
|
self.desired_state = "stopped"
|
|
self._clear_current()
|
|
|
|
async def maintain(self, *, stall_seconds: float = 12.0) -> dict[str, Any]:
|
|
"""Recover only unexpected failures while the desired state is playing."""
|
|
if self.desired_state != "playing" or not self.current_url:
|
|
return {"action": "none"}
|
|
generation = self.generation
|
|
now = time.time()
|
|
if not self.running():
|
|
progress = self.last_progress
|
|
self.process = None
|
|
if await self._load_current(start_at=progress, expected_generation=generation):
|
|
self.logger.warning(f"[mpv] 进程退出后已从 {progress:.1f} 秒恢复")
|
|
return {"action": "process_restarted", "progress": progress}
|
|
return {"action": "failed", "reason": "process_restart_failed"}
|
|
|
|
state = await self.snapshot()
|
|
if generation != self.generation or self.desired_state != "playing":
|
|
return {"action": "superseded", "snapshot": state}
|
|
progress = float((state.get("current") or {}).get("progress", 0) or 0)
|
|
duration = float((state.get("current") or {}).get("duration", 0) or 0)
|
|
effective_progress = max(progress, self.last_progress)
|
|
near_end = duration > 0 and effective_progress >= max(0.0, duration - 2.0)
|
|
unloaded_after_progress = (
|
|
state.get("idle")
|
|
and not str(state.get("path") or "")
|
|
and effective_progress >= 0.5
|
|
)
|
|
if state.get("eof") or (state.get("idle") and near_end) or unloaded_after_progress:
|
|
self.mark_ended()
|
|
return {"action": "ended", "progress": effective_progress, "duration": duration, "snapshot": state}
|
|
|
|
loading_grace = now - self.started_at < 2.5
|
|
if state.get("paused") and not loading_grace:
|
|
if await self._command(["set_property", "pause", False]) is not None:
|
|
self.last_progress_at = now
|
|
self.logger.warning("[mpv] 检测到非预期暂停，已自动继续播放")
|
|
return {"action": "resumed", "progress": progress, "snapshot": state}
|
|
|
|
if state.get("idle") and not loading_grace:
|
|
return {"action": "reload_required", "reason": "unexpected_idle", "progress": progress, "snapshot": state}
|
|
|
|
if self.last_progress_at and not loading_grace and now - self.last_progress_at >= max(3.0, stall_seconds):
|
|
self.recovery_count += 1
|
|
if self.recovery_count == 1:
|
|
await self._command(["set_property", "pause", False])
|
|
self.last_progress_at = now
|
|
return {"action": "unstalled", "progress": progress, "snapshot": state}
|
|
return {"action": "reload_required", "reason": "stalled", "progress": progress, "snapshot": state}
|
|
return {"action": "none", "snapshot": state}
|