1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
|
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, get_param, item_to_dict
def remove_duplicates_by_key(lst: list[ItemDict], key: str) -> list[ItemDict]:
return list({x[key]: x for x in lst}.values())
class Plugin(PluginInterface):
def __init__(self, id: str, params: Params) -> None:
super().__init__(id, params)
self.digest_title = get_param("digest_title", params)
self.digest_description = get_param("digest_description", params)
print(f"[DigestPlugin#{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 make_digest(self) -> list[Item]:
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": {"cutoff_timestamp": time.localtime(), "channels": {}},
}
)
state = data["state"]
channels: dict[str, list[ItemDict]] = state["channels"]
cutoff_timestamp = time.struct_time(state["cutoff_timestamp"])
aggregated_items: list[Item] = []
for data_source_id in channels:
item_dicts = channels[data_source_id]
items = list(map(dict_to_item, item_dicts))
for item in items:
if item.pub_date is not None and cutoff_timestamp < item.pub_date:
aggregated_items.append(item)
state["cutoff_timestamp"] = time.localtime()
digest_desc = (
f"{self.digest_description}<br><br>" if self.digest_description else ""
)
for item in aggregated_items:
pub_date_stamp = (
time.strftime("%a, %d %b %Y %H:%M:%S +0000", item.pub_date)
if item.pub_date
else ""
)
author = item.author if item.author else ""
digest_desc += f"<strong>{item.title}</strong><br>"
digest_desc += f"<small>{pub_date_stamp} <i>{author}</i></small><br><br>"
digest_desc += item.description if item.description else ""
digest_desc += "<br><br>"
digest_item: Item = Item(
title=self.digest_title,
description=digest_desc,
link=None,
author=None,
pub_date=time.localtime(),
category=None,
comments=None,
enclosures=[],
)
database_manager.plugin_states.upsert( # type: ignore
{"plugin_id": self.id, "state": state}, plugin_state_q
)
return [digest_item]
def add_to_state(self, source_id: str, items: list[Item]) -> list[Item]:
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": {"cutoff_timestamp": time.localtime(), "channels": {}},
}
)
state = data["state"]
items_as_dicts = list(map(item_to_dict, items))
state["channels"][source_id] = items_as_dicts
database_manager.plugin_states.upsert( # type: ignore
{"plugin_id": self.id, "state": state}, plugin_state_q
)
return []
def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
print(f"[DigestPlugin#{self.id}] process called, n={len(items)}")
if source_id is None:
return self.make_digest()
else:
return self.add_to_state(source_id, items)
|
