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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148
|
from typing import List, Optional
from gi.repository import GObject, GLib
from datetime import datetime
from gfeeds.confManager import ConfManager
from gfeeds.feed_item import FeedItem
import pytz
import gfeeds.feeds_manager as feeds_manager
from syndom import Feed as SynDomFeed, FeedItem as SynDomFeedItem
class Feed(GObject.Object):
__gsignals__ = {
'empty_changed': (
GObject.SignalFlags.RUN_FIRST, None, ()
)
}
__title = ''
__link = ''
__description = ''
__image_url = ''
__unread_count = 0
rss_link = ''
tags = list()
items = dict()
sd_feed = None
favicon_path = ''
init_time = None
def __init__(
self, rss_link: str, title: str, link: str, description: str,
image_url: Optional[str], favicon_path: Optional[str],
sd_feed: SynDomFeed, raw_entries: List[SynDomFeedItem]
):
super().__init__()
self.confman = ConfManager()
self.feedman = feeds_manager.FeedsManager()
self.tag_store = self.feedman.tag_store
self.__unread_count = 0
self.tags = list()
self.items = dict()
self.update(
rss_link, title, link, description, image_url, favicon_path,
sd_feed, raw_entries
)
def update(
self, rss_link: str, title: str, link: str, description: str,
image_url: Optional[str], favicon_path: Optional[str],
sd_feed: SynDomFeed, raw_entries: List[SynDomFeedItem]
):
self.rss_link = rss_link
self.__title = title
self.__link = link
self.__description = description
self.__image_url = image_url
self.favicon_path = favicon_path
self.sd_feed = sd_feed
unread_count = 0
self.init_time = pytz.UTC.localize(datetime.utcnow())
for entry in raw_entries:
n_item = FeedItem(entry, self)
uid = self.rss_link + n_item.identifier
item_age = self.init_time - n_item.pub_date # type: ignore
valid_age = item_age <= self.confman.max_article_age
if uid in self.items:
if not valid_age:
del self.items[uid]
else:
if valid_age:
self.items[uid] = n_item
if valid_age and not n_item.read:
unread_count += 1
if not valid_age and n_item.read:
read_items: List[str] = \
self.confman.nconf.read_items # type: ignore
read_items.remove(n_item.identifier)
self.confman.nconf.read_items = read_items
if self.rss_link in self.confman.nconf.feeds: # type: ignore
feed_conf = (self.get_conf_dict() or dict())
for tag_name in feed_conf.get('tags', []):
tag_obj = self.tag_store.get_tag(tag_name)
if tag_obj is not None and tag_obj not in self.tags:
self.tags.append(tag_obj)
# Set property, trigger signal (avoiding excess signals during init)
if unread_count != self.__unread_count:
def do():
self.unread_count = unread_count
GLib.idle_add(do)
def get_conf_dict(self) -> Optional[dict]:
return self.confman.nconf.feeds.get( # type: ignore
self.rss_link, None
)
@GObject.Property(type=str)
def title(self) -> str: # type: ignore
return self.__title
@GObject.Property(type=str)
def link(self) -> str: # type: ignore
return self.__link
@GObject.Property(type=str)
def description(self) -> str: # type: ignore
return self.__description
@GObject.Property(type=str)
def image_url(self) -> Optional[str]: # type: ignore
return self.__image_url
@image_url.setter
def image_url(self, n_image_url: Optional[str]):
self.__image_url = n_image_url
@GObject.Property(type=int, default=0)
def unread_count(self) -> int: # type: ignore
return self.__unread_count
@unread_count.setter
def unread_count(self, v: int):
prev = self.__unread_count
self.__unread_count = v
if self.__unread_count == 0 and prev > 0:
self.emit('empty_changed')
elif self.__unread_count > 0 and prev == 0:
self.emit('empty_changed')
change = self.__unread_count - prev
for tag in self.tags:
tag.increment_unread_count(change)
def __repr__(self):
return f'Feed Object `{self.__title}`; {len(self.items)} items'
def to_dict(self) -> dict:
return {
'title': self.title,
'link': self.link,
'description': self.description,
'image_url': self.image_url,
'rss_link': self.rss_link,
'favicon_path': self.favicon_path,
}
|