squash

job_thread/main.py

@@ -0,0 +1,133 @@
+import logging
+import sqlite3
+from os import environ
+from time import time
+import praw
+
+SECONDS_IN_WEEK = 60 * 60 * 24 * 7
+
+
+class Config:
+    DB_PATH = "posts.db"
+    SUBREDDIT = "developersindia"
+
+    POST_FLAIR = "Hiring"
+    POST_TITLE = "Don't Miss Out on These Job Opportunities | Weekly Job Openings Thread"
+    POST_TEXT = """\
+This thread has all the latest job openings that haven't been posted on previous weekly threads.
+
+If you have a job opening that you'd like to share with the community, you can post it using this link:- https://developersindia.in/post-a-job/
+
+For all the available job openings, check out the Job Board at:- https://developersindia.in/job-board/
+
+Stay tuned for updates on the latest job openings, and apply for the ones that interest you. Wishing you the best of luck in your job search!\
+"""
+
+    CLIENT_ID = environ["REDDIT_CLIENT_ID"]
+    CLIENT_SECRET = environ["REDDIT_CLIENT_SECRET"]
+    REDDIT_PASSWORD = environ["REDDIT_PASSWORD"]
+    USERNAME = environ["REDDIT_USERNAME"]
+    USER_AGENT = f"u/{USERNAME} Job Board"
+
+
+class Database:
+    def __init__(self, db_path):
+        self.conn = sqlite3.connect(db_path)
+        self.conn.row_factory = sqlite3.Row
+
+        self.cur = self.conn.cursor()
+
+        self._create()
+
+    def _create(self):
+        with self.conn:
+            self.cur.execute(
+                "CREATE TABLE IF NOT EXISTS Posts"
+                "(post_id TEXT PRIMARY KEY, time INTEGER NOT NULL DEFAULT(UNIXEPOCH()))"
+            )
+
+    def get_latest_post(self):
+        self.cur.execute("SELECT post_id, time from Posts ORDER BY time DESC")
+
+        if (result := self.cur.fetchone()) is not None:
+            return dict(result)
+
+    def insert_post(self, post_id: str, timestamp: int):
+        with self.conn:
+            self.cur.execute(
+                "INSERT INTO Posts (post_id, time) VALUES ((?), (?))",
+                (post_id, timestamp),
+            )
+
+
+def should_create_new_post(latest_post):
+    if latest_post is not None:
+        return (time() - latest_post["time"]) >= SECONDS_IN_WEEK
+
+    return True
+
+
+def create_job_post(subreddit):
+    # https://old.reddit.com/r/redditdev/comments/ovte4q/praw_flair_a_post/h7doqmd/?context=3
+    flair = next(
+        filter(
+            lambda flair: flair["flair_text"] == Config.POST_FLAIR,
+            subreddit.flair.link_templates.user_selectable(),
+        )
+    )
+
+    submission = subreddit.submit(
+        Config.POST_TITLE,
+        selftext=Config.POST_TEXT,
+        flair_id=flair["flair_template_id"],
+    )
+    submission.mod.sticky()
+
+    return submission
+
+
+def main():
+    logging.root.setLevel(logging.INFO)
+
+    db = Database(Config.DB_PATH)
+    reddit = praw.Reddit(
+        client_id=Config.CLIENT_ID,
+        client_secret=Config.CLIENT_SECRET,
+        password=Config.REDDIT_PASSWORD,
+        user_agent=Config.USER_AGENT,
+        username=Config.USERNAME,
+    )
+
+    subreddit = reddit.subreddit(Config.SUBREDDIT)
+
+    maybe_old_post = db.get_latest_post()
+
+    logging.info(f"Latest post in database {maybe_old_post}")
+
+    if should_create_new_post(maybe_old_post):
+        # Un-stick/pin the old post
+        if maybe_old_post is not None:
+            logging.info(f"Un-pinning old post")
+
+            try:
+                reddit.submission(maybe_old_post["post_id"]).mod.sticky(
+                    state=False
+                )
+            except Exception:
+                logging.warning(f"Failed to un-pin post!", exc_info=True)
+
+        new_submission = create_job_post(subreddit)
+
+        logging.info(
+            f"Created new post {new_submission.id} at {new_submission.created_utc}"
+        )
+
+        db.insert_post(new_submission.id, new_submission.created_utc)
+
+    submission = reddit.submission(db.get_latest_post()["post_id"])
+
+    logging.info(f"Fetched latest submission {submission.id}")
+
+
+if __name__ == "__main__":
+    main()