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

Sort functions alphabetical and fill out docstrings

Browse Source
This commit is contained in:
Joakim Hellsén
2022-12-05 12:02:41 +01:00
parent e045f07282
commit 4f92447915
3 changed files with 197 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
discord_rss_bot/feeds.py
View File

@ -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)

184
discord_rss_bot/main.py
View File

@ -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__":

59
discord_rss_bot/settings.py
View File

@ -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