Underground-Digital/Workflow-Engine

1import enum2import importlib.util3import json4import logging5import os6from pathlib import Path7from typing import Any, Optional8 9from pydantic import BaseModel10 11from core.helper.position_helper import sort_to_dict_by_position_map12 13 14class ExtensionModule(enum.Enum):15    MODERATION = "moderation"16    EXTERNAL_DATA_TOOL = "external_data_tool"17 18 19class ModuleExtension(BaseModel):20    extension_class: Any = None21    name: str22    label: Optional[dict] = None23    form_schema: Optional[list] = None24    builtin: bool = True25    position: Optional[int] = None26 27 28class Extensible:29    module: ExtensionModule30 31    name: str32    tenant_id: str33    config: Optional[dict] = None34 35    def __init__(self, tenant_id: str, config: Optional[dict] = None) -> None:36        self.tenant_id = tenant_id37        self.config = config38 39    @classmethod40    def scan_extensions(cls):41        extensions: list[ModuleExtension] = []42        position_map = {}43 44        # get the path of the current class45        current_path = os.path.abspath(cls.__module__.replace(".", os.path.sep) + ".py")46        current_dir_path = os.path.dirname(current_path)47 48        # traverse subdirectories49        for subdir_name in os.listdir(current_dir_path):50            if subdir_name.startswith("__"):51                continue52 53            subdir_path = os.path.join(current_dir_path, subdir_name)54            extension_name = subdir_name55            if os.path.isdir(subdir_path):56                file_names = os.listdir(subdir_path)57 58                # is builtin extension, builtin extension59                # in the front-end page and business logic, there are special treatments.60                builtin = False61                position = None62                if "__builtin__" in file_names:63                    builtin = True64 65                    builtin_file_path = os.path.join(subdir_path, "__builtin__")66                    if os.path.exists(builtin_file_path):67                        position = int(Path(builtin_file_path).read_text(encoding="utf-8").strip())68                    position_map[extension_name] = position69 70                if (extension_name + ".py") not in file_names:71                    logging.warning(f"Missing {extension_name}.py file in {subdir_path}, Skip.")72                    continue73 74                # Dynamic loading {subdir_name}.py file and find the subclass of Extensible75                py_path = os.path.join(subdir_path, extension_name + ".py")76                spec = importlib.util.spec_from_file_location(extension_name, py_path)77                if not spec or not spec.loader:78                    raise Exception(f"Failed to load module {extension_name} from {py_path}")79                mod = importlib.util.module_from_spec(spec)80                spec.loader.exec_module(mod)81 82                extension_class = None83                for name, obj in vars(mod).items():84                    if isinstance(obj, type) and issubclass(obj, cls) and obj != cls:85                        extension_class = obj86                        break87 88                if not extension_class:89                    logging.warning(f"Missing subclass of {cls.__name__} in {py_path}, Skip.")90                    continue91 92                json_data = {}93                if not builtin:94                    if "schema.json" not in file_names:95                        logging.warning(f"Missing schema.json file in {subdir_path}, Skip.")96                        continue97 98                    json_path = os.path.join(subdir_path, "schema.json")99                    json_data = {}100                    if os.path.exists(json_path):101                        with open(json_path, encoding="utf-8") as f:102                            json_data = json.load(f)103 104                extensions.append(105                    ModuleExtension(106                        extension_class=extension_class,107                        name=extension_name,108                        label=json_data.get("label"),109                        form_schema=json_data.get("form_schema"),110                        builtin=builtin,111                        position=position,112                    )113                )114 115        sorted_extensions = sort_to_dict_by_position_map(116            position_map=position_map, data=extensions, name_func=lambda x: x.name117        )118 119        return sorted_extensions120