136 lines
4.0 KiB
Python
Executable File
136 lines
4.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
import fcntl
|
||
import os
|
||
import time
|
||
import eyed3
|
||
import tempfile
|
||
import subprocess
|
||
from mpd import MPDClient
|
||
from pathlib import Path
|
||
from typing import Optional, Union
|
||
import hashlib
|
||
|
||
# lock file to prevent parallel runs / duplicate notifications
|
||
LOCK_FN = "/tmp/rmpc-notify.lock"
|
||
DUPLICATE_WINDOW = 5 # seconds to consider same file a duplicate
|
||
|
||
exception_buff = []
|
||
|
||
# Acquire non-blocking exclusive lock; if fails, another instance is running -> exit
|
||
_lock_fd = open(LOCK_FN, "a+")
|
||
try:
|
||
fcntl.flock(_lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
_lock_fd.seek(0)
|
||
_lock_fd.truncate()
|
||
_lock_fd.write(f"{os.getpid()} {time.time()}\n")
|
||
_lock_fd.flush()
|
||
except BlockingIOError:
|
||
# Another instance is running — exit silently
|
||
raise SystemExit(0)
|
||
|
||
|
||
_last_shown = {"id": None, "ts": 0}
|
||
|
||
def file_id(path: Path) -> str:
|
||
try:
|
||
st = path.stat()
|
||
return f"{path}:{st.st_mtime_ns}"
|
||
except Exception:
|
||
return hashlib.sha1(str(path).encode()).hexdigest()
|
||
|
||
def extract(track: Path) -> Optional[Path]:
|
||
try:
|
||
audio = eyed3.load(track)
|
||
except Exception as e:
|
||
exception_buff.append(f"eyed3.load error: {e}")
|
||
return None
|
||
|
||
if not audio or not audio.tag:
|
||
exception_buff.append("track data unavailable")
|
||
return None
|
||
|
||
# Защита от проблем с парсингом дат (например "0")
|
||
try:
|
||
rd = getattr(audio.tag, "recording_date", None)
|
||
if rd:
|
||
raw = getattr(rd, "text", None) or str(rd)
|
||
if raw == "0" or raw.strip() == "":
|
||
audio.tag.recording_date = None
|
||
except Exception:
|
||
pass
|
||
|
||
images = getattr(audio.tag, "images", None)
|
||
if not images:
|
||
return None
|
||
|
||
try:
|
||
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp:
|
||
temp.write(images[0].image_data)
|
||
return temp.name
|
||
except Exception as e:
|
||
exception_buff.append(f"image extract error: {e}")
|
||
return None
|
||
|
||
def notify(icon: Union[str, Path], title: str, body: str) -> None:
|
||
try:
|
||
subprocess.Popen(["notify-send", "-i", str(icon), "-a", "rmpc", "-t", "2500", title, body])
|
||
except Exception as e:
|
||
exception_buff.append(e)
|
||
|
||
def main():
|
||
global _last_shown
|
||
|
||
try:
|
||
client = MPDClient()
|
||
client.connect("localhost", 6600)
|
||
song = client.currentsong() or {}
|
||
except Exception as e:
|
||
exception_buff.append(f"mpd connection error: {e}")
|
||
song = {}
|
||
|
||
music_path = Path("/home/zloy_linux/Музыка/iTunes/")
|
||
track_rel = song.get('file')
|
||
if not track_rel:
|
||
exception_buff.append("no track file in mpd response")
|
||
track_path = None
|
||
else:
|
||
track_path = music_path / track_rel
|
||
|
||
cover_path = None
|
||
if not track_path or not track_path.exists():
|
||
exception_buff.append("track file not found")
|
||
else:
|
||
ident = file_id(track_path)
|
||
now = time.time()
|
||
# дедупликация по файлу+mtime в пределах окна
|
||
if _last_shown["id"] == ident and (now - _last_shown["ts"]) < DUPLICATE_WINDOW:
|
||
return # пропускаем повторный показ
|
||
cover_path = extract(track_path)
|
||
_last_shown["id"] = ident
|
||
_last_shown["ts"] = now
|
||
|
||
artist = song.get('artist', 'Unknown artist')
|
||
title = song.get('title', 'Unknown title')
|
||
body = f"[E] {', '.join(str(e) for e in exception_buff)}" if exception_buff else artist
|
||
|
||
notify(cover_path or "audio-x-generic", title, body)
|
||
|
||
if cover_path:
|
||
time.sleep(1)
|
||
try:
|
||
os.remove(cover_path)
|
||
except Exception as e:
|
||
# не критично — логим в exception_buff и выводим в stdout
|
||
print(f"[E] cleanup error: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
finally:
|
||
try:
|
||
# release lock by closing descriptor (file remains but unlocked)
|
||
_lock_fd.close()
|
||
except Exception:
|
||
pass
|
||
|