Added uncached downloader #16

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

Priky-one wants to merge 11 commits into westsurname:dev from Priky-one:dev

.env.template

-Original file line number
+Diff line change
@@ Expand Up / @@ -37,7 +37,8 @@ BLACKHOLE_RADARR_PATH="Movies" @@
     BLACKHOLE_SONARR_PATH="TV Shows"
     BLACKHOLE_FAIL_IF_NOT_CACHED=true
     BLACKHOLE_RD_MOUNT_REFRESH_SECONDS=200
-    BLACKHOLE_WAIT_FOR_TORRENT_TIMEOUT=60
+    BLACKHOLE_WAIT_FOR_TORRENT_TIMEOUT=51840
+    BLACKHOLE_WAIT_FOR_PROGRESS_CHANGE=720
     BLACKHOLE_HISTORY_PAGE_SIZE=500
     DISCORD_ENABLED=false
@@ Expand Down @@

Dockerfile.blackhole

-Original file line number
+Diff line change
@@ -1,4 +1,4 @@
-    FROM python:3.8-slim
+    FROM python:3.11-slim
     # Metadata labels
     LABEL org.opencontainers.image.source="https://github.com/westsurname/scripts"
@@ Expand Down @@

blackhole.py

Large diffs are not rendered by default.

blackhole_downloader.py

-Original file line number
+Diff line change
@@ -0,0 +1,262 @@
+    import asyncio
+    import os
+    import glob
+    from shared.discord import discordError, discordUpdate, discordStatusUpdate
+    from shared.shared import blackhole
+    import re
+    webhook = None
+    async def downloader(torrent, file, arr, torrentFile, shared_dict, lock):
+        from blackhole import refreshArr
+        availableHost = torrent.getAvailableHost()
+        activeTorrents = torrent.getActiveTorrents()
+        global webhook
+        while True:
+            if activeTorrents['limit'] - activeTorrents['nb'] -2 > 0:
+                allTorrents = torrent.getAllTorrents()
+                torrentExists = False
+                for at in allTorrents:
+                    if at["hash"] == torrent.getHash().lower():
+                        torrent.id = at["id"]
+                        torrentName = at["filename"]
+                        if at["status"] == "downloaded":
+                            print("File already exists")
+                        elif at["status"] == "downloading":
+                            print("File downloading")
+                        torrentExists = True
+                        lock.acquire()
+                        if torrentName in shared_dict:
+                            lock.release()
+                            remove_file(torrentFile, lock, shared_dict, "", current=True)
+                            return
+                        lock.release()
+                        break
+                if not torrentExists:
+                    while True:
+                        folder_path = os.path.dirname(torrentFile)
+                        all_files = glob.glob(os.path.join(folder_path, '*'))
+                        all_files.sort(key=os.path.getctime)
+                        top_4_files = all_files[:4]
+                        if torrentFile in top_4_files:
+                            torrent.getInstantAvailability()
+                            if torrent.addTorrent(availableHost) is None:
+                                remove_file(torrentFile, lock, shared_dict, "", current=True)
+                                return
+                            info = torrent.getInfo(refresh=True)
+                            torrentName = info['filename']
+                            lock.acquire()
+                            try:
+                                if shared_dict:
+                                    shared_dict.update({torrentName : "added"})
+                                    webhook = discordStatusUpdate(shared_dict, webhook, edit=True)
+                                else:
+                                    shared_dict.update({torrentName : "added"})
+                                    webhook = discordStatusUpdate(shared_dict, webhook)
+                            finally:
+                                lock.release()
+                            break
+                        if not os.path.exists(torrentFile):
+                            remove_file(torrentFile, lock, shared_dict, torrentName, current=True)
+                            return
+                        await asyncio.sleep(60)
+                break
+            await asyncio.sleep(60)
+        count = 0
+        progress=0
+        waitForProgress=0
+        while True:
+            count += 1
+            info = torrent.getInfo(refresh=True)
+            if "status" not in info:
+                remove_file(torrentFile, lock, shared_dict, torrentName, current=True)
+                return
+            status = info['status']
+            if torrentName != info['filename']:
+                lock.acquire()
+                try:
+                    if torrentName in shared_dict:
+                        del shared_dict[torrentName]
+                    if shared_dict and webhook and webhook.id:
+                        webhook = discordStatusUpdate(shared_dict, webhook, edit=True)
+                    elif not shared_dict and webhook and webhook.id:
+                        webhook = discordStatusUpdate(shared_dict, webhook, delete=True)
+                finally:
+                    lock.release()
+            torrentName = info['filename']
+            print('status:', status)
+            if not os.path.exists(torrentFile):
+                torrent.delete()
+                break
+            if status == 'waiting_files_selection':
+                if not torrent.selectFiles(uncached=True):
+                    torrent.delete()
+                    break
+            elif status == 'magnet_conversion' or status == 'queued' or status == 'compressing' or status == 'uploading':
+                await asyncio.sleep(1)
+            elif status == 'downloading':
+                if progress != info['progress']:
+                    progress = info['progress']
+                    waitForProgress = 0
+                elif waitForProgress >= blackhole['waitForProgressChange']:
+                    torrent.delete()
+                    remove_file(torrentFile, lock, shared_dict, torrentName, current=True)
+                    return
+                else:
+                    waitForProgress += 1
+                print(progress)
+                lock.acquire()
+                try:
+                    if shared_dict:
+                        shared_dict.update({torrentName : f"Downloading {progress}%"})
+                        webhook = discordStatusUpdate(shared_dict, webhook, edit=True)
+                    else:
+                        shared_dict.update({torrentName : f"Downloading {progress}%"})
+                        webhook = discordStatusUpdate(shared_dict, webhook)
+                finally:
+                    lock.release()
+                if torrent.incompatibleHashSize and torrent.failIfNotCached:
+                    print("Non-cached incompatible hash sized torrent")
+                    torrent.delete()
+                    break
+                await asyncio.sleep(60)
+            elif status == 'magnet_error' or status == 'error' or status == 'dead' or status == 'virus':
+                discordError(f"Error: {file.fileInfo.filenameWithoutExt}", info)
+                torrent.delete()
+                break
+            elif status == 'downloaded':
+                existsCount = 0
+                print('Waiting for folders to refresh...')
+                filename = info.get('filename')
+                originalFilename = info.get('original_filename')
+                folderPathMountFilenameTorrent = os.path.join(blackhole['rdMountTorrentsPath'], filename)
+                folderPathMountOriginalFilenameTorrent = os.path.join(blackhole['rdMountTorrentsPath'], originalFilename)
+                folderPathMountOriginalFilenameWithoutExtTorrent = os.path.join(blackhole['rdMountTorrentsPath'], os.path.splitext(originalFilename)[0])
+                while True:
+                    existsCount += 1
+                    if os.path.exists(folderPathMountFilenameTorrent) and os.listdir(folderPathMountFilenameTorrent):
+                        folderPathMountTorrent = folderPathMountFilenameTorrent
+                    elif os.path.exists(folderPathMountOriginalFilenameTorrent) and os.listdir(folderPathMountOriginalFilenameTorrent):
+                        folderPathMountTorrent = folderPathMountOriginalFilenameTorrent
+                    elif (originalFilename.endswith(('.mkv', '.mp4')) and
+                            os.path.exists(folderPathMountOriginalFilenameWithoutExtTorrent) and os.listdir(folderPathMountOriginalFilenameWithoutExtTorrent)):
+                        folderPathMountTorrent = folderPathMountOriginalFilenameWithoutExtTorrent
+                    else:
+                        folderPathMountTorrent = None
+                    if folderPathMountTorrent:
+                        multiSeasonRegex1 = r'(?<=[\W_][Ss]eason[\W_])[\d][\W_][\d]{1,2}(?=[\W_])'
+                        multiSeasonRegex2 = r'(?<=[\W_][Ss])[\d]{2}[\W_][Ss]?[\d]{2}(?=[\W_])'
+                        multiSeasonRegexCombined = f'{multiSeasonRegex1}|{multiSeasonRegex2}'
+                        multiSeasonMatch = re.search(multiSeasonRegexCombined, file.fileInfo.filenameWithoutExt)
+                        for root, dirs, files in os.walk(folderPathMountTorrent):
+                            relRoot = os.path.relpath(root, folderPathMountTorrent)
+                            for filename in files:
+                                # Check if the file is accessible
+                                # if not await is_accessible(os.path.join(root, filename)):
+                                #     print(f"Timeout reached when accessing file: {filename}")
+                                #     discordError(f"Timeout reached when accessing file", filename)
+                                    # Uncomment the following line to fail the entire torrent if the timeout on any of its files are reached
+                                    # fail(torrent)
+                                    # return
+                                if multiSeasonMatch:
+                                    seasonMatch = re.search(r'S([\d]{2})E[\d]{2}', filename)
+                                    if seasonMatch:
+                                        season = seasonMatch.group(1)
+                                        seasonShort = season[1:] if season[0] == '0' else season
+                                        seasonFolderPathCompleted = re.sub(multiSeasonRegex1, seasonShort, file.fileInfo.folderPathCompleted)
+                                        seasonFolderPathCompleted = re.sub(multiSeasonRegex2, season, seasonFolderPathCompleted)
+                                        os.makedirs(os.path.join(seasonFolderPathCompleted, relRoot), exist_ok=True)
+                                        os.symlink(os.path.join(root, filename), os.path.join(seasonFolderPathCompleted, relRoot, filename))
+                                        print('Season Recursive:', f"{os.path.join(seasonFolderPathCompleted, relRoot, filename)} -> {os.path.join(root, filename)}")
+                                        continue
+                                os.makedirs(os.path.join(file.fileInfo.folderPathCompleted, relRoot), exist_ok=True)
+                                os.symlink(os.path.join(root, filename), os.path.join(file.fileInfo.folderPathCompleted, relRoot, filename))
+                                print('Recursive:', f"{os.path.join(file.fileInfo.folderPathCompleted, relRoot, filename)} -> {os.path.join(root, filename)}")
+                        print('Refreshed')
+                        discordUpdate(f"Sucessfully processed {file.fileInfo.filenameWithoutExt}", f"Now available for immediate consumption! existsCount: {existsCount}")
+                        await refreshArr(arr)
+                        break
+                    if existsCount >= blackhole['rdMountRefreshSeconds'] + 1:
+                        print(f"Torrent folder not found in filesystem: {file.fileInfo.filenameWithoutExt}")
+                        discordError("Torrent folder not found in filesystem", file.fileInfo.filenameWithoutExt)
+                        return False
+                    await asyncio.sleep(1)
+                break
+            if torrent.failIfNotCached:
+                if count == 21 and status != "downloading":
+                    print('infoCount > 20')
+                    discordError(f"{file.fileInfo.filenameWithoutExt} info attempt count > 20", status)
+                elif count == blackhole['waitForTorrentTimeout']:
+                    print(f"infoCount == {blackhole['waitForTorrentTimeout']} - Failing")
+                    torrent.delete()
+                    break
+        remove_file(torrentFile, lock, shared_dict, torrentName)
+    def remove_file(torrentFile, lock, shared_dict, torrentName, current=False):
+        if os.path.exists(torrentFile):
+            folder_path = os.path.dirname(torrentFile)
+            all_files = glob.glob(os.path.join(folder_path, '*'))
+            all_files.sort(key=os.path.getctime)
+            try:
+                file_index = all_files.index(torrentFile)
+            except ValueError:
+                print("The file does not exist in the folder.")
+                file_index = -1
+            if file_index != -1:
+                if current:
+                    files_to_remove = [all_files[file_index]]
+                else:
+                    files_to_remove = all_files[file_index:]
+                for file in files_to_remove:
+                    try:
+                        os.remove(file)
+                        print(f"Removed: {file}")
+                    except Exception as e:
+                        print(f"Error removing {file}: {e}")
+                remaining_files = os.listdir(folder_path)
+                if not remaining_files:
+                    try:
+                        os.rmdir(folder_path)
+                        print(f"Removed folder: {folder_path}")
+                    except Exception as e:
+                        print(f"Error removing folder {folder_path}: {e}")
+                else:
+                    print(f"Folder is not empty: {folder_path}")
+            else:
+                print("The specified file is not in the folder.")
+        else:
+            print("The file does not exist.")
+        lock.acquire()
+        try:
+            global webhook
+            if torrentName in shared_dict:
+                del shared_dict[torrentName]
+            if shared_dict and webhook and webhook.id:
+                webhook = discordStatusUpdate(shared_dict, webhook, edit=True)
+            elif not shared_dict and webhook and webhook.id:
+                webhook = discordStatusUpdate(shared_dict, webhook, delete=True)
+        finally:
+            lock.release()

blackhole_watcher.py

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -1,28 +1,36 @@
  
    import asyncio

    from watchdog.observers import Observer

    from watchdog.events import FileSystemEventHandler

    from blackhole import start, getPath

    from blackhole import start, resumeUncached, getPath

    import threading

    class BlackholeHandler(FileSystemEventHandler):

        def __init__(self, is_radarr):

        def __init__(self, is_radarr, lock):

            super().__init__()

            self.is_processing = False

            self.is_radarr = is_radarr

            self.path_name = getPath(is_radarr, create=True)

            self.lock = lock

        def on_created(self, event):

            if not self.is_processing and not event.is_directory and event.src_path.lower().endswith((".torrent", ".magnet")):

                self.is_processing = True

                try:

                    start(self.is_radarr)

                    start(self.is_radarr, self.lock)

                finally:

                    self.is_processing = False

    async def scheduleResumeUncached(lock):

        await resumeUncached(lock)

    if __name__ == "__main__":

        print("Watching blackhole")

        lock = threading.Lock()

        radarr_handler = BlackholeHandler(is_radarr=True)

        sonarr_handler = BlackholeHandler(is_radarr=False)

        radarr_handler = BlackholeHandler(is_radarr=True, lock=lock)

        sonarr_handler = BlackholeHandler(is_radarr=False, lock=lock)

        radarr_observer = Observer()

        radarr_observer.schedule(radarr_handler, radarr_handler.path_name)

    @@ -33,6 +41,7 @@ def on_created(self, event):
  
        try:

            radarr_observer.start()

            sonarr_observer.start()

            asyncio.run(scheduleResumeUncached(lock))

        except KeyboardInterrupt:

            radarr_observer.stop()

            sonarr_observer.stop()

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added uncached downloader #16

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Added uncached downloader #16

Are you sure you want to change the base?

Uh oh!

Added uncached downloader #16

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!