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 --- plugins/ConcatPlugin.py | 50 ++++++++++++++++++++++++++ plugins/DigestPlugin.py | 0 plugins/FeedSinkPlugin.py | 86 +++++++++++++++++++++++++++++++++++++++++++++ plugins/FeedSourcePlugin.py | 58 ++++++++++++++++++++++++++++++ plugins/FilterPlugin.py | 20 +++++++++++ plugins/MapPlugin.py | 31 ++++++++++++++++ 6 files changed, 245 insertions(+) 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 (limited to 'plugins') 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