PapaTutuWawa/akibabot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Format using black

Browse Source
This commit is contained in:
PapaTutuWawa 2023-12-17 00:59:54 +01:00
parent 078e9ba247
commit 6887c3ef89
2 changed files with 29 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
akibabot/main.py
View File

@ -12,6 +12,7 @@ import akibapass_downloader.episode
from akibabot.xmpp import send_notification from akibabot.xmpp import send_notification
@dataclass @dataclass
class Show: class Show:
# List of episode numbers that are locally available. # List of episode numbers that are locally available.
@ -26,23 +27,29 @@ class Show:
# The base URL of the show. # The base URL of the show.
url: str url: str
def build_show_dir_name(name: str, year: int) -> str: def build_show_dir_name(name: str, year: int) -> str:
"""Build the Jellyfin-compatible directory name of the show.""" """Build the Jellyfin-compatible directory name of the show."""
return f"{name} ({year})" return f"{name} ({year})"
def get_episode_numbers(show_name: str, show_year: int, storage_path: Path) -> list[int]:
def get_episode_numbers(
show_name: str, show_year: int, storage_path: Path
) -> list[int]:
"""Computes the episode numbers that are locally available. """Computes the episode numbers that are locally available.
@show_name: The name of the show @show_name: The name of the show
@show_year: The release year of the show. @show_year: The release year of the show.
@storage_path: Path to where the show directories are. @storage_path: Path to where the show directories are.
""" """
episodes_path = storage_path / build_show_dir_name(show_name, show_year) episodes_path = storage_path / build_show_dir_name(show_name, show_year)
if not episodes_path.exists(): if not episodes_path.exists():
return [] return []
return [ return [
int(item.split(".")[0].split(" ")[1]) for item in os.listdir(episodes_path) if (episodes_path / item).is_file() int(item.split(".")[0].split(" ")[1])
for item in os.listdir(episodes_path)
if (episodes_path / item).is_file()
] ]
@ -114,20 +121,29 @@ def main():
logging.debug("Processing %s", show.name) logging.debug("Processing %s", show.name)
episodes_path = storage_path / build_show_dir_name(show.name, show.year) episodes_path = storage_path / build_show_dir_name(show.name, show.year)
if not episodes_path.exists(): if not episodes_path.exists():
logging.info("Episodes directory of %s does not exist. Creating...", show.name) logging.info(
"Episodes directory of %s does not exist. Creating...", show.name
)
episodes_path.mkdir(parents=True) episodes_path.mkdir(parents=True)
episodes_remote = akibapass_downloader.episode.list_episodes( episodes_remote = akibapass_downloader.episode.list_episodes(
show.url, show.url,
) )
episodes_to_download = set([episode.episode_nr for episode in episodes_remote]) - set(show.local_episodes) episodes_to_download = set(
[episode.episode_nr for episode in episodes_remote]
) - set(show.local_episodes)
for episode in episodes_remote: for episode in episodes_remote:
if episode.episode_nr in episodes_to_download: if episode.episode_nr in episodes_to_download:
logging.info(f"Downloading {episode.name} ({episode.episode_nr})...") logging.info(f"Downloading {episode.name} ({episode.episode_nr})...")
downloads = episode.get_downloads(cookies=cookies, filter_quality=akibapass_downloader.episode.Quality.UHD_1440P) downloads = episode.get_downloads(
cookies=cookies,
filter_quality=akibapass_downloader.episode.Quality.UHD_1440P,
)
if not downloads: if not downloads:
logging.warning("Failed to find UHD quality for episode %d", episode.episode_nr) logging.warning(
"Failed to find UHD quality for episode %d", episode.episode_nr
)
continue continue
download = downloads[0] download = downloads[0]

9
akibabot/xmpp.py
View File

@ -1,8 +1,12 @@
import aioxmpp import aioxmpp
async def send_notification(jid: str, password: str, to: str, show_name: str, episode: int):
async def send_notification(
jid: str, password: str, to: str, show_name: str, episode: int
):
"""Sends a notification of a new show download to @to using the account @jid:@password. """Sends a notification of a new show download to @to using the account @jid:@password.
@show_name is the name of show. @episode is the number of the episode that was just downloaded.""" @show_name is the name of show. @episode is the number of the episode that was just downloaded.
"""
client = aioxmpp.PresenceManagedClient( client = aioxmpp.PresenceManagedClient(
aioxmpp.JID.fromstr(jid), aioxmpp.JID.fromstr(jid),
aioxmpp.make_security_layer(password), aioxmpp.make_security_layer(password),
@ -18,4 +22,3 @@ async def send_notification(jid: str, password: str, to: str, show_name: str, ep
show_name, show_name,
) )
await stream.send(msg) await stream.send(msg)