KaikOut/kaikout/utilities.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import csv
from email.utils import parseaddr
import hashlib
from kaikout.database import Toml
from kaikout.log import Logger
#import kaikout.sqlite as sqlite
import os
import sys
import tomli_w
from urllib.parse import urlsplit

try:
    import tomllib
except:
    import tomli as tomllib

logger = Logger(__name__)


class Config:


    def get_default_data_directory():
        """
        Determine the directory path where data will be stored.
    
        * If $XDG_DATA_HOME is defined, use it;
        * else if $HOME exists, use it;
        * else if the platform is Windows, use %APPDATA%;
        * else use the current directory.

        Returns
        -------
        str
            Path to data directory.
        """
        if os.environ.get('HOME'):
            data_home = os.path.join(os.environ.get('HOME'), '.local', 'share')
            return os.path.join(data_home, 'kaikout')
        elif sys.platform == 'win32':
            data_home = os.environ.get('APPDATA')
            if data_home is None:
                return os.path.join(
                    os.path.dirname(__file__), 'kaikout_data')
        else:
            return os.path.join(os.path.dirname(__file__), 'kaikout_data')


    def get_default_config_directory():
        """
        Determine the directory path where configuration will be stored.
    
        * If $XDG_CONFIG_HOME is defined, use it;
        * else if $HOME exists, use it;
        * else if the platform is Windows, use %APPDATA%;
        * else use the current directory.

        Returns
        -------
        str
            Path to configuration directory.
        """
    #    config_home = xdg.BaseDirectory.xdg_config_home
        config_home = os.environ.get('XDG_CONFIG_HOME')
        if config_home is None:
            if os.environ.get('HOME') is None:
                if sys.platform == 'win32':
                    config_home = os.environ.get('APPDATA')
                    if config_home is None:
                        return os.path.abspath('.')
                else:
                    return os.path.abspath('.')
            else:
                config_home = os.path.join(
                    os.environ.get('HOME'), '.config'
                    )
        return os.path.join(config_home, 'kaikout')


    def get_setting_value(db_file, key):
        value = sqlite.get_setting_value(db_file, key)
        if value:
            value = value[0]
        else:
            value = Config.get_value('settings', 'Settings', key)
        return value


    def get_values(filename, key=None):
        config_dir = Config.get_default_config_directory()
        if not os.path.isdir(config_dir): config_dir = os.path.join('usr', 'share', 'kaikout')
        if not os.path.isdir(config_dir): config_dir = os.path.join(os.path.dirname(__file__), 'assets')
        config_file = os.path.join(config_dir, filename)
        with open(config_file, mode="rb") as f: result = tomllib.load(f)
        values = result[key] if key else result
        return values


class Documentation:


    def manual(filename, section=None, command=None):
        function_name = sys._getframe().f_code.co_name
        logger.debug('{}: filename: {}'.format(function_name, filename))
        config_dir = Config.get_default_config_directory()
        with open(config_dir + '/' + filename, mode="rb") as f: cmds = tomllib.load(f)
        if section == 'all':
            cmd_list = ''
            for cmd in cmds:
                for i in cmds[cmd]:
                    cmd_list += cmds[cmd][i] + '\n'
        elif command and section:
            try:
                cmd_list = cmds[section][command]
            except KeyError as e:
                logger.error(e)
                cmd_list = None
        elif section:
            try:
                cmd_list = []
                for cmd in cmds[section]:
                    cmd_list.extend([cmd])
            except KeyError as e:
                logger.error('KeyError:' + str(e))
                cmd_list = None
        else:
            cmd_list = []
            for cmd in cmds:
                cmd_list.extend([cmd])
        return cmd_list


class Log:


    def jid_exist(filename, fields):
        """
        Ceck whether Alias and Jabber ID exist.

        Parameters
        ----------
        filename : str
            Filename.
        fields : list
            jid, alias, timestamp.
    
        Returns
        -------
        None.
        """
        data_dir = Config.get_default_data_directory()
        if not os.path.isdir(data_dir): return False
        data_dir_logs = os.path.join(data_dir, 'logs')
        if not os.path.isdir(data_dir_logs): return False
        csv_file = os.path.join(data_dir_logs, f'{filename}.csv')
        if not os.path.exists(csv_file): return False
        with open(csv_file, 'r') as f:
            reader = csv.reader(f)
            for line in reader:
                if line[0] == fields[0]:
                    return True


    def alias_jid_exist(filename, fields):
        """
        Ceck whether Alias and Jabber ID exist.

        Parameters
        ----------
        filename : str
            Filename.
        fields : list
            jid, alias, timestamp.
    
        Returns
        -------
        None.
        """
        data_dir = Config.get_default_data_directory()
        if not os.path.isdir(data_dir): return False
        data_dir_logs = os.path.join(data_dir, 'logs')
        if not os.path.isdir(data_dir_logs): return False
        csv_file = os.path.join(data_dir_logs, f'{filename}.csv')
        if not os.path.exists(csv_file): return False
        with open(csv_file, 'r') as f:
            reader = csv.reader(f)
            for line in reader:
                if line[0] == fields[0] and line[1] == fields[1]:
                    return True
                    


    def csv_jid(filename, fields):
        """
        Log Jabber ID to CSV file.

        Parameters
        ----------
        filename : str
            Filename.
        fields : list
            jid, alias, timestamp.
    
        Returns
        -------
        None.
        """
        data_dir = Config.get_default_data_directory()
        if not os.path.isdir(data_dir): os.mkdir(data_dir)
        data_dir_logs = os.path.join(data_dir, 'logs')
        if not os.path.isdir(data_dir_logs): os.mkdir(data_dir_logs)
        csv_file = os.path.join(data_dir_logs, f'{filename}.csv')
        if not os.path.exists(csv_file):
            columns = ['jid', 'alias', 'timestamp']
            with open(csv_file, 'w') as f:
                writer = csv.writer(f)
                writer.writerow(columns)
        with open(csv_file, 'a') as f:
            writer = csv.writer(f)
            writer.writerow(fields)


    def csv(filename, fields):
        """
        Log message or presence to CSV file.

        Parameters
        ----------
        filename : str
            Filename.
        fields : list
            type, timestamp, alias, body, lang, and identifier.
    
        Returns
        -------
        None.
        """
        data_dir = Config.get_default_data_directory()
        if not os.path.isdir(data_dir): os.mkdir(data_dir)
        data_dir_logs = os.path.join(data_dir, 'logs')
        if not os.path.isdir(data_dir_logs): os.mkdir(data_dir_logs)
        csv_file = os.path.join(data_dir_logs, f'{filename}.csv')
        if not os.path.exists(csv_file):
            columns = ['type', 'timestamp', 'alias', 'body', 'lang', 'identifier']
            with open(csv_file, 'w') as f:
                writer = csv.writer(f)
                writer.writerow(columns)
        with open(csv_file, 'a') as f:
            writer = csv.writer(f)
            writer.writerow(fields)


    def toml(self, room, fields, stanza_type):
        """
        Log message to TOML file.

        Parameters
        ----------
        room : str
            Group chat Jabber ID.
        fields : list
            alias, room, identifier, timestamp.
        stanza_type: str
            message or presence.
    
        Returns
        -------
        None.
        """
        alias, content, identifier, timestamp = fields
        data_dir = Toml.get_default_data_directory()
        filename = Toml.get_data_file(data_dir, room)
        # filename = room + '.toml'
        entry = {}
        entry['alias'] = alias
        entry['body'] = content
        entry['id'] = identifier
        entry['timestamp'] = timestamp
        activity_type = 'activity_' + stanza_type
        message_activity_list = self.settings[room][activity_type] if activity_type in self.settings[room] else []
        while len(message_activity_list) > 20: message_activity_list.pop(0)
        message_activity_list.append(entry)
        self.settings[room][activity_type] = message_activity_list # NOTE This directive might not be needed
        data = self.settings[room]
        content = tomli_w.dumps(data)
        with open(filename, 'w') as f: f.write(content)


class BlockList:


    def get_filename():
        """
        Get pathname of filename.
        If filename does not exist, create it.

        Parameters
        ----------
        None.
    
        Returns
        -------
        filename : str
            Pathname.
        """
        data_dir = Config.get_default_data_directory()
        if not os.path.isdir(data_dir): os.mkdir(data_dir)
        filename = os.path.join(data_dir, r"blocklist.toml")
        if not os.path.exists(filename):
            data = {'entries' : {}}
            content = tomli_w.dumps(data)
            with open(filename, 'w') as f: f.write(content)
        return filename


    def load_blocklist(self):
        filename = BlockList.get_filename()
        with open(filename, 'rb') as f:
            self.blocklist = tomllib.load(f)


    def add_entry_to_blocklist(self, jabber_id, node_id, item_id):
        """
        Update blocklist file.

        Parameters
        ----------
        jabber_id : str
            Jabber ID.
        node_id : str
            Node name.
        item_id : str
            Item ID.
    
        Returns
        -------
        None.
        """
        if jabber_id not in self.blocklist['entries']:
            self.blocklist['entries'][jabber_id] = {}
        if node_id not in self.blocklist['entries'][jabber_id]:
            self.blocklist['entries'][jabber_id][node_id] = []
        self.blocklist['entries'][jabber_id][node_id].append(item_id)
        data = self.blocklist
        content = tomli_w.dumps(data)
        filename = BlockList.get_filename()
        with open(filename, 'w') as f: f.write(content)


class Url:


    def check_xmpp_uri(uri):
        """
        Check validity of XMPP URI.

        Parameters
        ----------
        uri : str
            URI.
    
        Returns
        -------
        jid : str
            JID or None.
        """
        jid = urlsplit(uri).path
        if parseaddr(jid)[1] != jid:
            jid = False
        return jid


class String:


    def md5_hash(url):
        """
        Hash URL string to MD5 checksum.

        Parameters
        ----------
        url : str
            URL.
    
        Returns
        -------
        url_digest : str
            Hashed URL as an MD5 checksum.
        """
        url_encoded = url.encode()
        url_hashed = hashlib.md5(url_encoded)
        url_digest = url_hashed.hexdigest()
        return url_digest