From 7f2e5ee5bf59e5814d62990401714e01cd69ba07 Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Sun, 3 Sep 2023 19:07:30 +0300 Subject: Refactor plugins to own dir --- app/ConcatPlugin.py | 50 -------------------------- app/FeedSinkPlugin.py | 86 --------------------------------------------- app/FeedSourcePlugin.py | 58 ------------------------------ app/FilterPlugin.py | 20 ----------- app/MapPlugin.py | 33 ----------------- app/PluginManager.py | 2 +- plugins/ConcatPlugin.py | 50 ++++++++++++++++++++++++++ plugins/DigestPlugin.py | 0 plugins/FeedSinkPlugin.py | 86 +++++++++++++++++++++++++++++++++++++++++++++ plugins/FeedSourcePlugin.py | 58 ++++++++++++++++++++++++++++++ plugins/FilterPlugin.py | 20 +++++++++++ plugins/MapPlugin.py | 31 ++++++++++++++++ 12 files changed, 246 insertions(+), 248 deletions(-) delete mode 100644 app/ConcatPlugin.py delete mode 100644 app/FeedSinkPlugin.py delete mode 100644 app/FeedSourcePlugin.py delete mode 100644 app/FilterPlugin.py delete mode 100644 app/MapPlugin.py create mode 100644 plugins/ConcatPlugin.py create mode 100644 plugins/DigestPlugin.py create mode 100644 plugins/FeedSinkPlugin.py create mode 100644 plugins/FeedSourcePlugin.py create mode 100644 plugins/FilterPlugin.py create mode 100644 plugins/MapPlugin.py diff --git a/app/ConcatPlugin.py b/app/ConcatPlugin.py deleted file mode 100644 index 43d7db2..0000000 --- a/app/ConcatPlugin.py +++ /dev/null @@ -1,50 +0,0 @@ -from typing import Any -import time - -from tinydb import Query -from app.Item import Item -from app.PluginInterface import Params, PluginInterface -from app.DatabaseManager import database_manager -from app.utils import ItemDict, dict_to_item, item_to_dict - - -class Plugin(PluginInterface): - def __init__(self, id: str, params: Params) -> None: - super().__init__(id, params) - print(f"[ConcatPlugin#{self.id}] initialized") - - def item_sort_key(self, item: Item) -> time.struct_time: - if item.pub_date is None: - return time.localtime() - - return item.pub_date - - 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") - - plugin_state_q = Query().plugin_id == self.id - _d: Any = database_manager.plugin_states.get(plugin_state_q) # type: ignore - data: dict[str, Any] = ( - _d if _d is not None else {"plugin_id": self.id, "state": {}} - ) - state = data["state"] - - items_as_dicts = list(map(item_to_dict, items)) - state[source_id] = items_as_dicts - - database_manager.plugin_states.upsert( # type: ignore - {"plugin_id": self.id, "state": state}, plugin_state_q - ) - - aggregated_item_dicts: list[ItemDict] = [] - for data_source_id in state: - item_dicts = state[data_source_id] - aggregated_item_dicts += item_dicts - - ret_items = list(map(dict_to_item, aggregated_item_dicts)) - ret = sorted(ret_items, key=self.item_sort_key) - - print(f"[ConcatPlugin#{self.id}] process returning items, n={len(ret)}") - return ret diff --git a/app/FeedSinkPlugin.py b/app/FeedSinkPlugin.py deleted file mode 100644 index c0fc4c9..0000000 --- a/app/FeedSinkPlugin.py +++ /dev/null @@ -1,86 +0,0 @@ -import time -from typing import Any -import xml.etree.ElementTree as ET - -from tinydb import Query - -from app.Item import Item -from app.PluginInterface import PluginInterface -from app.utils import get_param -from app.DatabaseManager import database_manager - - -class Plugin(PluginInterface): - def __init__(self, id: str, params: dict[str, Any]) -> None: - super().__init__(id, params) - self.feed_id: str = get_param("feed_id", params) - self.feed_title: str = get_param("feed_title", params) - self.feed_link: str | None = params.get("feed_link", None) - self.feed_description: str = get_param("feed_description", params) - print(f"[FeedSinkPlugin#{self.id}] initialized") - print(f"[FeedSinkPlugin#{self.id}] feed will be served at path /{self.feed_id}") - - def build_xml(self, items: list[Item]): - rss = ET.Element("rss", {"version": "2.0"}) - channel = ET.SubElement(rss, "channel") - title = ET.SubElement(channel, "title") - title.text = self.feed_title - if self.feed_link is not None: - link = ET.SubElement(channel, "link") - link.text = self.feed_link - description = ET.SubElement(channel, "description") - description.text = self.feed_description - - for item in items: - item_elem = ET.SubElement(channel, "item") - if item.title is not None: - item_title = ET.SubElement(item_elem, "title") - item_title.text = item.title - if item.link is not None: - item_link = ET.SubElement(item_elem, "link") - item_link.text = item.link - if item.description is not None: - item_description = ET.SubElement(item_elem, "description") - item_description.text = item.description - if item.category is not None: - item_category = ET.SubElement(item_elem, "category") - item_category.text = item.category - if item.comments is not None: - item_comments = ET.SubElement(item_elem, "comments") - item_comments.text = item.comments - if item.pub_date is not None: - item_pub_date = ET.SubElement(item_elem, "pubDate") - item_pub_date.text = time.strftime( - "%a, %d %b %Y %H:%M:%S +0000", item.pub_date - ) - if item.author is not None: - item_author = ET.SubElement(item_elem, "author") - item_author.text = item.author - - for enc in item.enclosures: - ET.SubElement( - item_elem, - "enclosure", - { - "length": enc.length, - "type": enc.type, - "url": enc.url, - }, - ) - - return ET.tostring(rss, "unicode") - - 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") - if database_manager.db is None: - raise Exception("Database is not initialized") - - ret = self.build_xml(items) - - Q = Query() - 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/FeedSourcePlugin.py b/app/FeedSourcePlugin.py deleted file mode 100644 index 575e5d6..0000000 --- a/app/FeedSourcePlugin.py +++ /dev/null @@ -1,58 +0,0 @@ -import feedparser # type: ignore -from typing import Any -from app.Item import Item, ItemEnclosure -from app.PluginInterface import Params, PluginInterface -from app.utils import ItemDict, get_param - - -class Plugin(PluginInterface): - def __init__(self, id: str, params: Params) -> None: - super().__init__(id, params) - self.feed_url: str = get_param("feed_url", params) - - print(f"[FeedSourcePlugin#{self.id}] initialized") - - 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 ItemSource {source_id}" - ) - - feed: Any = feedparser.parse(self.feed_url) # type: ignore - result_items: list[Item] = [] - if "bozo" in feed and feed["bozo"] == 1: - raise Exception( - f"[FeedSourcePlugin#{self.id}] malformed XML in feed {self.feed_url}" - ) - - for _d in feed.entries: - d: ItemDict = _d - item_enclosures: list[ItemEnclosure] = [] - for _e in d["enclosures"]: - e: dict[str, str] = _e # type: ignore - item_enclosures.append( - ItemEnclosure( - length=e["length"], - type=e["type"], - url=e["href"], - ) - ) - - result_items.append( - Item( - title=d.get("title", None), # type: ignore - link=d.get("link", None), # type: ignore - description=d.get("description", None), # type: ignore - author=d.get("author", None), # type: ignore - pub_date=d.get("published_parsed", None), # type: ignore - category=d.get("category", None), # type: ignore - comments=d.get("comments"), # type: ignore - enclosures=item_enclosures, - ) - ) - - 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 deleted file mode 100644 index 0f2b500..0000000 --- a/app/FilterPlugin.py +++ /dev/null @@ -1,20 +0,0 @@ -from app.Item import Item -from app.PluginInterface import Params, PluginInterface -from app.utils import get_param - - -class Plugin(PluginInterface): - def __init__(self, id: str, params: Params) -> None: - super().__init__(id, params) - self.filter_expr: str = get_param("filter_expr", params) - print(f"[FilterPlugin#{self.id}] initialized") - - 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)}") - return ret diff --git a/app/MapPlugin.py b/app/MapPlugin.py deleted file mode 100644 index abc016b..0000000 --- a/app/MapPlugin.py +++ /dev/null @@ -1,33 +0,0 @@ -from dataclasses import fields -from typing import Any -from app.Item import Item -from app.PluginInterface import Params, PluginInterface -from app.utils import get_param - - -def set_field(item: Item, k: str, v: Any): - field_names = [f.name for f in fields(Item)] - if k not in field_names: - raise Exception(f"Invalid set_field key: {k}") - - setattr(item, k, v) - return item - - -class Plugin(PluginInterface): - def __init__(self, id: str, params: Params) -> None: - super().__init__(id, params) - self.map_expr: str = get_param("map_expr", params) - print(f"[MapItemPlugin#{self.id}] initialized") - - def process(self, source_id: str | None, items: list[Item]) -> list[Item]: - print(f"[MapItemPlugin#{self.id}] process called, n={len(items)}") - if source_id is None: - raise Exception(f"[MapItemPlugin#{self.id}] can not be scheduled") - - expr = f"map(lambda item: {self.map_expr}, input_feed)" - print("expr:") - print(expr) - ret = list(eval(expr, {"input_feed": items, "set_field": set_field})) - print(f"[MapItemPlugin#{self.id}] process returning items, n={len(ret)}") - return ret diff --git a/app/PluginManager.py b/app/PluginManager.py index f50196e..68149fd 100644 --- a/app/PluginManager.py +++ b/app/PluginManager.py @@ -17,7 +17,7 @@ class PluginManager: self.config = config def load_plugin(self, plugin_name: str) -> None: - module: ModuleType = importlib.import_module(f"app.{plugin_name}") + module: ModuleType = importlib.import_module(f"plugins.{plugin_name}") plugin_class = module.Plugin self._plugins[plugin_name] = plugin_class diff --git a/plugins/ConcatPlugin.py b/plugins/ConcatPlugin.py new file mode 100644 index 0000000..43d7db2 --- /dev/null +++ b/plugins/ConcatPlugin.py @@ -0,0 +1,50 @@ +from typing import Any +import time + +from tinydb import Query +from app.Item import Item +from app.PluginInterface import Params, PluginInterface +from app.DatabaseManager import database_manager +from app.utils import ItemDict, dict_to_item, item_to_dict + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: Params) -> None: + super().__init__(id, params) + print(f"[ConcatPlugin#{self.id}] initialized") + + def item_sort_key(self, item: Item) -> time.struct_time: + if item.pub_date is None: + return time.localtime() + + return item.pub_date + + 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") + + plugin_state_q = Query().plugin_id == self.id + _d: Any = database_manager.plugin_states.get(plugin_state_q) # type: ignore + data: dict[str, Any] = ( + _d if _d is not None else {"plugin_id": self.id, "state": {}} + ) + state = data["state"] + + items_as_dicts = list(map(item_to_dict, items)) + state[source_id] = items_as_dicts + + database_manager.plugin_states.upsert( # type: ignore + {"plugin_id": self.id, "state": state}, plugin_state_q + ) + + aggregated_item_dicts: list[ItemDict] = [] + for data_source_id in state: + item_dicts = state[data_source_id] + aggregated_item_dicts += item_dicts + + ret_items = list(map(dict_to_item, aggregated_item_dicts)) + ret = sorted(ret_items, key=self.item_sort_key) + + print(f"[ConcatPlugin#{self.id}] process returning items, n={len(ret)}") + return ret diff --git a/plugins/DigestPlugin.py b/plugins/DigestPlugin.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/FeedSinkPlugin.py b/plugins/FeedSinkPlugin.py new file mode 100644 index 0000000..c0fc4c9 --- /dev/null +++ b/plugins/FeedSinkPlugin.py @@ -0,0 +1,86 @@ +import time +from typing import Any +import xml.etree.ElementTree as ET + +from tinydb import Query + +from app.Item import Item +from app.PluginInterface import PluginInterface +from app.utils import get_param +from app.DatabaseManager import database_manager + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: dict[str, Any]) -> None: + super().__init__(id, params) + self.feed_id: str = get_param("feed_id", params) + self.feed_title: str = get_param("feed_title", params) + self.feed_link: str | None = params.get("feed_link", None) + self.feed_description: str = get_param("feed_description", params) + print(f"[FeedSinkPlugin#{self.id}] initialized") + print(f"[FeedSinkPlugin#{self.id}] feed will be served at path /{self.feed_id}") + + def build_xml(self, items: list[Item]): + rss = ET.Element("rss", {"version": "2.0"}) + channel = ET.SubElement(rss, "channel") + title = ET.SubElement(channel, "title") + title.text = self.feed_title + if self.feed_link is not None: + link = ET.SubElement(channel, "link") + link.text = self.feed_link + description = ET.SubElement(channel, "description") + description.text = self.feed_description + + for item in items: + item_elem = ET.SubElement(channel, "item") + if item.title is not None: + item_title = ET.SubElement(item_elem, "title") + item_title.text = item.title + if item.link is not None: + item_link = ET.SubElement(item_elem, "link") + item_link.text = item.link + if item.description is not None: + item_description = ET.SubElement(item_elem, "description") + item_description.text = item.description + if item.category is not None: + item_category = ET.SubElement(item_elem, "category") + item_category.text = item.category + if item.comments is not None: + item_comments = ET.SubElement(item_elem, "comments") + item_comments.text = item.comments + if item.pub_date is not None: + item_pub_date = ET.SubElement(item_elem, "pubDate") + item_pub_date.text = time.strftime( + "%a, %d %b %Y %H:%M:%S +0000", item.pub_date + ) + if item.author is not None: + item_author = ET.SubElement(item_elem, "author") + item_author.text = item.author + + for enc in item.enclosures: + ET.SubElement( + item_elem, + "enclosure", + { + "length": enc.length, + "type": enc.type, + "url": enc.url, + }, + ) + + return ET.tostring(rss, "unicode") + + 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") + if database_manager.db is None: + raise Exception("Database is not initialized") + + ret = self.build_xml(items) + + Q = Query() + 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/plugins/FeedSourcePlugin.py b/plugins/FeedSourcePlugin.py new file mode 100644 index 0000000..575e5d6 --- /dev/null +++ b/plugins/FeedSourcePlugin.py @@ -0,0 +1,58 @@ +import feedparser # type: ignore +from typing import Any +from app.Item import Item, ItemEnclosure +from app.PluginInterface import Params, PluginInterface +from app.utils import ItemDict, get_param + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: Params) -> None: + super().__init__(id, params) + self.feed_url: str = get_param("feed_url", params) + + print(f"[FeedSourcePlugin#{self.id}] initialized") + + 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 ItemSource {source_id}" + ) + + feed: Any = feedparser.parse(self.feed_url) # type: ignore + result_items: list[Item] = [] + if "bozo" in feed and feed["bozo"] == 1: + raise Exception( + f"[FeedSourcePlugin#{self.id}] malformed XML in feed {self.feed_url}" + ) + + for _d in feed.entries: + d: ItemDict = _d + item_enclosures: list[ItemEnclosure] = [] + for _e in d["enclosures"]: + e: dict[str, str] = _e # type: ignore + item_enclosures.append( + ItemEnclosure( + length=e["length"], + type=e["type"], + url=e["href"], + ) + ) + + result_items.append( + Item( + title=d.get("title", None), # type: ignore + link=d.get("link", None), # type: ignore + description=d.get("description", None), # type: ignore + author=d.get("author", None), # type: ignore + pub_date=d.get("published_parsed", None), # type: ignore + category=d.get("category", None), # type: ignore + comments=d.get("comments"), # type: ignore + enclosures=item_enclosures, + ) + ) + + print( + f"[FeedSourcePlugin#{self.id}] process returns items, n={len(result_items)}" + ) + return result_items diff --git a/plugins/FilterPlugin.py b/plugins/FilterPlugin.py new file mode 100644 index 0000000..0f2b500 --- /dev/null +++ b/plugins/FilterPlugin.py @@ -0,0 +1,20 @@ +from app.Item import Item +from app.PluginInterface import Params, PluginInterface +from app.utils import get_param + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: Params) -> None: + super().__init__(id, params) + self.filter_expr: str = get_param("filter_expr", params) + print(f"[FilterPlugin#{self.id}] initialized") + + 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)}") + return ret diff --git a/plugins/MapPlugin.py b/plugins/MapPlugin.py new file mode 100644 index 0000000..8b5dae7 --- /dev/null +++ b/plugins/MapPlugin.py @@ -0,0 +1,31 @@ +from dataclasses import fields +from typing import Any +from app.Item import Item +from app.PluginInterface import Params, PluginInterface +from app.utils import get_param + + +def set_field(item: Item, k: str, v: Any): + field_names = [f.name for f in fields(Item)] + if k not in field_names: + raise Exception(f"Invalid set_field key: {k}") + + setattr(item, k, v) + return item + + +class Plugin(PluginInterface): + def __init__(self, id: str, params: Params) -> None: + super().__init__(id, params) + self.map_expr: str = get_param("map_expr", params) + print(f"[MapItemPlugin#{self.id}] initialized") + + def process(self, source_id: str | None, items: list[Item]) -> list[Item]: + print(f"[MapItemPlugin#{self.id}] process called, n={len(items)}") + if source_id is None: + raise Exception(f"[MapItemPlugin#{self.id}] can not be scheduled") + + expr = f"map(lambda item: {self.map_expr}, input_feed)" + ret = list(eval(expr, {"input_feed": items, "set_field": set_field})) + print(f"[MapItemPlugin#{self.id}] process returning items, n={len(ret)}") + return ret -- cgit v1.3