From 398799de0c1826c3544eb6eec681fa48096932a0 Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Fri, 1 Sep 2023 19:31:47 +0300 Subject: Refactor --- aggro.py | 25 +++++++++++++++++ app/AggroConfig.py | 7 +++++ app/FeedSourcePlugin.py | 28 +++++++++++++++++++ app/FilterPlugin.py | 22 +++++++++++++++ app/Item.py | 10 +++++++ app/PluginInterface.py | 18 +++++++++++++ app/PluginManager.py | 71 +++++++++++++++++++++++++++++++++++++++++++++++++ app/utils.py | 5 ++++ src/AggroConfig.py | 7 ----- src/FeedSourcePlugin.py | 28 ------------------- src/FilterPlugin.py | 22 --------------- src/Item.py | 10 ------- src/PluginInterface.py | 18 ------------- src/PluginManager.py | 71 ------------------------------------------------- src/aggro.py | 24 ----------------- src/utils.py | 5 ---- 16 files changed, 186 insertions(+), 185 deletions(-) create mode 100644 aggro.py create mode 100644 app/AggroConfig.py create mode 100644 app/FeedSourcePlugin.py create mode 100644 app/FilterPlugin.py create mode 100644 app/Item.py create mode 100644 app/PluginInterface.py create mode 100644 app/PluginManager.py create mode 100644 app/utils.py delete mode 100644 src/AggroConfig.py delete mode 100644 src/FeedSourcePlugin.py delete mode 100644 src/FilterPlugin.py delete mode 100644 src/Item.py delete mode 100644 src/PluginInterface.py delete mode 100644 src/PluginManager.py delete mode 100644 src/aggro.py delete mode 100644 src/utils.py diff --git a/aggro.py b/aggro.py new file mode 100644 index 0000000..8e4f908 --- /dev/null +++ b/aggro.py @@ -0,0 +1,25 @@ +import json +import os + +from app.AggroConfig import AggroConfig +from app.PluginManager import PluginManager + +if __name__ == "__main__": + print("Starting aggro. Press CTRL-C to exit.") + + aggrofile_path = os.environ.get("AGGRO_CONFIG_PATH", "Aggrofile") + + with open(aggrofile_path) as f: + aggrofile_content = json.loads(f.read()) + + aggro_config = AggroConfig( + plugins=aggrofile_content["plugins"], graph=aggrofile_content["graph"] + ) + + manager = PluginManager(aggro_config) + manager.build_plugin_instances() + + try: + manager.run() + except KeyboardInterrupt: + print("Exiting...") diff --git a/app/AggroConfig.py b/app/AggroConfig.py new file mode 100644 index 0000000..e933eec --- /dev/null +++ b/app/AggroConfig.py @@ -0,0 +1,7 @@ +from dataclasses import dataclass + + +@dataclass +class AggroConfig: + plugins: dict[str, dict[str, str]] + graph: dict[str, list[str]] diff --git a/app/FeedSourcePlugin.py b/app/FeedSourcePlugin.py new file mode 100644 index 0000000..af686b9 --- /dev/null +++ b/app/FeedSourcePlugin.py @@ -0,0 +1,28 @@ +import feedparser # type: ignore +from typing import Any +from app.Item import Item +from app.PluginInterface import PluginInterface +from app.utils import get_param + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: dict[str, Any]) -> None: + super().__init__(id, params) + self.feed_url: str = get_param("feed_url", params) + + print(f"[FeedSourcePlugin#{self.id}] initialized") + + def validate_input_n(self, n: int) -> bool: + return n == 0 + + def process(self, inputs: list[list[Item]]) -> list[Item]: + print(f"[FeedSourcePlugin#{self.id}] process called") + feed: Any = feedparser.parse(self.feed_url) # type: ignore + items: list[Item] = [] + for d in feed.entries: + items.append( + Item(title=d["title"], link=d["link"], description=d["description"]) + ) + + print(f"[FeedSourcePlugin#{self.id}] process returns items, n={len(items)}") + return items diff --git a/app/FilterPlugin.py b/app/FilterPlugin.py new file mode 100644 index 0000000..fd3984a --- /dev/null +++ b/app/FilterPlugin.py @@ -0,0 +1,22 @@ +from typing import Any +from app.Item import Item +from app.PluginInterface import PluginInterface +from app.utils import get_param + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: dict[str, Any]) -> None: + super().__init__(id, params) + self.filter_expr: str = get_param("filter_expr", params) + print(f"[FilterPlugin#{self.id}] initialized") + + def validate_input_n(self, n: int) -> bool: + return n == 1 + + def process(self, inputs: list[list[Item]]) -> list[Item]: + input_feed: list[Item] = inputs[0] + print(f"[FilterPlugin#{self.id}] process called, n={len(input_feed)}") + expr = f"filter(lambda item: {self.filter_expr}, input_feed)" + ret = list(eval(expr, {"input_feed": input_feed})) + print(f"[FilterPlugin#{self.id}] process returning items, n={len(ret)}") + return ret diff --git a/app/Item.py b/app/Item.py new file mode 100644 index 0000000..e74239f --- /dev/null +++ b/app/Item.py @@ -0,0 +1,10 @@ +from dataclasses import dataclass + + +@dataclass +class Item: + """RSS Item""" + + title: str + link: str + description: str diff --git a/app/PluginInterface.py b/app/PluginInterface.py new file mode 100644 index 0000000..6194436 --- /dev/null +++ b/app/PluginInterface.py @@ -0,0 +1,18 @@ +from abc import ABC, abstractmethod +from typing import Any + +from app.Item import Item + + +class PluginInterface(ABC): + def __init__(self, id: str, params: dict[str, Any]): + self.id = id + self.params = params + + @abstractmethod + def validate_input_n(self, n: int) -> bool: + raise NotImplemented("abstract method not implemented") + + @abstractmethod + def process(self, inputs: list[list[Item]]) -> list[Item]: + pass diff --git a/app/PluginManager.py b/app/PluginManager.py new file mode 100644 index 0000000..8d60bc1 --- /dev/null +++ b/app/PluginManager.py @@ -0,0 +1,71 @@ +import importlib +import schedule +import time +from types import ModuleType +from typing import Any +from app.Item import Item +from app.PluginInterface import PluginInterface +from app.AggroConfig import AggroConfig +from app.utils import get_param + + +class PluginManager: + def __init__(self, config: AggroConfig) -> None: + self._plugins: dict[str, Any] = {} + self.plugin_instances: dict[str, PluginInterface] = {} + self.running = False + self.config = config + + def load_plugin(self, plugin_name: str) -> None: + module: ModuleType = importlib.import_module(f"app.{plugin_name}") + plugin_class = module.Plugin + self._plugins[plugin_name] = plugin_class + + def propagate(self, id: str, items: list[Item]): + next_nodes: list[str] + if id not in self.config.graph: + next_nodes = [] + else: + next_nodes = self.config.graph[id] + + for next_node_id in next_nodes: + self.run_plugin_job(next_node_id, items) + + def run_plugin_job(self, id: str, items: list[Item] = []): + if not self.running: + return + + plugin: PluginInterface = self.plugin_instances[id] + ret_items: list[Item] = plugin.process([items]) + self.propagate(id, ret_items) + + def build_plugin_instances(self): + self.graph: dict[str, PluginInterface] = {} + for id in self.config.plugins: + params: dict[str, str] = self.config.plugins[id] + + plugin_name = get_param("plugin", params) + trigger_type = get_param("trigger_type", params) + + if plugin_name not in self._plugins: + self.load_plugin(plugin_name) + + match trigger_type: + case "schedule": + schedule_expr = get_param("schedule_expr", params) + job: schedule.Job = eval(schedule_expr, {"schedule": schedule}) + job.do(self.run_plugin_job, id) # type: ignore + case "input_change": + pass + case _: + raise Exception("unknown trigger_type: " + trigger_type) + + PluginClass: Any = self._plugins[plugin_name] + plugin: PluginInterface = PluginClass(id=id, params=params) + self.plugin_instances[id] = plugin + + def run(self) -> None: + self.running = True + while self.running: + schedule.run_pending() + time.sleep(1) diff --git a/app/utils.py b/app/utils.py new file mode 100644 index 0000000..ac07bf7 --- /dev/null +++ b/app/utils.py @@ -0,0 +1,5 @@ +def get_param(key: str, params: dict[str, str]) -> str: + if key not in params: + raise Exception(f"no {key} field in config entry: " + str(params)) + + return params[key] diff --git a/src/AggroConfig.py b/src/AggroConfig.py deleted file mode 100644 index e933eec..0000000 --- a/src/AggroConfig.py +++ /dev/null @@ -1,7 +0,0 @@ -from dataclasses import dataclass - - -@dataclass -class AggroConfig: - plugins: dict[str, dict[str, str]] - graph: dict[str, list[str]] diff --git a/src/FeedSourcePlugin.py b/src/FeedSourcePlugin.py deleted file mode 100644 index e6e4366..0000000 --- a/src/FeedSourcePlugin.py +++ /dev/null @@ -1,28 +0,0 @@ -import feedparser # type: ignore -from typing import Any -from Item import Item -from PluginInterface import PluginInterface -from utils import get_param - - -class Plugin(PluginInterface): - def __init__(self, id: str, params: dict[str, Any]) -> None: - super().__init__(id, params) - self.feed_url: str = get_param("feed_url", params) - - print(f"[FeedSourcePlugin#{self.id}] initialized") - - def validate_input_n(self, n: int) -> bool: - return n == 0 - - def process(self, inputs: list[list[Item]]) -> list[Item]: - print(f"[FeedSourcePlugin#{self.id}] process called") - feed: Any = feedparser.parse(self.feed_url) # type: ignore - items: list[Item] = [] - for d in feed.entries: - items.append( - Item(title=d["title"], link=d["link"], description=d["description"]) - ) - - print(f"[FeedSourcePlugin#{self.id}] process returns items, n={len(items)}") - return items diff --git a/src/FilterPlugin.py b/src/FilterPlugin.py deleted file mode 100644 index 6524d20..0000000 --- a/src/FilterPlugin.py +++ /dev/null @@ -1,22 +0,0 @@ -from typing import Any -from Item import Item -from PluginInterface import PluginInterface -from utils import get_param - - -class Plugin(PluginInterface): - def __init__(self, id: str, params: dict[str, Any]) -> None: - super().__init__(id, params) - self.filter_expr: str = get_param("filter_expr", params) - print(f"[FilterPlugin#{self.id}] initialized") - - def validate_input_n(self, n: int) -> bool: - return n == 1 - - def process(self, inputs: list[list[Item]]) -> list[Item]: - input_feed: list[Item] = inputs[0] - print(f"[FilterPlugin#{self.id}] process called, n={len(input_feed)}") - expr = f"filter(lambda item: {self.filter_expr}, input_feed)" - ret = list(eval(expr, {"input_feed": input_feed})) - print(f"[FilterPlugin#{self.id}] process returning items, n={len(ret)}") - return ret diff --git a/src/Item.py b/src/Item.py deleted file mode 100644 index e74239f..0000000 --- a/src/Item.py +++ /dev/null @@ -1,10 +0,0 @@ -from dataclasses import dataclass - - -@dataclass -class Item: - """RSS Item""" - - title: str - link: str - description: str diff --git a/src/PluginInterface.py b/src/PluginInterface.py deleted file mode 100644 index c6be538..0000000 --- a/src/PluginInterface.py +++ /dev/null @@ -1,18 +0,0 @@ -from abc import ABC, abstractmethod -from typing import Any - -from Item import Item - - -class PluginInterface(ABC): - def __init__(self, id: str, params: dict[str, Any]): - self.id = id - self.params = params - - @abstractmethod - def validate_input_n(self, n: int) -> bool: - raise NotImplemented("abstract method not implemented") - - @abstractmethod - def process(self, inputs: list[list[Item]]) -> list[Item]: - pass diff --git a/src/PluginManager.py b/src/PluginManager.py deleted file mode 100644 index 4a9bd13..0000000 --- a/src/PluginManager.py +++ /dev/null @@ -1,71 +0,0 @@ -import importlib -import schedule -import time -from types import ModuleType -from typing import Any -from AggroConfig import AggroConfig -from Item import Item -from PluginInterface import PluginInterface -from utils import get_param - - -class PluginManager: - def __init__(self, config: AggroConfig) -> None: - self._plugins: dict[str, Any] = {} - self.plugin_instances: dict[str, PluginInterface] = {} - self.running = False - self.config = config - - def load_plugin(self, plugin_name: str) -> None: - module: ModuleType = importlib.import_module(plugin_name) - plugin_class = module.Plugin - self._plugins[plugin_name] = plugin_class - - def propagate(self, id: str, items: list[Item]): - next_nodes: list[str] - if id not in self.config.graph: - next_nodes = [] - else: - next_nodes = self.config.graph[id] - - for next_node_id in next_nodes: - self.run_plugin_job(next_node_id, items) - - def run_plugin_job(self, id: str, items: list[Item] = []): - if not self.running: - return - - plugin: PluginInterface = self.plugin_instances[id] - ret_items: list[Item] = plugin.process([items]) - self.propagate(id, ret_items) - - def build_plugin_instances(self): - self.graph: dict[str, PluginInterface] = {} - for id in self.config.plugins: - params: dict[str, str] = self.config.plugins[id] - - plugin_name = get_param("plugin", params) - trigger_type = get_param("trigger_type", params) - - if plugin_name not in self._plugins: - self.load_plugin(plugin_name) - - match trigger_type: - case "schedule": - schedule_expr = get_param("schedule_expr", params) - job: schedule.Job = eval(schedule_expr, {"schedule": schedule}) - job.do(self.run_plugin_job, id) # type: ignore - case "input_change": - pass - case _: - raise Exception("unknown trigger_type: " + trigger_type) - - PluginClass: Any = self._plugins[plugin_name] - plugin: PluginInterface = PluginClass(id=id, params=params) - self.plugin_instances[id] = plugin - - def run(self) -> None: - self.running = True - while self.running: - schedule.run_pending() - time.sleep(1) diff --git a/src/aggro.py b/src/aggro.py deleted file mode 100644 index 22e8c0b..0000000 --- a/src/aggro.py +++ /dev/null @@ -1,24 +0,0 @@ -import json -from AggroConfig import AggroConfig -from PluginManager import PluginManager -import os - -if __name__ == "__main__": - print("Starting aggro. Press CTRL-C to exit.") - - aggrofile_path = os.environ.get("AGGRO_CONFIG_PATH", "Aggrofile") - - with open(aggrofile_path) as f: - aggrofile_content = json.loads(f.read()) - - aggro_config = AggroConfig( - plugins=aggrofile_content["plugins"], graph=aggrofile_content["graph"] - ) - - manager = PluginManager(aggro_config) - manager.build_plugin_instances() - - try: - manager.run() - except KeyboardInterrupt: - print("Exiting...") diff --git a/src/utils.py b/src/utils.py deleted file mode 100644 index ac07bf7..0000000 --- a/src/utils.py +++ /dev/null @@ -1,5 +0,0 @@ -def get_param(key: str, params: dict[str, str]) -> str: - if key not in params: - raise Exception(f"no {key} field in config entry: " + str(params)) - - return params[key] -- cgit v1.3