Merge pull request #177 from arabcoders/dev

Refactoring the code to use shared manager

.github/workflows/main.yml

@@ -45,6 +45,8 @@ jobs:
         with:
           filename: "/app/library/version.py"
           placeholder: "dev-master"
+          with_date: "true"
+          with_branch: "true"
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v3
@@ -92,7 +94,7 @@ jobs:
           cache-to: type=gha, scope=${{ github.workflow }}
 
       - name: Version tag
-        if: github.event_name == 'disabled_now'
+        if: github.event_name == 'push'
         uses: arabcoders/action-python-autotagger@master
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
@@ -100,9 +102,10 @@ jobs:
           path: app/library/version.py
           variable: APP_VERSION
 
+
   dockerhub-sync-readme:
     needs: push-build
-    if: github.event_name == 'push'
+    if: github.ref == 'refs/heads/master' && github.event_name == 'push'
     runs-on: ubuntu-latest
     steps:
       - name: Sync README
@@ -120,12 +123,14 @@ jobs:
   create_release:
     needs: push-build
     runs-on: ubuntu-latest
+    # Only run on master push
     if: github.ref == 'refs/heads/master' && github.event_name == 'push' && success()
+
     steps:
-      - name: Checkout
+      - name: Check out code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0 # so we can git log the full history
+          fetch-depth: 0 # so we can do git log for the full history
 
       - name: Get commits between old and new
         id: commits
@@ -147,37 +152,24 @@ jobs:
           echo "$LOG" >> "$GITHUB_ENV"
           echo "EOF" >> "$GITHUB_ENV"
 
-      - name: Create local tag matching Docker's raw pattern
-        id: createtag
+      - name: Fetch the most recent tag
+        id: last_tag
         run: |
-          # We'll replicate raw format: branch + base_ref + date + short sha
-          # If base_ref is empty (normal push to master, no PR?), it won't appear.
-          BRANCH_NAME="${{ github.ref_name }}"
-          BASE_REF="${{ github.base_ref }}"
-          CURRENT_SHA="${{ github.event.after }}"
-          DATE=$(date +%Y%m%d)
-          SHORT_SHA=$(echo "$CURRENT_SHA" | cut -c1-7)
+          # Make sure we have all tags
+          git fetch --tags
 
-          TAG="${BRANCH_NAME}${BASE_REF}-${DATE}-${SHORT_SHA}"
+          # Get the latest tag by commit date (most recent)
+          # If there are no tags at all, set a fallback
+          LAST_TAG=$(git describe --tags --abbrev=0 $(git rev-list --tags --max-count=1) 2>/dev/null || echo "no-tags-found")
 
-          echo "Using tag: $TAG"
+          echo "Latest tag found: $LAST_TAG"
+          echo "LAST_TAG=$LAST_TAG" >> "$GITHUB_OUTPUT"
 
-          git config user.name "github-actions"
-          git config user.email "github-actions@github.com"
-
-          # Create a lightweight local tag
-          git tag "$TAG" "$CURRENT_SHA"
-
-          # Push the tag back to GitHub
-          git push origin "$TAG"
-
-          # Expose as an output
-          echo "TAG=$TAG" >> "$GITHUB_OUTPUT"
-      - name: Create GitHub Release
+      - name: Create or update GitHub Release using last tag
         uses: softprops/action-gh-release@master
         with:
-          tag_name: ${{ steps.createtag.outputs.TAG }}
-          name: "${{ steps.createtag.outputs.TAG }}"
+          tag_name: ${{ steps.last_tag.outputs.LAST_TAG }}
+          name: "${{ steps.last_tag.outputs.LAST_TAG }}"
           body: ${{ env.LOG }}
           draft: false
           prerelease: false

app/library/Download.py

@@ -6,8 +6,6 @@ import multiprocessing
 import os
 import shutil
 import time
-
-from multiprocessing.managers import SyncManager
 from email.utils import formatdate
 
 import yt_dlp
@@ -15,9 +13,9 @@ import yt_dlp
 from .AsyncPool import Terminator
 from .config import Config
 from .Emitter import Emitter
+from .ffprobe import ffprobe
 from .ItemDTO import ItemDTO
 from .Utils import get_opts, jsonCookie, mergeConfig
-from .ffprobe import ffprobe
 
 LOG = logging.getLogger("download")
 
@@ -38,13 +36,14 @@ class Download:
     debug: bool = False
     tempPath: str = None
     emitter: Emitter = None
-    manager: SyncManager = None
     canceled: bool = False
     is_live: bool = False
     info_dict: dict = None
     "yt-dlp metadata dict."
     update_task = None
 
+    cancel_in_progress: bool = False
+
     bad_live_options: list = [
         "concurrent_fragment_downloads",
         "fragment_retries",
@@ -188,9 +187,8 @@ class Download:
         LOG.info(f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" completed.')
 
     async def start(self, emitter: Emitter):
-        self.manager = multiprocessing.Manager()
         self.emitter = emitter
-        self.status_queue = self.manager.Queue()
+        self.status_queue = Config.get_manager().Queue()
 
         # Create temp dir for each download.
         self.tempPath = os.path.join(self.temp_dir, hashlib.shake_256(f"D-{self.info.id}".encode("utf-8")).hexdigest(5))
@@ -223,12 +221,15 @@ class Download:
         Close download process.
         This method MUST be called to clear resources.
         """
-        if not self.started():
+        if not self.started() or self.cancel_in_progress:
             return False
 
+        self.cancel_in_progress = True
         procId = self.proc.ident
+
+        LOG.info(f"Closing PID='{procId}' download process.")
+
         try:
-            LOG.info(f"Closing download process: '{procId}'.")
             try:
                 if "update_task" in self.__dict__ and self.update_task.done() is False:
                     self.update_task.cancel()
@@ -239,19 +240,16 @@ class Download:
             self.kill()
 
             loop = asyncio.get_running_loop()
-            tasks = []
 
             if self.proc.is_alive():
-                tasks.append(loop.run_in_executor(None, self.proc.join))
+                LOG.debug(f"Waiting for PID='{procId}' to close.")
+                await loop.run_in_executor(None, self.proc.join)
+                LOG.debug(f"PID='{procId}' closed.")
 
             if self.status_queue:
+                LOG.debug(f"Closing status queue for PID='{procId}'.")
                 self.status_queue.put(Terminator())
-
-            if self.manager:
-                tasks.append(loop.run_in_executor(None, self.manager.shutdown))
-
-            if len(tasks) > 0:
-                await asyncio.gather(*tasks)
+                LOG.debug(f"Closed status queue for PID='{procId}'.")
 
             if self.proc:
                 self.proc.close()
@@ -259,7 +257,7 @@ class Download:
 
             self.delete_temp()
 
-            LOG.debug(f"Closed download process: '{procId}'.")
+            LOG.debug(f"Closed PID='{procId}' download process.")
 
             return True
         except Exception as e:
@@ -293,7 +291,7 @@ class Download:
         if self.tempKeep is True or not self.tempPath:
             return
 
-        if self.info.status != "finished" and self.is_live:
+        if "finished" != self.info.status and self.is_live:
             LOG.warning(
                 f"Keeping live temp folder '{self.tempPath}', as the reported status is not finished '{self.info.status}'."
             )
@@ -320,11 +318,9 @@ class Download:
                 self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get)
                 status = await self.update_task
             except (asyncio.CancelledError, OSError, FileNotFoundError):
-                LOG.debug(f"Closing progress update for: {self.info._id=}.")
                 return
 
-            if status is None or status.__class__ is Terminator:
-                LOG.debug(f"Closing progress update for: {self.info._id=}.")
+            if status is None or isinstance(status, Terminator):
                 return
 
             if status.get("id") != self.id or len(status) < 2:
@@ -352,7 +348,7 @@ class Download:
             self.info.status = status.get("status", self.info.status)
             self.info.msg = status.get("msg")
 
-            if self.info.status == "error" and "error" in status:
+            if "error" == self.info.status and "error" in status:
                 self.info.error = status.get("error")
                 asyncio.create_task(
                     self.emitter.error(message=self.info.error, data=self.info), name=f"emitter-e-{self.id}"
@@ -361,7 +357,10 @@ class Download:
             if "downloaded_bytes" in status:
                 total = status.get("total_bytes") or status.get("total_bytes_estimate")
                 if total:
-                    self.info.percent = status["downloaded_bytes"] / total * 100
+                    try:
+                        self.info.percent = status["downloaded_bytes"] / total * 100
+                    except ZeroDivisionError:
+                        self.info.percent = 0
                     self.info.total_bytes = total
 
             self.info.speed = status.get("speed")

app/library/config.py

@@ -1,16 +1,18 @@
 import json
 import logging
+import multiprocessing
 import os
 import re
 import sys
 import time
+from multiprocessing.managers import SyncManager
 from pathlib import Path
 
 import coloredlogs
 from dotenv import load_dotenv
 from yt_dlp.version import __version__ as YTDLP_VERSION
 
-from .Utils import load_file, mergeDict, IGNORED_KEYS
+from .Utils import IGNORED_KEYS, load_file, mergeDict
 from .version import APP_VERSION
 
 
@@ -131,11 +133,20 @@ class Config:
         "default_preset",
     )
 
+    _manager: SyncManager | None = None
+
     @staticmethod
     def get_instance():
         """Static access method."""
         return Config() if not Config.__instance else Config.__instance
 
+    @staticmethod
+    def get_manager() -> SyncManager:
+        if not Config._manager:
+            Config._manager = multiprocessing.Manager()
+
+        return Config._manager
+
     def __init__(self):
         """Virtually private constructor."""
         if Config.__instance is not None: