group_manage/ruoyi_common/ruoyi/registry.py

#!/usr/bin/env python
# -- coding: utf-8 --
# @Time : 2024/09/18

import os.path
from importlib import import_module
from pathlib import Path
import sys
from types import ModuleType
from flask import Blueprint, Flask

from ..base.signal import module_initailize
from .config import CONFIG_CACHE


def path_to_module(file_path:str, root_path:str) -> str:
    """
    文件路径转为模块名称
    
    Args:
        file_path (str): 文件路径
        root_path (str): 根路径
    
    Returns:
        str: 模块名称
    """
    file_path = Path(file_path)
    root_path = Path(root_path).resolve()
    file_relative = file_path.relative_to(root_path)
    module_path = str(file_relative.with_suffix('')).replace(os.sep, '.').replace('\\', '.')
    return module_path


class RuoYiModuleRegistry(object):
    
    module_prefix = "ruoyi_"
    controller_name = "controller"
    exclude_modules = ["ruoyi_ui"]
    
    def __init__(self, app:Flask=None, proot:str=None):
        self.app = app
        self.proot = proot
        url_prefix = CONFIG_CACHE["ruoyi.api.version"]
        if not url_prefix.startswith("/"):
            raise ValueError("url_prefix must start with /")
        self.api = Blueprint(
            "api", 
            __name__, 
            url_prefix=url_prefix
        )
    
    def register_modules(self):
        '''
        注册所有模块
        '''
        self.import_modules()
        self.register_controllers()
    
    def import_modules(self):
        '''
        导入所有模块
        '''
        for modname in os.listdir(self.proot):
            if not modname.startswith(self.module_prefix):
                continue
            if modname in self.exclude_modules:
                continue
            if not os.path.exists(
                os.path.join(self.proot,modname,"__init__.py")
            ):
                self.app.logger.warning(f"模块路径不存在__init__.py文件: {modname}")
                continue
            self.import_module(modname)
        
    def import_module(self, modname:str):
        '''
        导入模块
        
        Args:
            modname (str): 模块名称
        '''
        modpath = os.path.join(self.proot,modname)
        mod = import_module(path_to_module(modpath,self.proot))
        module_initailize.send(mod,registry=self)
    
    def is_registered_module(self, modname:str) -> bool:
        '''
        检查模块是否已经注册
        
        Args:
            modname (str): 模块名称
        Returns:
            bool: 是否已经注册
        '''
        flag = modname in sys.modules
        return flag
        
    def unregister_module(self, mod:ModuleType):
        '''
        注销模块
        
        Args:
            mod (ModuleType): 模块对象
        '''
        if mod in self.default_modules:
            return
        # todo

    def register_controllers(self):
        '''
        注册所有控制层路由
        '''
        for modname in os.listdir(self.proot):
            
            if not modname.startswith(self.module_prefix):
                continue
            if modname in self.exclude_modules:
                continue
            if not self.is_registered_module(modname):
                continue
            self.register_controller(modname)
        self.app.register_blueprint(self.api)
    
    def register_controller(self, modname:str):
        '''
        注册控制层路由
        
        Args:
            mod (str): 模块名称
        '''
        modpath = os.path.join(self.proot,modname)
        mod_con_path = os.path.join(modpath,self.controller_name)
        if not os.path.exists(mod_con_path):
            return  
        try:
            self._register_rules(mod_con_path)
        except Exception as e:
            raise Exception(
                "在模块中的路由，注册失败: {}，原因: {}".format(modname,str(e))
            )
    
    def unregister_controller(self, modname:str):
        '''
        注销控制层路由
        
        Args:
            modname (str): 模块名称
        '''
        # todo
    
    
    def _register_rules(self,path:str):
        '''
        注册路由规则
        
        Args:
            path (str): 路由路径
        '''
        for rule_name in os.listdir(path):
            con_path = os.path.join(path,rule_name)
            if os.path.isfile(con_path) and rule_name.endswith(".py") \
                and rule_name != "__init__.py":
                rule_path = path_to_module(con_path,self.proot)
                import_module(rule_path)
            elif os.path.isdir(con_path) and \
                os.path.exists(os.path.join(con_path,"__init__.py")):
                for sub_rule_name in os.listdir(con_path):
                    sub_path = os.path.join(con_path,sub_rule_name)
                    if os.path.isfile(sub_path) and sub_rule_name.endswith(".py") \
                        and sub_rule_name != "__init__.py":
                        rule_path = path_to_module(sub_path,self.proot)
                        import_module(path_to_module(sub_path,self.proot))
            else:
                continue