Improve scraper errors

3 files changed, 143 insertions, 88 deletions

diff --git a/app/PluginManager.py b/app/PluginManager.py
index 86da22f..fdc81a9 100644
--- a/app/PluginManager.py
+++ b/app/PluginManager.py
@@ -42,10 +42,15 @@ class PluginManager:
             return
 
         try:
-            plugin: PluginInterface = self.plugin_instances[id]
-            ret_items: list[Item] = plugin.process(source_id, items)
-            self.propagate(id, ret_items)
-        except Exception as ex:
+            try:
+                plugin: PluginInterface = self.plugin_instances[id]
+                ret_items: list[Item] = plugin.process(source_id, items)
+                self.propagate(id, ret_items)
+            except Exception as ex:
+                ex.add_note(f"context: plugin id: {id}, source id: {source_id}")
+                raise
+
+        except:
             exc = traceback.format_exc()
             print(exc, file=sys.stderr)
             if self.email_alerter:
diff --git a/app/utils.py b/app/utils.py
index 5401c5c..8dc4972 100644
--- a/app/utils.py
+++ b/app/utils.py
@@ -50,3 +50,11 @@ def item_to_dict(item: Item) -> ItemDict:
 def dict_to_item(d: ItemDict) -> Item:
     pub_date, rest = (lambda pub_date, **rest: (pub_date, rest))(**d)  # type: ignore
     return Item(pub_date=datetime.fromisoformat(pub_date), **rest)  # type: ignore
+
+
+def dump_to_file(prefix: str, text: str):
+    now = datetime.now().isoformat()
+    dump_file_name = f"dump/{prefix}__{now}"
+    with open(dump_file_name, "w") as f:
+        f.write(text)
+    return dump_file_name
diff --git a/plugins/ScraperSourcePlugin.py b/plugins/ScraperSourcePlugin.py
index 79fe69c..eb5e526 100644
--- a/plugins/ScraperSourcePlugin.py
+++ b/plugins/ScraperSourcePlugin.py
@@ -2,12 +2,12 @@ from bs4 import BeautifulSoup
 import time
 import hashlib
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, TypeVar, cast, overload
 import re
 import requests
 from app.Item import Item, ItemEnclosure, ItemGUID, ItemMediaContent
 from app.PluginInterface import Params, PluginInterface
-from app.utils import ItemDict, get_config, get_config_or_default
+from app.utils import ItemDict, dump_to_file, get_config, get_config_or_default
 
 
 class Plugin(PluginInterface):
@@ -39,12 +39,57 @@ class Plugin(PluginInterface):
 
         print(f"[ScraperSourcePlugin#{self.id}] initialized")
 
-    def absolute_link(self, link: str) -> str:
+    def absolute_link(self, link: str | list[str]) -> str:
+        if type(link) == list:
+            link = link[0]
+
+        link = cast(str, link)
         if link.startswith("/") or link.startswith("#"):
             link = self.url + link
 
         return link
 
+    def get_url(self, session: requests.Session, url: str):
+        try:
+            resp = session.get(url, allow_redirects=True)
+            if resp.status_code >= 400:
+                raise Exception("status_code >= 400")
+        except Exception as ex:
+            ex.add_note(f"{self.log_prefix} failed to fetch page, url: {url}")
+            raise
+
+        return resp
+
+    def parse_html(self, from_url: str, html: str):
+        try:
+            elem = BeautifulSoup(html, "html.parser")
+        except Exception as ex:
+            dump_file_name = dump_to_file(self.id, html)
+            ex.add_note(
+                f"{self.log_prefix} failed to parse HTML from {from_url}, HTML dumped to {dump_file_name}"
+            )
+            raise
+
+        return elem
+
+    def eval_selector(
+        self, selector: str | None, ctx: dict[str, Any]
+    ) -> list[BeautifulSoup]:
+        if not selector:
+            return []
+
+        try:
+            elems = eval(selector, ctx)
+        except Exception as ex:
+            dump_file_name = dump_to_file(self.id, str(ctx))
+            ex.add_note(
+                f"{self.log_prefix} failed to eval selector:\n  {selector}\n"
+                + f"HTML dumped to {dump_file_name}"
+            )
+            raise
+
+        return elems
+
     def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
         if source_id is not None:
             raise Exception(
@@ -73,98 +118,99 @@ class Plugin(PluginInterface):
         with requests.session() as session:
             session.headers.update(headers)
 
-            page_resp = session.get(self.url, allow_redirects=True)
-            page_elem = BeautifulSoup(page_resp.text, "html.parser")
-            post_elems = eval(self.selector_post, {"page": page_elem})
+            # Fetch the page
+            page_resp = self.get_url(session, url=self.url)
+
+            # Parse the page
+            page_elem = self.parse_html(from_url=self.url, html=page_resp.text)
+
+            # Evaluate post selector on parsed page
+            ctx: dict[str, BeautifulSoup | None] = {"page": page_elem}
+            post_elems = self.eval_selector(self.selector_post, ctx=ctx)
+
+            if post_elems is None:
+                post_elems = []
 
+            # For each post on page
             for post_elem in post_elems:
-                if self.selector_link:
-                    link_elem = eval(
-                        self.selector_link, {"page": page_elem, "post": post_elem}
-                    )[0]
-                    detail_page_url = self.absolute_link(link_elem["href"])
-                else:
-                    detail_page_url = None
+                ctx_with_post: dict[str, BeautifulSoup | None] = {
+                    "page": page_elem,
+                    "post": post_elem,
+                }
+
+                # Evaluate link selector on post
+                link_elems = self.eval_selector(self.selector_link, ctx=ctx_with_post)
+                detail_page_url = (
+                    self.absolute_link(link_elems[0]["href"])
+                    if len(link_elems) > 0
+                    else None
+                )
 
+                # Fetch and parse post detail page
                 if detail_page_url:
-                    detail_page_resp = session.get(
-                        detail_page_url, allow_redirects=True
-                    )
-                    detail_page_elem = BeautifulSoup(
-                        detail_page_resp.text, "html.parser"
+                    # Fetch
+                    detail_page_resp = self.get_url(session, url=detail_page_url)
+                    detail_page_elem = self.parse_html(
+                        from_url=detail_page_url, html=detail_page_resp.text
                     )
+
                     guid = ItemGUID(detail_page_url, is_perma_link=True)
                 else:
                     detail_page_elem = None
                     guid = None
 
-                if self.selector_title:
-                    title_elem = eval(
-                        self.selector_title,
-                        {
-                            "page": page_elem,
-                            "post": post_elem,
-                            "detail_page": detail_page_elem,
-                        },
-                    )[0]
-                    title = title_elem.get_text().strip()
-                else:
-                    title = None
+                ctx_with_post_detail: dict[str, BeautifulSoup | None] = {
+                    "page": page_elem,
+                    "post": post_elem,
+                    "detail_page": detail_page_elem,
+                }
 
-                if self.selector_description:
-                    description_elem = eval(
-                        self.selector_description,
-                        {
-                            "page": page_elem,
-                            "post": post_elem,
-                            "detail_page": detail_page_elem,
-                        },
-                    )[0]
-                    description = str(description_elem).strip()
-                else:
-                    description = None
+                # Evaluate title selector on post
+                title_elems = self.eval_selector(
+                    self.selector_title, ctx=ctx_with_post_detail
+                )
+                title = (
+                    title_elems[0].get_text().strip() if len(title_elems) > 0 else None
+                )
 
-                if self.selector_date:
-                    date_elem = eval(
-                        self.selector_date,
-                        {
-                            "page": page_elem,
-                            "post": post_elem,
-                            "detail_page": detail_page_elem,
-                        },
-                    )[0]
-                    date = date_elem.get_text().strip()
-                else:
-                    date = None
+                # Evaluate description selector on post
+                description_elems = self.eval_selector(
+                    self.selector_description, ctx=ctx_with_post_detail
+                )
+                description = (
+                    str(description_elems[0]).strip()
+                    if len(description_elems) > 0
+                    else None
+                )
 
-                if self.selector_author:
-                    author_elem = eval(
-                        self.selector_author,
-                        {
-                            "page": page_elem,
-                            "post": post_elem,
-                            "detail_page": detail_page_elem,
-                        },
-                    )[0]
-                    author = author_elem.get_text().strip()
-                else:
-                    author = None
+                # Evaluate date selector on post
+                date_elems = self.eval_selector(
+                    self.selector_date, ctx=ctx_with_post_detail
+                )
+                date = date_elems[0].get_text().strip() if len(date_elems) > 0 else None
+
+                # Evaluate author selector on post
+                author_elems = self.eval_selector(
+                    self.selector_author, ctx=ctx_with_post_detail
+                )
+                author = (
+                    author_elems[0].get_text().strip()
+                    if len(author_elems) > 0
+                    else None
+                )
 
-                if self.selector_image:
-                    image_elem = eval(
-                        self.selector_image,
-                        {
-                            "page": page_elem,
-                            "post": post_elem,
-                            "detail_page": detail_page_elem,
-                        },
-                    )[0]
+                image_elems = self.eval_selector(
+                    self.selector_image, ctx=ctx_with_post_detail
+                )
+                image_elem = image_elems[0] if len(image_elems) > 0 else None
+                image_src: str | None = None
+                if image_elem:
                     if image_elem.has_attr("src"):
-                        image_src = image_elem["src"]
+                        image_src = cast(str, image_elem["src"])
                     elif image_elem.has_attr("content"):
-                        image_src = image_elem["content"]
+                        image_src = cast(str, image_elem["content"])
                     elif image_elem.has_attr("href"):
-                        image_src = image_elem["href"]
+                        image_src = cast(str, image_elem["href"])
                     elif (
                         image_elem.has_attr("style")
                         and "background-image:" in image_elem["style"]
@@ -177,11 +223,7 @@ class Plugin(PluginInterface):
                             raise Exception(
                                 f"{self.log_prefix} weird regex result when looking for background-image"
                             )
-                        image_src = match.group("url").strip("'\"")
-                    else:
-                        image_src = None
-                else:
-                    image_src = None
+                        image_src = cast(str, match.group("url").strip("'\""))
 
                 if guid is None:
                     if title:
@@ -194,7 +236,7 @@ class Plugin(PluginInterface):
                             f"{self.log_prefix} both title and description are None"
                         )
 
-                if image_src is not None:
+                if image_src:
                     image_src = self.absolute_link(image_src)
                     image_html = f'<img src="{image_src}"><br><br>'
                     if description and self.show_image_in_description: