Newswire Notifications via Webhook

main.py

from threading import Timer
from requests import request
import sqlite3, os

api_url = "https://socialclub.rockstargames.com/events/eventlisting?pageId=1&gameId=GTAV"
article_url = "https://socialclub.rockstargames.com/events/{}/{}"
webhook_url = ""

def create_connection():
  file = os.path.join(os.getcwd(), 'database.sqlite')
  return sqlite3.connect(file)

def create_table():
  conn = create_connection()
  curr = conn.cursor()
  curr.execute("CREATE TABLE IF NOT EXISTS articles (hash varchar(16) PRIMARY KEY, slug text);")
  conn.close()

def article_exists(conn, article):
  curr = conn.cursor()
  curr.execute("SELECT * FROM articles WHERE hash = ?;", (article["urlHash"], ))
  return curr.fetchone() != None

def insert_article(conn, article):
  curr = conn.cursor()
  curr.execute("INSERT INTO articles (hash, slug) VALUES (?, ?);", (article["urlHash"], article["slug"]))
  conn.commit()

def sent_webhook(article):
  request("post", webhook_url, json={
    "username": "Newswire",
    "avatar_url": "https://editors.dexerto.com/wp-content/uploads/2021/11/17/GTA-Vice-City-Rockstar-Logo.jpg",
    "content": "[{}]({})".format(article["headerText"], article_url.format(article["urlHash"], article["slug"]))
  })

def main():
  conn = create_connection()
  req = request("get", api_url)
  data = req.json()
  
  for article in data["events"]:
    if article["isLive"] == True:
      if not article_exists(conn, article):
        insert_article(conn, article)
        sent_webhook(article)

  conn.close()
  Timer(60, main).start()

if __name__ == "__main__":
  create_table()
  main()

requirements.txt

requests==2.28.1
pysqlite3==0.4.7