helpers.py

# Copyright (C) 2003-2014 Yann Leboulanger <asterix AT lagaule.org>
# Copyright (C) 2005-2006 Dimitur Kirov <dkirov AT gmail.com>
#                         Nikos Kouremenos <kourem AT gmail.com>
# Copyright (C) 2006 Alex Mauer <hawke AT hawkesnest.net>
# Copyright (C) 2006-2007 Travis Shirk <travis AT pobox.com>
# Copyright (C) 2006-2008 Jean-Marie Traissard <jim AT lapin.org>
# Copyright (C) 2007 Lukas Petrovicky <lukas AT petrovicky.net>
#                    James Newton <redshodan AT gmail.com>
#                    Julien Pivotto <roidelapluie AT gmail.com>
# Copyright (C) 2007-2008 Stephan Erb <steve-e AT h3c.de>
# Copyright (C) 2008 Brendan Taylor <whateley AT gmail.com>
#                    Jonathan Schleifer <js-gajim AT webkeks.org>
#
# This file is part of Gajim.
#
# Gajim is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; version 3 only.
#
# Gajim is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.

from typing import Any  # pylint: disable=unused-import
from typing import Dict  # pylint: disable=unused-import

import sys
import re
import os
import subprocess
import base64
import hashlib
import shlex
import socket
import logging
import json
import copy
import collections
import platform
import functools
from collections import defaultdict
import random
import weakref
import inspect
import string
import webbrowser
from string import Template
import urllib
from urllib.parse import unquote
from encodings.punycode import punycode_encode
from functools import wraps
from pathlib import Path
from packaging.version import Version as V

from nbxmpp.namespaces import Namespace
from nbxmpp.const import Role
from nbxmpp.const import ConnectionProtocol
from nbxmpp.const import ConnectionType
from nbxmpp.structs import ProxyData
from nbxmpp.protocol import JID
from nbxmpp.protocol import InvalidJid
from OpenSSL.crypto import load_certificate
from OpenSSL.crypto import FILETYPE_PEM
from gi.repository import Gio
from gi.repository import GLib
import precis_i18n.codec  # pylint: disable=unused-import

from gajim.common import app
from gajim.common import configpaths
from gajim.common.i18n import Q_
from gajim.common.i18n import _
from gajim.common.i18n import ngettext
from gajim.common.i18n import get_rfc5646_lang
from gajim.common.const import ShowConstant
from gajim.common.const import Display
from gajim.common.const import URIType
from gajim.common.const import URIAction
from gajim.common.const import GIO_TLS_ERRORS
from gajim.common.const import SHOW_LIST
from gajim.common.regex import INVALID_XML_CHARS_REGEX
from gajim.common.regex import STH_AT_STH_DOT_STH_REGEX
from gajim.common.structs import URI


log = logging.getLogger('gajim.c.helpers')

special_groups = (_('Transports'),
                  _('Not in contact list'),
                  _('Observers'),
                  _('Group chats'))

URL_REGEX = re.compile(
    r"(www\.(?!\.)|[a-z][a-z0-9+.-]*://)[^\s<>'\"]+[^!,\.\s<>\)'\"\]]")


class InvalidFormat(Exception):
    pass


def parse_jid(jidstring):
    try:
        return str(validate_jid(jidstring))
    except Exception as error:
        raise InvalidFormat(error)

def idn_to_ascii(host):
    """
    Convert IDN (Internationalized Domain Names) to ACE (ASCII-compatible
    encoding)
    """
    from encodings import idna
    labels = idna.dots.split(host)
    converted_labels = []
    for label in labels:
        if label:
            converted_labels.append(idna.ToASCII(label).decode('utf-8'))
        else:
            converted_labels.append('')
    return ".".join(converted_labels)

def ascii_to_idn(host):
    """
    Convert ACE (ASCII-compatible encoding) to IDN (Internationalized Domain
    Names)
    """
    from encodings import idna
    labels = idna.dots.split(host)
    converted_labels = []
    for label in labels:
        converted_labels.append(idna.ToUnicode(label))
    return ".".join(converted_labels)

def puny_encode_url(url):
    _url = url
    if '//' not in _url:
        _url = '//' + _url
    try:
        o = urllib.parse.urlparse(_url)
        p_loc = idn_to_ascii(o.hostname)
    except Exception:
        log.debug('urlparse failed: %s', url)
        return False
    return url.replace(o.hostname, p_loc)

def parse_resource(resource):
    """
    Perform stringprep on resource and return it
    """
    if not resource:
        return None

    try:
        return resource.encode('OpaqueString').decode('utf-8')
    except UnicodeError:
        raise InvalidFormat('Invalid character in resource.')

def windowsify(word):
    if os.name == 'nt':
        return word.capitalize()
    return word

def get_uf_show(show, use_mnemonic=False):
    """
    Return a userfriendly string for dnd/xa/chat and make all strings
    translatable

    If use_mnemonic is True, it adds _ so GUI should call with True for
    accessibility issues
    """
    if isinstance(show, ShowConstant):
        show = show.name.lower()

    if show == 'dnd':
        if use_mnemonic:
            uf_show = _('_Busy')
        else:
            uf_show = _('Busy')
    elif show == 'xa':
        if use_mnemonic:
            uf_show = _('_Not Available')
        else:
            uf_show = _('Not Available')
    elif show == 'chat':
        if use_mnemonic:
            uf_show = _('_Free for Chat')
        else:
            uf_show = _('Free for Chat')
    elif show == 'online':
        if use_mnemonic:
            uf_show = Q_('?user status:_Available')
        else:
            uf_show = Q_('?user status:Available')
    elif show == 'connecting':
        uf_show = _('Connecting')
    elif show == 'away':
        if use_mnemonic:
            uf_show = _('A_way')
        else:
            uf_show = _('Away')
    elif show == 'offline':
        if use_mnemonic:
            uf_show = _('_Offline')
        else:
            uf_show = _('Offline')
    elif show == 'not in roster':
        uf_show = _('Not in contact list')
    elif show == 'requested':
        uf_show = Q_('?contact has status:Unknown')
    else:
        uf_show = Q_('?contact has status:Has errors')
    return uf_show

def get_uf_sub(sub):
    if sub == 'none':
        uf_sub = Q_('?Subscription we already have:None')
    elif sub == 'to':
        uf_sub = _('To')
    elif sub == 'from':
        uf_sub = _('From')
    elif sub == 'both':
        uf_sub = _('Both')
    else:
        uf_sub = _('Unknown')

    return uf_sub

def get_uf_ask(ask):
    if ask is None:
        uf_ask = Q_('?Ask (for Subscription):None')
    elif ask == 'subscribe':
        uf_ask = _('Subscribe')
    else:
        uf_ask = ask

    return uf_ask

def get_uf_role(role, plural=False):
    ''' plural determines if you get Moderators or Moderator'''
    if not isinstance(role, str):
        role = role.value

    if role == 'none':
        role_name = Q_('?Group Chat Contact Role:None')
    elif role == 'moderator':
        if plural:
            role_name = _('Moderators')
        else:
            role_name = _('Moderator')
    elif role == 'participant':
        if plural:
            role_name = _('Participants')
        else:
            role_name = _('Participant')
    elif role == 'visitor':
        if plural:
            role_name = _('Visitors')
        else:
            role_name = _('Visitor')
    return role_name

def get_uf_affiliation(affiliation, plural=False):
    '''Get a nice and translated affilition for muc'''
    if not isinstance(affiliation, str):
        affiliation = affiliation.value

    if affiliation == 'none':
        affiliation_name = Q_('?Group Chat Contact Affiliation:None')
    elif affiliation == 'owner':
        if plural:
            affiliation_name = _('Owners')
        else:
            affiliation_name = _('Owner')
    elif affiliation == 'admin':
        if plural:
            affiliation_name = _('Administrators')
        else:
            affiliation_name = _('Administrator')
    elif affiliation == 'member':
        if plural:
            affiliation_name = _('Members')
        else:
            affiliation_name = _('Member')
    return affiliation_name

def get_sorted_keys(adict):
    keys = sorted(adict.keys())
    return keys

def to_one_line(msg):
    msg = msg.replace('\\', '\\\\')
    msg = msg.replace('\n', '\\n')
    # s1 = 'test\ntest\\ntest'
    # s11 = s1.replace('\\', '\\\\')
    # s12 = s11.replace('\n', '\\n')
    # s12
    # 'test\\ntest\\\\ntest'
    return msg

def from_one_line(msg):
    # (?<!\\) is a lookbehind assertion which asks anything but '\'
    # to match the regexp that follows it

    # So here match '\\n' but not if you have a '\' before that
    expr = re.compile(r'(?<!\\)\\n')
    msg = expr.sub('\n', msg)
    msg = msg.replace('\\\\', '\\')
    # s12 = 'test\\ntest\\\\ntest'
    # s13 = re.sub('\n', s12)
    # s14 s13.replace('\\\\', '\\')
    # s14
    # 'test\ntest\\ntest'
    return msg

def get_uf_chatstate(chatstate):
    """
    Remove chatstate jargon and returns user friendly messages
    """
    if chatstate == 'active':
        return _('is paying attention to the conversation')
    if chatstate == 'inactive':
        return _('is doing something else')
    if chatstate == 'composing':
        return _('is composing a message…')
    if chatstate == 'paused':
        #paused means he or she was composing but has stopped for a while
        return _('paused composing a message')
    if chatstate == 'gone':
        return _('has closed the chat window or tab')
    return ''

def exec_command(command, use_shell=False, posix=True):
    """
    execute a command. if use_shell is True, we run the command as is it was
    typed in a console. So it may be dangerous if you are not sure about what
    is executed.
    """
    if use_shell:
        subprocess.Popen('%s &' % command, shell=True).wait()
    else:
        args = shlex.split(command, posix=posix)
        process = subprocess.Popen(args)
        app.thread_interface(process.wait)

def build_command(executable, parameter):
    # we add to the parameter (can hold path with spaces)
    # "" so we have good parsing from shell
    parameter = parameter.replace('"', '\\"') # but first escape "
    command = '%s "%s"' % (executable, parameter)
    return command

def get_file_path_from_dnd_dropped_uri(uri):
    path = urllib.parse.unquote(uri) # escape special chars
    path = path.strip('\r\n\x00') # remove \r\n and NULL
    # get the path to file
    if re.match('^file:///[a-zA-Z]:/', path): # windows
        path = path[8:] # 8 is len('file:///')
    elif path.startswith('file://'): # nautilus, rox
        path = path[7:] # 7 is len('file://')
    elif path.startswith('file:'): # xffm
        path = path[5:] # 5 is len('file:')
    return path

def sanitize_filename(filename):
    """
    Make sure the filename we will write does contain only acceptable and latin
    characters, and is not too long (in that case hash it)
    """
    # 48 is the limit
    if len(filename) > 48:
        hash_ = hashlib.md5(filename.encode('utf-8'))
        filename = base64.b64encode(hash_.digest()).decode('utf-8')

    # make it latin chars only
    filename = punycode_encode(filename).decode('utf-8')
    filename = filename.replace('/', '_')
    if os.name == 'nt':
        filename = filename.replace('?', '_').replace(':', '_')\
                .replace('\\', '_').replace('"', "'").replace('|', '_')\
                .replace('*', '_').replace('<', '_').replace('>', '_')

    return filename

def reduce_chars_newlines(text, max_chars=0, max_lines=0):
    """
    Cut the chars after 'max_chars' on each line and show only the first
    'max_lines'

    If any of the params is not present (None or 0) the action on it is not
    performed
    """
    def _cut_if_long(string_):
        if len(string_) > max_chars:
            string_ = string_[:max_chars - 3] + '…'
        return string_

    if max_lines == 0:
        lines = text.split('\n')
    else:
        lines = text.split('\n', max_lines)[:max_lines]
    if max_chars > 0:
        if lines:
            lines = [_cut_if_long(e) for e in lines]
    if lines:
        reduced_text = '\n'.join(lines)
        if reduced_text != text:
            reduced_text += '…'
    else:
        reduced_text = ''
    return reduced_text

def get_account_status(account):
    status = reduce_chars_newlines(account['status_line'], 100, 1)
    return status

def get_contact_dict_for_account(account):
    """
    Create a dict of jid, nick -> contact with all contacts of account.

    Can be used for completion lists
    """
    contacts_dict = {}
    for jid in app.contacts.get_jid_list(account):
        contact = app.contacts.get_contact_with_highest_priority(account, jid)
        contacts_dict[jid] = contact
        name = contact.name
        if name in contacts_dict:
            contact1 = contacts_dict[name]
            del contacts_dict[name]
            contacts_dict['%s (%s)' % (name, contact1.jid)] = contact1
            contacts_dict['%s (%s)' % (name, jid)] = contact
        elif contact.name:
            if contact.name == app.get_nick_from_jid(jid):
                del contacts_dict[jid]
            contacts_dict[name] = contact
    return contacts_dict

def play_sound(event):
    if not app.settings.get('sounds_on'):
        return
    play_sound_file(app.settings.get_soundevent_settings(event)['path'])

def check_soundfile_path(file_, dirs=None):
    """
    Check if the sound file exists

    :param file_: the file to check, absolute or relative to 'dirs' path
    :param dirs: list of knows paths to fallback if the file doesn't exists
                                     (eg: ~/.gajim/sounds/, DATADIR/sounds...).
    :return      the path to file or None if it doesn't exists.
    """
    if not file_:
        return None
    if Path(file_).exists():
        return Path(file_)

    if dirs is None:
        dirs = [configpaths.get('MY_DATA'),
                configpaths.get('DATA')]

    for dir_ in dirs:
        dir_ = dir_ / 'sounds' / file_
        if dir_.exists():
            return dir_
    return None

def strip_soundfile_path(file_, dirs=None, abs_=True):
    """
    Remove knowns paths from a sound file

    Filechooser returns an absolute path.
    If path is a known fallback path, we remove it.
    So config has no hardcoded path to DATA_DIR and text in textfield is
    shorther.
    param: file_: the filename to strip
    param: dirs: list of knowns paths from which the filename should be stripped
    param: abs_: force absolute path on dirs
    """

    if not file_:
        return None

    if dirs is None:
        dirs = [configpaths.get('MY_DATA'),
                configpaths.get('DATA')]

    file_ = Path(file_)
    name = file_.name
    for dir_ in dirs:
        dir_ = dir_ / 'sounds' / name
        if abs_:
            dir_ = dir_.absolute()
        if file_ == dir_:
            return name
    return file_

def play_sound_file(path_to_soundfile):
    path_to_soundfile = check_soundfile_path(path_to_soundfile)
    if path_to_soundfile is None:
        return

    path_to_soundfile = str(path_to_soundfile)
    if sys.platform == 'win32':
        import winsound
        try:
            winsound.PlaySound(path_to_soundfile,
                               winsound.SND_FILENAME|winsound.SND_ASYNC)
        except Exception:
            log.exception('Sound Playback Error')

    elif sys.platform == 'darwin':
        try:
            from AppKit import NSSound
        except ImportError:
            log.exception('Sound Playback Error')
            return

        sound = NSSound.alloc()
        sound.initWithContentsOfFile_byReference_(path_to_soundfile, True)
        sound.play()

    elif app.is_installed('GSOUND'):
        try:
            app.gsound_ctx.play_simple({'media.filename' : path_to_soundfile})
        except GLib.Error as error:
            log.error('Could not play sound: %s', error.message)

def get_connection_status(account):
    con = app.connections[account]
    if con.state.is_reconnect_scheduled:
        return 'error'

    if con.state.is_connecting or con.state.is_connected:
        return 'connecting'

    if con.state.is_disconnected:
        return 'offline'
    return con.status

def get_global_show():
    maxi = 0
    for account in app.connections:
        if not app.settings.get_account_setting(account,
                                                'sync_with_global_status'):
            continue
        status = get_connection_status(account)
        index = SHOW_LIST.index(status)
        if index > maxi:
            maxi = index
    return SHOW_LIST[maxi]

def get_global_status_message():
    maxi = 0
    for account, con in app.connections.items():
        if not app.settings.get_account_setting(account,
                                                'sync_with_global_status'):
            continue
        index = SHOW_LIST.index(con.status)
        if index > maxi:
            maxi = index
            status_message = con.status_message
    return status_message

def statuses_unified():
    """
    Test if all statuses are the same
    """
    reference = None
    for account, con in app.connections.items():
        if not app.settings.get_account_setting(account,
                                                'sync_with_global_status'):
            continue
        if reference is None:
            reference = con.status
        elif reference != con.status:
            return False
    return True

def get_icon_name_to_show(contact, account=None):
    """
    Get the icon name to show in online, away, requested, etc
    """
    if account and app.events.get_nb_roster_events(account, contact.jid):
        return 'event'
    if account and app.events.get_nb_roster_events(account,
                                                   contact.get_full_jid()):
        return 'event'
    if account and account in app.interface.minimized_controls and \
    contact.jid in app.interface.minimized_controls[account] and app.interface.\
            minimized_controls[account][contact.jid].get_nb_unread_pm() > 0:
        return 'event'
    if account and contact.jid in app.gc_connected[account]:
        if app.gc_connected[account][contact.jid]:
            return 'muc-active'
        return 'muc-inactive'
    if contact.jid.find('@') <= 0: # if not '@' or '@' starts the jid ==> agent
        return contact.show
    if contact.sub in ('both', 'to'):
        return contact.show
    if contact.ask == 'subscribe':
        return 'requested'
    transport = app.get_transport_name_from_jid(contact.jid)
    if transport:
        return contact.show
    if contact.show in SHOW_LIST:
        return contact.show
    return 'notinroster'

def get_full_jid_from_iq(iq_obj):
    """
    Return the full jid (with resource) from an iq
    """
    jid = iq_obj.getFrom()
    if jid is None:
        return None
    return parse_jid(str(iq_obj.getFrom()))

def get_jid_from_iq(iq_obj):
    """
    Return the jid (without resource) from an iq
    """
    jid = get_full_jid_from_iq(iq_obj)
    return app.get_jid_without_resource(jid)

def get_auth_sha(sid, initiator, target):
    """
    Return sha of sid + initiator + target used for proxy auth
    """
    return hashlib.sha1(("%s%s%s" % (sid, initiator, target)).encode('utf-8')).\
        hexdigest()

def remove_invalid_xml_chars(string_):
    if string_:
        string_ = re.sub(INVALID_XML_CHARS_REGEX, '', string_)
    return string_

def get_random_string(count=16):
    """
    Create random string of count length

    WARNING: Don't use this for security purposes
    """
    allowed = string.ascii_uppercase + string.digits
    return ''.join(random.choice(allowed) for char in range(count))

@functools.lru_cache(maxsize=1)
def get_os_info():
    info = 'N/A'
    if sys.platform in ('win32', 'darwin'):
        info = f'{platform.system()} {platform.release()}'

    elif sys.platform == 'linux':
        try:
            import distro
            info = distro.name(pretty=True)
        except ImportError:
            info = platform.system()
    return info

def allow_showing_notification(account):
    if not app.settings.get('show_notifications'):
        return False
    if app.settings.get('autopopupaway'):
        return True
    if app.account_is_available(account):
        return True
    return False

def allow_popup_window(account):
    """
    Is it allowed to popup windows?
    """
    autopopup = app.settings.get('autopopup')
    autopopupaway = app.settings.get('autopopupaway')
    if autopopup and (autopopupaway or \
    app.connections[account].status in ('online', 'chat')):
        return True
    return False

def allow_sound_notification(account, sound_event):
    if (app.settings.get('sounddnd') or
            app.connections[account].status != 'dnd' and
            app.settings.get_soundevent_settings(sound_event)['enabled']):
        return True
    return False

def get_notification_icon_tooltip_dict():
    """
    Return a dict of the form {acct: {'show': show, 'message': message,
    'event_lines': [list of text lines to show in tooltip]}
    """
    # How many events before we show summarized, not per-user
    max_ungrouped_events = 10

    accounts = get_accounts_info()

    # Gather events. (With accounts, when there are more.)
    for account in accounts:
        account_name = account['name']
        account['event_lines'] = []
        # Gather events per-account
        pending_events = app.events.get_events(account=account_name)
        messages, non_messages = {}, {}
        total_messages, total_non_messages = 0, 0
        for jid in pending_events:
            for event in pending_events[jid]:
                if event.type_.count('file') > 0:
                    # This is a non-messagee event.
                    messages[jid] = non_messages.get(jid, 0) + 1
                    total_non_messages = total_non_messages + 1
                else:
                    # This is a message.
                    messages[jid] = messages.get(jid, 0) + 1
                    total_messages = total_messages + 1
        # Display unread messages numbers, if any
        if total_messages > 0:
            if total_messages > max_ungrouped_events:
                text = ngettext('%d message pending',
                                '%d messages pending',
                                total_messages,
                                total_messages,
                                total_messages)
                account['event_lines'].append(text)
            else:
                for jid, msg in messages.items():
                    text = ngettext('%d message pending',
                                    '%d messages pending',
                                    msg, msg, msg)
                    contact = app.contacts.get_first_contact_from_jid(
                        account['name'], jid)
                    text += ' '
                    if jid in app.gc_connected[account['name']]:
                        text += _('from group chat %s') % (jid)
                    elif contact:
                        name = contact.get_shown_name()
                        text += _('from user %s') % (name)
                    else:
                        text += _('from %s') % (jid)
                    account['event_lines'].append(text)

        # Display unseen events numbers, if any
        if total_non_messages > 0:
            if total_non_messages > max_ungrouped_events:
                text = ngettext('%d event pending',
                                '%d events pending',
                                total_non_messages,
                                total_non_messages,
                                total_non_messages)
                account['event_lines'].append(text)
            else:
                for jid, msg in non_messages.items():
                    text = ngettext('%d event pending',
                                    '%d events pending',
                                    msg, msg, msg)
                    text += ' ' + _('from user %s') % (jid)
                    account[account]['event_lines'].append(text)

    return accounts

def get_accounts_info():
    """
    Helper for notification icon tooltip
    """
    accounts = []
    accounts_list = sorted(app.contacts.get_accounts())
    for account in accounts_list:

        status = get_connection_status(account)
        message = app.connections[account].status_message
        single_line = get_uf_show(status)
        if message is None:
            message = ''
        else:
            message = message.strip()
        if message != '':
            single_line += ': ' + message
        account_label = app.get_account_label(account)
        accounts.append({'name': account,
                         'account_label': account_label,
                         'status_line': single_line,
                         'show': status,
                         'message': message})
    return accounts

def get_current_show(account):
    if account not in app.connections:
        return 'offline'
    return app.connections[account].status

def get_optional_features(account):
    features = []

    if app.settings.get_account_setting(account, 'request_user_data'):
        features.append(Namespace.MOOD + '+notify')
        features.append(Namespace.ACTIVITY + '+notify')
        features.append(Namespace.TUNE + '+notify')
        features.append(Namespace.LOCATION + '+notify')

    features.append(Namespace.NICK + '+notify')

    if app.connections[account].get_module('Bookmarks').nativ_bookmarks_used:
        features.append(Namespace.BOOKMARKS_1 + '+notify')
    elif app.connections[account].get_module('Bookmarks').pep_bookmarks_used:
        features.append(Namespace.BOOKMARKS + '+notify')
    if app.is_installed('AV'):
        features.append(Namespace.JINGLE_RTP)
        features.append(Namespace.JINGLE_RTP_AUDIO)
        features.append(Namespace.JINGLE_RTP_VIDEO)
        features.append(Namespace.JINGLE_ICE_UDP)

    # Give plugins the possibility to add their features
    app.plugin_manager.extension_point('update_caps', account, features)
    return features

def jid_is_blocked(account, jid):
    con = app.connections[account]
    return jid in con.get_module('Blocking').blocked

def get_subscription_request_msg(account=None):
    s = app.settings.get_account_setting(account, 'subscription_request_msg')
    if s:
        return s
    s = _('I would like to add you to my contact list.')
    if account:
        s = _('Hello, I am $name.') + ' ' + s
        s = Template(s).safe_substitute({'name': app.nicks[account]})
        return s

def get_user_proxy(account):
    proxy_name = app.settings.get_account_setting(account, 'proxy')
    if not proxy_name:
        return None
    return get_proxy(proxy_name)

def get_proxy(proxy_name):
    try:
        settings = app.settings.get_proxy_settings(proxy_name)
    except ValueError:
        return None

    username, password = None, None
    if settings['useauth']:
        username, password = settings['user'], settings['pass']

    return ProxyData(type=settings['type'],
                     host='%s:%s' % (settings['host'], settings['port']),
                     username=username,
                     password=password)

def version_condition(current_version, required_version):
    if V(current_version) < V(required_version):
        return False
    return True

def get_available_emoticon_themes():
    files = []
    for folder in configpaths.get('EMOTICONS').iterdir():
        if not folder.is_dir():
            continue
        files += [theme for theme in folder.iterdir() if theme.is_file()]

    my_emots = configpaths.get('MY_EMOTS')
    if my_emots.is_dir():
        files += list(my_emots.iterdir())

    emoticons_themes = ['font']
    emoticons_themes += [file.stem for file in files if file.suffix == '.png']
    return sorted(emoticons_themes)

def call_counter(func):
    def helper(self, restart=False):
        if restart:
            self._connect_machine_calls = 0
        self._connect_machine_calls += 1
        return func(self)
    return helper

def load_json(path, key=None, default=None):
    try:
        with path.open('r') as file:
            json_dict = json.loads(file.read())
    except Exception:
        log.exception('Parsing error')
        return default

    if key is None:
        return json_dict
    return json_dict.get(key, default)

def ignore_contact(account, jid):
    jid = str(jid)
    known_contact = app.contacts.get_contacts(account, jid)
    ignore = app.settings.get_account_setting(account,
                                              'ignore_unknown_contacts')
    if ignore and not known_contact:
        log.info('Ignore unknown contact %s', jid)
        return True
    return False

class AdditionalDataDict(collections.UserDict):
    def __init__(self, initialdata=None):
        collections.UserDict.__init__(self, initialdata)

    @staticmethod
    def _get_path_childs(full_path):
        path_childs = [full_path]
        if ':' in full_path:
            path_childs = full_path.split(':')
        return path_childs

    def set_value(self, full_path, key, value):
        path_childs = self._get_path_childs(full_path)
        _dict = self.data
        for path in path_childs:
            try:
                _dict = _dict[path]
            except KeyError:
                _dict[path] = {}
                _dict = _dict[path]
        _dict[key] = value

    def get_value(self, full_path, key, default=None):
        path_childs = self._get_path_childs(full_path)
        _dict = self.data
        for path in path_childs:
            try:
                _dict = _dict[path]
            except KeyError:
                return default
        try:
            return _dict[key]
        except KeyError:
            return default

    def remove_value(self, full_path, key):
        path_childs = self._get_path_childs(full_path)
        _dict = self.data
        for path in path_childs:
            try:
                _dict = _dict[path]
            except KeyError:
                return
        try:
            del _dict[key]
        except KeyError:
            return

    def copy(self):
        return copy.deepcopy(self)


def save_roster_position(window):
    if not app.settings.get('save-roster-position'):
        return
    if app.is_display(Display.WAYLAND):
        return
    x_pos, y_pos = window.get_position()
    log.debug('Save roster position: %s %s', x_pos, y_pos)
    app.settings.set('roster_x-position', x_pos)
    app.settings.set('roster_y-position', y_pos)


class Singleton(type):