Skip to content

Instantly share code, notes, and snippets.

@hydrobeam
Last active February 12, 2022 05:14
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save hydrobeam/fa5b084b032fc205db567e8e0eb8247e to your computer and use it in GitHub Desktop.
Save hydrobeam/fa5b084b032fc205db567e8e0eb8247e to your computer and use it in GitHub Desktop.
Stackoverflow Discord webhook
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Final
import requests
from apscheduler.schedulers.blocking import BlockingScheduler
from bs4 import BeautifulSoup
from dhooks import Embed, Webhook
from stackapi import StackAPI
from stackapi.stackapi import StackAPIError
# retrieved by copying the link when pasting a link to stackoverflow on discord
STACK_OVERFLOW_LOGO_LINK: Final = "https://images-ext-1.discordapp.net/external/bmwCfJoD44JOslIoT_HOIkmcq908os0A03x6SNLwV9U/https/images-ext-1.discordapp.net/external/VbfwnzN2MM794XNccNxDzrB1YeuPrxR53y11bwRfflY/%253Fv%253D73d79a89bded/https/cdn.sstatic.net/Sites/stackoverflow/Img/apple-touch-icon%25402.png?width=96&height=96"
@dataclass
class Question:
"""An object representing a question on stackoveflow
Parameters
---------
link
The link to the post
title
The title of the post (the name of the question)
description
The short description that shows up when pasting a link to a stackoverflow post.
tags
A list of associated tags.
"""
link: str
title: str
description: str
tags: list[str]
@classmethod
def from_dict(cls, content_dict: dict):
"""Creates a Question object from a dict.
Below is a sample dictionary that would be passed into this method.
{'tags': ['python', 'python-3.x', 'manim'],
'owner': {'reputation': 370,
'user_id': 9690045,
'user_type': 'registered',
'profile_image': 'https://i.stack.imgur.com/oA7vH.jpg?s=256&g=1',
'display_name': 'SMMousaviSP',
'link': 'https://stackoverflow.com/users/9690045/smmousavisp'},
'is_answered': False,
'view_count': 8,
'answer_count': 0,
'score': 0,
'last_activity_date': 1644604745,
'creation_date': 1644604745,
'question_id': 71085043,
'content_license': 'CC BY-SA 4.0',
'link': 'https://stackoverflow.com/questions/71085043/add-buff-only-in-one-end-of-a-manim-line-arrow',
'title': 'Add buff only in one end of a manim line / arrow'}
"""
tags: list[str] = content_dict["tags"]
title: str = content_dict["title"]
link: str = content_dict["link"]
req = requests.get(link)
soup = BeautifulSoup(req.text, "html.parser")
# a meta tag with a twitter:description attribute holds the information
# that would be typically shown in embed
# (i.e. if you were to paste a stackoverflow link in discord)
meta_tag = soup.find("meta", attrs={"name": "twitter:description"})
description: str = meta_tag.get("content")
return cls(link=link, title=title, description=description, tags=tags)
def create_embed(self) -> Embed:
"""Creates an embed according to discord.py (now defunct) semantics.
refer to https://cog-creators.github.io/discord-embed-sandbox/ to create your own.
"""
embed = Embed(
title=self.title,
url=self.link,
color=0xFEA306, # orange-ish
description=self.description,
)
embed.set_thumbnail(url=STACK_OVERFLOW_LOGO_LINK)
embed.set_footer(text=f"Tags: {', '.join(self.tags)}")
return embed
@dataclass
class MainProcess:
"""The heart of the procedure. Manages the main func that's repeated on an increment
and sends out posts to the webhook channel.
Parameters
----------
scheduler
An apscheduler scheduler. Doesn't really have to be a blocking scheduler.
Manages the cron job that is called in `run`
webhook_obj
The webhook object, provided by dhooks. Could be done manually but this offers some
convienence classes and methods. Especially `Embed`.
tag
The tag we're going to be searching for. Use a semicolon to search for combined tags
tag = "python;javascript"
is equivalent to Python AND Javascript. Not sure how to do OR.
https://meta.stackexchange.com/questions/279044/is-it-possible-to-use-the-search-api-to-lookup-many-tags-at-the-same-time
That ^ didn't help.
site
The stack exchange site you want to be searching.
"""
scheduler: BlockingScheduler
webhook_obj: Webhook
tag: str
question_fetch_increment: timedelta = timedelta(minutes=15)
site: StackAPI = StackAPI("stackoverflow")
def run(self):
# https://apscheduler.readthedocs.io/en/3.x/modules/triggers/cron.html?highlight=cron#module-apscheduler.triggers.cron
scheduler.add_job(
self.main_func,
"cron",
# run every n minutes.
minute=f"*/{self.question_fetch_increment.seconds//60}",
)
scheduler.start()
def main_func(self):
"""The main function. Called every `self.question_fetch_increment.seconds` seconds.
Calls the stackoverflow API via StackAPI and checks if new questions have been asked.
If so, create Questions, then Embeds from the json."""
try:
queries: list[dict] = self.query_posts()["items"]
if queries:
questions = self.create_Questions(queries)
self.post_embeds(questions)
else:
print(f"No Posts Found.")
except StackAPIError as e:
# copied from https://stackapi.readthedocs.io/en/latest/user/quickstart.html#errors
print(" Error URL: {}".format(e.url))
print(" Error Code: {}".format(e.code))
print(" Error Error: {}".format(e.error))
print(" Error Message: {}".format(e.message))
def query_posts(self) -> dict[str, Any]:
"""Ask stackoverflow if new posts have been made under a tag."""
manim_questions = self.site.fetch(
"questions",
tagged=self.tag,
fromdate=datetime.now() - self.question_fetch_increment,
)
return manim_questions
def create_Questions(self, item_data: list[dict]) -> list[Question]:
"""Creates a list of questions. Mostly relies on Question.from_dict ."""
question_list = []
for question_data in item_data:
temp_q = Question.from_dict(question_data)
question_list.append(temp_q)
return question_list
def post_embeds(self, questions: list[Question]) -> None:
"""Posts the embeds generated from a list of Questions to the channel."""
embeds = [question.create_embed() for question in questions]
for embed in embeds:
# should probably do error checking here but I don't know which errors to expect
self.webhook_obj.send(embed=embed)
if __name__ == "__main__":
url = WEBHOOK_URL
hook = Webhook(url)
scheduler = BlockingScheduler()
# every 15 minutes
main = MainProcess(
webhook_obj=hook,
scheduler=scheduler,
question_fetch_increment=timedelta(seconds=900),
tag="python",
)
main.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment