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 +- 6 files changed, 1 insertion(+), 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 (limited to 'app') 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 -- cgit v1.3