Spoupsify - Mute Spotify ads automatically

spoupsify.py

#!/usr/bin/env python3
#
# Spoupsify - Mute Spotify ads automatically
# WTFPL - github @dece - twitter @postdroned
#
# Run this script to automatically mute the Linux Spotify desktop application
# when it plays an audio ad. If your Spotify is not in English, you will need to
# add to the TITLES_TO_MUTE list the Spotify window title shown during ads.
# 
# Please note that this script needs the Spotify window to be kept open or
# reduced to the task bar, but will not work if reduced to the systray.
#
# Requirements (along with their Debian package):
# - Python ^3.4
# - xprop (x11-utils)
# - xdotool (xdotool)
# - pactl (pulseaudio-utils)

import re
import subprocess
import time


CLASSNAME = "spotify"  # Class name used in the X windows properties.
BINARY_NAME = "spotify"  # Used to identify Pulseaudio sink input exe source.
TITLES_TO_MUTE = ["Spotify", "Advertisement"]  # Window titles to mute.

SEARCH_ID_CMD = ['xdotool', 'search', '-classname']
PROPERTIES_CMD = ['xprop', '-id']
WMNAME_RE = re.compile(r'WM_NAME\(STRING\) = \"(.*)\"')
SEARCH_PACLIENT_CMD = ['pactl', 'list', 'sink-inputs']
SINK_INPUT_RE = re.compile(r'Sink Input #(\d+)')
SINK_INPUT_NAME_RE = re.compile(r'\s+application.process.binary = \"(.*)\"')
MUTE_CMD = ['pactl', 'set-sink-input-mute', None, '1']
UNMUTE_CMD = ['pactl', 'set-sink-input-mute', None, '0']


class Spoupsify(object):

    def __init__(self):
        self.window_id = 0
        self.previous_title = None
        self.current_title = None
        self.sink_ids = []
        self.muted = False
    
    def run(self):
        self.find_window_id(CLASSNAME)
        if self.window_id == 0:
            return

        while True:
            self.find_window_title()
            if self.current_title in TITLES_TO_MUTE and not self.muted:
                print('Muting', self.current_title)
                self.find_pulse_sink_input(BINARY_NAME)
                self.mute()
            elif self.current_title != self.previous_title and self.muted:
                print('Unmuting after a slight delay...')
                time.sleep(1)
                self.find_pulse_sink_input(BINARY_NAME)
                self.unmute()
            time.sleep(0.1)

    def find_window_id(self, classname):
        """ Store the highest ID found in window_id, or 0 on error. """
        result = run_command(SEARCH_ID_CMD + [classname])
        if result is None:
            self.window_id = 0
            return

        ids = [int(wid.strip()) for wid in result.split()]
        self.window_id = max(ids)

    def find_window_title(self):
        """ Refresh the window current and previous title. """
        command = PROPERTIES_CMD + [str(self.window_id)]
        result = run_command(command)
        if result is None:
            time.sleep(5)
            return

        for line in result.splitlines():
            if not line.startswith('WM_NAME'):
                continue
            match = WMNAME_RE.match(line)
            if match:
                self.previous_title = self.current_title
                self.current_title = match.group(1)
                return

    def find_pulse_sink_input(self, app_name):
        """ Set the current Pulseaudio sink input ID. """
        result = run_command(SEARCH_PACLIENT_CMD)
        if result is None:
            self.sink_ids = []
            return

        sink_ids = []
        for line in result.splitlines():
            match = SINK_INPUT_RE.match(line)
            if match:
                sink_id = int(match.group(1))
                continue
            match = SINK_INPUT_NAME_RE.match(line)
            if match and match.group(1) == app_name:
                sink_ids.append(sink_id)
        print("Found sink input IDs", ",".join([str(sink) for sink in sink_ids]))
        self.sink_ids = sink_ids

    def mute(self):
        for sink in self.sink_ids:
            MUTE_CMD[2] = str(sink)
            subprocess.run(MUTE_CMD)
        self.muted = True

    def unmute(self):
        for sink in self.sink_ids:
            UNMUTE_CMD[2] = str(sink)
            subprocess.run(UNMUTE_CMD)
        self.muted = False


def run_command(command):
    try:
        return subprocess.check_output(command).decode()
    except subprocess.CalledProcessError as exc:
        print("Command failed:", str(exc))


def main():
    while True:
        try:
            Spoupsify().run()
        except KeyboardInterrupt:
            print("Interrupted, exiting.")
            return
        time.sleep(5)


if __name__ == '__main__':
    main()