yasd/yasd.py

#!/usr/bin/env python
# (c) 2023 Perro Tuerto <hi@perrotuerto.blog>.
# Founded by Mexican Academy of Language <https://academia.org.mx>.
# Licensed under GPLv3 <https://www.gnu.org/licenses/gpl-3.0.en.html>.
# Requirements: python > 3.10, pyyaml, lxml, bs4, xmlschema, rich

import sys
import yaml
import argparse
import xmlschema
import urllib.request
from pathlib import Path
from bs4 import BeautifulSoup
from bs4.formatter import XMLFormatter
from rich.console import Console
from rich.markdown import Markdown


class YASD:
    """
    YASD actions performer.
    """

    def do(
        action="check",
        indata=None,
        outfile=None,
        quiet=False,
        log=False,
        stdout=False,
        validate=False,
    ):
        """
        Performs YASD actions directly.

        Intented for YASDCLI, but can also be used programmatically.

        :param str action: YASD action to perform; 'check' by default
        :param indata: YASD input; 'None' by default
        :type indata: None or Path or dict
        :param outfile: YASD output file path; 'None' by default
        :type outfile: None or Path
        :param quiet: If messages are print or not; 'False' by default
        :type quiet: True or False
        :param log: If messages are write in a file or not; 'False' by default
        :type log: True or False
        :param stdout: if conversion goes to stdout or not; 'False' by default
        :type stdout: True or False
        :param validate: if XSD is validated; 'False' by default
        :type validate: True or False
        :return: Output data; str on 'document'; bs4.element.Tag on 'convert'
            or 'sample'; YAML dict on 'check'
        :rtype: str or bs4.element.Tag or dict
        """
        yasd = YASD(indata, outfile, quiet, log, stdout, validate)
        yasd.msgr.run(f"action_{action}")
        match action:
            case "document":
                return yasd.document()
            case "convert":
                return yasd.convert()
            case "sample":
                return yasd.sample()
            case "check":
                return yasd.yaml

    def __init__(
        self,
        indata=None,
        outfile=None,
        quiet=False,
        log=False,
        stdout=False,
        validate=False,
    ):
        """
        Inits YASD object.

        :param indata: YASD input; 'None' by default
        :type indata: None or Path or dict
        :param outfile: YASD output file path; 'None' by default
        :type outfile: None or Path
        :param quiet: If messages are print or not; 'False' by default
        :type quiet: True or False
        :param log: If messages are write in a file or not; 'False' by default
        :type log: True or False
        :param stdout: if conversion goes to stdout or not; 'False' by default
        :type stdout: True or False
        :param validate: if XSD is validated; 'False' by default
        :type validate: True or False
        """
        self.msgr = YASDMessenger(quiet=quiet, log=log)
        self.yaml = YASDCheck(indata, self.msgr).yaml
        self.formatter = XMLFormatter(indent=2)
        self.stdout = stdout
        self.validate = validate
        if outfile is None:
            self.outfile = None
        else:
            self.outfile = YASDCheck.file(outfile, self.msgr)

    def convert(self):
        """
        Converts YASD to XSD.

        :return: XSD element
        :rtype: bs4.element.Tag
        """
        self.xsd = YASDXSD(self.yaml, self.msgr).xsd
        out = self.__output(self.xsd)
        if self.validate:
            return YASDCheck.xsd(out, self.msgr)
        else:
            return out

    def sample(self):
        """
        Generates XML sample from YASD.

        :return: XML element
        :rtype: bs4.element.Tag
        """
        self.xml = YASDXML(self.yaml, self.msgr).xml
        return self.__output(self.xml, extname=".xml")

    def document(self):
        """
        Generates RST documentation.

        :return: RST document
        :rtype: str
        """
        self.rst = YASDRST(self.yaml, self.msgr).rst
        return self.__output(self.rst, extname=".rst")

    def __output(self, outdata, extname=".xsd"):
        """
        Prints in the terminal or writes into a file.

        :return: Output data
        :rtype: str
        """
        if type(outdata) is not str:
            outdata = outdata.prettify(formatter=self.formatter)
        if self.stdout:
            if self.outfile is None:
                print(outdata)
            else:
                suffix = self.outfile.suffix
                if len(suffix) > 0 and suffix == suffix.replace(" ", ""):
                    extname = suffix
                filename = f"{self.outfile.stem}{extname}"
                filename = Path(self.outfile.parent / filename)
                filename.write_text(outdata)
        return outdata


class YASDXSD:
    """
    YASD convertor to XSD.
    """

    # TODO refactor so it fits XSD grammar, cfr. reference:
    # https://www.w3schools.com/xml/schema_elements_ref.asp

    def __init__(self, yml=None, messenger=None):
        """
        Inits YASD convertor.
        """
        self.msgr = YASDCheck.messenger(messenger)
        self.yaml = YASDCheck.yaml(yml, self.msgr)
        self.xsd = BeautifulSoup(parser="xml")
        self.__build_schema()
        self.__build_elements()
        self.__build_attributes()
        self.__build_groups()

    def __build_schema(self):
        """
        Builds root node for XSD.
        """
        for key in ["version", "schemaLocation"]:
            del self.yaml["schema"][key]
        schema = self.xsd.new_tag("schema", nsprefix="xs")
        schema["xmlns:xs"] = "http://www.w3.org/2001/XMLSchema"
        for key, val in self.yaml["schema"].items():
            schema[key] = val
        self.xsd.append(schema)

    def __build_elements(self):
        """
        Builds element nodes for XSD.

        Element nodes can be simple or complex types.
        """
        for el in self.yaml["elements"]:
            el = self.__sanitize(el)
            if el["type"] == "simple":
                self.__build_simple(el)
            else:
                self.__build_complex(el)

    def __build_attributes(self):
        """
        Builds attribute nodes for XSD.

        Attributes are always simple types.
        """
        for el in self.yaml["attributeElements"]:
            self.__build_simple(self.__sanitize(el), tag="attribute")

    def __build_groups(self):
        """
        Builds group nodes for XSD.
        """
        for el in self.yaml["groups"]:
            if "attribute_group" in el.keys():
                # TODO build attributeGroup
                ...
            else:
                element = self.xsd.new_tag("group", nsprefix="xs")
                indicator = self.__build_indicator(el)
                element.append(indicator)
            self.xsd.schema.append(element)

    def __build_simple(self, el, tag="element"):
        """
        Builds simple node for XSD.

        :param dict el: YASD element
        :param str tag: tag name for XSD node
        """
        element = self.xsd.new_tag(tag, nsprefix="xs")
        if "default" in el.keys() and "fixed" in el.keys():
            del el["fixed"]
        for key, val in el.items():
            if key == "datatype":
                element["type"] = f"xs:{val}"
            elif key == "restriction":
                self.__build_restriction(element, val)
            else:
                element[key] = val
        self.xsd.schema.append(element)

    def __build_complex(self, el):
        """
        Builds complex node for XSD.

        :param dict el: YASD element
        """
        element = self.__build_complex_root(el)
        complex_type = self.__build_complex_type(el)
        self.__add_references(complex_type, el, is_attr=True)
        if "children" in el.keys():
            indicator = self.__build_indicator(el)
            complex_type.append(indicator)
        element.append(complex_type)
        self.xsd.schema.append(element)

    def __build_complex_root(self, el):
        """
        Builds root complex node for XSD.

        :param dict el: YASD element
        :return: root complex node
        :rtype: bs4.element.Tag
        """
        element = self.xsd.new_tag("element", nsprefix="xs")
        element["name"] = el["name"]
        return element

    def __build_complex_type(self, el):
        """
        Builds complex type node for XSD.

        :param dict el: YASD element
        :return: root complex node
        :rtype: bs4.element.Tag
        """
        container = self.xsd.new_tag("complexType", nsprefix="xs")
        simple_content = self.__build_simple_content(el)
        if simple_content is not None:
            container.append(simple_content)
        if el["type"] == "mixed":
            container["mixed"] = "true"
        return container

    def __build_simple_content(self, el):
        """
        Builds simple content node for XSD.
        """
        simple_content = None
        if el["type"] == "no_elements":
            simple_content = self.xsd.new_tag("simpleContent", nsprefix="xs")
            extension = self.xsd.new_tag("extension", nsprefix="xs")
            extension["base"] = f"xs:{el['datatype']}"
            self.__add_references(extension, el, is_attr=True)
            simple_content.append(extension)
        return simple_content

    def __build_restriction(self, root, restrs, simple=True):
        """
        Builds restriction node for XSD.

        :param bs4.element.Tag root: root node that requires restriction node
        :param dict restrs: restrictions for root node
        :param str container_tag: name of container tag for restriction
        """
        if simple:
            container = self.xsd.new_tag("simpleType", nsprefix="xs")
        else:
            container = self.xsd.new_tag("complexContent", nsprefix="xs")
        restriction = self.xsd.new_tag("restriction", nsprefix="xs")
        restriction["base"] = self.__get_base(restrs)
        for restr in restrs:
            for key, val in restr.items():
                constrain = self.xsd.new_tag(key, nsprefix="xs", value=val)
                restriction.append(constrain)
        container.append(restriction)
        root.append(container)

    def __build_indicator(self, el):
        """
        Builds indicator node for XSD.

        :param dict el: YASD element
        """
        other_tag = el["children_order"]
        indicator = self.xsd.new_tag(other_tag, nsprefix="xs")
        self.__add_references(indicator, el)
        return indicator

    def __get_base(self, restrictions):
        """
        Gets restriction data type.

        It uses the first restriction to get the data type. A valid restriction
        node always have the same data type for all its restrictions.

        :param dict restrictions: restrictions as a dict
        :return: 'xs:string' or 'xs:integer'
        :rtype: str
        """
        key = list(restrictions[0].keys())[0]
        strings = "enumeration pattern whiteSpace length minLength maxLength"
        if key in strings.split():
            return "xs:string"
        else:
            return "xs:integer"

    def __get_references(self, el, is_attr):
        """
        Gets required variables values for references.

        :param dict el: YASD element
        :param is_attr: if is and attribute reference
        :type is_attr: True or False
        """
        key, tag = "children", "element"
        if is_attr:
            key, tag = "attributes", "attribute"
        if key in el.keys() and "group" in el[key][0].keys():
            tag, name = "group", "group"
        else:
            name = "ref"
        return key, tag, name

    def __add_references(self, root, el, is_attr=False):
        """
        Adds element or attribute references to root node.

        :param bs4.element.Tag root: root node that requires references
        :param dict el: YASD element
        :param is_attr: if is an attribute reference; 'False' by default
        :type is_attr: True or False
        """
        key, tag, name = self.__get_references(el, is_attr)
        if key in el.keys():
            for element in el[key]:
                node = self.xsd.new_tag(tag, nsprefix="xs")
                node["ref"] = element[name]
                if "maxOccurs" in element.keys():
                    node["maxOccurs"] = element["maxOccurs"]
                if "minOccurs" in element.keys():
                    node["minOccurs"] = element["minOccurs"]
                root.append(node)
            del el[key]

    def __sanitize(self, el):
        """
        Prepares element or attribute for conversion.

        It eliminates 'description' key.

        :param dict el: Element or attribute as a dictionary
        :return: Sanitized element or attribute
        :rtype: dict
        """
        if "description" in el.keys():
            del el["description"]
        return el


class YASDXML:
    """
    YASD sampler to XML.
    """

    # TODO XML Sample

    def __init__(self, yml=None, messenger=None):
        """
        Inits YASD sampler.
        """
        self.msgr = YASDCheck.messenger(messenger)
        self.yaml = YASDCheck.yaml(yml, self.msgr)
        self.xml = "XML sample"


class YASDRST:
    """
    YASD document generator to RST.
    """

    # TODO RST document

    def __init__(self, yml=None, messenger=None):
        """
        Inits YASD document generator.
        """
        self.msgr = YASDCheck.messenger(messenger)
        self.yaml = YASDCheck.yaml(yml, self.msgr)
        self.rst = "RST document"


class YASDCheck:
    """
    YASD validator.

    Validates everything related to YASD classes.
    """

    def messenger(msgr=None):
        """
        Verifies if messenger was initialize.

        :param messenger: Messenger object
        :type messenger: None or YASDMessenger
        """
        if msgr is None:
            return YASDMessenger()
        else:
            return msgr

    def yaml(yml=None, msgr=None):
        """
        Verifies if yaml exists.

        :param dict yml: YAML object
        :param msgr: Messenger object
        :type msgr: None or YASDMessenger
        """
        msgr = YASDCheck.messenger(msgr)
        if yml is None or type(yml) is not dict:
            msgr.run("no_yaml", level="error")
        else:
            return yml

    def file(filepath=None, msgr=None):
        """
        Verifies if file exists.

        :param filepath: File path
        :type filepath: None or Path
        :param msgr: Messenger object
        :type msgr: None or YASDMessenger
        """
        msgr = YASDCheck.messenger(msgr)
        if type(filepath).__module__ != "pathlib":
            msgr.run("no_input", level="error")
        elif not filepath.exists() or not filepath.is_file():
            msgr.run("invalid_input", level="error", file=filepath)
        return filepath.resolve()

    def url(key="readme", msgr=None):
        """
        Verifies if remote file exists

        :param str key: YASDMessenger string key
        :param msgr: Messenger object
        :type msgr: None or YASDMessenger
        """
        # TODO if YASD becomes pip package, the fetched files should be local
        # Remove urllib import if that is the case
        msgr = YASDCheck.messenger(msgr)
        try:
            url = YASDMessenger.keys()[key]
            msgr.run("fetching", url=url)
            return urllib.request.urlopen(url).read().decode("utf-8")
        except Exception:
            msgr.run("no_url", level="error", url=url)

    def xsd(xsd, msgr=None):
        """
        Validates XSD.

        :param str xsd: XSD as XML string
        :param msgr: Messenger object
        :type msgr: None or YASDMessenger
        """
        msgr = YASDCheck.messenger(msgr)
        msgr.run("validating")
        try:
            xmlschema.XMLSchema(xsd)
        except xmlschema.validators.exceptions.XMLSchemaParseError as error:
            error = str(error).replace("\n", "\n  ")
            msgr.run("no_valid", error=error, level="error")

    def __init__(self, indata=None, messenger=None):
        """
        Inits YASD validator.

        :param indata: YASD input
        :type indata: None or Path or dict
        :param messenger: Object for print or save messages
        :type messenger: None or YASDMessenger
        """
        self.msgr = YASDCheck.messenger(messenger)
        if type(indata) is dict:
            self.yaml = indata
        else:
            self.yaml = self.parse_file(YASDCheck.file(indata, self.msgr))
        self.check_structure()

    def parse_file(self, filepath):
        """
        Attempts YASD file parsing.

        :param filepath: YASD file path
        :type filepath: Path
        """
        raw = filepath.read_text(encoding="utf8")
        try:
            return yaml.safe_load(raw)
        except yaml.YAMLError:
            self.msgr.run("invalid_yaml", level="error")

    def check_structure(self):
        """
        Verifies YASD structure.

        :return: YASD structure
        :rtype: dict
        """
        # TODO extra checks for self.yaml
        ...


class YASDMessenger:
    """
    YASD printer or writer.
    """

    def keys():
        """
        Messages keys dictionary.
        """
        # TODO internationalization with: https://github.com/sectasy0/pyi18n
        return {
            "prog": "yasd",
            "description": """
            YASD, Yet Another Schema Definition. YASD is a YAML format for
            human writable XSDs (XML Schema Definition), humans declare what is
            indispensable, leaving the machines to do the rest of the
            unreadable <syntaxis who_can_read_this="?" />.
            """,
            "epilog": """
            (c) 2023 Perro Tuerto <hi@perrotuerto.blog>. Founded by Mexican
            Academy of Language <https://academia.org.mx>. Licensed under GPLv3
            <https://www.gnu.org/licenses/gpl-3.0.en.html>.
            """,
            "readme": "https://gitlab.com/perrotuerto_personal/codigo/yasd/-/raw/no-masters/README.md",
            "w3": "https://www.w3.org/2001/XMLSchema.xsd",
            "help_action": "action to perform",
            "help_input": "input file in YAML format",
            "help_output": "output file",
            "help_quiet": "enable quiet mode",
            "help_log": "write log",
            "action_convert": "creating XSD schema",
            "action_check": "checking YASD structure",
            "action_sample": "creating XML sample",
            "action_document": "creating RST documentation",
            "invalid_level": "invalid log level '@lvl'",
            "invalid_input": "invalid file '@file'",
            "invalid_yaml": "invalid YAML structure",
            "no_url": "failed to fetch '@url'",
            "no_input": "input file needed",
            "no_yaml": "YAML dict needed",
            "no_valid": "XSD schema has the following error:\n  @error",
            "fetching": "fetching '@url'",
            "validating": "validating XSD schema",
        }

    def __init__(self, quiet=False, log=False):
        """
        Inits YASD Messenger.
        """
        self.quiet = quiet
        self.log = log

    def run(self, key="", level="info", **kwargs):
        """
        Prints or writes messages.

        '**kwargs' are the keys for message text replacements.

        :param str key: Message key
        :param str level: Log level; 'info' by default
        """
        self.__check_level(level)
        name = YASDMessenger.keys()["prog"]
        msg = self.__get_msg(key, **kwargs)
        msg = f"{name}: {level}: {msg}"
        # TODO print or save depending on self.quiet and self.log
        print(msg)
        if level in ["error", "fatal"]:
            sys.exit(1)

    def __check_level(self, level):
        """
        Verifies log level.

        Prints warning if log level doesn't exist.

        :param str level: Log level
        """
        if level not in ["trace", "debug", "info", "warn", "error", "fatal"]:
            YASDMessenger().run("invalid_level", level="warn", lvl=level)

    def __get_msg(self, key, **kwargs):
        """
        Gets message based on key.

        '**kwargs' are the keys for message text replacements.

        :param str key: Message key
        :return: Message or key if message key doesn't exist.
        :rtype: str
        """
        if key in YASDMessenger.keys().keys():
            msg = YASDMessenger.keys()[key]
            for key, value in kwargs.items():
                msg = msg.replace(f"@{key}", str(value))
            return msg
        else:
            return key


class YASDCLI:
    """
    YASD command-line interface.
    """

    def print_man():
        """
        Prints README as manual.
        """
        raw = YASDCheck.url()
        raw = raw.replace("## Table of Contents\n\n[TOC]\n\n", "")
        md = Markdown(raw)
        console = Console()
        with console.pager(styles=True):
            console.print(md)

    def __init__(self):
        """
        Inits YASD CLI.
        """
        self.__init_parser()
        args = self.parser.parse_args()
        if args.action == "man":
            YASDCLI.print_man()
        else:
            YASD.do(
                args.action,
                args.input,
                args.output,
                args.quiet,
                args.log,
                stdout=True,
                validate=True,
            )

    def __init_parser(self):
        """
        Inits argument parser.
        """
        msg = YASDMessenger.keys()
        self.parser = argparse.ArgumentParser(
            prog=msg["prog"],
            description=msg["description"],
            epilog=msg["epilog"],
        )
        self.parser.add_argument(
            "action",
            choices=["convert", "check", "sample", "document", "man"],
            help=msg["help_action"],
        )
        self.parser.add_argument(
            "input",
            type=Path,
            nargs="?",
            default=None,
            help=msg["help_input"],
        )
        self.parser.add_argument(
            "-q", "--quiet", action="store_true", help=msg["help_quiet"]
        )
        self.parser.add_argument(
            "-l", "--log", action="store_true", help=msg["help_log"]
        )
        self.parser.add_argument(
            "-o",
            "--output",
            type=Path,
            default=None,
            help=msg["help_output"],
        )


if __name__ == "__main__":
    YASDCLI()