Remove unneeded function
This commit is contained in:
@ -5,7 +5,7 @@ from fastapi import HTTPException
|
|||||||
from reader import Entry, Feed, FeedExistsError, Reader, TagNotFoundError
|
from reader import Entry, Feed, FeedExistsError, Reader, TagNotFoundError
|
||||||
from requests import Response
|
from requests import Response
|
||||||
|
|
||||||
from discord_rss_bot import custom_message, settings
|
from discord_rss_bot import custom_message
|
||||||
from discord_rss_bot.filter.blacklist import should_be_skipped
|
from discord_rss_bot.filter.blacklist import should_be_skipped
|
||||||
from discord_rss_bot.filter.whitelist import has_white_tags, should_be_sent
|
from discord_rss_bot.filter.whitelist import has_white_tags, should_be_sent
|
||||||
from discord_rss_bot.settings import default_custom_message, get_reader
|
from discord_rss_bot.settings import default_custom_message, get_reader
|
||||||
@ -22,7 +22,7 @@ def send_entry_to_discord(entry: Entry, custom_reader: Reader | None = None):
|
|||||||
reader: Reader = get_reader() if custom_reader is None else custom_reader
|
reader: Reader = get_reader() if custom_reader is None else custom_reader
|
||||||
|
|
||||||
# Get the webhook URL for the entry.
|
# Get the webhook URL for the entry.
|
||||||
webhook_url: str = settings.get_webhook_for_entry(reader, entry)
|
webhook_url: str = str(reader.get_tag(entry.feed_url, "webhook", ""))
|
||||||
if not webhook_url:
|
if not webhook_url:
|
||||||
return "No webhook URL found."
|
return "No webhook URL found."
|
||||||
|
|
||||||
@ -116,7 +116,7 @@ def send_to_discord(custom_reader: Reader | None = None, feed: Feed | None = Non
|
|||||||
reader.set_entry_read(entry, True)
|
reader.set_entry_read(entry, True)
|
||||||
|
|
||||||
# Get the webhook URL for the entry. If it is None, we will continue to the next entry.
|
# Get the webhook URL for the entry. If it is None, we will continue to the next entry.
|
||||||
webhook_url: str = settings.get_webhook_for_entry(reader, entry)
|
webhook_url: str = str(reader.get_tag(entry.feed_url, "webhook", ""))
|
||||||
if not webhook_url:
|
if not webhook_url:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
@ -2,7 +2,7 @@ import os
|
|||||||
from functools import lru_cache
|
from functools import lru_cache
|
||||||
|
|
||||||
from platformdirs import user_data_dir
|
from platformdirs import user_data_dir
|
||||||
from reader import Entry, Reader, TagNotFoundError, make_reader # type: ignore
|
from reader import Reader, make_reader # type: ignore
|
||||||
|
|
||||||
data_dir: str = user_data_dir(appname="discord_rss_bot", appauthor="TheLovinator", roaming=True)
|
data_dir: str = user_data_dir(appname="discord_rss_bot", appauthor="TheLovinator", roaming=True)
|
||||||
os.makedirs(data_dir, exist_ok=True)
|
os.makedirs(data_dir, exist_ok=True)
|
||||||
@ -19,30 +19,6 @@ default_custom_embed: dict[str, str] = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def get_webhook_for_entry(custom_reader: Reader, entry: Entry) -> str:
|
|
||||||
"""
|
|
||||||
Get the webhook from the database.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
custom_reader: If we should use a custom reader, or the default one.
|
|
||||||
entry: The entry to get the webhook for.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Webhook URL if it has one, returns None if not or error.
|
|
||||||
"""
|
|
||||||
# Get the default reader if we didn't get a custom one.
|
|
||||||
reader: Reader = get_reader() if custom_reader is None else custom_reader
|
|
||||||
|
|
||||||
# Get the webhook from the feed.
|
|
||||||
# Is None if not found or error.
|
|
||||||
try:
|
|
||||||
webhook_url: str = str(reader.get_tag(entry.feed_url, "webhook"))
|
|
||||||
except TagNotFoundError:
|
|
||||||
webhook_url = ""
|
|
||||||
|
|
||||||
return webhook_url
|
|
||||||
|
|
||||||
|
|
||||||
@lru_cache()
|
@lru_cache()
|
||||||
def get_db_location(custom_location: str = "") -> str:
|
def get_db_location(custom_location: str = "") -> str:
|
||||||
"""Where we store the database file.
|
"""Where we store the database file.
|
||||||
|
@ -5,13 +5,7 @@ import tempfile
|
|||||||
from platformdirs import user_data_dir
|
from platformdirs import user_data_dir
|
||||||
from reader import Reader
|
from reader import Reader
|
||||||
|
|
||||||
from discord_rss_bot.settings import (
|
from discord_rss_bot.settings import data_dir, default_custom_message, get_db_location, get_reader
|
||||||
data_dir,
|
|
||||||
default_custom_message,
|
|
||||||
get_db_location,
|
|
||||||
get_reader,
|
|
||||||
get_webhook_for_entry,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_db_location() -> None:
|
def test_get_db_location() -> None:
|
||||||
@ -72,9 +66,6 @@ def test_get_webhook_for_entry() -> None:
|
|||||||
custom_reader.add_feed("https://www.reddit.com/r/movies.rss")
|
custom_reader.add_feed("https://www.reddit.com/r/movies.rss")
|
||||||
custom_reader.update_feed("https://www.reddit.com/r/movies.rss")
|
custom_reader.update_feed("https://www.reddit.com/r/movies.rss")
|
||||||
|
|
||||||
for entry in custom_reader.get_entries():
|
|
||||||
assert get_webhook_for_entry(custom_reader=custom_reader, entry=entry) == ""
|
|
||||||
|
|
||||||
# Add a webhook to the database.
|
# Add a webhook to the database.
|
||||||
custom_reader.set_tag("https://www.reddit.com/r/movies.rss", "webhook", "https://example.com") # type: ignore
|
custom_reader.set_tag("https://www.reddit.com/r/movies.rss", "webhook", "https://example.com") # type: ignore
|
||||||
our_tag: str = custom_reader.get_tag("https://www.reddit.com/r/movies.rss", "webhook") # type: ignore
|
our_tag: str = custom_reader.get_tag("https://www.reddit.com/r/movies.rss", "webhook") # type: ignore
|
||||||
|
Reference in New Issue
Block a user