<commit_before># coding=utf-8
import json
import logging
import os
import traceback
from typing import Dict, List

from . import Plugin
from .utils import getPluginId, pluginFilePattern
from ..enums import PluginType, API
from ..exceptions.plugin import InvalidPluginSyntaxError

logger = logging.getLogger(__name__)

class PluginManager:
    def __init__(self, core):
        self.core = core
        self.plugins: Dict[str, List[Plugin]] = None
        self.unloadPlugins()

    def unloadPlugins(self) -> None:
        # noinspection PyTypeChecker
        self.plugins: Dict[str, List[Plugin]] = {x.name: [] for x in PluginType}

    def loadPlugin(self, path: str) -> bool:
        plugin: Plugin = Plugin(path=path)
        try:
            plugin.load()
        except InvalidPluginSyntaxError:
            logger.warning(
                f"Plugin {path} "
                f"could not be loaded because it has invalid syntax. "
                f"Error Detail:\n{traceback.format_exc()}."
            )
            return False

        found: bool = False
        # noinspection PyTypeChecker
        for pluginType in PluginType:
            for i, loadedPlugin in enumerate(self.plugins[pluginType.name]):
                if plugin.meta.id == loadedPlugin.meta.id:
                    found = True
                    self.plugins[pluginType.name][i] = plugin

                    if PluginType.Thread == pluginType:
                        pass
                        # TODO: kill the thread

                    break
            if found:
                break
        else:
            self.plugins[plugin.meta.type.name].append(plugin)

        # noinspection PyTypeChecker
        self.plugins = {
            pluginType.name: sorted(
                    self.plugins[pluginType.name],
                    key=lambda x: x.meta.priority, reverse=True
            ) for pluginType in PluginType}

        with open(f"{self.core.config.directory.api}/{API.Plugins.value}", "w") as f:
            # noinspection PyTypeChecker
            json.dump([
                plugin.meta.__dict__
                for pluginType in PluginType for plugin in self.plugins[pluginType.name]
            ], f, sort_keys=True, indent=4)
        return True

    def unloadPlugin(self, path: str) -> bool:
        pluginId: str = getPluginId(path)

        # noinspection PyTypeChecker
        for pluginType in PluginType:
            for i, plugin in enumerate(self.plugins[pluginType.name]):
                if plugin.meta.id == pluginId:
                    del self.plugins[pluginType.name][i]
                    for j, x in enumerate(self.core.TM.willExecutePlugins):
                        if x.meta.id == pluginId:
                            del self.core.TM.willExecutePlugins[j]
                    # TODO: kill the thread

                    logger.info(
                        f"[Unloaded] Plugin \"{plugin.meta.name}\"({plugin.meta.path}) "
                        f"has been unloaded successfully."
                    )
                    return True
        logger.warning(
            f"Plugin \"{plugin.meta.name}\"({plugin.meta.path}) "
            f"could not be unloaded because it is not loaded."
        )
        return False

    def loadPluginsFromDir(self) -> bool:
        self.unloadPlugins()

        for pluginFile in os.listdir(self.core.config.directory.plugins):
            pluginPath: str = f"{self.core.config.directory.plugins}/{pluginFile}"
            if pluginFilePattern.match(pluginPath):
                self.loadPlugin(pluginPath)

        return True<commit_msg>Better implement: listdir to scandir<commit_after># coding=utf-8
import json
import logging
import os
import traceback
from typing import Dict, List

from . import Plugin
from .utils import getPluginId, pluginFilePattern
from ..enums import PluginType, API
from ..exceptions.plugin import InvalidPluginSyntaxError

logger = logging.getLogger(__name__)

class PluginManager:
    def __init__(self, core):
        self.core = core
        self.plugins: Dict[str, List[Plugin]] = None
        self.unloadPlugins()

    def unloadPlugins(self) -> None:
        # noinspection PyTypeChecker
        self.plugins: Dict[str, List[Plugin]] = {x.name: [] for x in PluginType}

    def loadPlugin(self, path: str) -> bool:
        plugin: Plugin = Plugin(path=path)
        try:
            plugin.load()
        except InvalidPluginSyntaxError:
            logger.warning(
                f"Plugin {path} "
                f"could not be loaded because it has invalid syntax. "
                f"Error Detail:\n{traceback.format_exc()}."
            )
            return False

        found: bool = False
        # noinspection PyTypeChecker
        for pluginType in PluginType:
            for i, loadedPlugin in enumerate(self.plugins[pluginType.name]):
                if plugin.meta.id == loadedPlugin.meta.id:
                    found = True
                    self.plugins[pluginType.name][i] = plugin

                    if PluginType.Thread == pluginType:
                        pass
                        # TODO: kill the thread

                    break
            if found:
                break
        else:
            self.plugins[plugin.meta.type.name].append(plugin)

        # noinspection PyTypeChecker
        self.plugins = {
            pluginType.name: sorted(
                    self.plugins[pluginType.name],
                    key=lambda x: x.meta.priority, reverse=True
            ) for pluginType in PluginType}

        with open(f"{self.core.config.directory.api}/{API.Plugins.value}", "w") as f:
            # noinspection PyTypeChecker
            json.dump([
                plugin.meta.__dict__
                for pluginType in PluginType for plugin in self.plugins[pluginType.name]
            ], f, sort_keys=True, indent=4)
        return True

    def unloadPlugin(self, path: str) -> bool:
        pluginId: str = getPluginId(path)

        # noinspection PyTypeChecker
        for pluginType in PluginType:
            for i, plugin in enumerate(self.plugins[pluginType.name]):
                if plugin.meta.id == pluginId:
                    del self.plugins[pluginType.name][i]
                    for j, x in enumerate(self.core.TM.willExecutePlugins):
                        if x.meta.id == pluginId:
                            del self.core.TM.willExecutePlugins[j]
                    # TODO: kill the thread

                    logger.info(
                        f"[Unloaded] Plugin \"{plugin.meta.name}\"({plugin.meta.path}) "
                        f"has been unloaded successfully."
                    )
                    return True
        logger.warning(
            f"Plugin \"{plugin.meta.name}\"({plugin.meta.path}) "
            f"could not be unloaded because it is not loaded."
        )
        return False

    def loadPluginsFromDir(self) -> bool:
        self.unloadPlugins()

        with os.scandir(self.core.config.directory.plugins) as it:
            for entry in it:
                if pluginFilePattern.match(entry.name) and entry.is_file():
                    self.loadPlugin(f"{self.core.config.directory.plugins}/{entry.name}")

        return True
{'fraction_non_alphanumeric': 0.0912227435826663, 'fraction_numerical': 0.0005520287054926856, 'mean_word_length': 1.6781226903178124, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '773214', 'n_tokens_mistral': 1964, 'n_tokens_neox': 1860, 'n_words': 478}