Migrate to python (#265)

* Add initial python migration

* Add pylint

* Add pre-commit

* Add Dockerfile

* Add expression matching

* Use shelve to store previous matches

* Add notifications

* Calculate post age

* Update README.md

rfd_notify/cli.py

@@ -0,0 +1,40 @@
+import argparse
+import os
+
+from config import load_yaml_file
+from rfd import get_topics, look_for_matches
+from loguru import logger
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Process some configuration.")
+
+    parser.add_argument(
+        "-c", "--config", type=str, required=True, help="path to configuration file"
+    )
+
+    parser.add_argument(
+        "-s",
+        "--storage-path",
+        type=str,
+        required=False,
+        default="previous_matches",
+        help="path to persistent storage",
+    )
+
+    args = parser.parse_args()
+
+    config_path = args.config
+
+    apprise_url = os.getenv("APPRISE_URL")
+
+    config = load_yaml_file(config_path)
+
+    topics = get_topics(forum_id=9, pages=2)
+
+    logger.debug(f"config: {config}")
+
+    look_for_matches(topics, config, args.storage_path, apprise_url)
+
+
+main()

rfd_notify/config.py

@@ -0,0 +1,19 @@
+import yaml
+from loguru import logger
+
+
+class Config:
+    def __init__(self, expressions):
+        self.expressions = expressions
+
+    def __repr__(self):
+        return f"Config(expressions={self.expressions})"
+
+
+def load_yaml_file(filename: str) -> Config:
+    with open(filename, "r", encoding="utf-8") as file:
+        try:
+            data = yaml.safe_load(file)
+        except yaml.YAMLError as err:
+            logger.error(f"Error loading config file: {err}")
+    return Config(**data)

rfd_notify/constants.py

@@ -0,0 +1 @@
+API_BASE_URL = "https://forums.redflagdeals.com"

rfd_notify/models/offer.py

@@ -0,0 +1,8 @@
+class Offer:
+    # pylint: disable=unused-argument
+    def __init__(self, dealer_name, url, **kwargs):
+        self.dealer_name = dealer_name
+        self.url = url
+
+    def __repr__(self):
+        return f"Offer({self.url})"

rfd_notify/models/post.py

@@ -0,0 +1,8 @@
+class Post:
+    # pylint: disable=unused-argument
+    def __init__(
+        self,
+        body: str,
+        **kwargs,
+    ):
+        self.body = body

rfd_notify/models/topic.py

@@ -0,0 +1,23 @@
+from .offer import Offer
+
+
+class Topic:
+    # pylint: disable=unused-argument
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        topic_id: int,
+        title: str,
+        post_time: str,
+        web_path: str,
+        offer: dict,
+        **kwargs,
+    ):
+        self.topic_id = topic_id
+        self.title = title
+        self.post_time = post_time
+        self.web_path = web_path
+        self.offer = Offer(**offer)
+
+    def __repr__(self):
+        return f"Topic({self.title})"

rfd_notify/notifications.py

@@ -0,0 +1,42 @@
+from typing import List
+from datetime import datetime, timezone
+
+import apprise
+from models.topic import Topic
+from models.post import Post
+from loguru import logger
+from constants import API_BASE_URL
+
+
+def send_notification(
+    topic: Topic, posts: List[Post], expression: str, servers: str
+) -> None:
+    apobj = apprise.Apprise()
+    apobj.add(servers)
+
+    subject = topic.title
+    body = f"""\
+    <b>Post age:</b> {datetime.now(timezone.utc) - datetime.fromisoformat(topic.post_time)}
+    <br>
+    <br>
+    <b>Dealer:</b> {topic.offer.dealer_name}
+    <br>
+    <br>
+    <b>Deal:</b> {topic.title}
+    <br>
+    <br>
+    <b>Post:</b> {API_BASE_URL}{topic.web_path}\
+    <br>
+    <br>
+    <b>Body:</b> {posts[0].body}
+    <br>
+    <br>
+    <b>Matched by expression:</b> {expression}
+    """
+
+    logger.debug("Sending notification")
+
+    apobj.notify(
+        body=body,
+        title=subject,
+    )

rfd_notify/rfd.py

@@ -0,0 +1,87 @@
+from json.decoder import JSONDecodeError
+from typing import List, Any
+import re
+import shelve
+import requests
+from loguru import logger
+from constants import API_BASE_URL
+
+from config import Config
+from notifications import send_notification
+from models.topic import Topic
+from models.post import Post
+
+
+class RfdTopicsException(Exception):
+    def __init__(self, message):
+        self.message = message
+
+
+def get_topic(topic_id: int) -> List[Post]:
+    posts = []
+    try:
+        response = requests.get(
+            f"{API_BASE_URL}/api/topics/{topic_id}/posts?per_page=1&page=1",
+            timeout=30,
+        )
+        if response.status_code != 200:
+            raise RfdTopicsException(
+                f"Received status code {response.status_code} when getting topic."
+            )
+        for post in response.json().get("posts"):
+            posts.append(Post(**post))
+    except JSONDecodeError as err:
+        logger.error("Unable to decode topics. %s", err)
+    return posts
+
+
+def get_topics(forum_id: int, pages: int) -> List[Topic]:
+    topics = []
+    try:
+        for page in range(1, pages + 1):
+            response = requests.get(
+                f"{API_BASE_URL}/api/topics?forum_id={forum_id}&per_page=40&page={page}",
+                timeout=30,
+            )
+            if response.status_code != 200:
+                raise RfdTopicsException(
+                    f"Received status code {response.status_code} when getting topics."
+                )
+            for topic in response.json().get("topics"):
+                topics.append(Topic(**topic))
+    except JSONDecodeError as err:
+        logger.error("Unable to decode topics. %s", err)
+    return topics
+
+
+def look_for_matches(
+    topics: List[Topic], config: Config, storage_path: str, apprise_url: Any
+):
+    with shelve.open(storage_path) as previous_matches:
+        for topic in topics:
+            found_match = False
+            for expression in config.expressions:
+                expression = expression.lower()
+                topic_title = topic.title.lower()
+                dealer_name = topic.offer.dealer_name.lower()
+                if re.search(expression, topic_title):
+                    found_match = True
+                    logger.debug(
+                        f"Expression {expression} matched title '{topic.title}'"
+                    )
+                elif re.search(expression, dealer_name):
+                    found_match = True
+                    logger.debug(
+                        f"Expression {expression} matched dealer '{dealer_name}' - '{topic.title}'"
+                    )
+                if not found_match:
+                    continue
+                if str(topic.topic_id) not in previous_matches:
+                    posts = get_topic(topic.topic_id)
+                    previous_matches[str(topic.topic_id)] = 1
+                    send_notification(topic, posts, expression, apprise_url)
+                else:
+                    logger.debug(
+                        f"Already matched topic '{topic.offer.dealer_name} - {topic.title}'"
+                    )
+                break