Changed program to be a CLI for nyaa.si

nyaacli/colors.py

@@ -0,0 +1,42 @@
+from inquirer.themes import Theme
+from blessings import Terminal
+from colorama import init
+
+init()
+
+
+def red(text: str) -> str:
+    return f'\033[31m{text}\033[0m'
+
+
+def green(text: str) -> str:
+    return f'\33[92m{text}\033[0m'
+
+
+def yellow(text: str) -> str:
+    return f'\33[33m{text}\033[0m'
+
+
+def blue(text: str) -> str:
+    return f'\33[34m{text}\033[0m'
+
+
+class PromptTheme(Theme):
+    def __init__(self):
+        super(PromptTheme, self).__init__()
+
+        term = Terminal()
+
+        self.Question.mark_color = term.yellow
+        self.Question.brackets_color = term.normal
+        self.Question.default_color = term.normal
+        self.Editor.opening_prompt_color = term.bright_black
+        self.Checkbox.selection_color = term.blue
+        self.Checkbox.selection_icon = '❯'
+        self.Checkbox.selected_icon = '◉'
+        self.Checkbox.selected_color = term.yellow + term.bold
+        self.Checkbox.unselected_color = term.normal
+        self.Checkbox.unselected_icon = '◯'
+        self.List.selection_color = term.blue
+        self.List.selection_cursor = '❯'
+        self.List.unselected_color = term.normal

nyaacli/nyaa_search.py

@@ -0,0 +1,165 @@
+from urllib import request
+from typing import Optional, List, Tuple
+from dataclasses import dataclass
+import os
+import feedparser
+
+from guessit import guessit
+import inquirer
+
+from nyaacli.colors import red, yellow, green, PromptTheme
+
+
+def get_file_extension(path: str) -> str:
+    """
+    Gets the File extension from the path to a file
+
+    'asd.txt' -> '.txt'
+    'asd' -> ''
+    '/path/to/asd.mkv' -> '.mkv'
+    """
+    filename, extension = os.path.splitext(path)
+
+    return extension
+
+
+def text_break() -> None:
+    print('-' * 10, end='\n')
+
+
+@dataclass
+class Entry:
+    title: str
+    link: str
+    original_title: str
+    extension: str
+    full_title: Optional[str]
+    episode: Optional[int]
+    other: Optional[str]
+    release_group: Optional[str]
+    screen_size: Optional[str]
+    alternative_title: Optional[str]
+    display_title: Optional[str]
+    season: Optional[str]
+    seeders: Optional[str]
+    size: Optional[str]
+    episode_title: Optional[str]
+
+
+def search_torrent(search: str, episode: Optional[int] = None, dub: bool = False) -> Optional[Tuple[str, str]]:
+    """
+    Results a tuple with (Path to .torrent file, Result name of video file)
+    Nyaa.si rss search flags
+    https://github.com/nyaadevs/nyaa/blob/a38e5d5b53805ecb1d94853d849826f948f07aad/nyaa/views/main.py#L65
+    """
+
+    search_query = f"{search}".replace(' ', '%20')
+    if episode:
+        search_query += f" {episode}".replace(' ', '%20')
+
+    # Parse Nyaa.si rss feed search
+    feed: feedparser.FeedParserDict = feedparser.parse(f"https://nyaa.si/rss?c=1_2&q={search_query}&s=seeders&o=desc")
+
+    entries: List[Entry] = []
+
+    for entry in feed['entries']:
+        title = guessit(entry['title'])
+
+        if not title.get('screen_size'):
+            # Ignore entries without Screen size property
+            continue
+
+        if not dub and 'dub' in entry['title'].lower():
+            # Ignore entries with 'dub' in their titles if dub=False
+            continue
+
+        # Screen size needs to be higher than 480p
+        good_size = int(title.get('screen_size').replace('p', '')) > 480
+
+        if title.get('episode') == episode and title.get('type') == 'episode' and good_size:
+            entries.append(Entry(
+                link=entry['link'],
+                size=entry['nyaa_size'],
+                original_title=entry['title'],
+                display_title=entry['title'],
+                extension=get_file_extension(entry['title']),
+                title=title.get('title'),
+                season=title.get('season'),
+                seeders=entry.get('nyaa_seeders'),
+                full_title='',
+                episode=title.get('episode'),
+                episode_title=title.get('episode_title'),
+                other=title.get('other'),
+                release_group=title.get('release_group'),
+                screen_size=title.get('screen_size'),
+                alternative_title=title.get('alternative_title')
+            ))
+
+    if not entries:
+        print(red(f'No results found for search: \'{search_query.replace("%20", " ")}\''))
+        return None
+
+    for entry in entries[:5]:
+        entry_title = entry.title
+        entry.full_title = entry.title
+
+        if entry.alternative_title:
+            entry_title += f" - {entry.alternative_title}"
+            entry.full_title += f" - {entry.alternative_title}"
+
+        if entry.episode or entry.episode_title:
+            if entry.episode:
+                entry_title += f" - Episode {entry.episode}"
+                entry.full_title += f" - Episode {entry.episode}"
+            else:
+                entry_title += f" - Episode {entry.episode_title}"
+                entry.full_title += f" - Episode {entry.episode_title}"
+
+        if entry.season:
+            entry_title += f" - Season {entry.season}"
+            entry.full_title += f" - Season {entry.season}"
+
+        if entry.release_group:
+            entry_title += f" ({entry.release_group})"
+
+        if entry.screen_size:
+            entry_title += f" {entry.screen_size}"
+
+        if entry.seeders:
+            seeders = f'{entry.seeders} Seeders'
+
+            if int(entry.seeders) > 40:
+                entry_title += f" - {green(seeders)}"
+            elif 40 > int(entry.seeders) > 20:
+                entry_title += f" - {yellow(seeders)}"
+            else:
+                entry_title += f" - {red(seeders)}"
+
+        if entry.size:
+            entry_title += f" - {entry.size}"
+
+        entry.display_title = entry_title
+
+    questions = [
+        inquirer.List(
+            'entry',
+            message=green("Select one of the entries below"),
+            choices=[(str(entry.display_title), index) for index, entry in enumerate(entries[:5])],
+        ),
+    ]
+
+    answer = inquirer.prompt(questions, theme=PromptTheme())
+
+    index_choice = answer['entry'] - 1
+
+    entry_choice = entries[index_choice]
+
+    final_path = entry_choice.full_title
+
+    torrent_path = f'/tmp/{final_path}.torrent'
+
+    print(f"{green('[Downloading]')} '{torrent_path}'")
+
+    request.urlretrieve(entry_choice.link, torrent_path)
+
+    return torrent_path, final_path + entry_choice.extension

nyaacli/torrenting.py

@@ -0,0 +1,66 @@
+import os
+import sys
+import time
+
+import libtorrent
+
+from nyaacli.colors import red, green
+
+
+def download_torrent(filename: str, result_filename: str = None, show_progress: bool = True, base_path: str = 'Anime'):
+    session = libtorrent.session({'listen_interfaces': '0.0.0.0:6881'})
+
+    base_path = os.path.expanduser(base_path)
+
+    info = libtorrent.torrent_info(filename)
+
+    handle: libtorrent.torrent_handle = session.add_torrent({
+        'ti': info,
+        'save_path': base_path
+    })
+
+    status: libtorrent.session_status = handle.status()
+
+    print(f"\n{green('◉ Started downloading torrent:')} {status.name}", end='\n\n')
+
+    if show_progress:
+        while not status.is_seeding:
+            status = handle.status()
+
+            progress = green(f"{status.progress * 100:.2f}%")
+            download_rate = green(f"{status.download_rate / 1000:.1f} kB/s")
+
+            sys.stdout.write(
+                f'\r◉ {green(str(status.state).title())} - {progress} '
+                f'(Download: {download_rate} - Upload: {status.upload_rate / 1000:.1f} kB/s - '
+                f'Peers: {status.num_peers}) '
+            )
+
+            alerts = session.pop_alerts()
+
+            alert: libtorrent.alert
+            for alert in alerts:
+                if alert.category() & libtorrent.alert.category_t.error_notification:
+                    sys.stdout.write(f"\r{red('[Alert]')} {alert} ")
+
+            sys.stdout.flush()
+
+            time.sleep(1)
+
+        print(green(f'\n\nFinished downloading {handle.name()}'), end='\n\n')
+
+    if result_filename:
+        old_name = f'{base_path}/{status.name}'
+        new_name = f'{base_path}/{result_filename}'
+
+        os.rename(old_name, new_name)
+
+        print(f'Finished download, at: \'{green(new_name)}\' ')
+
+
+if __name__ == '__main__':
+    if len(sys.argv) < 2:
+        print('You need to pass in the .torrent file path.')
+        sys.exit(1)
+
+    download_torrent(sys.argv[1])