Modified vsvinav Spotify Ads Blocker for Linux

adspotter

#!/usr/bin/env python3
__version__ = '0.0.1'
__author__ = 'vsvinav'

""" 
depedency
sudo apt-get install libdbus-1-dev libdbus-glib-1-dev

requirements.txt
dbus-python==1.2.8
mpris2==1.0.2
colorama==0.4.1
"""

try:
    import os
    import sys
    import dbus
    import time
    import mpris2
    import signal
    import getpass
    import subprocess
    from colorama import Fore, Style
except:
    print('\033[1;31m[ERROR]: \033[1;m' + 'missing requirements')
    exit()

user = getpass.getuser()
red, green, reset, bold = Fore.RED, Fore.GREEN, Style.RESET_ALL, "\033[1;31m"
error = '\r' + red + bold + "[ERROR]:" + reset + ' '
status = bold + green + "[STATUS]:" + reset + ' '
open_spotify = "setsid spotify >/dev/null"
close_spotify = "ps -ef | grep \'spotify\' | grep -v grep | awk \'{print $2}\' | xargs -r kill -9"


def type(text, speed, next):
    for char in text + next:
        sys.stdout.write(char)
        sys.stdout.flush()
        time.sleep(1. / speed)


def keynterrupt():
    type(error + 'interrupted by ' + user, 50, '\n\n')
    exit()

def is_minimized(command_check):
    try:
        w_list = [l.split() for l in subprocess.check_output(["wmctrl", "-lp"]).decode("utf-8").splitlines()]
        proc = subprocess.check_output(["pgrep", "-f", command_check]).decode("utf-8").strip().split()
        match = sum([[l[0] for l in w_list if p in l] for p in proc], [])
        window_info = subprocess.check_output(["xwininfo", "-all", "-id", match[0]]).decode("utf-8").strip().split()
        if "Hidden" in window_info:
            return True
        else:
            return False
    except (IndexError, subprocess.CalledProcessError):
        return False

def exec_minimized(command):
    command_check = 'spotify'
    subprocess.Popen(["/bin/bash", "-c", command])

    t = 1
    while t < 30:
        try:
            w_list = [l.split() for l in subprocess.check_output(["wmctrl", "-lp"]).decode("utf-8").splitlines()]
            proc = subprocess.check_output(["pgrep", "-f", command_check]).decode("utf-8").strip().split()
            match = sum([[l[0] for l in w_list if p in l] for p in proc], [])
            subprocess.Popen(["xdotool", "windowminimize", match[0]])
            break
        except (IndexError, subprocess.CalledProcessError):
            pass
        t += 1
        time.sleep(1)
    
def main():
    os.system('clear')
    type(bold + green + r'''
   _        _   ___                _     _
  /_\    __| | / __|  _ __   ___  | |_  | |_   ___   _ _
 / _ \  / _` | \__ \ | '_ \ / _ \ |  _| |  _| / -_) | '_|
/_/ \_\ \__,_| |___/ | .__/ \___/  \__|  \__| \___| |_|
                     |_|'''[1:] + "v" + __version__, 300,"\n\n")
    
    open_spotify = "setsid spotify &>/dev/null"
    try:
        uri = dbus.String('org.mpris.MediaPlayer2.spotify')
        player = mpris2.Player(dbus_interface_info={'dbus_uri': uri})
    except:
        type(error + "spotify isn't running, lets open it", 50, '\n\n')
        os.system(open_spotify)
        time.sleep(2)
    type(status + 'waiting for ads...', 10, '\n')
    uri = dbus.String('org.mpris.MediaPlayer2.spotify')
    player = mpris2.Player(dbus_interface_info={'dbus_uri': uri})
    kill_spotify = "ps -ef | grep \'spotify\' | grep -v grep | awk \'{print $2}\' | xargs -r kill -9"
    while True:
        title = str(dict(player.Metadata).get(dbus.String('xesam:title')))
        signal.signal(signal.SIGINT, lambda x, y: keynterrupt())
        try:
            if title in ['Advertisement', 'Spotify', 'spotify', 'Ad']:
                print(status + 'Detected Ad')
                window_minimized = is_minimized("spotify")
                os.system(kill_spotify)

                print(status + 'Blocking Ad')
                if window_minimized:
                    exec_minimized(open_spotify)
                else:
                    os.system(open_spotify)
                player.Next()
        except:
            time.sleep(3)
            uri = dbus.String('org.mpris.MediaPlayer2.spotify')
            player = mpris2.Player(dbus_interface_info={'dbus_uri': uri})
            player.Next()
    time.sleep(2)

if __name__ == '__main__':
    main()