Files
mediaplayer/media_player.py
2026-08-12 13:07:48 +08:00

1391 lines
46 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""跨平台媒体播放器Windows / Linux / macOS
基于 tkinter播放内核按优先级为VLCpython-vlc、纯 Python 解码
PyAV + sounddevice无需安装任何外部软件、pygame 音频回退。
用法:
python media_player.py [媒体文件或文件夹...]
常用快捷键:
Space 播放 / 暂停
← / → 后退 / 前进 10 秒
Shift+← / → 后退 / 前进 60 秒
↑ / ↓ 音量 +5 / -5
M 静音 / 恢复
Ctrl+← / → 上一首 / 下一首macOS 也支持 Command+← / →)
N / P 下一首 / 上一首
[ / ] 减速 / 加速0.5x - 2.0x
L 切换循环模式(关 / 单曲 / 列表)
F 视频全屏切换
Esc 退出全屏
Ctrl+O 打开文件macOS 也支持 Command+O
Delete 从播放列表移除所选曲目
Enter 播放列表中选中的曲目
"""
import collections
import json
import os
import queue
import sys
import threading
import time
from pathlib import Path
import tkinter as tk
from tkinter import filedialog, messagebox, ttk
try:
from PIL import Image as PILImage, ImageTk
except ImportError:
PILImage = ImageTk = None
try:
import vlc
except Exception:
vlc = None
try:
import pygame
except ImportError:
pygame = None
IS_MAC = sys.platform == "darwin"
IS_WIN = sys.platform == "win32"
IS_LINUX = sys.platform.startswith("linux")
AUDIO_EXTS = {
".mp3", ".wav", ".ogg", ".flac", ".m4a", ".aac",
".opus", ".wma", ".aiff", ".ape",
}
VIDEO_EXTS = {
".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv",
".webm", ".m4v", ".ts", ".mpg", ".mpeg", ".3gp",
}
SUPPORTED_EXTS = AUDIO_EXTS | VIDEO_EXTS
LOOP_LABELS = {"off": "循环: 关", "one": "循环: 单曲", "list": "循环: 列表"}
SPEED_PRESETS = [0.5, 0.75, 1.0, 1.25, 1.5, 2.0]
SEEK_STEP = 10
SEEK_STEP_BIG = 60
def fmt_time(ms):
"""把毫秒格式化为 mm:ss 或 h:mm:ss。"""
if ms is None or ms < 0:
return "--:--"
total = int(ms // 1000)
hours, remainder = divmod(total, 3600)
minutes, seconds = divmod(remainder, 60)
if hours:
return f"{hours}:{minutes:02d}:{seconds:02d}"
return f"{minutes:02d}:{seconds:02d}"
class _BaseEngine:
"""播放内核统一接口。"""
EMBEDDABLE = False
supports_rate = True
def __init__(self):
self._events = queue.Queue()
def play(self, path):
raise NotImplementedError
def toggle_pause(self):
pass
def stop(self):
pass
def seek(self, ms):
return False
def get_time(self):
return -1
def get_length(self):
return -1
def is_active(self):
return False
def is_playing(self):
return False
def set_volume(self, volume):
pass
def set_rate(self, rate):
pass
def set_fullscreen(self, on):
pass
def embed(self, window_id):
return False
def get_frame(self):
"""返回当前应显示的帧PIL 图像),没有新帧时返回 None。"""
return None
def poll_events(self):
events = []
while True:
try:
events.append(self._events.get_nowait())
except queue.Empty:
return events
def release(self):
pass
class VlcEngine(_BaseEngine):
"""基于 VLC 的内核:支持音视频、定位、变速与全屏。"""
EMBEDDABLE = not IS_MAC
def __init__(self):
super().__init__()
args = ["--no-video-title-show", "--quiet"]
if IS_MAC:
args.append("--vout=macosx")
self.instance = vlc.Instance(args)
self.player = self.instance.media_player_new()
self._media = None
self._rate = 1.0
self._volume = 80
manager = self.player.event_manager()
manager.event_attach(vlc.EventType.MediaPlayerEndReached, self._on_event_end)
manager.event_attach(vlc.EventType.MediaPlayerEncounteredError, self._on_event_error)
def _on_event_end(self, _event):
self._events.put("end")
def _on_event_error(self, _event):
self._events.put("error")
def play(self, path):
media = self.instance.media_new(str(path))
self.player.set_media(media)
self._media = media
self.player.play()
self.player.set_rate(self._rate)
self.player.audio_set_volume(self._volume)
return True
def toggle_pause(self):
if not self.is_active():
return
if self.is_playing():
self.player.set_pause(1)
else:
self.player.set_pause(0)
def stop(self):
self.player.stop()
self._media = None
def seek(self, ms):
if not self.is_active():
return False
self.player.set_time(max(int(ms), 0))
return True
def get_time(self):
return self.player.get_time() if self.is_active() else -1
def get_length(self):
if not self.is_active():
return -1
length = self.player.get_length()
return length if length > 0 else -1
def is_active(self):
return self._media is not None
def is_playing(self):
return bool(self.player.is_playing())
def set_volume(self, volume):
self._volume = max(0, min(100, int(volume)))
self.player.audio_set_volume(self._volume)
def set_rate(self, rate):
self._rate = float(rate)
self.player.set_rate(self._rate)
def set_fullscreen(self, on):
self.player.set_fullscreen(bool(on))
def embed(self, window_id):
if IS_WIN:
self.player.set_hwnd(int(window_id))
elif IS_LINUX:
self.player.set_xwindow(int(window_id))
return True
def release(self):
try:
self.player.stop()
self.player.release()
except Exception:
pass
class PygameEngine(_BaseEngine):
"""基于 pygame 的回退内核:无 VLC 时播放音频文件。"""
EMBEDDABLE = False
supports_rate = False
def __init__(self):
super().__init__()
pygame.mixer.init()
self._volume = 80
self._paused = False
self._started = False
self._stopped = True
self._pos = 0
self._end_reported = False
def play(self, path):
self.stop()
pygame.mixer.music.load(str(path))
pygame.mixer.music.play()
pygame.mixer.music.set_volume(self._volume / 100.0)
self._started = True
self._stopped = False
self._paused = False
self._pos = 0
self._end_reported = False
return True
def toggle_pause(self):
if self._stopped:
return
if self._paused:
pygame.mixer.music.unpause()
self._paused = False
else:
pygame.mixer.music.pause()
self._paused = True
def stop(self):
pygame.mixer.music.stop()
self._started = False
self._stopped = True
self._paused = False
self._pos = 0
def seek(self, ms):
try:
pygame.mixer.music.set_pos(max(ms, 0) / 1000.0)
self._pos = max(ms, 0)
return True
except pygame.error:
return False
def get_time(self):
if self._stopped:
return -1
if not self._paused and pygame.mixer.music.get_busy():
pos = pygame.mixer.music.get_pos()
if pos > 0:
self._pos = pos
return int(self._pos)
def get_length(self):
return -1
def is_active(self):
return self._started
def is_playing(self):
return self._started and not self._paused and pygame.mixer.music.get_busy()
def set_volume(self, volume):
self._volume = max(0, min(100, int(volume)))
pygame.mixer.music.set_volume(self._volume / 100.0)
def set_rate(self, rate):
pass
def set_fullscreen(self, on):
pass
def poll_events(self):
events = super().poll_events()
if (
self._started
and not self._paused
and not self._stopped
and not pygame.mixer.music.get_busy()
and not self._end_reported
):
self._end_reported = True
events.append("end")
return events
def release(self):
try:
pygame.mixer.music.stop()
pygame.mixer.quit()
except Exception:
pass
class AvEngine(_BaseEngine):
"""纯 Python 解码内核:基于 PyAV 与 sounddevice无需安装 VLC。
跨平台Windows / macOS / Linux可用PyAV 自带 FFmpeg 解码能力,
sounddevice 自带 PortAudio 音频输出,视频画面由主窗口负责绘制。
"""
EMBEDDABLE = True
supports_rate = False
AUDIO_RATE = 44100
def __init__(self):
super().__init__()
try:
import av
import numpy # noqa: F401 to_ndarray 依赖
import sounddevice as sd
from PIL import Image # noqa: F401
except ImportError as exc:
raise ImportError("缺少纯 Python 播放依赖av / sounddevice / numpy / Pillow%s" % exc)
self._av = av
self._sd = sd
self._lock = threading.Lock()
self._reset_state()
def _reset_state(self):
self._container = None
self._video_stream = None
self._audio_stream = None
self._video_deque = collections.deque()
self._video_deque_max = 6
self._audio_queue = queue.Queue(maxsize=64)
self._audio_leftover = None
self._audio_stream_obj = None
self._audio_ok = False
self._audio_broken = False
self._audio_written = 0
self._resampler = None
self._thread = None
self._stop_flag = threading.Event()
self._seek_request = None
self._gen = 0
self._seek_after = 0.0
self._eof = False
self._ended = False
self._paused = False
self._volume = 80
self._clock_base = 0.0
self._clock_started = time.monotonic()
self._clock_running = False
def play(self, path):
self.stop()
container = self._av.open(str(path))
self._container = container
videos = [s for s in container.streams if s.type == "video"]
if videos:
from av.stream import Disposition
self._video_stream = max(
videos,
key=lambda s: (0 if s.disposition & Disposition.attached_pic else 1, s.frames or 0),
)
else:
self._video_stream = None
self._audio_stream = next((s for s in container.streams if s.type == "audio"), None)
self._gen += 1
self._eof = False
self._ended = False
self._paused = False
self._seek_after = 0.0
self._resampler = None
self._audio_ok = False
self._audio_broken = False
if self._audio_stream is not None:
try:
self._audio_stream_obj = self._sd.OutputStream(
samplerate=self.AUDIO_RATE,
channels=2,
dtype="float32",
callback=self._audio_callback,
)
self._audio_stream_obj.start()
self._audio_ok = True
except Exception:
self._audio_stream_obj = None
self._audio_written = 0
self._clock_base = 0.0
self._clock_started = time.monotonic()
self._clock_running = self._audio_stream_obj is None
self._stop_flag = threading.Event()
self._thread = threading.Thread(target=self._decode_loop, daemon=True)
self._thread.start()
return True
def toggle_pause(self):
if not self.is_active() or self._ended:
return
self._paused = not self._paused
if self._paused:
if self._clock_running:
self._clock_base = self._noaudio_pos()
self._clock_running = False
else:
self._clock_started = time.monotonic()
if self._audio_stream_obj is None:
self._clock_running = True
def stop(self):
self._stop_flag.set()
if self._thread is not None:
self._thread.join(timeout=1.0)
if self._audio_stream_obj is not None:
try:
self._audio_stream_obj.stop()
self._audio_stream_obj.close()
except Exception:
pass
if self._container is not None:
try:
self._container.close()
except Exception:
pass
self._reset_state()
def seek(self, ms):
if not self.is_active():
return False
target = max(0.0, int(ms) / 1000.0)
with self._lock:
self._seek_request = target
self._seek_after = target
self._gen += 1
self._video_deque.clear()
self._drain_queue(self._audio_queue)
self._audio_leftover = None
self._audio_written = int(target * self.AUDIO_RATE)
self._clock_base = target
self._clock_started = time.monotonic()
self._ended = False
if self._audio_stream_obj is not None:
try:
self._audio_stream_obj.stop()
self._audio_stream_obj.start()
except Exception:
pass
if self._thread is None or not self._thread.is_alive():
self._eof = False
self._resampler = None
self._stop_flag = threading.Event()
self._thread = threading.Thread(target=self._decode_loop, daemon=True)
self._thread.start()
return True
def get_time(self):
if not self.is_active():
return -1
if self._audio_stream_obj is not None:
with self._lock:
return int(self._audio_written * 1000.0 / self.AUDIO_RATE)
return int(self._noaudio_pos() * 1000.0)
def _disable_audio(self, reason=""):
"""音频不可用时降级为无声播放,视频播放不受影响。"""
stream = self._audio_stream_obj
self._audio_stream_obj = None
self._audio_ok = False
self._audio_broken = False
self._drain_queue(self._audio_queue)
self._audio_leftover = None
if stream is not None:
try:
stream.stop()
stream.close()
except Exception:
pass
with self._lock:
self._clock_base = self._audio_written / self.AUDIO_RATE
self._clock_started = time.monotonic()
self._clock_running = not self._paused
if reason:
print("音频不可用,已切换为无声播放:", reason)
def _noaudio_pos(self):
if self._clock_running:
return self._clock_base + (time.monotonic() - self._clock_started)
return self._clock_base
def get_length(self):
if not self.is_active():
return -1
duration = self._container.duration
if duration and duration > 0:
return int(duration / 1000)
return -1
def is_active(self):
return self._container is not None
def is_playing(self):
return self.is_active() and not self._paused and not self._ended
def set_volume(self, volume):
self._volume = max(0, min(100, int(volume)))
def set_rate(self, rate):
pass
def set_fullscreen(self, on):
pass
def embed(self, window_id):
return False
def get_frame(self):
if not self.is_active():
return None
pos = self.get_time() / 1000.0
latest = None
with self._lock:
while self._video_deque:
gen, secs, img = self._video_deque[0]
if gen != self._gen:
self._video_deque.popleft()
continue
if secs > pos + 0.05:
break
latest = img
self._video_deque.popleft()
return latest
def poll_events(self):
events = super().poll_events()
if self._audio_broken and self._audio_stream_obj is not None:
self._disable_audio("音频回调异常")
if not self._ended and self.is_active() and not self._paused and self._eof:
with self._lock:
video_done = not self._video_deque
audio_done = (
self._audio_stream_obj is None
or (self._audio_queue.empty() and self._audio_leftover is None)
)
if video_done and audio_done:
self._ended = True
events.append("end")
return events
def release(self):
self.stop()
def _decode_loop(self):
try:
container = self._container
video = self._video_stream
audio = self._audio_stream
streams = [s for s in (video, audio) if s is not None]
demux = container.demux(*streams)
while not self._stop_flag.is_set():
if self._seek_request is not None:
with self._lock:
target = self._seek_request
self._seek_request = None
seek_stream = video if video is not None else audio
if seek_stream is not None:
offset = int(round(target / float(seek_stream.time_base)))
try:
container.seek(offset, stream=seek_stream)
except Exception:
pass
demux = container.demux(*streams)
self._resampler = None
continue
gen = self._gen
try:
packet = next(demux)
except StopIteration:
self._eof = True
break
except Exception:
self._eof = True
break
if self._stop_flag.is_set():
break
if packet.stream.type == "video" and video is not None and packet.size:
try:
frames = packet.decode()
except Exception:
continue
for frame in frames:
if self._stop_flag.is_set():
break
if frame.pts is None:
continue
secs = float(frame.pts * video.time_base)
if secs + 0.05 < self._seek_after:
continue
try:
img = frame.to_image()
except Exception:
continue
while not self._stop_flag.is_set():
with self._lock:
if len(self._video_deque) < self._video_deque_max:
self._video_deque.append((gen, secs, img))
break
time.sleep(0.005)
elif packet.stream.type == "audio" and audio is not None and packet.size:
if not self._audio_ok:
continue
try:
self._decode_audio_packet(packet, gen)
except Exception as exc:
self._disable_audio("音频解码出错(%s" % exc)
if self._resampler is not None and not self._stop_flag.is_set():
self._flush_resampler()
except Exception as exc:
self._eof = True
print("解码线程异常退出:", exc)
def _decode_audio_packet(self, packet, gen):
try:
frames = packet.decode()
except Exception:
return
atb = self._audio_stream.time_base
for frame in frames:
if self._stop_flag.is_set():
return
if frame.pts is not None:
secs = float(frame.pts * atb)
if secs + 0.05 < self._seek_after:
continue
if self._resampler is None:
self._resampler = self._av.AudioResampler(
format="fltp", layout="stereo", rate=self.AUDIO_RATE
)
for out in self._resampler.resample(frame):
try:
chunk = out.to_ndarray().T
except Exception:
continue
self._put_audio((gen, chunk))
def _put_audio(self, item):
while not self._stop_flag.is_set():
try:
self._audio_queue.put(item, timeout=0.1)
return
except queue.Full:
if item[0] != self._gen:
return
def _flush_resampler(self):
try:
for out in self._resampler.resample(None):
self._put_audio((self._gen, out.to_ndarray().T))
except Exception:
pass
def _audio_callback(self, outdata, frames, time_info, status):
try:
self._audio_callback_inner(outdata, frames)
except Exception:
self._audio_broken = True
outdata[:] = 0
def _audio_callback_inner(self, outdata, frames):
if self._paused:
outdata[:] = 0
return
outdata[:] = 0
n = 0
leftover = self._audio_leftover
if leftover is not None and len(leftover):
take = min(len(leftover), frames)
outdata[:take] = leftover[:take]
n = take
self._audio_leftover = leftover[take:] if take < len(leftover) else None
while n < frames:
try:
item = self._audio_queue.get_nowait()
except queue.Empty:
break
gen, chunk = item
if gen != self._gen:
continue
take = min(len(chunk), frames - n)
outdata[n:n + take] = chunk[:take]
n += take
if take < len(chunk):
self._audio_leftover = chunk[take:]
outdata *= self._volume / 100.0
with self._lock:
self._audio_written += frames
@staticmethod
def _drain_queue(q):
while True:
try:
q.get_nowait()
except queue.Empty:
return
class SeekBar(tk.Canvas):
"""自绘进度条:支持点击与拖动定位,跨平台表现一致。"""
TRACK_COLOR = "#4a4a4a"
FILL_COLOR = "#3f8efc"
def __init__(self, master, height=24, command=None):
super().__init__(
master,
height=height,
bg="#2b2b2b",
highlightthickness=0,
cursor="hand2",
)
self._command = command
self._frac = 0.0
self._dragging = False
self.bind("<Configure>", lambda _e: self._redraw())
self.bind("<ButtonPress-1>", self._on_press)
self.bind("<B1-Motion>", self._on_drag)
self.bind("<ButtonRelease-1>", self._on_release)
self._redraw()
def _width(self):
return max(self.winfo_width(), 1)
def _x_of_frac(self, frac):
return 4 + frac * (self._width() - 8)
def _frac_of_event(self, event):
return min(max((event.x - 4) / max(self._width() - 8, 1), 0.0), 1.0)
def set_frac(self, frac):
if not self._dragging:
self._frac = min(max(frac, 0.0), 1.0)
self._redraw()
def _redraw(self):
self.delete("all")
width, height = self._width(), max(self.winfo_height(), 1)
y = height // 2
self.create_rectangle(2, y - 3, width - 2, y + 3, fill=self.TRACK_COLOR, outline="")
x = self._x_of_frac(self._frac)
if x > 4:
self.create_rectangle(2, y - 3, x, y + 3, fill=self.FILL_COLOR, outline="")
radius = 6
self.create_oval(x - radius, y - radius, x + radius, y + radius, fill=self.FILL_COLOR, outline="")
def _on_press(self, event):
self._dragging = True
self._frac = self._frac_of_event(event)
self._redraw()
if self._command:
self._command(self._frac, "start")
def _on_drag(self, event):
if not self._dragging:
return
self._frac = self._frac_of_event(event)
self._redraw()
if self._command:
self._command(self._frac, "drag")
def _on_release(self, event):
if not self._dragging:
return
self._dragging = False
self._frac = self._frac_of_event(event)
self._redraw()
if self._command:
self._command(self._frac, "end")
class MediaPlayerApp(tk.Tk):
"""播放器主窗口。"""
def __init__(self):
super().__init__()
self.title("媒体播放器")
self.geometry("960x600")
self.minsize(720, 460)
self.playlist = []
self.current_index = -1
self.volume = 80
self.rate = 1.0
self.loop_mode = "list"
self._seeking = False
self._switching_until = 0.0
self._fullscreen = False
self._muted_volume = None
self._photo = None
self.engine = self._create_engine()
if self.engine is None:
messagebox.showerror(
"无法启动",
"未找到可用的播放内核。\n"
"请先安装 python-vlc并安装 VLC 播放器),或安装 pygame。",
)
self.destroy()
return
self._build_ui()
self._bind_shortcuts()
if isinstance(self.engine, PygameEngine):
self.status_var.set("pygame 音频模式:未检测到 VLC视频功能不可用")
elif isinstance(self.engine, AvEngine):
self.status_var.set("纯 Python 解码内核:无需安装 VLC可直接播放音视频")
self._load_playlist()
self.protocol("WM_DELETE_WINDOW", self._on_close)
self.after(100, self._maybe_embed)
self.after(200, self._tick)
def _create_engine(self):
if vlc is not None:
try:
return VlcEngine()
except Exception as exc:
print("VLC 内核初始化失败,回退到纯 Python 内核:", exc)
try:
return AvEngine()
except Exception as exc:
print("纯 Python 内核初始化失败,回退到 pygame", exc)
if pygame is not None:
try:
return PygameEngine()
except Exception as exc:
print("pygame 内核初始化失败:", exc)
return None
def _build_ui(self):
style = ttk.Style(self)
if "clam" in style.theme_names():
style.theme_use("clam")
toolbar = ttk.Frame(self, padding=(8, 8))
toolbar.pack(fill="x")
ttk.Button(toolbar, text="打开文件…", command=self.open_files, takefocus=0).pack(side="left", padx=(0, 4))
ttk.Button(toolbar, text="打开文件夹…", command=self.open_folder, takefocus=0).pack(side="left", padx=(0, 8))
self.btn_prev = ttk.Button(toolbar, text="上一首", command=self.prev_track, takefocus=0)
self.btn_prev.pack(side="left", padx=2)
self.btn_play = ttk.Button(toolbar, text="播放", command=self.toggle_play, width=8, takefocus=0)
self.btn_play.pack(side="left", padx=2)
self.btn_stop = ttk.Button(toolbar, text="停止", command=self.stop_playback, takefocus=0)
self.btn_stop.pack(side="left", padx=2)
self.btn_next = ttk.Button(toolbar, text="下一首", command=self.next_track, takefocus=0)
self.btn_next.pack(side="left", padx=(2, 10))
self.btn_loop = ttk.Button(toolbar, text=LOOP_LABELS[self.loop_mode], command=self.cycle_loop, width=9, takefocus=0)
self.btn_loop.pack(side="left", padx=2)
self.btn_speed = ttk.Button(toolbar, text="速度 1x", command=self.cycle_speed, width=9, takefocus=0)
self.btn_speed.pack(side="left", padx=2)
ttk.Label(toolbar, text="音量").pack(side="right", padx=(10, 4))
self.volume_slider = ttk.Scale(
toolbar,
from_=0,
to=100,
length=110,
value=self.volume,
command=self._on_volume_slider,
takefocus=0,
)
self.volume_slider.pack(side="right", padx=(0, 4))
main = ttk.Frame(self, padding=(8, 0))
main.pack(fill="both", expand=True)
main.columnconfigure(0, weight=1)
main.rowconfigure(0, weight=1)
self.video_frame = tk.Frame(main, bg="#101010")
self.video_frame.grid(row=0, column=0, sticky="nsew", padx=(0, 8))
if isinstance(self.engine, AvEngine):
self.video_canvas = tk.Canvas(self.video_frame, bg="#101010", highlightthickness=0)
self.video_canvas.pack(fill="both", expand=True)
hint_text = "打开文件开始播放\n(视频将显示在此窗口)"
if not self.engine.EMBEDDABLE:
hint_text = "打开文件开始播放\n(视频将在独立窗口中显示)"
self.video_hint = tk.Label(self.video_frame, text=hint_text, bg="#101010", fg="#8a8a8a")
self.video_hint.place(relx=0.5, rely=0.5, anchor="center")
side = ttk.Frame(main, width=260)
side.grid(row=0, column=1, sticky="ns")
side.columnconfigure(0, weight=1)
side.rowconfigure(0, weight=1)
self.listbox = tk.Listbox(
side,
activestyle="none",
selectmode="extended",
selectbackground="#3f8efc",
selectforeground="#ffffff",
exportselection=False,
highlightthickness=0,
relief="flat",
)
self.listbox.grid(row=0, column=0, sticky="nsew")
scrollbar = ttk.Scrollbar(side, orient="vertical", command=self.listbox.yview)
scrollbar.grid(row=0, column=1, sticky="ns")
self.listbox.configure(yscrollcommand=scrollbar.set)
self._playlist_menu = tk.Menu(self, tearoff=0)
self._playlist_menu.add_command(label="播放选中曲目", command=self.play_selected)
self._playlist_menu.add_separator()
self._playlist_menu.add_command(label="删除选中曲目", command=self.remove_selected)
self._playlist_menu.add_command(label="清空播放列表", command=self.clear_playlist)
if IS_MAC:
self.listbox.bind("<Button-2>", self._show_playlist_menu)
self.listbox.bind("<Button-3>", self._show_playlist_menu)
self.listbox.bind("<Control-Button-1>", self._show_playlist_menu)
else:
self.listbox.bind("<Button-3>", self._show_playlist_menu)
bottom = ttk.Frame(self, padding=(8, 6))
bottom.pack(fill="x")
self.seekbar = SeekBar(bottom, command=self._on_seek_command)
self.seekbar.pack(fill="x")
self.time_var = tk.StringVar(value="00:00 / --:--")
self.status_var = tk.StringVar(value="就绪 — 打开文件或按 Ctrl+O")
ttk.Label(bottom, textvariable=self.time_var).pack(side="left", anchor="w")
ttk.Label(bottom, textvariable=self.status_var, anchor="e").pack(side="right", fill="x", expand=True)
def _bind_shortcuts(self):
shortcuts = [
("<space>", self._on_space),
("<Left>", self._on_left),
("<Right>", self._on_right),
("<Shift-Left>", lambda _e: self.seek_relative(-SEEK_STEP_BIG)),
("<Shift-Right>", lambda _e: self.seek_relative(SEEK_STEP_BIG)),
("<Control-Left>", lambda _e: self.prev_track()),
("<Control-Right>", lambda _e: self.next_track()),
("<Command-Left>", lambda _e: self.prev_track()),
("<Command-Right>", lambda _e: self.next_track()),
("<Up>", self._on_up),
("<Down>", self._on_down),
("<m>", lambda _e: self.toggle_mute()),
("<M>", lambda _e: self.toggle_mute()),
("<Control-o>", lambda _e: self.open_files()),
("<Command-o>", lambda _e: self.open_files()),
("<Control-Shift-O>", lambda _e: self.open_folder()),
("<Command-Shift-O>", lambda _e: self.open_folder()),
("<Control-s>", lambda _e: self.stop_playback()),
("<Command-s>", lambda _e: self.stop_playback()),
("<s>", lambda _e: self.stop_playback()),
("<S>", lambda _e: self.stop_playback()),
("<n>", lambda _e: self.next_track()),
("<N>", lambda _e: self.next_track()),
("<p>", lambda _e: self.prev_track()),
("<P>", lambda _e: self.prev_track()),
("<l>", lambda _e: self.cycle_loop()),
("<L>", lambda _e: self.cycle_loop()),
("<bracketleft>", lambda _e: self.step_speed(-1)),
("<bracketright>", lambda _e: self.step_speed(1)),
("<f>", lambda _e: self.toggle_fullscreen()),
("<F>", lambda _e: self.toggle_fullscreen()),
("<Escape>", lambda _e: self.leave_fullscreen()),
("<Delete>", lambda _e: self.remove_selected()),
("<Return>", lambda _e: self.play_selected()),
]
for sequence, handler in shortcuts:
self.bind(sequence, handler)
self.listbox.bind(sequence, handler)
self.listbox.bind("<Double-Button-1>", lambda _e: self.play_selected())
def _on_space(self, _event):
self.toggle_play()
return "break"
def _on_left(self, _event):
self.seek_relative(-SEEK_STEP)
return "break"
def _on_right(self, _event):
self.seek_relative(SEEK_STEP)
return "break"
def _on_up(self, _event):
self.adjust_volume(5)
return "break"
def _on_down(self, _event):
self.adjust_volume(-5)
return "break"
def _maybe_embed(self):
if self.engine.EMBEDDABLE:
try:
self.engine.embed(self.video_frame.winfo_id())
except Exception:
pass
def _tick(self):
if self.engine is not None:
for event in self.engine.poll_events():
if event == "end":
self._on_media_end()
elif event == "error":
self.status_var.set("无法播放该文件,请检查文件格式或 VLC 是否已安装")
img = self.engine.get_frame()
if img is not None:
self._show_frame(img)
pos = self.engine.get_time()
length = self.engine.get_length()
if pos >= 0 and not self._seeking:
self.time_var.set(f"{fmt_time(pos)} / {fmt_time(length)}")
if length > 0:
self.seekbar.set_frac(pos / length)
playing = self.engine.is_playing()
self.btn_play.config(text="暂停" if playing else "播放")
self.after(33, self._tick)
def _show_frame(self, img):
if ImageTk is None:
return
canvas = getattr(self, "video_canvas", None)
if canvas is None:
return
width = canvas.winfo_width()
height = canvas.winfo_height()
if width < 2 or height < 2:
return
iw, ih = img.size
if iw < 1 or ih < 1:
return
scale = min(width / iw, height / ih)
size = (max(1, int(iw * scale)), max(1, int(ih * scale)))
if size != img.size:
img = img.resize(size, PILImage.Resampling.BILINEAR)
self._photo = ImageTk.PhotoImage(img)
canvas.delete("all")
canvas.create_image(width // 2, height // 2, image=self._photo, anchor="center")
def open_files(self):
filetypes = [
("媒体文件", " ".join("*" + ext for ext in sorted(SUPPORTED_EXTS))),
("所有文件", "*"),
]
paths = filedialog.askopenfilenames(title="选择媒体文件", filetypes=filetypes)
if paths:
self.add_paths(list(paths))
def open_folder(self):
folder = filedialog.askdirectory(title="选择包含媒体文件的文件夹")
if folder:
self.add_paths([folder])
def add_paths(self, paths):
start = len(self.playlist)
added = []
for item in paths:
path = Path(item)
if path.is_dir():
for child in sorted(path.rglob("*")):
if child.is_file() and child.suffix.lower() in SUPPORTED_EXTS:
added.append(str(child))
elif path.is_file() and path.suffix.lower() in SUPPORTED_EXTS:
added.append(str(path))
if not added:
self.status_var.set("所选路径中没有支持的媒体文件")
return
self.playlist.extend(added)
for name in added:
self.listbox.insert(tk.END, Path(name).name)
if not self.engine.is_active():
self.play_index(start)
else:
self._update_status()
self._save_playlist()
def _playlist_path(self):
return Path.home() / ".mediaplayer_playlist.json"
def _save_playlist(self):
try:
data = {"playlist": list(self.playlist), "current_index": self.current_index}
self._playlist_path().write_text(
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
)
except Exception as exc:
print("保存播放列表失败:", exc)
def _load_playlist(self):
path = self._playlist_path()
if not path.exists():
return
try:
data = json.loads(path.read_text(encoding="utf-8"))
items = data.get("playlist", [])
except Exception as exc:
print("读取播放列表失败:", exc)
return
added = []
for item in items:
p = Path(item)
if p.is_file() and p.suffix.lower() in SUPPORTED_EXTS:
added.append(str(p))
if not added:
return
self.playlist.extend(added)
for name in added:
self.listbox.insert(tk.END, Path(name).name)
saved_index = data.get("current_index", -1)
if 0 <= saved_index < len(self.playlist):
self.current_index = saved_index
self.listbox.selection_set(saved_index)
self.listbox.see(saved_index)
msg = f"已恢复上次的播放列表({len(added)} 个文件)"
dropped = len(items) - len(added)
if dropped:
msg += f",已忽略 {dropped} 个不存在的文件"
self.status_var.set(msg)
def play_index(self, index):
if not (0 <= index < len(self.playlist)):
return
self.current_index = index
path = self.playlist[index]
self._switching_until = time.monotonic() + 0.8
if self.engine.EMBEDDABLE:
try:
self.engine.embed(self.video_frame.winfo_id())
except Exception:
pass
try:
ok = self.engine.play(path)
except Exception:
ok = False
if not ok:
self.status_var.set(f"无法播放:{Path(path).name}")
return
self.listbox.selection_clear(0, tk.END)
self.listbox.selection_set(index)
self.listbox.see(index)
if Path(path).suffix.lower() in VIDEO_EXTS and self.engine.EMBEDDABLE:
self.video_hint.place_forget()
else:
self.video_hint.place(relx=0.5, rely=0.5, anchor="center")
self.title(f"{Path(path).name} — 媒体播放器")
self._update_status()
self._save_playlist()
def toggle_play(self):
if not self.playlist:
self.open_files()
return
if not self.engine.is_active():
self.play_index(self.current_index if self.current_index >= 0 else 0)
return
self.engine.toggle_pause()
self._update_status()
def stop_playback(self):
self.engine.stop()
self._seeking = False
self.time_var.set("00:00 / --:--")
self.seekbar.set_frac(0.0)
self.btn_play.config(text="播放")
canvas = getattr(self, "video_canvas", None)
if canvas is not None:
canvas.delete("all")
self._photo = None
self.video_hint.place(relx=0.5, rely=0.5, anchor="center")
self._update_status()
def next_track(self):
if not self.playlist:
self.open_files()
return
self.play_index((self.current_index + 1) % len(self.playlist))
def prev_track(self):
if not self.playlist:
self.open_files()
return
self.play_index((self.current_index - 1) % len(self.playlist))
def seek_relative(self, seconds):
pos = self.engine.get_time()
if pos < 0:
return
self.seek_to(pos + seconds * 1000)
def seek_to(self, ms):
ms = max(0, int(ms))
if not self.engine.seek(ms):
self.status_var.set("当前文件格式不支持定位")
def _on_seek_command(self, frac, phase):
length = self.engine.get_length()
if length <= 0:
return
target = int(frac * length)
if phase in ("start", "drag"):
self._seeking = True
self.time_var.set(f"{fmt_time(target)} / {fmt_time(length)}")
else:
self._seeking = False
self.seek_to(target)
def adjust_volume(self, delta):
self.volume = max(0, min(100, self.volume + delta))
self.volume_slider.set(self.volume)
self.engine.set_volume(self.volume)
if self.volume > 0:
self._muted_volume = None
self._update_status()
def _on_volume_slider(self, value):
self.volume = max(0, min(100, int(float(value))))
self.engine.set_volume(self.volume)
if self.volume > 0:
self._muted_volume = None
self._update_status()
def toggle_mute(self):
if self.volume > 0:
self._muted_volume = self.volume
self.adjust_volume(-self.volume)
else:
self.adjust_volume(self._muted_volume or 80)
def cycle_loop(self):
order = ["off", "one", "list"]
self.loop_mode = order[(order.index(self.loop_mode) + 1) % len(order)]
self.btn_loop.config(text=LOOP_LABELS[self.loop_mode])
self._update_status()
def cycle_speed(self):
self.step_speed(1)
def step_speed(self, direction):
if not getattr(self.engine, "supports_rate", True):
self.status_var.set("当前播放内核不支持变速")
return
index = SPEED_PRESETS.index(self.rate) if self.rate in SPEED_PRESETS else SPEED_PRESETS.index(1.0)
index = (index + direction) % len(SPEED_PRESETS)
self.rate = SPEED_PRESETS[index]
self.engine.set_rate(self.rate)
self.btn_speed.config(text=f"速度 {self.rate:g}x")
self._update_status()
def toggle_fullscreen(self):
self._fullscreen = not self._fullscreen
if isinstance(self.engine, AvEngine):
self.attributes("-fullscreen", self._fullscreen)
else:
self.engine.set_fullscreen(self._fullscreen)
def leave_fullscreen(self):
self._fullscreen = False
if isinstance(self.engine, AvEngine):
self.attributes("-fullscreen", False)
else:
self.engine.set_fullscreen(False)
def play_selected(self):
selection = self.listbox.curselection()
if selection:
self.play_index(selection[0])
def remove_selected(self):
selection = list(self.listbox.curselection())
if not selection:
return
if self.current_index in selection:
self.engine.stop()
self.current_index = -1
for index in reversed(selection):
self.playlist.pop(index)
self.listbox.delete(index)
self._update_status()
self._save_playlist()
def _show_playlist_menu(self, event):
has_selection = bool(self.listbox.curselection())
self._playlist_menu.entryconfig("播放选中曲目", state="normal" if has_selection else "disabled")
self._playlist_menu.entryconfig("删除选中曲目", state="normal" if has_selection else "disabled")
try:
self._playlist_menu.tk_popup(event.x_root, event.y_root)
finally:
self._playlist_menu.grab_release()
def clear_playlist(self):
if not self.playlist:
return
if not messagebox.askyesno("清空播放列表", "确定要清空整个播放列表吗?"):
return
self.stop_playback()
self.playlist.clear()
self.listbox.delete(0, tk.END)
self.current_index = -1
self._save_playlist()
self.status_var.set("播放列表已清空")
def _update_status(self):
parts = [f"循环:{LOOP_LABELS[self.loop_mode].split(':')[1].strip()}"]
parts.append(f"速度 {self.rate:g}x")
parts.append(f"音量 {self.volume}")
if isinstance(self.engine, PygameEngine):
parts.append("pygame 音频模式(未检测到 VLC视频不可用")
elif isinstance(self.engine, AvEngine):
parts.append("纯 Python 解码内核(无需 VLC")
if 0 <= self.current_index < len(self.playlist):
name = Path(self.playlist[self.current_index]).name
total = len(self.playlist)
parts.insert(0, f"正在播放:{name}(第 {self.current_index + 1}/{total} 首)")
self.status_var.set("".join(parts))
def _on_media_end(self):
if time.monotonic() < self._switching_until:
return
if not self.playlist or not self.engine.is_active():
return
if self.loop_mode == "one":
self.play_index(self.current_index)
return
next_index = self.current_index + 1
if next_index < len(self.playlist):
self.play_index(next_index)
elif self.loop_mode == "list":
self.play_index(0)
else:
self.stop_playback()
self.status_var.set("播放结束")
def _on_close(self):
self._save_playlist()
self.engine.release()
self.destroy()
def main():
app = MediaPlayerApp()
if app.engine is None:
return
if sys.argv[1:]:
app.add_paths(sys.argv[1:])
app.mainloop()
if __name__ == "__main__":
main()