amr/discorss
amr/discorss
1
1
Fork 0
Code Issues Pull requests Projects Releases 3 Packages Wiki Activity Actions

class refactor: Squashed commit of the following:

Browse source 
commit 597f383724
Author: A.M. Rowsell <amrowsell@frozenelectronics.ca>
Date:   Sun Apr 20 16:04:00 2025 -0400

    fix: typo in log setup, error when replacing w/self

commit 810cbd6f3d
Author: A.M. Rowsell <amrowsell@frozenelectronics.ca>
Date:   Sun Apr 20 13:51:41 2025 -0400

    refactor: transformed into class-based app
This commit is contained in:
A.M. Rowsell 2025-04-20 16:06:16 -04:00
commit 087a6339c8
Signed by: amr
GPG key ID: 0B6E2D8375CF79A9
1 changed files with 204 additions and 188 deletions
Download patch file Download diff file Expand all files Collapse all files

140
discorss.py
View file

@ -22,33 +22,31 @@ import sys
import argparse import argparse
import re import re
config_dir = os.environ.get("XDG_CONFIG_HOME")
class Discorss:
def __init__(self):
self.config_dir = os.environ.get("XDG_CONFIG_HOME")
home_dir = Path.home() home_dir = Path.home()
if config_dir is None: if self.config_dir is None:
config_file_path = str(home_dir) + "/.config/discorss/discorss.conf" self.config_file_path = str(home_dir) + "/.config/discorss/discorss.conf"
config_dir = str(home_dir) + "/.config/discorss" self.config_dir = str(home_dir) + "/.config/discorss"
else: else:
config_file_path = config_dir + r"/discorss/discorss.conf" self.config_file_path = self.config_dir + r"/discorss/discorss.conf"
log_dir = r"/var/log/discorss" self.log_dir = r"/var/log/discorss"
log_file_path = r"/app.log" self.log_file_path = r"/app.log"
# Yes, I know you "can't parse HTML with regex", but # Yes, I know you "can't parse HTML with regex", but
# just watch me. # just watch me.
html_filter = re.compile(r"\<\/?([A-Za-z0-9 \:\.\-\/\"\=])*\>") self.html_filter = re.compile(r"\<\/?([A-Za-z0-9 \:\.\-\/\"\=])*\>")
success_codes = [200, 201, 202, 203, 204, 205, 206] self.success_codes = [200, 201, 202, 203, 204, 205, 206]
app_config = {} self.app_config = {}
# IDEA: Consider making this into a class-based program
# This would solve a couple issues around global variables and generally
# make things a bit neater
# This function gets and formats the brief excerpt that goes in the embed # This function gets and formats the brief excerpt that goes in the embed
# Different feeds put summaries in different fields, so we pick the best # Different feeds put summaries in different fields, so we pick the best
# one and limit it to 250 characters. # one and limit it to 250 characters.
def get_description(feed, length=250, min_length=150, addons=None): def get_description(self, feed, length=250, min_length=150, addons=None):
try: try:
temporary_string = str(feed["summary_detail"]["value"]) temporary_string = str(feed["summary_detail"]["value"])
temporary_string = html_filter.sub("", temporary_string) temporary_string = self.html_filter.sub("", temporary_string)
while length > min_length: while length > min_length:
if temporary_string[length - 1 : length] == " ": if temporary_string[length - 1 : length] == " ":
break break
@ -56,7 +54,7 @@ def get_description(feed, length=250, min_length=150, addons=None):
length -= 1 length -= 1
except KeyError: except KeyError:
temporary_string = str(feed["description"]) temporary_string = str(feed["description"])
temporary_string = html_filter.sub("", temporary_string) temporary_string = self.html_filter.sub("", temporary_string)
while length > min_length: while length > min_length:
if temporary_string[length - 1 : length] == " ": if temporary_string[length - 1 : length] == " ":
break break
@ -68,41 +66,46 @@ def get_description(feed, length=250, min_length=150, addons=None):
desc = desc + str(addons) desc = desc + str(addons)
return desc return desc
def setupPaths(self):
def setupPaths():
global app_config
global logger
# Check for log and config files/paths, create empty directories if needed # Check for log and config files/paths, create empty directories if needed
# TODO: make this cleaner # TODO: make this cleaner
if not Path(log_dir).exists(): if not Path(self.log_dir).exists():
print("No log file path exists. Yark! We'll try and make {}...".format(log_dir)) print(
"No log file path exists. Yark! We'll try and make {}...".format(
self.log_dir
)
)
try: try:
Path(log_dir).mkdir(parents=True, exist_ok=True) Path(self.log_dir).mkdir(parents=True, exist_ok=True)
except FileExistsError: except FileExistsError:
print("The path {} already exists and is not a directory!".format(log_dir)) print(
if not Path(config_file_path).exists(): "The path {} already exists and is not a directory!".format(
self.log_dir
)
)
if not Path(self.config_file_path).exists():
print( print(
"No config file at {}! Snarf. We'll try and make {}...".format( "No config file at {}! Snarf. We'll try and make {}...".format(
config_file_path, config_dir self.config_file_path, self.config_dir
) )
) )
try: try:
Path(config_dir).mkdir(parents=True, exist_ok=True) Path(self.config_dir).mkdir(parents=True, exist_ok=True)
except FileExistsError: except FileExistsError:
print( print(
"The config dir {} already exists and is not a directory! Please fix manually. Quitting!".format( "The config dir {} already exists and is not a directory! Please fix manually. Quitting!".format(
config_dir self.config_dir
) )
) )
sys.exit(255) sys.exit(255)
return return
# Loading the config file # Loading the config file
with open(config_file_path, "r") as config_file: with open(self.config_file_path, "r") as config_file:
app_config = json.load(config_file) self.app_config = json.load(config_file)
# Set up logging # Set up logging
logger = logging.getLogger(__name__) self.logger = logging.getLogger(__name__)
logging.basicConfig( logging.basicConfig(
filename=str(log_dir + log_file_path), filename=str(self.log_dir + self.log_file_path),
encoding="utf-8", encoding="utf-8",
level=logging.DEBUG, level=logging.DEBUG,
datefmt="%m/%d/%Y %H:%M:%S", datefmt="%m/%d/%Y %H:%M:%S",
@ -110,23 +113,26 @@ def setupPaths():
) )
return return
def process(self):
def main():
os.environ["TZ"] = "America/Toronto" os.environ["TZ"] = "America/Toronto"
time.tzset() time.tzset()
now = time.mktime(time.localtime()) now = time.mktime(time.localtime())
setupPaths() # Handle the config and log paths self.setupPaths() # Handle the config and log paths
try: try:
last_check = app_config["lastupdate"] last_check = self.app_config["lastupdate"]
except KeyError: except KeyError:
last_check = now - 21600 # first run, no lastupdate, check up to 6 hours ago last_check = (
for i, hook in enumerate(app_config["feeds"]): # Feed loop start now - 21600
logger.debug("Parsing feed %s...", hook["name"]) ) # first run, no lastupdate, check up to 6 hours ago
feeds = feedparser.parse(hook["url"]) for i, hook in enumerate(self.app_config["feeds"]): # Feed loop start
latest_post = [] self.logger.debug("Parsing feed %s...", hook["name"])
self.feeds = feedparser.parse(hook["url"])
self.latest_post = []
prev_best = 0 prev_best = 0
logger.debug("About to sort through entries for feed %s ...", hook["name"]) self.logger.debug(
for feed in feeds["entries"]: "About to sort through entries for feed %s ...", hook["name"]
)
for feed in self.feeds["entries"]:
try: try:
bad_time = False bad_time = False
published_time = time.mktime(feed["published_parsed"]) published_time = time.mktime(feed["published_parsed"])
@ -140,32 +146,32 @@ def main():
else: else:
continue continue
if bad_time is True: if bad_time is True:
logger.debug( self.logger.debug(
"Feed %s doesn't supply a published time, using updated time instead", "Feed %s doesn't supply a published time, using updated time instead",
hook["name"], hook["name"],
) )
# Hash the title and time of the latest post and use that to determine if it's been posted # Hash the title and time of the latest post and use that to determine if it's been posted
# Yes, SHA3-512 is totally unnecessary for this purpose, but I love SHA3 # Yes, SHA3-512 is totally unnecessary for this purpose, but I love SHA3
logger.debug("About to hash %s ...", latest_post["title"]) self.logger.debug("About to hash %s ...", latest_post["title"])
try: try:
new_hash = hashlib.sha3_512( new_hash = hashlib.sha3_512(
bytes(latest_post["title"] + str(published_time), "utf-8") bytes(latest_post["title"] + str(published_time), "utf-8")
).hexdigest() ).hexdigest()
except TypeError: except TypeError:
logger.error("Title of %s isn't hashing correctly", hook["name"]) self.logger.error("Title of %s isn't hashing correctly", hook["name"])
continue continue
try: try:
if hook["lasthash"] != new_hash: if hook["lasthash"] != new_hash:
app_config["feeds"][i]["lasthash"] = new_hash self.app_config["feeds"][i]["lasthash"] = new_hash
else: else:
continue continue
except KeyError: except KeyError:
app_config["feeds"][i]["lasthash"] = new_hash self.app_config["feeds"][i]["lasthash"] = new_hash
logger.info( self.logger.info(
"Feed %s has no existing hash, likely a new feed!", hook["name"] "Feed %s has no existing hash, likely a new feed!", hook["name"]
) )
# Generate the webhook # Generate the webhook
logger.info( self.logger.info(
"Publishing webhook for %s. Last check was %d, now is %d", "Publishing webhook for %s. Last check was %d, now is %d",
hook["name"], hook["name"],
last_check, last_check,
@ -188,7 +194,7 @@ def main():
"fields": [ "fields": [
{ {
"name": "Excerpt from post:", "name": "Excerpt from post:",
"value": get_description(latest_post), "value": self.get_description(latest_post),
} }
], ],
# "timestamp": str(now), # "timestamp": str(now),
@ -202,25 +208,35 @@ def main():
} }
webhook_string = json.dumps(webhook) webhook_string = json.dumps(webhook)
logger.debug("About to run POST for %s", hook["name"]) self.logger.debug("About to run POST for %s", hook["name"])
r = requests.post(hook["webhook"], data=webhook_string, headers=custom_header) r = requests.post(
if r.status_code not in success_codes: hook["webhook"], data=webhook_string, headers=custom_header
logger.error( )
if r.status_code not in self.success_codes:
self.logger.error(
"Error %d while trying to post %s", r.status_code, hook["name"] "Error %d while trying to post %s", r.status_code, hook["name"]
) )
else: else:
logger.debug("Got %d when posting %s", r.status_code, hook["name"]) self.logger.debug("Got %d when posting %s", r.status_code, hook["name"])
# End of feed loop # End of feed loop
# Dump updated config back to json file # Dump updated config back to json file
logger.debug("Dumping config back to %s", str(config_file_path)) self.logger.debug("Dumping config back to %s", str(self.config_file_path))
app_config["lastupdate"] = now self.app_config["lastupdate"] = now
with open(config_file_path, "w") as config_file: with open(self.config_file_path, "w") as config_file:
json.dump(app_config, config_file, indent=4) json.dump(self.app_config, config_file, indent=4)
return return
# end of Discorss class
def main():
app = Discorss()
app.process()
if __name__ == "__main__": if __name__ == "__main__":
main() main()