funkwlmpv/funkwlplay.py

import requests
import concurrent.futures
import argparse
from pathlib import Path

tracks_stor = []
with open('instances') as instances:
    instances = instances.read().strip().split('\n')


parser = argparse.ArgumentParser(
                    prog='funkwhale playlist',
                    description='Create playlist from query or just random playlist tracks from funkwhale instances')
parser.add_argument('-s', '--search')
parser.add_argument('-t', '--tag')
parser.add_argument('-i', '--instance')
args = parser.parse_args()
if args.instance:
    instances = [args.instance]


def create_playlist_file(track_list):
    with open('playlist.m3u8', 'w') as file:
        file.write('#EXTM3U\n')
        for i in track_list:
            file.write('\n' + i)


def filter_tracks(tracks):
    Path('filter_tags').touch()
    Path('filter_artists').touch()
    Path('filter_raw_urls').touch()
    with open('filter_tags') as tags_file:
        block_tags = tags_file.read().strip().split('\n')

    with open('filter_artists') as artists_file:
        block_artists = artists_file.read().strip().split('\n')

    with open('filter_raw_urls') as raw_urls_file:
        block_raw_urls = raw_urls_file.read().strip().split('\n')

    for i in tracks:
        if [tag.lower() for tag in i['tags']] in block_tags:
            continue
        if i['artist']['name'].lower() in block_artists:
            continue
        if i['listen_url'].lower() in block_raw_urls:
            continue
        tracks_stor.append(i)


def search_tracks_on_instance(instance, tag='', query=''):
    tracks = requests.get(f'https://{instance}/api/v1/tracks', params={'tag': tag, 'q': query,
                                                                       'local': True, 'playable': True,
                                                                       'ordering': 'random'}, timeout=10).json()
    tracks_replacer = []
    for track in tracks['results']:
        track['listen_url'] = f'https://{instance}' + track['listen_url']
        tracks_replacer.append(track)
    tracks['results'] = tracks_replacer
    return tracks


with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
    res = [executor.submit(search_tracks_on_instance, instance, args.tag, args.search) for instance in instances]
    concurrent.futures.wait(res)
    playlist_files = []
    for idx, instance in enumerate(instances):
        try:
            tracks = res[idx].result()
            filter_tracks(tracks['results'])
        except Exception as E:
            print(E)
    for track in tracks_stor:
        artist, album, title, play_url = track['artist']['name'], track['album']['title'], track['title'], track['listen_url']
        playlist_files.append(f'#EXTINF:-1,{artist} - {album} - {title}\n{play_url}')
    create_playlist_file(playlist_files)