Script to skip/mute Spotify ads

Pipfile

[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[dev-packages]

[packages]
pydbus = "*"
pulsectl = "*"

[requires]
python_version = "3"

Pipfile.lock

{
    "_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": {}
}

skipify.py

#!/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

skipify.service

[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