funkwlmpv/src/fw_instances.py

from src.fw_api import current_instance
import src.settings as settings
from pyfzf.pyfzf import FzfPrompt
from shlex import quote
from loguru import logger
import json
import time
import concurrent
import requests

fzf = FzfPrompt()


@logger.catch
def get_new_funkwhale_servers():
    # Uses official API network.funkwhale.audio for getting new instances
    public_server_api = 'https://network.funkwhale.audio/dashboards/api/tsdb/query'
    now = int(time.time())
    timeback = now - 86400

    request_public_servers = {
        'from': f"{timeback}",
        'to': f"{now}",
        'queries': [
            {
                'refId': "A",
                'intervalMs': 60000,
                'maxDataPoints': 1174,
                'datasourceId': 1,
                'rawSql': "SELECT * FROM (\n  SELECT\n    DISTINCT on (c.domain) c.domain as \"Name\",\n    c.up as \"Is up\",\n    coalesce(c.open_registrations, false) as \"Open registrations\",\n    coalesce(anonymous_can_listen, false) as \"Anonymous can listen\",\n    coalesce(c.usage_users_total, 0) as \"Total users\",\n    coalesce(c.usage_users_active_month, 0) as \"Active users (this month)\",\n    coalesce(c.software_version_major, 0)::text || '.' || coalesce(c.software_version_minor, 0)::text || '.' || coalesce(c.software_version_patch, 0)::text as \"Version\",\n    c.time as \"Last checked\",\n    d.first_seen as \"First seen\"\n  FROM checks as c\n  INNER JOIN domains AS d ON d.name = c.domain\n  WHERE d.blocked = false AND c.up = true AND c.time > now() - INTERVAL '7 days'\n      AND c.anonymous_can_listen IN ('true')\n      AND c.open_registrations IN ('true','false')\n\n  ORDER BY c.domain, c.time DESC\n) as t ORDER BY \"Active users (this month)\"  DESC",
                'format': "table"
            }
        ]
    }
    try:
        r = requests.post(public_server_api, json=request_public_servers)
        results = r.json()
        new_instances = {}
        if results:
            new_instances_list = results['results']['A']['tables'][0]['rows']
            for i in new_instances_list:
                anonymousCanListen = i[1]
                if anonymousCanListen:
                    new_instances[i[0]] = f'{anonymousCanListen} | ?'

        for i in get_new_funkwhale_servers_fediverse_observer():
            new_instances[i] = "?"
        return new_instances
    except: # If any errors then return empty list
        return {}


def get_new_funkwhale_servers_fediverse_observer():
    try:
        graphQL_request = {
                'query':
                '{\n  nodes(softwarename: \"funkwhale\") {\n    domain\n    metanodeinfo\n  }\n}'
                }
        r = requests.post('https://api.fediverse.observer/',
                          headers={'Accept-Encoding': 'gzip, deflate'},
                          json=graphQL_request)
        new_instances = []
        for i in r.json()['data']['nodes']:
            if i.get('metanodeinfo'):
                auth_no_required = json.loads(i['metanodeinfo'])['library']['anonymousCanListen']
                if auth_no_required and i['domain']:
                    new_instances.append(i['domain'])
        return new_instances
    except:
        return []


def fetch_instances_nodeinfo_and_avalaibility(instances):
    extended_instances_info = {}

    def request_nodeinfo(instance):
        return requests.get('https://' + instance + '/api/v1/instance/nodeinfo/2.0/',
                            headers={
                                'Accept-Encoding': 'gzip, brotli, deflate',
                                'User-Agent': 'funkwlmpv/latest-commit; +https://git.phreedom.club/localhost_frssoft/funkwlmpv'},
                            timeout=10).json()

    with concurrent.futures.ThreadPoolExecutor() as executor: # optimally defined number of threads
        res = [executor.submit(request_nodeinfo, instance) for instance in instances]
        concurrent.futures.wait(res)
    for idx, v in enumerate(instances):
        try:
            data_for_instance = res[idx].result()
            anon = data_for_instance['metadata']['library']['anonymousCanListen']
            tracks = data_for_instance['metadata']['library']['tracks']['total']
            extended_instances_info[v] = f'{anon} | {tracks}'
        except:
            extended_instances_info[v] = 'fail'
    return extended_instances_info


def instances_menu(fetch_manually=False, fetch_node_info=False):
    with open('config.json', 'rt') as f:
        conf = json.loads(f.read())
        if conf.get('automatic_fetch_new_instances') or fetch_manually:
            public_server_list_instances = get_new_funkwhale_servers()
            new_ins_count = len(public_server_list_instances)
        else:
            public_server_list_instances = {}
            new_ins_count = 'Disabled'

    list_instances = conf.get('public_list_instances_extended')
    if public_server_list_instances != {}:
        list_instances_merge = {**list_instances, **public_server_list_instances}
        settings.set_config('public_list_instances_extended', list_instances_merge)
        list_instances = list_instances_merge

    map_in_extend_mode = ''
    if fetch_node_info:
        list_instances = fetch_instances_nodeinfo_and_avalaibility([instance.split('|')[0].strip() for instance in list_instances.keys()])
        settings.set_config('public_list_instances_extended', list_instances)
        map_in_extend_mode = '\nmap: instance | anonymousCanListen | tracks'
    instance_menu_selector = ['Add new instance',
                              'Fetch new instances',
                              'Fetch nodeinfo and avalaibility',
                              'Remove unreachible instances',
                              'Shuffle']

    instance = fzf.prompt(
            instance_menu_selector +
            [f'{instance} | {info}' for instance, info in list_instances.items()],
            '--header='+quote(f'Select instance\nNew instances: {new_ins_count}{map_in_extend_mode}'))
    if instance == []:
        return
    else:
        instance = instance[0].split('|')[0].strip()
    if instance == 'Add new instance':
        new = input('example.com\n').strip()
        list_instances[new] = 'added by user'
        settings.set_config('public_list_instances_extended', list_instances)
        instance = new
    if instance == 'Fetch new instances':
        return instances_menu(fetch_manually=True)
    if instance == 'Fetch nodeinfo and avalaibility':
        return instances_menu(fetch_node_info=True)
    if instance == 'Shuffle':
        import random
        instance = random.choice(list(list_instances.keys()))
    if instance == 'Remove unreachible instances':
        clean_unreach = {}
        for ins, info in list_instances.items():
            if 'fail' not in info.split():
                clean_unreach[ins] = info
        settings.set_config('public_list_instances_extended', clean_unreach)
        return instances_menu()
    current_instance.select_instance(instance)