diff options
Diffstat (limited to 'app')
| -rw-r--r-- | app/AggroConfig.py | 1 | ||||
| -rw-r--r-- | app/FeedSinkPlugin.py | 5 | ||||
| -rw-r--r-- | app/FeedSourcePlugin.py | 11 | ||||
| -rw-r--r-- | app/FilterPlugin.py | 5 | ||||
| -rw-r--r-- | app/PluginInterface.py | 2 | ||||
| -rw-r--r-- | app/PluginManager.py | 22 | ||||
| -rw-r--r-- | app/database.py | 16 |
7 files changed, 42 insertions, 20 deletions
diff --git a/app/AggroConfig.py b/app/AggroConfig.py index e933eec..6ff6801 100644 --- a/app/AggroConfig.py +++ b/app/AggroConfig.py @@ -3,5 +3,6 @@ from dataclasses import dataclass @dataclass class AggroConfig: + db_path: str plugins: dict[str, dict[str, str]] graph: dict[str, list[str]] diff --git a/app/FeedSinkPlugin.py b/app/FeedSinkPlugin.py index c35a811..6b0125d 100644 --- a/app/FeedSinkPlugin.py +++ b/app/FeedSinkPlugin.py @@ -54,8 +54,11 @@ class Plugin(PluginInterface): return ET.tostring(rss, "unicode") - def process(self, items: list[Item]) -> list[Item]: + def process(self, source_id: str | None, items: list[Item]) -> list[Item]: print(f"[FeedSinkPlugin#{self.id}] process called, n={len(items)}") + if source_id is None: + raise Exception(f"FeedSinkPlugin#{self.id} can not be scheduled") + ret = self.build_xml(items) print(f"[FeedSinkPlugin#{self.id}] process returning XML:") print(ret) diff --git a/app/FeedSourcePlugin.py b/app/FeedSourcePlugin.py index 5453f80..7690c8d 100644 --- a/app/FeedSourcePlugin.py +++ b/app/FeedSourcePlugin.py @@ -12,8 +12,13 @@ class Plugin(PluginInterface): print(f"[FeedSourcePlugin#{self.id}] initialized") - def process(self, items: list[Item]) -> list[Item]: + def process(self, source_id: str | None, items: list[Item]) -> list[Item]: print(f"[FeedSourcePlugin#{self.id}] process called") + if source_id is not None: + raise Exception( + f"FeedSourcePlugin#{self.id} can only be scheduled, trying to process items from source {source_id}" + ) + feed: Any = feedparser.parse(self.feed_url) # type: ignore result_items: list[Item] = [] if "bozo" in feed and feed["bozo"] == 1: @@ -26,5 +31,7 @@ class Plugin(PluginInterface): Item(title=d["title"], link=d["link"], description=d["description"]) ) - print(f"[FeedSourcePlugin#{self.id}] process returns items, n={len(items)}") + print( + f"[FeedSourcePlugin#{self.id}] process returns items, n={len(result_items)}" + ) return result_items diff --git a/app/FilterPlugin.py b/app/FilterPlugin.py index 735540c..4e6ee3b 100644 --- a/app/FilterPlugin.py +++ b/app/FilterPlugin.py @@ -10,8 +10,11 @@ class Plugin(PluginInterface): self.filter_expr: str = get_param("filter_expr", params) print(f"[FilterPlugin#{self.id}] initialized") - def process(self, items: list[Item]) -> list[Item]: + def process(self, source_id: str | None, items: list[Item]) -> list[Item]: print(f"[FilterPlugin#{self.id}] process called, n={len(items)}") + if source_id is None: + raise Exception(f"[FilterPlugin#{self.id}] can not be scheduled") + expr = f"filter(lambda item: {self.filter_expr}, input_feed)" ret = list(eval(expr, {"input_feed": items})) print(f"[FilterPlugin#{self.id}] process returning items, n={len(ret)}") diff --git a/app/PluginInterface.py b/app/PluginInterface.py index f334034..6492ed1 100644 --- a/app/PluginInterface.py +++ b/app/PluginInterface.py @@ -10,5 +10,5 @@ class PluginInterface(ABC): self.params = params @abstractmethod - def process(self, items: list[Item]) -> list[Item]: + def process(self, source_id: str | None, items: list[Item]) -> list[Item]: pass diff --git a/app/PluginManager.py b/app/PluginManager.py index 3ec9d44..da85428 100644 --- a/app/PluginManager.py +++ b/app/PluginManager.py @@ -21,8 +21,6 @@ class PluginManager: plugin_class = module.Plugin self._plugins[plugin_name] = plugin_class - # TODO instead of propagating procedurally, - # insert data into an next_node_id-identified input queue in tinydb def propagate(self, id: str, items: list[Item]): next_nodes: list[str] if id not in self.config.graph: @@ -31,14 +29,14 @@ class PluginManager: next_nodes = self.config.graph[id] for next_node_id in next_nodes: - self.run_plugin_job(next_node_id, items) + self.run_plugin_job(next_node_id, id, items) - def run_plugin_job(self, id: str, items: list[Item] = []): + def run_plugin_job(self, id: str, source_id: str | None, items: list[Item] = []): if not memory_state.running: return plugin: PluginInterface = self.plugin_instances[id] - ret_items: list[Item] = plugin.process(items) + ret_items: list[Item] = plugin.process(source_id, items) self.propagate(id, ret_items) def build_plugin_instances(self): @@ -47,20 +45,14 @@ class PluginManager: params: dict[str, str] = self.config.plugins[id] plugin_name = get_param("plugin", params) - trigger_type = get_param("trigger_type", params) + schedule_expr: str | None = params.get("schedule_expr", None) 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) + if schedule_expr is not None: + job: schedule.Job = eval(schedule_expr, {"schedule": schedule}) + job.do(self.run_plugin_job, id, None) # type: ignore PluginClass: Any = self._plugins[plugin_name] plugin: PluginInterface = PluginClass(id=id, params=params) diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..fdeac1a --- /dev/null +++ b/app/database.py @@ -0,0 +1,16 @@ +from tinydb import TinyDB + +from app.AggroConfig import AggroConfig + + +def setup_db(config: AggroConfig): + global database_manager + database_manager = DatabaseManager(config) + + +class DatabaseManager: + def __init__(self, config: AggroConfig): + self.db = TinyDB(config.db_path) + + +database_manager: DatabaseManager |