_dispatch.py

# -*- coding: UTF-8 -*-
"""
Code Dispatcher
===============

Author
------
Yuchen Jin (cainmagi)
cainmagi@gmail.com

License
-------
MIT License

Description
-----------
A dispatcher script used for copying codes to the corresponding places when the project
is large.
"""

import os
import re
import abc
import shutil
import logging

from typing import Union, IO

try:
    from typing import Sequence
    from typing import List
except ImportError:
    from collections.abc import Sequence
    from builtins import list as List

from typing_extensions import ClassVar, Literal


PathLike = Union[str, os.PathLike]
"""A string of a path object that can be used for locating file(s)."""

log = logging.LoggerAdapter(
    logging.getLogger("code-dispatcher"), extra={"className": ""}
)
"""(Logger) Code dispatcher of the complicated Dash project."""

__version__ = "1.0.0"
"""Version of this script."""


class PatternDispatcher(abc.ABC):
    """The pattern used in dispatchers."""

    def __init__(self, pattern: str) -> None:
        """Initialization.

        Arguments
        ---------
        pattern: `str`
            The definition of the pattern.
        """
        self.__pattern = re.compile(str(pattern))

    @property
    @abc.abstractmethod
    def type(self) -> Literal["include", "exclude"]:
        """The type of this dispatcher."""
        raise NotImplementedError

    @property
    def pattern(self):
        return self.__pattern


class PatternInclude(PatternDispatcher):
    """The pattern specifying file(s) that should be included."""

    @property
    def type(self) -> Literal["include"]:
        """The type of this dispatcher."""
        return "include"


class PatternExclude(PatternDispatcher):
    """The pattern specifying file(s) that should be excluded."""

    @property
    def type(self) -> Literal["exclude"]:
        """The type of this dispatcher."""
        return "exclude"


class CodeDispatcher:
    patterns: ClassVar[List[PatternDispatcher]] = [
        PatternInclude(r"^.*?\.(?:py|pyc|js|json|js\.map|css|css\.map)$"),
        PatternExclude(r"^_(?:_init_|_main_)_\.py$"),
    ]
    """Regex patterns used for matching the names the files that should be included."""

    patterns_comps: ClassVar[List[PatternDispatcher]] = [
        PatternInclude(r"^.*?\.py$"),
        PatternExclude(r"^_(?:_init_|_main_|imports)_\.py$"),
    ]
    """Regex patterns used for matching the names of python scripts that are the
    definitions of the Dash components. Note that these patterns are used only when
    the `patterns` used for including files match the file names."""

    pattern_code_namespace: ClassVar[str] = (
        r"^(\s*?_namespace = ['\"])(.*?)(_components['\"])\s*?$"
    )
    """Regex pattern used for capturing the namespace to be modified."""

    def __init__(self) -> None:
        """Initialization."""
        self._pattern_code_namespace = re.compile(self.pattern_code_namespace)
        self.log = logging.LoggerAdapter(
            log.logger, extra={"className": self.__class__.__name__}
        )

    def parse_codes(self, fin: IO[str], fout: IO[str]) -> None:
        """The implementation of the code parser.

        This parser is performed on the Dash component modules.

        Arguments
        ---------
        fin: `IO[str]`
            The codes generated by Dash.

        fout: `IO[str]`
            The modified codes that will be written in the destination.
        """
        while line := fin.readline():
            matchobj = self._pattern_code_namespace.fullmatch(line)
            if matchobj is None:
                fout.write(line)
            else:
                prefix = matchobj.group(1)
                fout.write(prefix + matchobj.group(2) + prefix[-1] + "\n")

    @staticmethod
    def is_pattern_match(file_name: str, patterns: Sequence[PatternDispatcher]) -> bool:
        """Check whether the given file_name matches the rules defined by patterns.

        Will return `False` if no pattern is provided.
        """
        if not patterns:
            return False
        for pattern in patterns:
            if pattern.type == "include":
                if pattern.pattern.search(file_name) is None:
                    return False
            else:
                if pattern.pattern.search(file_name) is not None:
                    return False
        return True

    def copy(self, src: PathLike, dst: PathLike) -> None:
        """Copy all included files in `src` to the destination folder `dst`.

        The component scripts will be modified.

        Arguments
        ---------
        src: `str | os.PathLike`
            The path to the source folder, where the codes generated by dash are
            stored.

        dst: `str | os.PathLike`
            The destination where the components codes will be copied to.
        """
        if not os.path.isdir(src):
            raise FileNotFoundError(src)

        if os.path.isfile(dst):
            raise FileExistsError(dst)
        os.makedirs(dst, exist_ok=True)

        src = os.path.normpath(str(src)).strip()
        dst = os.path.normpath(str(dst)).strip()
        for src_root, dirs, files in os.walk(src):
            dst_root = os.path.normpath(src_root.replace(src, dst))
            for dir in dirs:
                path_dir = os.path.join(dst_root, dir)
                self.log.info("Create folder: {0}".format(path_dir))
                os.makedirs(path_dir, exist_ok=True)
            for file in files:
                file = file.strip()
                if not self.is_pattern_match(file, self.patterns):
                    continue
                src_file = os.path.join(src_root, file)
                dst_file = os.path.join(dst_root, file)
                if self.is_pattern_match(file, self.patterns_comps):
                    self.log.info("Copy module codes: {0}".format(dst_file))
                    with open(src_file, "r", encoding="utf-8") as fin, open(
                        dst_file, "w", encoding="utf-8"
                    ) as fout:
                        self.parse_codes(fin, fout)
                else:
                    self.log.info("Copy file: {0}".format(dst_file))
                    shutil.copyfile(src_file, dst_file)


if __name__ == "__main__":
    import argparse

    logging.basicConfig(
        level=log.logger.level,
        format=(
            "%(module)s - %(className)s.%(funcName)s @ %(asctime)s: %(levelname)s %(message)s"
        ),
    )

    parser = argparse.ArgumentParser(
        description=("Code Dispatcher by cainmagi. {0}.".format(__version__)),
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )

    def parse_args(arg_parser: argparse.ArgumentParser) -> argparse.Namespace:
        """Configure the arguments of the CLI."""
        arg_optional = arg_parser._action_groups.pop()
        arg_optional.add_argument(
            "-v",
            "--version",
            version=__version__,
            action="version",
            help="If specified, show the current version.",
        )
        arg_required = arg_parser.add_argument_group("required arguments")
        arg_parser._action_groups.append(arg_optional)

        arg_required.add_argument(
            "-src",
            "--source",
            type=str,
            required=True,
            metavar="str",
            help=(
                'The source path of the folder created by "dash-generate-components".'
            ),
        )
        arg_required.add_argument(
            "-dst",
            "--destination",
            type=str,
            required=True,
            metavar="str",
            help=(
                "The target path of the subfolder where the source files will be "
                "copied."
            ),
        )

        return arg_parser.parse_args()

    args = vars(parse_args(parser))

    CodeDispatcher().copy(
        "dash_file_cache_components", os.path.join("dash_file_cache", "components")
    )