Mainichi poster

main.py

import requests
import feedparser
import sqlite3
import os
import logging
import dotenv
import time

dotenv.load_dotenv()
logging.basicConfig(level=logging.INFO)

ACCESS_TOKEN = os.environ.get("ACCESS_TOKEN")
MASTODON_URL = os.environ.get("MASTODON_URL")
SLEEP_TIME = 5

con = sqlite3.connect(os.environ.get("MAINICHI_DB", "mainichi.db"))


def setup_db():
    cur = con.cursor()
    logging.info("Creating posts table")
    cur.execute("CREATE TABLE IF NOT EXISTS posts(url)")
    cur.close()


def does_post_exist(url):
    cur = con.cursor()
    if cur.execute("SELECT url FROM posts WHERE url=?", [url]).fetchone():
        return True

    logging.info("Storing post url=%s", url)
    cur.execute("INSERT INTO posts VALUES(?)", [url])
    cur.close()


def post(title, url):
    res = requests.post(url=f"{MASTODON_URL}/api/v1/statuses", json={
        "status": f"{title} {url}"
    }, headers={
        "Authorization": f"Bearer {ACCESS_TOKEN}"
    })

    if res.status_code >= 400:
        logging.error("Failed to post url=%s because=%s", url, res.text)
        return

    logging.error("Successfully posted url=%s", url)


def main():
    feed = feedparser.parse("https://mainichi.jp/rss/etc/english_latest.rss")
    for article in feed.entries:
        url = article["link"]
        summary = article["title"]

        if does_post_exist(url=article["link"]):
            logging.info("Skipping url=%s because it has already been posted", url)
            continue

        post(title=summary, url=url)
        time.sleep(SLEEP_TIME)

    con.commit()


if __name__ == '__main__':
    setup_db()
    main()