ody_migr_utils.py

#!/usr/bin/env python
"""This module includes utility functions."""


import base64
import json
import hashlib
import logging
import sys

from functools import wraps
from errno import EACCES, EPERM, ENOENT

LOGGER = logging.getLogger(__name__)


def get_ith_batch(_list, batch_index, batch_size):
    """Utility function to obtain a ith batch of data from list

    Args:
        _list:          list to choose batch from
        batch_index:    batch index
        batch_size:     size of batch to select

    Returns:
        batch (sub list of _list) of size batch_size
    """
    start = batch_index * batch_size
    end = min(batch_index * batch_size + batch_size, len(_list))
    return _list[start:end]


def encode_base64(_data):
    """Encode content into using base64 encoding

    Args:
        _data (str): input data to encode

    Returns:
        base64 encoded bytes
    """
    return base64.b64encode(_data.encode(encoding="utf-8"))


def decode_base64(_encode):
    """Decode base64 encoded content to string

    Args:
        _encode : bytes-like object

    Returns:
        decoded string
    """
    return base64.b64decode(_encode).decode(encoding="utf-8")


def get_valid_json(_obj, **kwargs):
    """Utility function to print pretty json string

    Args:
        _obj:       python dictionary
        **kwargs:   optional keywords arguments
                    possible values:
                        pretty: for pretty printing
    Returns:
        prints formatted json string

    """
    pretty = kwargs.get("pretty", False)
    if pretty:
        # formatted print layout
        return json.dumps(_obj, indent=4, sort_keys=True,
                          separators=(',', ':'))
    # compact encoding
    return json.dumps(_obj)


def sanitize_insert_values(_values):
    """Cleanup list values

    Args:
        _values: list of values

    Returns:
        list of normalized list
    """
    assert type(_values) == list
    sanitized_values = []
    for val in _values:
        if hasattr(val, 'strip'):
            val = val.strip()
        sanitized_values.append(val)
    return sanitized_values


def sanitize_dict(records_dict):
    """Sanitize dictionary values

    Args:
        records_dict: dictionary values

    Returns:
        list of stripped, sanitized dictionary
    """
    for k in records_dict.keys():
        v = records_dict[k]
        if hasattr(v, 'strip'):
            v = v.strip()
        records_dict[k] = v
    return records_dict


def sanitize_insert_collection(_collection):
    """Cleanup list values

    Args:
        _collection: list of dictionary values

    Returns:
        list of stripped, sanitized list of dictionary values
    """
    assert type(_collection) == list
    if len(_collection) > 0:
        assert type(_collection[0]) == dict

    for records_dict in _collection:
        records_dict = sanitize_dict(records_dict)
    return _collection


def generate_mdx_hash(_cu, _key):
    """Generate and return MDX new_id hash

    Args:
        _cu:        Current CU
        _key:       key to append with _cu

    Returns:
        SHA1 encoded hexdigest code
    """
    assert _key is not None
    assert type(_key) == str
    message = "{}{}".format(_cu, _key)
    return hashlib.sha1(bytes(message, encoding='utf-8')).hexdigest()


def generate_hash(_action, _cu, _secret_key):
    """Generate and return authentication hash code

    Args:
        _action:    Data target
        _cu:        Current CU

    Returns:
        SHA1 encoded hexdigest code
    """
    assert _secret_key is not None
    assert type(_secret_key) == str
    message = "".join([_action, _cu, _secret_key])
    return hashlib.sha1(bytes(message, encoding='utf-8')).hexdigest()


def format_error_permission(_e, _file__or_script):
    """Display error messages for FileNotFound/Permission Denied errors.

    Args:
        _e:                 error message object
        _file_or_script:    script/file that caused error

    Returns:
        Formatted PermissionError/FileNotFoundError error message
    """
    error_msg = ""
    # PermissionError
    if _e.errno == EPERM or _e.errno == EACCES:
        error_msg = "PermissionError - error({}): {} for: {}\t".format(
            _e.errno, _e.strerror, _file__or_script)
    # FileNotFoundError
    elif _e.errno == ENOENT:
        error_msg = "FileNotFoundError - error({}): {} as:{}\t".format(
            _e.errno, _e.strerror, _file__or_script)
    # returning for logging purpose whenever needed
    return error_msg


def file_exc_decorator(original_file_handler):
    """decorator to handle all file operation exceptions"""
    @wraps(original_file_handler)
    def wrapper(*args, **kwargs):
        try:
            return original_file_handler(*args, **kwargs)
        except (IOError, NameError) as e:
            error_msg = format_error_permission(e, "")
            LOGGER.error(error_msg)
            raise SystemExit(error_msg)
        except Exception as e:
            LOGGER.error(e)
            raise SystemExit(e)
        except:
            # sys.exc_info() returns a tuple (type, value, traceback)
            error_msg = sys.exc_info()[1].args[0]
            LOGGER.error(error_msg)
            raise SystemExit(error_msg)
    return wrapper


def get_strip_value_dict(_k, _dict, **kwargs):
    """Get or Pop an element from a dictionary

    Args:
        _k:         key to get or pop
        _dict:      dictionary to get or pop an element from
        **kwargs:   Options:
                    default: default value to return
                    pop:     do we `get` or `pop` from the dict

    Returns:
        value associated with key in a dictionary

    Raises:
        ValueError, TypeError if the invalid castto is encountered
    """
    default = kwargs.get("default", None)
    pop = kwargs.get("pop", False)

    if pop:
        val = _dict.pop(_k, default)
    else:
        val = _dict.get(_k, default)

    if val in [None, ''] and default not in [None, '']:
        val = default

    if hasattr(val, 'strip'):
        val = val.strip()

    return val


def log_progress(_progress_dict):
    """Log progress information to be used in progressbar in Monitor

    Args:
        _progress_dict: dictionary that contains completed and total stat
    """
    LOGGER.info("[PROGRESS] Tables imported: {}/{}".format(
        _progress_dict["completed"], _progress_dict["total"]))


def custom_css_parser(_content):
    """Simple css parser to convert css attributes to python dict.

    Args:
        _content (str): main css content

    Returns:
        cssdict (dict): parsed python dictionary

    Alternative advanced css parsing using tinycss
    **implementation starts**
    import tinycss
    parser = tinycss.make_parser()
    stylesheet = parser.parse_stylesheet(_content)
    cssdict = {}
    for rule in stylesheet.rules:
        this_rule = rule.selector.as_css()
        cssdict[this_rule] = {}
        print (this_rule)
        for css_attr in rule.declarations:
            # print (css_attr)
            cssdict[this_rule][css_attr.name] = css_attr.value.as_css()
            print (css_attr.name, css_attr.value.as_css())
    print(stylesheet.errors)
    return cssdict
    **implementation ends**
    """

    # are we inside /* and */ block?
    inside_comment_block = False

    # are we inside { and  } block?
    reading_values = False

    # main dict to store parsed css file as a dict
    cssdict = {}
    # running css rule value eg. body
    running_css_rule = ""
    # running css attribute values of current css rule
    # eg. attributes inside body
    running_css_attrs_dict = {}

    # iterate through each line in the css content ("\n" delimited)
    for line in _content.splitlines():
        # only after "{" is read
        if reading_values:
            # re-initiate running variabels and stop parsing attributes
            # for the current rule
            if "}" in line:
                assert running_css_rule != ""

                cssdict[running_css_rule] = running_css_attrs_dict

                running_css_rule = ""
                running_css_attrs_dict = {}
                reading_values = False
                continue

            # otherwise, keep adding attribtues to the dict
            # corresponding to the running css rule
            k_v = line.split(":", 2)

            k = k_v[0].strip()
            # handle special case url(https://..)
            v = ":".join(k_v[1:]).strip()[:-1]  # do not include ;

            running_css_attrs_dict[k] = v

        # very naive assumption about comment blocks
        if "/* **" in line and "** */" in line:
            continue

        if "/*" in line:
            inside_comment_block = True
            continue

        if "*/" in line:
            inside_comment_block = False
            continue

        # assumption is made that comment blocks are not found inside
        # { and } blocks
        if inside_comment_block is False:
            # we start parsing attribtues from next line
            if "{" in line:
                running_css_rule = line.split("{")[0].strip()
                assert running_css_attrs_dict == {}

                reading_values = True

    return cssdict