From 4f924479155b3d0e6dae53640616d9c1c9778f03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Mon, 5 Dec 2022 12:02:41 +0100 Subject: [PATCH] Sort functions alphabetical and fill out docstrings --- discord_rss_bot/feeds.py | 125 ++++++++++++++---------- discord_rss_bot/main.py | 184 ++++++++++++++++++------------------ discord_rss_bot/settings.py | 59 ++++++------ 3 files changed, 197 insertions(+), 171 deletions(-) diff --git a/discord_rss_bot/feeds.py b/discord_rss_bot/feeds.py index e9c1265..ed48189 100644 --- a/discord_rss_bot/feeds.py +++ b/discord_rss_bot/feeds.py @@ -29,22 +29,27 @@ from requests import Response from discord_rss_bot.settings import logger, reader -def check_feeds() -> None: - """Check all feeds""" - reader.update_feeds() - entries = reader.get_entries(read=False) - send_to_discord(entries) +class IfFeedError(BaseModel): + """Update a feed. - -def check_feed(feed_url: str) -> None: - """Check a single feed""" - reader.update_feeds() - entry = reader.get_entries(feed=feed_url, read=False) - send_to_discord(entry) + Attributes: + feed_url: The feed to update. + webhook: The webhook to use. + error: True if error, False if no error. + err_msg: The error message, if any. + exception: The exception, if any. + """ + feed_url: str + webhook: str + error: bool + err_msg: str = "" + exception: str = "" class NoWebhookFoundError(Exception): - """No webhook found error.""" + """Raises an exception if no webhook is found. + + Used in send_to_discord().""" def __init__(self, message): self.message = message @@ -53,13 +58,73 @@ class NoWebhookFoundError(Exception): return self.message +def add_feed(feed_url: str, webhook: str, exist_ok=False, allow_invalid_url=False) -> IfFeedError: + """ + Add a feed to reader. If error occurs, it will return IfFeedError with error=True. + + Args: + feed_url: The feed to add. + webhook: The webhook to use. + exist_ok: If the feed already exists, do nothing. + allow_invalid_url: If the feed url is invalid, add it anyway. + + Returns: + IfFeedError: Error or not. + """ + try: + reader.add_feed(feed=feed_url, exist_ok=exist_ok, allow_invalid_url=allow_invalid_url) + except FeedExistsError as error: + error_msg = "Feed already exists" + logger.error(f"{error_msg}: {error}") + return IfFeedError(error=True, err_msg=error_msg, feed_url=feed_url, webhook=webhook, exception=error.message) + + except InvalidFeedURLError as error: + error_msg = "Invalid feed URL" + logger.error(f"{error_msg}: {error}") + return IfFeedError(error=True, err_msg=error_msg, feed_url=feed_url, webhook=webhook, exception=error.message) + + return IfFeedError(error=False, feed_url=feed_url, webhook=webhook) + + +def check_feed(feed_url: str) -> None: + """Update a single feed and send its unread entries to Discord. + + We don't need to mark entries as read here, because send_to_discord() does that when sending entries to Discord + if it was successful. + + Args: + feed_url: The feed to check. + """ + reader.update_feed(feed_url) + entries = reader.get_entries(feed=feed_url, read=False) + for entry in entries: + send_to_discord(entry) + + +def check_feeds() -> None: + """Update all feeds and send all the entries that are unread to Discord. + + We don't need to mark entries as read here, because send_to_discord() does that when sending entries to Discord + if it was successful. + """ + reader.update_feeds() + entries = reader.get_entries(read=False) + for entry in entries: + send_to_discord(entry) + + def send_to_discord(entry) -> Response: """ Send entries to Discord. + If response was not ok, we will log the error and mark the entry as unread, so it will be sent again next time. + Args: entry: The entry to send. + Raises: + NoWebhookFoundError: If no webhook is found. + Returns: Response: The response from the webhook. """ @@ -77,21 +142,11 @@ def send_to_discord(entry) -> Response: f"{entry.link}", rate_limit_retry=True) response = webhook.execute() if not response.ok: - # TODO: Send error to discord logger.error(f"Error: {response.status_code} {response.reason}") reader.mark_entry_as_unread(entry) return response -class IfFeedError(BaseModel): - """Update a feed.""" - feed_url: str - webhook: str - error: bool - err_msg: str = "" - exception: str = "" - - def update_feed(feed_url: str, webhook: str) -> IfFeedError: """ Update a feed. @@ -122,29 +177,3 @@ def update_feed(feed_url: str, webhook: str) -> IfFeedError: return IfFeedError(error=True, err_msg=error_msg, feed_url=feed_url, webhook=webhook, exception=error.message) return IfFeedError(error=False, feed_url=feed_url, webhook=webhook) - - -def add_feed(feed_url: str, webhook: str, exist_ok=False, allow_invalid_url=False) -> IfFeedError: - """ - Add a feed. - - Args: - feed_url: The feed to add. - webhook: The webhook to use. - exist_ok: If the feed already exists, do nothing. - allow_invalid_url: If the feed url is invalid, add it anyway. - - Returns: - IfFeedError: Error or not. - """ - try: - reader.add_feed(feed=feed_url, exist_ok=exist_ok, allow_invalid_url=allow_invalid_url) - except FeedExistsError as error: - error_msg = "Feed already exists" - logger.error(f"{error_msg}: {error}") - return IfFeedError(error=True, err_msg=error_msg, feed_url=feed_url, webhook=webhook, exception=error.message) - - except InvalidFeedURLError as error: - error_msg = "Invalid feed URL" - logger.error(f"{error_msg}: {error}") - return IfFeedError(error=True, err_msg=error_msg, feed_url=feed_url, webhook=webhook, exception=error.message) diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py index 52266c7..8519f03 100644 --- a/discord_rss_bot/main.py +++ b/discord_rss_bot/main.py @@ -45,13 +45,6 @@ app = FastAPI() templates = Jinja2Templates(directory="templates") -@cache -@app.get("/favicon.ico", include_in_schema=False) -async def favicon(): - """Return favicon.""" - return FileResponse('static/favicon.ico') - - @app.post("/check", response_class=HTMLResponse) def check_feed(request: Request, feed_url: str = Form()): """Check all feeds""" @@ -65,21 +58,91 @@ def check_feed(request: Request, feed_url: str = Form()): return templates.TemplateResponse("feed.html", {"request": request, "feed": feed}) -@app.on_event('startup') -def startup(): - """This is called when the server starts. +@app.post("/add") +async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()): + """ + Add a feed to the database. - It reads the settings file and starts the scheduler.""" + Args: + feed_url: The feed to add. + webhook_dropdown: The webhook to use. + + Returns: + dict: The feed that was added. + """ + logger.info(f"Add feed: {feed_url}") + logger.info(f"Webhook: {webhook_dropdown}") + + # Update a single feed. The feed will be updated even if updates are disabled for it. + updated_feed = update_feed(feed_url, webhook_dropdown) + + # Add a new feed to the database. + added_feed = add_feed(feed_url, webhook_dropdown) + + if updated_feed.error or added_feed.error: + error_dict = {"error": updated_feed.error, "feed": updated_feed.feed_url, "webhook": updated_feed.webhook, + "exception": updated_feed.exception} + return HTTPException(status_code=500, detail=error_dict) + + # Check if set_hook_by_name() was successful. + if isinstance(set_hook_by_name(name=webhook_dropdown, feed_url=feed_url), ResourceNotFoundError): + return set_hook_by_name(name=webhook_dropdown, feed_url=feed_url) + + new_tag = reader.get_tag(feed_url, "webhook") + logger.info(f"New tag: {new_tag}") + return {"feed_url": str(feed_url), "status": "added"} + + +def create_list_of_webhooks(): + """List with webhooks.""" + logger.info("Creating list with webhooks.") settings = read_settings_file() + list_of_webhooks = dict() + for hook in settings["webhooks"]: + logger.info(f"Webhook name: {hook} with URL: {settings['webhooks'][hook]}") + list_of_webhooks[hook] = settings["webhooks"][hook] - if not settings["webhooks"]: - logger.critical("No webhooks found in settings file.") - sys.exit() - for key in settings["webhooks"]: - logger.info(f"Webhook name: {key} with URL: {settings['webhooks'][key]}") + logger.info(f"List of webhooks: {list_of_webhooks}") + return enum.Enum("DiscordWebhooks", list_of_webhooks) - scheduler = BackgroundScheduler() - scheduler.start() + +@cache +@app.get("/favicon.ico", include_in_schema=False) +async def favicon(): + """Return favicon.""" + return FileResponse('static/favicon.ico') + + +@app.get("/add", response_class=HTMLResponse) +def get_add(request: Request): + """ + Page for adding a new feed. + + Args: + request: The request. + + Returns: + HTMLResponse: The HTML response. + """ + context = make_context_index(request) + return templates.TemplateResponse("add.html", context) + + +@app.post("/feed", response_class=HTMLResponse) +async def get_feed(request: Request, feed_url: str = Form()): + """ + Get a feed by URL. + + Args: + request: The request. + feed_url: The feed to add. + + Returns: + HTMLResponse: The HTML response. + """ + logger.info(f"Get feed: {feed_url}") + feed = reader.get_feed(feed_url) + return templates.TemplateResponse("feed.html", {"request": request, "feed": feed}) @app.get("/", response_class=HTMLResponse) @@ -148,84 +211,21 @@ async def remove_feed(request: Request, feed_url: str = Form()): return templates.TemplateResponse("index.html", {"request": request, "feed": feed}) -@app.post("/feed", response_class=HTMLResponse) -async def get_feed(request: Request, feed_url: str = Form()): - """ - Get a feed by URL. +@app.on_event('startup') +def startup(): + """This is called when the server starts. - Args: - request: The request. - feed_url: The feed to add. - - Returns: - HTMLResponse: The HTML response. - """ - logger.info(f"Get feed: {feed_url}") - feed = reader.get_feed(feed_url) - return templates.TemplateResponse("feed.html", {"request": request, "feed": feed}) - - -def create_list_of_webhooks(): - """List with webhooks.""" - logger.info("Creating list with webhooks.") + It reads the settings file and starts the scheduler.""" settings = read_settings_file() - list_of_webhooks = dict() - for hook in settings["webhooks"]: - logger.info(f"Webhook name: {hook} with URL: {settings['webhooks'][hook]}") - list_of_webhooks[hook] = settings["webhooks"][hook] - logger.info(f"List of webhooks: {list_of_webhooks}") - return enum.Enum("DiscordWebhooks", list_of_webhooks) + if not settings["webhooks"]: + logger.critical("No webhooks found in settings file.") + sys.exit() + for key in settings["webhooks"]: + logger.info(f"Webhook name: {key} with URL: {settings['webhooks'][key]}") - -@app.post("/add") -async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()): - """ - Add a feed to the database. - - Args: - feed_url: The feed to add. - webhook_dropdown: The webhook to use. - - Returns: - dict: The feed that was added. - """ - logger.info(f"Add feed: {feed_url}") - logger.info(f"Webhook: {webhook_dropdown}") - - # Update a single feed. The feed will be updated even if updates are disabled for it. - updated_feed = update_feed(feed_url, webhook_dropdown) - - # Add a new feed to the database. - added_feed = add_feed(feed_url, webhook_dropdown) - - if updated_feed.error or added_feed.error: - error_dict = {"error": updated_feed.error, "feed": updated_feed.feed_url, "webhook": updated_feed.webhook, - "exception": updated_feed.exception} - return HTTPException(status_code=500, detail=error_dict) - - # Check if set_hook_by_name() was successful. - if isinstance(set_hook_by_name(name=webhook_dropdown, feed_url=feed_url), ResourceNotFoundError): - return set_hook_by_name(name=webhook_dropdown, feed_url=feed_url) - - new_tag = reader.get_tag(feed_url, "webhook") - logger.info(f"New tag: {new_tag}") - return {"feed_url": str(feed_url), "status": "added"} - - -@app.get("/add", response_class=HTMLResponse) -def get_add(request: Request): - """ - Page for adding a new feed. - - Args: - request: The request. - - Returns: - HTMLResponse: The HTML response. - """ - context = make_context_index(request) - return templates.TemplateResponse("add.html", context) + scheduler = BackgroundScheduler() + scheduler.start() if __name__ == "__main__": diff --git a/discord_rss_bot/settings.py b/discord_rss_bot/settings.py index 2b0ecb1..77bdb6f 100644 --- a/discord_rss_bot/settings.py +++ b/discord_rss_bot/settings.py @@ -24,16 +24,40 @@ from platformdirs import user_data_dir from reader import make_reader from tomlkit import TOMLDocument, comment, document, parse, table -logging.basicConfig( - level=logging.DEBUG, - format="[%(asctime)s] [%(funcName)s:%(lineno)d] %(message)s", -) +logging.basicConfig(level=logging.DEBUG, format="[%(asctime)s] [%(funcName)s:%(lineno)d] %(message)s") logger = logging.getLogger(__name__) # For get_data_dir() data_directory = user_data_dir(appname="discord_rss_bot", appauthor="TheLovinator", roaming=True) +def create_settings_file(settings_file) -> None: + """Create the settings file if it doesn't exist.""" + logger.debug(f"Settings file: {settings_file}") + + # [webhooks] + # Both options are commented out by default. + webhooks = table() + webhooks.add(comment('"First webhook" = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"')) + webhooks.add(comment('"Second webhook" = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"')) + + # [database] + # Option is commented out by default. + database = table() + database.add(comment('"location" = "/path/to/database/file"')) + + doc = document() + doc.add("webhooks", webhooks) + doc.add("database", database) + + logger.debug(f"Settings file: {doc}") + logger.debug(f"Settings file as TOML: {doc.as_string()}") + + # Write the settings file + with open(settings_file, "w") as f: + f.write(doc.as_string()) + + def get_data_dir(data_dir: str = data_directory) -> Path: """ Get the data directory. This is where the database file and config file are stored. @@ -81,33 +105,6 @@ def get_db_file(custom_db_name: str = "db.sqlite") -> Path: return Path(db_file) -def create_settings_file(settings_file) -> None: - """Create the settings file if it doesn't exist.""" - logger.debug(f"Settings file: {settings_file}") - - # [webhooks] - # Both options are commented out by default. - webhooks = table() - webhooks.add(comment('"First webhook" = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"')) - webhooks.add(comment('"Second webhook" = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"')) - - # [database] - # Option is commented out by default. - database = table() - database.add(comment('"location" = "/path/to/database/file"')) - - doc = document() - doc.add("webhooks", webhooks) - doc.add("database", database) - - logger.debug(f"Settings file: {doc}") - logger.debug(f"Settings file as TOML: {doc.as_string()}") - - # Write the settings file - with open(settings_file, "w") as f: - f.write(doc.as_string()) - - def read_settings_file(custom_settings_name: str = "settings.toml") -> TOMLDocument: """Read the settings file