X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;ds=inline;f=generic_news_rss_renderer.py;h=f1261cb59e171cfa2a8348487839cb7edc2dd124;hb=4ee891c0fc29a002291c7c88f8d8fb5d14c28f92;hp=849c1ad4a643bd724d952215c994567a609409cc;hpb=75b27cc68871343681f01e3f5b04cae84b1b7b2a;p=kiosk.git diff --git a/generic_news_rss_renderer.py b/generic_news_rss_renderer.py index 849c1ad..f1261cb 100644 --- a/generic_news_rss_renderer.py +++ b/generic_news_rss_renderer.py @@ -1,188 +1,325 @@ +#!/usr/bin/env python3 + +from abc import abstractmethod +import datetime +from dateutil.parser import parse +import http.client +import logging +import re +from typing import Dict, List, Optional, Union +import xml.etree.ElementTree as ET + +from scottutilz import profanity_filter + import file_writer import grab_bag import renderer -import http.client import page_builder -import profanity_filter -import random -import re -import xml.etree.ElementTree as ET -class generic_news_rss_renderer(renderer.debuggable_abstaining_renderer): - def __init__(self, name_to_timeout_dict, feed_site, feed_uris, page_title): - super(generic_news_rss_renderer, self).__init__(name_to_timeout_dict, False) - self.debug = 1 + +logger = logging.getLogger(__name__) + + +class generic_news_rss_renderer(renderer.abstaining_renderer): + def __init__( + self, + name_to_timeout_dict: Dict[str, int], + feed_site: str, + feed_uris: List[str], + page_title: str, + ): + super().__init__(name_to_timeout_dict) self.feed_site = feed_site self.feed_uris = feed_uris self.page_title = page_title self.news = grab_bag.grab_bag() self.details = grab_bag.grab_bag() - self.filter = profanity_filter.profanity_filter() + self.filter = profanity_filter.ProfanityFilter() - def debug_prefix(self): + @abstractmethod + def get_headlines_page_prefix(self) -> str: pass - def get_headlines_page_prefix(self): + @abstractmethod + def get_details_page_prefix(self) -> str: pass - def get_details_page_prefix(self): - pass + def get_headlines_page_priority(self) -> str: + return "4" + + def get_details_page_priority(self) -> str: + return "6" - def should_use_https(self): + @abstractmethod + def should_use_https(self) -> bool: pass - def should_profanity_filter(self): - return False + def should_profanity_filter(self) -> bool: + return True - def find_title(self, item): - return item.findtext('title') + def find_title(self, item: ET.Element) -> Optional[str]: + return item.findtext("title") - def munge_title(self, title): + def munge_title(self, title: str, item: ET.Element) -> str: return title - def find_description(self, item): - return item.findtext('description') + def find_description(self, item: ET.Element) -> Optional[str]: + return item.findtext("description") - def munge_description(self, description): - description = re.sub('<[^>]+>', '', description) + def munge_description(self, description: str, item: ET.Element) -> str: + description = re.sub("<[^>]+>", "", description) return description - def find_link(self, item): - return item.findtext('link') + def find_link(self, item: ET.Element) -> Optional[str]: + return item.findtext("link") - def find_image(self, item): - return item.findtext('image') + def munge_link(self, link: str) -> str: + return link - def item_is_interesting_for_headlines(self, title, description, item): - pass + def find_image(self, item: ET.Element) -> Optional[str]: + return item.findtext("image") - def item_is_interesting_for_article(self, title, description, item): - pass + def munge_image(self, image: str) -> str: + return image + + def find_pubdate(self, item: ET.Element) -> Optional[str]: + return item.findtext("pubDate") + + def munge_pubdate(self, pubdate: str) -> str: + return pubdate + + def item_is_interesting_for_headlines( + self, title: str, description: str, item: ET.Element + ) -> bool: + return True - def periodic_render(self, key): + def do_headlines(self) -> bool: + return True + + def do_details(self) -> bool: + return True + + def is_item_older_than_n_days( + self, item: ET.Element, n: int, default: bool = False + ) -> bool: + pubdate = self.find_pubdate(item) + if pubdate is None: + return default + pubdatetime = parse(pubdate) + tzinfo = pubdatetime.tzinfo + now = datetime.datetime.now(tzinfo) + delta = (now - pubdatetime).total_seconds() / (60 * 60 * 24) + return delta > n + + def item_is_interesting_for_article( + self, title: str, description: str, item: ET.Element + ) -> bool: + return True + + def periodic_render(self, key: str) -> bool: if key == "Fetch News": return self.fetch_news() elif key == "Shuffle News": return self.shuffle_news() else: - raise error('Unexpected operation') - - def shuffle_news(self): - headlines = page_builder.page_builder() - headlines.set_layout(page_builder.page_builder.LAYOUT_FOUR_ITEMS) - headlines.set_title("%s" % self.page_title) - subset = self.news.subset(4) - if subset is None: - self.debug_print("Not enough messages to choose from.") - return False - for msg in subset: - headlines.add_item(msg) - f = file_writer.file_writer('%s_4_none.html' % ( - self.get_headlines_page_prefix())) - headlines.render_html(f) - f.close() - - details = page_builder.page_builder() - details.set_layout(page_builder.page_builder.LAYOUT_ONE_ITEM) - details.set_title("%s" % self.page_title) - subset = self.details.subset(1) - if subset is None: - self.debug_print("Not enough details to choose from."); - return False - for msg in subset: - blurb = msg - blurb += "\n" - details.add_item(blurb) - g = file_writer.file_writer('%s_6_none.html' % ( - self.get_details_page_prefix())) - details.render_html(g) - g.close() + raise Exception + + def shuffle_news(self) -> bool: + if self.do_headlines(): + headlines = page_builder.page_builder() + headlines.set_layout(page_builder.page_builder.LAYOUT_FOUR_ITEMS) + headlines.set_title("%s" % self.page_title) + subset = self.news.subset(4) + if subset is None: + logger.warning("Not enough messages to select from in shuffle_news?!") + return False + for msg in subset: + headlines.add_item(msg) + headlines.set_custom_html( + """ + """ + ) + _ = f"{self.get_headlines_page_prefix()}_{self.get_headlines_page_priority()}_25900.html" + with file_writer.file_writer(_) as f: + headlines.render_html(f) + + if self.do_details(): + details = page_builder.page_builder() + details.set_layout(page_builder.page_builder.LAYOUT_ONE_ITEM) + details.set_custom_html( + """ + """ + ) + details.set_title(self.page_title) + subset = self.details.subset(1) + if subset is None: + logger.warning("Not enough details to choose from in do_details") + logger.debug("Not enough details to choose from.") + return False + for msg in subset: + blurb = msg + blurb += "" + details.add_item(blurb) + _ = f"{self.get_details_page_prefix()}_{self.get_details_page_priority()}_86400.html" + with file_writer.file_writer(_) as g: + details.render_html(g) return True - def fetch_news(self): + def fetch_news(self) -> bool: count = 0 self.news.clear() self.details.clear() + self.conn: Optional[ + Union[http.client.HTTPConnection, http.client.HTTPSConnection] + ] = None for uri in self.feed_uris: + url = None if self.should_use_https(): - self.debug_print("Fetching: https://%s%s" % (self.feed_site, uri)) - self.conn = http.client.HTTPSConnection(self.feed_site) + url = f"https://{self.feed_site}{uri}" + logger.info(f"Fetching: {url}") + self.conn = http.client.HTTPSConnection(self.feed_site, timeout=10) else: - self.debug_print("Fetching: http://%s%s" % (self.feed_site, uri)) - self.conn = http.client.HTTPConnection(self.feed_site) + url = f"http://{self.feed_site}{uri}" + logger.info(f"Fetching: {url}") + self.conn = http.client.HTTPConnection(self.feed_site, timeout=10) + assert self.conn is not None + assert url is not None self.conn.request( "GET", uri, None, - {"Accept-Charset": "utf-8"}) - response = self.conn.getresponse() + { + "Accept": "*/*", + "Cache-control": "max-age=50", + }, + ) + try: + response = self.conn.getresponse() + except Exception: + logger.exception( + f"Exception in generic RSS renderer HTTP connection fetching {url}; giving up." + ) + return False + if response.status != 200: - print(("%s: RSS fetch_news error, response: %d" % (self.page_title, - response.status))) - self.debug_print(response.read()) + logger.error( + f"Unexpected status {response.status} while fetching {url}: {response.reason}; giving up." + ) + print(dir(response)) + print(response.headers) return False - rss = ET.fromstring(response.read()) + raw = response.read() + logger.info(f"Status 200: got {len(raw)} bytes back from {url}") + rss = ET.fromstring(raw) channel = rss[0] - for item in channel.getchildren(): + title_filter = set() + for item in list(channel): title = self.find_title(item) + description = item.findtext("description") if title is not None: - title = self.munge_title(title) - description = item.findtext('description') + title = self.munge_title(title, item) + else: + logger.info("Skipping RSS feed item with no title.") + continue + logger.debug(f"Considering RSS item {title}...") if description is not None: - description = self.munge_description(description) - link = item.findtext('link') - image = item.findtext('image') - - if (title is None or - not self.item_is_interesting_for_headlines(title, - description, - item)): - self.debug_print('Item "%s" is not interesting' % title) + description = self.munge_description(description, item) + else: + description = "" + image = self.find_image(item) + if image is not None: + image = self.munge_image(image) + link = item.findtext("link") + if link is not None: + link = self.munge_link(link) + if not self.item_is_interesting_for_headlines(title, description, item): + logger.info(f"Skipping {title} because it's not interesting.") continue - if (self.should_profanity_filter() and - (self.filter.contains_bad_words(title) or - self.filter.contains_bad_words(description))): - self.debug_print('Found bad words in item "%s"' % title) + if self.should_profanity_filter() and ( + self.filter.contains_bad_word(title) + or self.filter.contains_bad_word(description) + ): + logger.info(f"Skipping {title} because it contains profanity.") continue - #print u"Title: %s\nDescription: %s\nLink: %s\nImage: %s\n" % ( - # title, description, link, image) + if title in title_filter: + logger.info( + f"Skipping {title} because we already saw an item with the same title." + ) + continue + title_filter.add(title) blurb = """
%s' % title
+ blurb += f''
+
+ if link is None:
+ blurb += f"
{title}" + else: + blurb += f'
{title}'
+
+ pubdate = self.find_pubdate(item)
+ if pubdate is not None:
+ logger.debug(f"Raw pubdate={pubdate}")
+ pubdate = self.munge_pubdate(pubdate)
+ ts = parse(pubdate)
+ logger.debug(f"Translated pubdate into: {ts}")
+ blurb += f' {ts.strftime("%b %d")}'
- if (description is not None and
- self.item_is_interesting_for_article(title, description, item)):
+ if self.item_is_interesting_for_article(title, description, item):
+ logger.info(
+ f"Item {title} is also interesting as an article details page; creating..."
+ )
longblurb = blurb
longblurb += "
"
longblurb += description
longblurb += "