diff options
| -rw-r--r-- | aggro.py | 45 | ||||
| -rw-r--r-- | app/ConcatPlugin.py | 44 | ||||
| -rw-r--r-- | app/DatabaseManager.py (renamed from app/database.py) | 3 | ||||
| -rw-r--r-- | app/FeedSinkPlugin.py | 4 | ||||
| -rw-r--r-- | app/server.py | 4 | ||||
| -rw-r--r-- | app/utils.py | 11 |
6 files changed, 100 insertions, 11 deletions
@@ -1,13 +1,17 @@ import json import os +import hashlib # from multiprocessing import Process from threading import Thread import time +from typing import Any + +from tinydb import Query from app.AggroConfig import AggroConfig from app.MemoryState import memory_state from app.PluginManager import PluginManager -from app.database import database_manager +from app.DatabaseManager import database_manager from app.server import run_web_server @@ -29,18 +33,45 @@ if __name__ == "__main__": aggrofile_path = os.environ.get("AGGRO_CONFIG_PATH", "Aggrofile") with open(aggrofile_path) as f: - aggrofile_content = json.loads(f.read()) + aggrofile_content = f.read() + aggrofile = json.loads(aggrofile_content) aggro_config = AggroConfig( - server_host=aggrofile_content.get("server_host", "localhost"), - server_port=aggrofile_content.get("server_port", 8080), - db_path=aggrofile_content.get("db_path", "db.json"), - plugins=aggrofile_content["plugins"], - graph=aggrofile_content["graph"], + server_host=aggrofile.get("server_host", "localhost"), + server_port=aggrofile.get("server_port", 8080), + db_path=aggrofile.get("db_path", "db.json"), + plugins=aggrofile["plugins"], + graph=aggrofile["graph"], ) database_manager.setup(aggro_config) + aggrofile_hash_q = Query().key == "aggrofile_hash" + aggrofile_stored_hash: str | None + if database_manager.meta_info.contains(aggrofile_hash_q): + aggrofile_stored_hash_obj: Any = database_manager.meta_info.get( # type: ignore + aggrofile_hash_q + ) + aggrofile_stored_hash = aggrofile_stored_hash_obj["value"] + else: + aggrofile_stored_hash = None + + aggrofile_current_hash = hashlib.sha256(aggrofile_content.encode("utf-8")) + aggrofile_current_hash_dig = aggrofile_current_hash.hexdigest() + + if ( + aggrofile_stored_hash is not None + and aggrofile_stored_hash != aggrofile_current_hash_dig + ): + # Aggrofile has been changed + print("Aggrofile has been changed. Truncating plugin states in DB...") + database_manager.plugin_states.truncate() + + database_manager.meta_info.upsert( # type: ignore + {"key": "aggrofile_hash", "value": aggrofile_current_hash_dig}, + aggrofile_hash_q, + ) + manager = PluginManager(aggro_config) manager.build_plugin_instances() diff --git a/app/ConcatPlugin.py b/app/ConcatPlugin.py new file mode 100644 index 0000000..14307ac --- /dev/null +++ b/app/ConcatPlugin.py @@ -0,0 +1,44 @@ +from typing import Any + +from tinydb import Query +from app.Item import Item +from app.PluginInterface import PluginInterface +from app.DatabaseManager import database_manager +from app.utils import dict_to_item, item_to_dict + +# TODO + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: dict[str, Any]) -> None: + super().__init__(id, params) + print(f"[ConcatPlugin#{self.id}] initialized") + + def process(self, source_id: str | None, items: list[Item]) -> list[Item]: + print(f"[ConcatPlugin#{self.id}] process called, n={len(items)}") + if source_id is None: + raise Exception(f"[ConcatPlugin#{self.id}] can not be scheduled") + + Q = Query() + q_result = database_manager.plugin_states.search(Q.plugin_id == self.id) + if len(q_result) > 1: + raise Exception( + f"[ConcatPlugin#{self.id}] invalid database state, found more than 1 plugin state: {len(q_result)}" + ) + + d: Any = q_result[0] if len(q_result) == 1 else {} + data: dict[str, list[dict[str, str]]] = d + + if source_id in data: + items_as_dicts = list(map(item_to_dict, items)) + d[source_id] = items_as_dicts + + aggregated_item_dicts: list[dict[str, str]] = [] + for data_source_id in data: + item_dicts = data[data_source_id] + aggregated_item_dicts += item_dicts + + ret = list(map(dict_to_item, aggregated_item_dicts)) + + print(f"[ConcatPlugin#{self.id}] process returning items, n={len(ret)}") + return ret diff --git a/app/database.py b/app/DatabaseManager.py index b796dc8..0ef3b49 100644 --- a/app/database.py +++ b/app/DatabaseManager.py @@ -9,6 +9,9 @@ class DatabaseManager: def setup(self, config: AggroConfig): self.db = TinyDB(config.db_path) + self.plugin_states = self.db.table("plugin_states") # type: ignore + self.feeds = self.db.table("feeds") # type: ignore + self.meta_info = self.db.table("meta_info") # type: ignore database_manager: DatabaseManager = DatabaseManager() diff --git a/app/FeedSinkPlugin.py b/app/FeedSinkPlugin.py index 2c85167..237214f 100644 --- a/app/FeedSinkPlugin.py +++ b/app/FeedSinkPlugin.py @@ -6,7 +6,7 @@ from tinydb import Query from app.Item import Item from app.PluginInterface import PluginInterface from app.utils import get_param -from app.database import database_manager +from app.DatabaseManager import database_manager class Plugin(PluginInterface): @@ -52,6 +52,6 @@ class Plugin(PluginInterface): ret = self.build_xml(items) Q = Query() - database_manager.db.upsert({"feed_id": self.feed_id, "feed_xml": ret}, Q.feed_id == self.feed_id) # type: ignore + database_manager.feeds.upsert({"feed_id": self.feed_id, "feed_xml": ret}, Q.feed_id == self.feed_id) # type: ignore print(f"[FeedSinkPlugin#{self.id}] processed") return [] diff --git a/app/server.py b/app/server.py index 785e30f..5de3018 100644 --- a/app/server.py +++ b/app/server.py @@ -2,7 +2,7 @@ from typing import Any import bottle as _bottle # type: ignore from tinydb import Query -from app.database import database_manager +from app.DatabaseManager import database_manager bottle: Any = _bottle @@ -13,7 +13,7 @@ def index(feed_id: str): raise Exception("Database is not initialized") Q = Query() - res = database_manager.db.search(Q.feed_id == feed_id) + res = database_manager.feeds.search(Q.feed_id == feed_id) if len(res) == 0: bottle.abort(400, f"No feed found with id {feed_id}") diff --git a/app/utils.py b/app/utils.py index ac07bf7..d513ae9 100644 --- a/app/utils.py +++ b/app/utils.py @@ -1,5 +1,16 @@ +from app.Item import Item + + 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] + + +def item_to_dict(item: Item) -> dict[str, str]: + raise NotImplemented() + + +def dict_to_item(d: dict[str, str]) -> Item: + raise NotImplemented() |