Skip to content

Instantly share code, notes, and snippets.

@johnnyg
Last active July 10, 2022 01:44
Show Gist options
  • Save johnnyg/da264f522ab8d509eaa988cba9ede1d6 to your computer and use it in GitHub Desktop.
Save johnnyg/da264f522ab8d509eaa988cba9ede1d6 to your computer and use it in GitHub Desktop.
Script to skip/mute Spotify ads
[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true
[dev-packages]
[packages]
pydbus = "*"
pulsectl = "*"
[requires]
python_version = "3"
{
"_meta": {
"hash": {
"sha256": "9d5ee5ed4b4230285ebd27ce0f1682e28d7a2685e6d60433175782410572818f"
},
"pipfile-spec": 6,
"requires": {
"python_version": "3"
},
"sources": [
{
"name": "pypi",
"url": "https://pypi.org/simple",
"verify_ssl": true
}
]
},
"default": {
"pulsectl": {
"hashes": [
"sha256:5042b9b69733ae691ed2ed7194f775e95a4ae7063542e4fd06ec8954f0fa61b7",
"sha256:b347983fb78baab168f4dc4804ab2c59ca5b813bf62f8146dfb5fcb6ab6c8ba2"
],
"index": "pypi",
"version": "==21.10.5"
},
"pydbus": {
"hashes": [
"sha256:4207162eff54223822c185da06c1ba8a34137a9602f3da5a528eedf3f78d0f2c",
"sha256:66b80106352a718d80d6c681dc2a82588048e30b75aab933e4020eb0660bf85e"
],
"index": "pypi",
"version": "==0.6.0"
}
},
"develop": {}
}
#!/usr/bin/env python3
# Requires system python-gobject (need to manually install),
# pulsectl and pydbus (provided by Pipfile)
from gi.repository import GLib
from pydbus import SessionBus
import pulsectl
from datetime import timedelta
import threading
import time
import sys
def log(msg):
print(msg, file=sys.stderr, flush=True)
class Notifer(object):
def __init__(self, app_name="Skipify", app_icon="", replaces_id=0, expire_timeout=5000):
self.app_name = app_name
self.expire_timeout = expire_timeout
self.replaces_id = replaces_id
self.__notifications = None
def notify(self, summary, body, app_icon="", actions=[], hints={}):
if self.__notifications is None:
self.__notifications = SessionBus().get('.Notifications')
self.replaces_id = self.__notifications.Notify(self.app_name, self.replaces_id, app_icon,
summary, body, actions, hints, self.expire_timeout)
class ContextualEvent(threading.Event):
def __enter__(self):
self.wait()
def __exit__(self, *args):
self.clear()
class Spotify(object):
class Track(object):
def __init__(self, metadata, start_time, last_track):
self.__metadata = metadata
self.start_time = start_time
self.is_first_song_after_ad = last_track.is_ad and not self.is_ad
@property
def id(self):
return self.__metadata.get('mpris:trackid')
@property
def title(self):
return self.__metadata.get('xesam:title')
@property
def is_ad(self):
return self.id and self.id.startswith("spotify:ad:")
def __eq__(self, other):
return isinstance(other, self.__class__) and self.id == other.id
class NoneTrack(Track):
def __init__(self):
super().__init__({}, None, self)
DBUS_IFACE = 'org.mpris.MediaPlayer2.spotify'
DBUS_PATH = '/org/mpris/MediaPlayer2'
def __init__(self):
self.__bus = SessionBus()
self.__bus.subscribe(
sender=self.DBUS_IFACE,
signal="PropertiesChanged",
signal_fired=self.__on_properties_changed,
)
self.__loop = GLib.MainLoop()
self.__player = None
self.__current_track = Spotify.NoneTrack()
self.__last_track = Spotify.NoneTrack()
self.track_changed = ContextualEvent()
def __update_track(self, metadata=None, start_time=None):
if metadata is None:
metadata = self.__player.Metadata
if start_time is None:
start_time = time.monotonic()
track = Spotify.Track(metadata, start_time, self.__current_track)
if track != self.__current_track:
self.__last_track, self.__current_track = self.__current_track, track
self.track_changed.set()
def __on_properties_changed(self, sender, obj, iface, signal, params):
start_time = time.monotonic()
interface, changed, invalidated = params
if changed:
log(f"Properties changed: {changed}")
if invalidated:
log(f"Properties invalidated: {invalidated}")
if self.__player is None:
self.__player = self.__bus.get(self.DBUS_IFACE, self.DBUS_PATH)
self.__update_track(changed['Metadata'], start_time)
@property
def last_track(self):
self.__update_track()
return self.__last_track
@property
def current_track(self):
self.__update_track()
return self.__current_track
def __getattr__(self, attr):
return getattr(self.__player, attr.title().replace('_', ''))
def __set_mute(self, mute):
with pulsectl.Pulse('Skipify') as pulse:
spotify_inputs = [sink for sink in pulse.sink_input_list() if sink.name == "Spotify"]
for spotify in spotify_inputs:
pulse.mute(spotify, mute)
def connect(self):
return self.__loop.run()
mute = property(None, __set_mute)
@property
def is_playing(self):
return self.__player.PlaybackStatus == "Playing"
class AdSkipper(threading.Thread):
def __init__(self, spotify, min_skip_time=timedelta(milliseconds=900), sleep_time=1):
super().__init__(daemon=True)
self.spotify = spotify
self.timer = None
self.min_skip_time = min_skip_time
self.sleep_time = sleep_time
self.notifier = Notifer()
def handle_track_change(self):
spotify = self.spotify
log(f"Track changed: {spotify.current_track.title}")
try:
self.timer.cancel()
except AttributeError:
pass
else:
self.timer = None
if spotify.current_track.is_ad:
spotify.mute = True
spotify.next()
log(f"Skipping ad '{spotify.current_track.title}'; will check again in {self.sleep_time}s")
self.notifier.notify("Skipping ad", f"Skipping ad '{spotify.current_track.title}'")
self.timer = threading.Timer(self.sleep_time, spotify.track_changed.set)
self.timer.start()
else:
if spotify.last_track.is_ad:
log(f"Ad over! Now playing '{spotify.current_track.title}'")
self.notifier.notify("Skipped over ad", f"Now playing '{spotify.current_track.title}'")
elif spotify.last_track.is_first_song_after_ad:
log(f"First song after ad was '{spotify.last_track.title}'")
diff_start_time_in_seconds = spotify.current_track.start_time - spotify.last_track.start_time
diff_start_time = timedelta(seconds=diff_start_time_in_seconds)
log(f"Last song was started {diff_start_time} ago")
if diff_start_time <= self.min_skip_time:
# We've skipped over a real song so let's go back
log(f"Oops! We skipped over '{spotify.last_track.title}'; going back")
self.notifier.notify("Skipped over ad", f"Now playing '{spotify.last_track.title}'")
spotify.previous()
spotify.mute = False
def run(self):
while True:
with self.spotify.track_changed:
self.handle_track_change()
if __name__ == '__main__':
spotify = Spotify()
skipper = AdSkipper(spotify)
skipper.start()
try:
spotify.connect()
except KeyboardInterrupt:
pass
[Unit]
Description=skipify spotify ad suppressor
Requires=dbus.service
After=dbus.service graphical-session.service
[Service]
Type=simple
ExecStart=/home/johnnyg/.virtualenvs/skipify/bin/python /home/johnnyg/skipify/skipify.py
[Install]
WantedBy=default.target
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment