TheLovinator/discord-rss-bot
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move update_feed and add_feed to own function, and create set_hook_by_name

Browse Source
This commit is contained in:
Joakim Hellsén
2022-12-05 11:15:53 +01:00
parent 3cb38f8714
commit 551644de5e
3 changed files with 154 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
discord_rss_bot/feeds.py
View File

@ -1,4 +1,7 @@
from discord_webhook import DiscordWebhook
from pydantic import BaseModel
from reader import FeedExistsError, FeedNotFoundError, InvalidFeedURLError, ParseError, StorageError
from requests import Response
from discord_rss_bot.settings import logger, reader
@ -7,23 +10,45 @@ def check_feeds() -> None:
"""Check all feeds"""
reader.update_feeds()
entries = reader.get_entries(read=False)
_check_feed(entries)
send_to_discord(entries)
def check_feed(feed_url: str) -> None:
"""Check a single feed"""
reader.update_feeds()
entry = reader.get_entries(feed=feed_url, read=False)
_check_feed(entry)
send_to_discord(entry)
def _check_feed(entries) -> None:
for entry in entries:
class NoWebhookFoundError(Exception):
"""No webhook found error."""
def __init__(self, message):
self.message = message
def __str__(self):
return self.message
def send_to_discord(entry) -> Response:
"""
Send entries to Discord.
Args:
entry: The entry to send.
Returns:
Response: The response from the webhook.
"""
reader.mark_entry_as_read(entry)
logger.debug(f"New entry: {entry.title}")
webhook_url = reader.get_tag(entry.feed.url, "webhook")
if webhook_url:
if not webhook_url:
logger.error(f"No webhook found for feed: {entry.feed.url}")
raise NoWebhookFoundError(f"No webhook found for feed: {entry.feed.url}")
logger.debug(f"Sending to webhook: {webhook_url}")
webhook = DiscordWebhook(url=str(webhook_url), content=f":robot: :mega: New entry: {entry.title}\n"
f"{entry.link}", rate_limit_retry=True)
@ -32,3 +57,71 @@ def _check_feed(entries) -> None:
# 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.
Args:
feed_url: The feed to update.
webhook: The webhook to use.
Returns:
IfFeedError: Error or not.
"""
try:
reader.update_feed(feed_url)
except FeedNotFoundError as error:
error_msg = "Feed not found"
logger.error(error_msg, exc_info=True)
return IfFeedError(error=True, err_msg=error_msg, feed_url=feed_url, webhook=webhook, exception=error.message)
except ParseError as error:
error_msg = "An error occurred while getting/parsing feed"
logger.error(error_msg, exc_info=True)
return IfFeedError(error=True, err_msg=error_msg, feed_url=feed_url, webhook=webhook, exception=error.message)
except StorageError as error:
error_msg = "An exception was raised by the underlying storage"
logger.error(error_msg, exc_info=True)
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)

41
discord_rss_bot/main.py
View File

@ -4,13 +4,14 @@ from functools import cache
import uvicorn
from apscheduler.schedulers.background import BackgroundScheduler
from fastapi import FastAPI, Form, Request
from fastapi import FastAPI, Form, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.templating import Jinja2Templates
from reader import FeedExistsError
from reader import ResourceNotFoundError
from discord_rss_bot.feeds import _check_feed
from discord_rss_bot.feeds import add_feed, send_to_discord, update_feed
from discord_rss_bot.settings import logger, read_settings_file, reader
from discord_rss_bot.webhooks import set_hook_by_name
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@ -28,7 +29,7 @@ def check_feed(request: Request, feed_url: str = Form()):
"""Check all feeds"""
reader.update_feeds()
entry = reader.get_entries(feed=feed_url, read=False)
_check_feed(entry)
send_to_discord(entry)
logger.info(f"Get feed: {feed_url}")
feed = reader.get_feed(feed_url)
@ -149,13 +150,6 @@ def create_list_of_webhooks():
return enum.Enum("DiscordWebhooks", list_of_webhooks)
def get_hook_by_name(name):
"""Get a webhook by name."""
settings = read_settings_file()
logger.debug(f"Webhook name: {name} with URL: {settings['webhooks'][name]}")
return settings["webhooks"][name]
@app.post("/add")
async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()):
"""
@ -170,18 +164,25 @@ async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()):
"""
logger.info(f"Add feed: {feed_url}")
logger.info(f"Webhook: {webhook_dropdown}")
try:
reader.add_feed(feed_url)
except FeedExistsError as error:
logger.error(f"Feed already exists: {error}")
return {"error": "Feed already exists."}
reader.update_feed(feed_url)
webhook_url = get_hook_by_name(webhook_dropdown)
reader.set_tag(feed_url, "webhook", webhook_url)
# 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", "webhook": webhook_url}
return {"feed_url": str(feed_url), "status": "added"}
@app.get("/add", response_class=HTMLResponse)

24
discord_rss_bot/webhooks.py Normal file
View File

@ -0,0 +1,24 @@
from fastapi import HTTPException
from reader import ResourceNotFoundError
from discord_rss_bot.settings import logger, read_settings_file, reader
def set_hook_by_name(name: str, feed_url: str) -> None or HTTPException:
"""Set a webhook by name.
Args:
name: The name of the webhook.
feed_url: The feed to set the webhook for.
Returns:
HTTPException: The HTTP exception if the webhook was not found, otherwise None.
"""
settings = read_settings_file()
logger.debug(f"Webhook name: {name} with URL: {settings['webhooks'][name]}")
webhook_url = settings["webhooks"][name]
try:
reader.set_tag(feed_url, "webhook", webhook_url)
except ResourceNotFoundError as e:
logger.error(f"ResourceNotFoundError: {e}")
return HTTPException(status_code=500, detail=f"ResourceNotFoundError: Could not set webhook: {e}")