aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--app/PluginInterface.py11
-rw-r--r--plugins/ConcatPlugin.py10
-rw-r--r--plugins/DigestPlugin.py12
-rw-r--r--plugins/FacebookSourcePlugin.py15
-rw-r--r--plugins/FeedSinkPlugin.py14
-rw-r--r--plugins/FeedSourcePlugin.py12
-rw-r--r--plugins/FilterPlugin.py10
-rw-r--r--plugins/JsonApiSourcePlugin.py17
-rw-r--r--plugins/MapPlugin.py11
-rw-r--r--plugins/ScraperSourcePlugin.py17
10 files changed, 71 insertions, 58 deletions
diff --git a/app/PluginInterface.py b/app/PluginInterface.py
index 6bc5513..23d76f8 100644
--- a/app/PluginInterface.py
+++ b/app/PluginInterface.py
@@ -7,10 +7,17 @@ Params: TypeAlias = dict[str, str]
class PluginInterface(ABC):
- def __init__(self, id: str, params: Params):
+ name: str
+
+ def __init__(self, plugin_type: str, id: str, params: Params):
+ self.plugin_type = plugin_type
self.id = id
self.params = params
+ self.log_prefix = f"[{self.plugin_type}#{self.id}]"
+
+ def log(self, msg) -> None:
+ print(f"{self.log_prefix} {msg}")
@abstractmethod
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
- pass
+ return NotImplemented
diff --git a/plugins/ConcatPlugin.py b/plugins/ConcatPlugin.py
index a0317cf..3c48728 100644
--- a/plugins/ConcatPlugin.py
+++ b/plugins/ConcatPlugin.py
@@ -10,8 +10,8 @@ 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")
+ super().__init__("ConcatPlugin", id, params)
+ self.log("initialized")
def item_sort_key(self, item: Item) -> datetime:
if item.pub_date is None:
@@ -20,9 +20,9 @@ class Plugin(PluginInterface):
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)}")
+ self.log(f"adding {len(items)} to concatenated stream")
if source_id is None:
- raise Exception(f"[ConcatPlugin#{self.id}] can not be scheduled")
+ raise Exception(f"{self.log_prefix} can not be scheduled")
plugin_state_q = Query().plugin_id == self.id
_d: Any = database_manager.plugin_states.get(plugin_state_q) # type: ignore
@@ -46,5 +46,5 @@ class Plugin(PluginInterface):
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)}")
+ self.log(f"concatenated {len(ret)} posts into one stream")
return ret
diff --git a/plugins/DigestPlugin.py b/plugins/DigestPlugin.py
index cf9a108..3675264 100644
--- a/plugins/DigestPlugin.py
+++ b/plugins/DigestPlugin.py
@@ -98,7 +98,7 @@ def parse_interval(interval: str) -> tuple[int, str]:
class Plugin(PluginInterface):
def __init__(self, id: str, params: Params) -> None:
- super().__init__(id, params)
+ super().__init__("DigestPlugin", id, params)
self.digest_title_prefix = get_config(params, "digest_title_prefix")
self.digest_description = get_config_or_default(
params, "digest_description", None
@@ -109,12 +109,12 @@ class Plugin(PluginInterface):
interval = get_config(params, "interval")
self.interval_pair = parse_interval(interval)
self.max_length = int(get_config_or_default(params, "max_length", "1000"))
- print(f"[DigestPlugin#{self.id}] initialized")
+ self.log("initialized")
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
- print(f"[DigestPlugin#{self.id}] process called, n={len(items)}")
+ self.log(f"building digests from {len(items)} posts")
if source_id is None:
- raise Exception(f"[DigestPlugin#{self.id}] can not be scheduled")
+ raise Exception(f"{self.log_prefix} can not be scheduled")
plugin_state_q = Query().plugin_id == self.id
_d: Any = database_manager.plugin_states.get(plugin_state_q) # type: ignore
@@ -189,8 +189,6 @@ class Plugin(PluginInterface):
digest_items.append(digest_item)
- print(
- f"[DigestPlugin#{self.id}] process returning items, n={len(digest_items)}"
- )
+ self.log(f"digested {len(digest_items)} posts")
return digest_items
diff --git a/plugins/FacebookSourcePlugin.py b/plugins/FacebookSourcePlugin.py
index 8d9a8e4..ea33615 100644
--- a/plugins/FacebookSourcePlugin.py
+++ b/plugins/FacebookSourcePlugin.py
@@ -211,25 +211,28 @@ def fetch_page_posts(email: str, password: str, page_id: str, limit: int) -> lis
class Plugin(PluginInterface):
+ def log(self, msg) -> None:
+ return super().log(msg)
+
def __init__(self, id: str, params: Params) -> None:
- super().__init__(id, params)
+ super().__init__("FacebookSourcePlugin", id, params)
self.login_email = get_config(params, "login_email")
self.login_password = get_config(params, "login_password")
self.page_id = get_config(params, "page_id")
self.limit = int(get_config_or_default(params, "limit", "10"))
-
- print(f"[FacebookSourcePlugin#{self.id}] initialized")
+ self.log("initialized")
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
- print(f"[FacebookSourcePlugin#{self.id}] process called")
if source_id is not None:
raise Exception(
- f"FacebookSourcePlugin#{self.id} can only be scheduled, trying to process items from source {source_id}"
+ f"{self.log_prefix} can only be scheduled, trying to propagate items from source {source_id}"
)
+ self.log(f'scraping posts from FB page id "{self.page_id}"')
+
posts = fetch_page_posts(
self.login_email, self.login_password, self.page_id, self.limit
)
- print(f"[FacebookSourcePlugin#{self.id}] process returns items, n={len(posts)}")
+ self.log(f'scraped {len(posts)} posts from FB page id "{self.page_id}"')
return posts
diff --git a/plugins/FeedSinkPlugin.py b/plugins/FeedSinkPlugin.py
index 09eb572..b51d64d 100644
--- a/plugins/FeedSinkPlugin.py
+++ b/plugins/FeedSinkPlugin.py
@@ -12,13 +12,13 @@ from app.DatabaseManager import database_manager
class Plugin(PluginInterface):
def __init__(self, id: str, params: dict[str, Any]) -> None:
- super().__init__(id, params)
+ super().__init__("FeedSinkPlugin", id, params)
self.feed_id: str = get_config(params, "feed_id")
self.feed_title: str = get_config(params, "feed_title")
self.feed_link: str | None = get_config_or_default(params, "feed_link", None)
self.feed_description: str = get_config(params, "feed_description")
- print(f"[FeedSinkPlugin#{self.id}] initialized")
- print(f"[FeedSinkPlugin#{self.id}] feed will be served at path /{self.feed_id}")
+ self.log("initialized")
+ self.log("feed will be served at path /{self.feed_id}")
def build_xml(self, items: list[Item]):
rss = ET.Element(
@@ -110,16 +110,16 @@ class Plugin(PluginInterface):
}
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
- print(f"[FeedSinkPlugin#{self.id}] process called, n={len(items)}")
+ self.log(f'building feed id "{self.feed_id}" from {len(items)} posts')
if source_id is None:
- raise Exception(f"FeedSinkPlugin#{self.id} can not be scheduled")
+ raise Exception(f"{self.log_prefix} can not be scheduled")
if database_manager.db is None:
- raise Exception("Database is not initialized")
+ raise Exception(f"{self.log_prefix} database is not initialized")
ret = self.build_xml(items)
Q = Query()
database_manager.feeds.upsert(ret, Q.feed_id == self.feed_id) # type: ignore
- print(f"[FeedSinkPlugin#{self.id}] processed")
+ self.log('build complete for feed id "{self.feed_id}"')
return []
diff --git a/plugins/FeedSourcePlugin.py b/plugins/FeedSourcePlugin.py
index 01692b3..2b27ade 100644
--- a/plugins/FeedSourcePlugin.py
+++ b/plugins/FeedSourcePlugin.py
@@ -16,24 +16,22 @@ def struct_time_to_utc_datetime(struct_time: time.struct_time) -> datetime:
class Plugin(PluginInterface):
def __init__(self, id: str, params: Params) -> None:
- super().__init__(id, params)
+ super().__init__("FeedSourcePlugin", id, params)
self.feed_url: str = get_config(params, "feed_url")
- print(f"[FeedSourcePlugin#{self.id}] initialized")
+ self.log("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 source {source_id}"
)
+ self.log(f'fetching feed "{self.feed_url}"')
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}"
- )
+ raise Exception(f"{self.log_prefix} malformed XML in feed {self.feed_url}")
if "image" in feed:
image_url = feed["image"]["href"]
@@ -83,6 +81,6 @@ class Plugin(PluginInterface):
)
print(
- f"[FeedSourcePlugin#{self.id}] process returns items, n={len(result_items)}"
+ f'{self.log_prefix} fetched {len(result_items)} from the feed "{self.feed_url}"'
)
return result_items
diff --git a/plugins/FilterPlugin.py b/plugins/FilterPlugin.py
index 11cbc68..85c2229 100644
--- a/plugins/FilterPlugin.py
+++ b/plugins/FilterPlugin.py
@@ -5,16 +5,16 @@ from app.utils import get_config
class Plugin(PluginInterface):
def __init__(self, id: str, params: Params) -> None:
- super().__init__(id, params)
+ super().__init__("FilterPlugin", id, params)
self.filter_expr: str = get_config(params, "filter_expr")
- print(f"[FilterPlugin#{self.id}] initialized")
+ self.log("initialized")
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
- print(f"[FilterPlugin#{self.id}] process called, n={len(items)}")
+ self.log(f"filtering {len(items)} with given filter expression")
if source_id is None:
- raise Exception(f"[FilterPlugin#{self.id}] can not be scheduled")
+ raise Exception(f"{self.log_prefix} 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)}")
+ self.log(f"filtering done, {len(ret)} passed filter")
return ret
diff --git a/plugins/JsonApiSourcePlugin.py b/plugins/JsonApiSourcePlugin.py
index bd06bb1..309bc9e 100644
--- a/plugins/JsonApiSourcePlugin.py
+++ b/plugins/JsonApiSourcePlugin.py
@@ -9,7 +9,7 @@ from app.utils import ItemDict, get_config, get_config_or_default
class Plugin(PluginInterface):
def __init__(self, id: str, params: Params) -> None:
- super().__init__(id, params)
+ super().__init__("JsonApiSourcePlugin", id, params)
self.url: str = get_config(params, "url").strip("/")
self.selector_post: str = get_config(params, "selector_post")
self.selector_title: str | None = get_config_or_default(
@@ -34,7 +34,7 @@ class Plugin(PluginInterface):
params, "show_image_in_description", True
)
- print(f"[JsonApiSourcePlugin#{self.id}] initialized")
+ self.log("initialized")
def absolute_link(self, link: str) -> str:
if link.startswith("/") or link.startswith("#"):
@@ -43,7 +43,12 @@ class Plugin(PluginInterface):
return link
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
- print(f"[JsonApiSourcePlugin#{self.id}] process called")
+ if source_id is not None:
+ raise Exception(
+ f"{self.log_prefix} can only be scheduled, trying to propagate items from source {source_id}"
+ )
+
+ self.log(f'fetching data from JSON API at URL "{self.url}"')
result_items: list[Item] = []
with requests.session() as session:
@@ -104,7 +109,7 @@ class Plugin(PluginInterface):
guid = ItemGUID(f"aggro__{self.id}__{digest[:32]}")
else:
raise Exception(
- f"[JsonApiSourcePlugin#{self.id}] both title and description are None"
+ f"{self.log_prefix} both title and description are None"
)
if image_src is not None:
@@ -142,7 +147,5 @@ class Plugin(PluginInterface):
)
result_items.append(item)
- print(
- f"[JsonApiSourcePlugin#{self.id}] process returns items, n={len(result_items)}"
- )
+ self.log(f"fetched {len(result_items)} items from JSON API")
return result_items
diff --git a/plugins/MapPlugin.py b/plugins/MapPlugin.py
index d924f2f..806d9a1 100644
--- a/plugins/MapPlugin.py
+++ b/plugins/MapPlugin.py
@@ -16,16 +16,17 @@ def set_field(item: Item, k: str, v: Any):
class Plugin(PluginInterface):
def __init__(self, id: str, params: Params) -> None:
- super().__init__(id, params)
+ super().__init__("MapPlugin", id, params)
self.map_expr: str = get_config(params, "map_expr")
- print(f"[MapItemPlugin#{self.id}] initialized")
+ self.log("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")
+ raise Exception(f"{self.log_prefix} can not be scheduled")
+
+ self.log(f"mapping over {len(items)} posts with given map expression")
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)}")
+ self.log(f"mapped over {len(ret)} posts")
return ret
diff --git a/plugins/ScraperSourcePlugin.py b/plugins/ScraperSourcePlugin.py
index edafb49..d782bf3 100644
--- a/plugins/ScraperSourcePlugin.py
+++ b/plugins/ScraperSourcePlugin.py
@@ -12,7 +12,7 @@ from app.utils import ItemDict, get_config, get_config_or_default
class Plugin(PluginInterface):
def __init__(self, id: str, params: Params) -> None:
- super().__init__(id, params)
+ super().__init__("ScraperSourcePlugin", id, params)
self.url: str = get_config(params, "url").strip("/")
self.selector_post: str = get_config(params, "selector_post")
self.selector_title: str | None = get_config_or_default(
@@ -46,7 +46,12 @@ class Plugin(PluginInterface):
return link
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
- print(f"[ScraperSourcePlugin#{self.id}] process called")
+ if source_id is not None:
+ raise Exception(
+ f"{self.log_prefix} can only be scheduled, trying to propagate items from source {source_id}"
+ )
+
+ self.log(f'starting to scrape posts from URL "{self.url}"')
result_items: list[Item] = []
with requests.session() as session:
@@ -151,7 +156,7 @@ class Plugin(PluginInterface):
)
if match is None:
raise Exception(
- f"[ScraperSourcePlugin#{self.id}] weird regex result when looking for background-image"
+ f"{self.log_prefix} weird regex result when looking for background-image"
)
image_src = match.group("url")
else:
@@ -167,7 +172,7 @@ class Plugin(PluginInterface):
guid = ItemGUID(f"aggro__{self.id}__{digest[:32]}")
else:
raise Exception(
- f"[ScraperSourcePlugin#{self.id}] both title and description are None"
+ f"{self.log_prefix} both title and description are None"
)
if image_src is not None:
@@ -205,7 +210,5 @@ class Plugin(PluginInterface):
)
result_items.append(item)
- print(
- f"[ScraperSourcePlugin#{self.id}] process returns items, n={len(result_items)}"
- )
+ self.log(f'scraped {len(result_items)} posts from URL "{self.url}"')
return result_items

Want to contribute? Host your changes somewhere and send me the link: jan (at) sydanviita.fi. Also, check out my website.


forge.jan.systems — powered by cgit