LDPy/ldpy.py

#!/usr/bin/env python3
# coding: utf8

"""A basic opensearch.org client.

All start here.

Author: frederic zind 2022-08-02
Licenses: GNU AGPL v3: http://www.gnu.org/licenses/
"""


import argparse
import logging
import os
from pprint import pformat as pf
import sys

import client  # pylint: disable=import-error

# Ranqe for the request retrieving the last entries.
LAST_E_CHOICES = range(1, 51)

logger = logging.getLogger(os.path.splitext(os.path.basename(sys.argv[0]))[0])


def setup_logging(options):
    """Configure logging."""
    root = logging.getLogger("")
    logger.setLevel(options.debug and logging.DEBUG or logging.INFO)

    if options.debug:
        # Logging with --debug option
        # Logging on terminal standard output

        handler_tty = logging.StreamHandler()
        handler_tty.setFormatter(
            logging.Formatter("%(name)s::%(levelname)s::%(message)s")
        )
        root.addHandler(handler_tty)

    else:
        # Logging default setings

        handler_tty = logging.StreamHandler()
        handler_tty.setFormatter(logging.Formatter("%(levelname)s\t%(message)s"))
        root.addHandler(handler_tty)


class CustomFormatter(
    argparse.RawDescriptionHelpFormatter, argparse.ArgumentDefaultsHelpFormatter
):
    """This convert file docstring into help text."""

    pass


def parse_args(args=sys.argv[1:]):
    """Argument parser.

    Put values & describe parameters here not in your code
    Ease usage with help & docs
    """
    parser = argparse.ArgumentParser(
        description=sys.modules[__name__].__doc__, formatter_class=CustomFormatter
    )

    group = parser.add_argument_group("log setting")
    group.add_argument(
        "-d",
        "--debug",
        action="store_true",
        default=False,
        help="Log activity in console",
    )

    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "-m",
        "--mapping",
        action="store_true",
        help="Show the stream mapping",
    )
    group.add_argument(
        "-l",
        "--last",
        choices=LAST_E_CHOICES,
        const="3",
        help="Last entries of the stream",
        metavar="INT",
        nargs="?",
        type=int,
    )

    return parser.parse_args(args)


def get_last_entries(entries_nb):
    """Get the last n entries from a stream."""
    if entries_nb in LAST_E_CHOICES:
        logger.debug("Wait before getting '%s' entries!", entries_nb)
        query = {"size": entries_nb}
        last_entries = client.opnsrch_clt.search(body=query)
        logger.debug(pf(last_entries))

    else:
        logger.critical("'%s' is not in '%s'", entries_nb, LAST_E_CHOICES)
        raise ValueError

    return last_entries


def get_map_props():
    """Get stream mapping.

    'demo' stream did not have an LDP_ALIAS

    LDP indices are rolling and LDP do not implement the call
    to get indices related to an alias.
    """
    map_props = []

    if client.LDP_CRED_TYPE == "token":
        indices = list(client.opnsrch_clt.indices.get_alias("*"))
        mapping = client.opnsrch_clt.indices.get_mapping(client.LDP_ALIAS)[indices[0]][
            "mappings"
        ]
        map_props = sorted(list(mapping["properties"]))
        logger.debug("Found mapping!")
    else:
        logger.critical(
            "Stream mapping unavailable with credential type: '%s'",
            client.LDP_CRED_TYPE,
        )

    return map_props


def strip_demo_entries(raw_data):
    """Remove keys in entries to ease human reading.

    Returns a list populated with a dict for each entry.
    This is a specific function for demo stream /!\
    """
    stripped_data = []

    for raw_hit in raw_data["hits"]["hits"]:

        stripped_hit = {}
        for key in [
            "source",
            "category",
            "title",
            "message",
            "rating_num",
            "timestamp",
        ]:
            stripped_hit[key] = raw_hit["_source"][key]

        stripped_data.append(stripped_hit)

    return stripped_data


def main(options):
    """Execute as script. Functions related to the arguments passed.

    Data stripper use a function build only for demo stream: `strip_demo_entries()`
    """
    if options.mapping:
        response = get_map_props()
        logger.debug("Mapping for '%s' stream: %s", client.LDP_STREAM_NAME, response)
    elif options.last:
        response = pf(strip_demo_entries(get_last_entries(options.last)))
        logger.debug(
            "Last '%s' entries for '%s' stream:", options.last, client.LDP_STREAM_NAME
        )
        logger.debug(response)
    else:
        raise NotImplementedError

    return response


if __name__ == "__main__":
    pargs = parse_args()
    setup_logging(pargs)
    result = main(pargs)

    if not pargs.debug:
        print(result)