Created a downloader.

Author: MahmoudAtef999

utils/downloader/__init__.py

@@ -0,0 +1,40 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from .models import Base, DownloadItem
+from datetime import datetime
+
+class DownloadDB:
+    def __init__(self, db_url="sqlite:///downloads.db"):
+        self.engine = create_engine(db_url, echo=False)
+        Base.metadata.create_all(self.engine)
+        self.Session = sessionmaker(bind=self.engine)
+
+    def upsert(self, item_data: dict):
+        session = self.Session()
+        try:
+            obj = session.query(DownloadItem).filter_by(id=item_data["id"]).first()
+            if obj:
+                for key, value in item_data.items():
+                    setattr(obj, key, value)
+            else:
+                item_data.setdefault("created_at", datetime.utcnow())
+                obj = DownloadItem(**item_data)
+                session.add(obj)
+            session.commit()
+        finally:
+            session.close()
+
+    def all(self):
+        session = self.Session()
+        try:
+            return session.query(DownloadItem).all()
+        finally:
+            session.close()
+
+    def delete(self, download_id: str):
+        session = self.Session()
+        try:
+            session.query(DownloadItem).filter_by(id=download_id).delete()
+            session.commit()
+        finally:
+            session.close()

utils/downloader/db.py

@@ -0,0 +1,9 @@
+from enum import Enum
+
+class DownloadStatus(Enum):
+    PAUSED = "paused"
+    RESUMED = "resumed"
+    CANCELLED = "cancelled"
+    COMPLETED = "completed"
+    ERROR = "error"
+    PENDING = "pending"

utils/downloader/enums.py

@@ -0,0 +1,121 @@
+import os
+import uuid
+from PyQt6.QtCore import QObject, QThreadPool, pyqtSignal
+from .worker import DownloadWorker
+from .enums import DownloadStatus
+from .db import DownloadDB
+
+class DownloaderManager(QObject):
+    progress = pyqtSignal(str, str, int, int, int)
+    finished = pyqtSignal(str, str)
+    error = pyqtSignal(str, str)
+    status = pyqtSignal(str, DownloadStatus)
+
+    def __init__(self, urls=None, default_folder="downloads", max_workers=3):
+        super().__init__()
+        self.default_folder = default_folder
+        self._pause_all = False
+        self._cancel_all = False
+        self.db = DownloadDB()
+        self.pool = QThreadPool.globalInstance()
+        self.pool.setMaxThreadCount(max_workers)
+
+        self._downloads = {}  # id → metadata
+        self._load_state()
+        if urls:
+            self._add_new_downloads(urls)
+
+    def _load_state(self):
+        for item in self.db.all():
+            self._downloads[item.id] = {
+                "id": item.id,
+                "url": item.url,
+                "filename": item.filename,
+                "folder_path": item.folder_path,
+                "status": item.status,
+                "downloaded_bytes": item.downloaded_bytes,
+                "total_bytes": item.total_bytes,
+                "file_hash": item.file_hash
+            }
+
+    def _add_new_downloads(self, urls):
+        for url in urls:
+            filename = os.path.basename(url)
+            folder = self.default_folder
+            download_id = str(uuid.uuid4())
+            new_entry = {
+                "id": download_id,
+                "url": url,
+                "filename": filename,
+                "folder_path": folder,
+                "status": DownloadStatus.PENDING.value
+            }
+            self.db.upsert(new_entry)
+            self._downloads[download_id] = new_entry
+
+    def start(self):
+        for download_id, info in self._downloads.items():
+            if info["status"] in (DownloadStatus.COMPLETED.value, DownloadStatus.CANCELLED.value):
+                continue
+            worker = DownloadWorker(info, {
+                "progress": self._on_progress,
+                "finished": self._on_finished,
+                "status": self._on_status,
+                "error": self._on_error,
+            }, manager=self)
+            self._downloads[download_id]["worker"] = worker
+            self.pool.start(worker)
+
+    def _on_progress(self, download_id, filename, downloaded, total, percent):
+        self.progress.emit(download_id, filename, downloaded, total, percent)
+
+    def _on_status(self, download_id, status):
+        self._downloads[download_id]["status"] = status.value
+        self.status.emit(download_id, status)
+
+    def _on_error(self, download_id, message):
+        self.error.emit(download_id, message)
+
+    def _on_finished(self, download_id, filename):
+        self.finished.emit(download_id, filename)
+
+    def pause(self, download_id):
+        if worker := self._downloads.get(download_id, {}).get("worker"):
+            worker.pause()
+
+    def resume(self, download_id):
+        if worker := self._downloads.get(download_id, {}).get("worker"):
+            worker.resume()
+
+    def cancel(self, download_id):
+        if worker := self._downloads.get(download_id, {}).get("worker"):
+            worker.cancel()
+
+    def pause_all(self):
+        self._pause_all = True
+        for info in self._downloads.values():
+            if info.get("worker"):
+                info["worker"].pause()
+
+    def resume_all(self):
+        self._pause_all = False
+        for info in self._downloads.values():
+            if info.get("worker"):
+                info["worker"].resume()
+
+    def cancel_all(self):
+        self._cancel_all = True
+        for info in self._downloads.values():
+            if info.get("worker"):
+                info["worker"].cancel()
+
+    def get_download_map(self):
+        return {
+            download_id: {
+                "url": data["url"],
+                "filename": data["filename"],
+                "folder_path": data["folder_path"],
+                "status": data["status"]
+            }
+            for download_id, data in self._downloads.items()
+        }

utils/downloader/manager.py

@@ -0,0 +1,18 @@
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import Column, String, Integer, DateTime
+from datetime import datetime
+
+Base = declarative_base()
+
+class DownloadItem(Base):
+    __tablename__ = "downloads"
+
+    id = Column(String, primary_key=True)
+    url = Column(String, nullable=False)
+    filename = Column(String)
+    folder_path = Column(String)
+    status = Column(String)
+    downloaded_bytes = Column(Integer, default=0)
+    total_bytes = Column(Integer, default=0)
+    file_hash = Column(String, nullable=True)
+    created_at = Column(DateTime, default=datetime.now)

utils/downloader/models.py

@@ -0,0 +1,83 @@
+import os
+import requests
+from PyQt6.QtCore import QRunnable, pyqtSlot
+from .enums import DownloadStatus
+from .db import DownloadDB
+
+class DownloadWorker(QRunnable):
+    def __init__(self, item_data: dict, signals, manager):
+        super().__init__()
+        self.item = item_data
+        self.signals = signals
+        self.manager = manager
+        self.db = DownloadDB()
+
+        self.filename = self.item["filename"]
+        self.folder_path = self.item["folder_path"]
+        self.final_path = os.path.join(self.folder_path, self.filename)
+        self.temp_path = self.final_path + ".part"
+        self.url = self.item["url"]
+        self.download_id = self.item["id"]
+
+        self._paused = False
+        self._cancelled = False
+
+    @pyqtSlot()
+    def run(self):
+        try:
+            self._download_file()
+        except Exception as e:
+            self.signals["status"].emit(self.download_id, DownloadStatus.ERROR)
+            self.signals["error"].emit(self.download_id, str(e))
+
+    def _download_file(self):
+        os.makedirs(self.folder_path, exist_ok=True)
+        resume_header = {}
+        file_mode = "ab" if os.path.exists(self.temp_path) else "wb"
+        downloaded_size = os.path.getsize(self.temp_path) if os.path.exists(self.temp_path) else 0
+
+        if downloaded_size > 0:
+            resume_header["Range"] = f"bytes={downloaded_size}-"
+
+        with requests.get(self.url, stream=True, headers=resume_header) as r:
+            r.raise_for_status()
+            total_size = int(r.headers.get("content-length", 0)) + downloaded_size
+
+            with open(self.temp_path, file_mode) as f:
+                for chunk in r.iter_content(chunk_size=64 * 1024):
+                    if self._cancelled or self.manager._cancel_all:
+                        self.signals["status"].emit(self.download_id, DownloadStatus.CANCELLED)
+                        return
+                    while self._paused or self.manager._pause_all:
+                        self.manager.msleep(100)
+
+                    if chunk:
+                        f.write(chunk)
+                        downloaded_size += len(chunk)
+                        percent = int((downloaded_size / total_size) * 100) if total_size else 0
+                        self.signals["progress"].emit(
+                            self.download_id, self.filename, downloaded_size, total_size, percent
+                        )
+                        self.db.upsert({
+                            **self.item,
+                            "downloaded_bytes": downloaded_size,
+                            "total_bytes": total_size,
+                            "status": DownloadStatus.RESUMED.value
+                        })
+
+        if not self._cancelled and not self.manager._cancel_all:
+            os.replace(self.temp_path, self.final_path)
+            self.signals["status"].emit(self.download_id, DownloadStatus.COMPLETED)
+            self.signals["finished"].emit(self.download_id, self.filename)
+            self.db.upsert({**self.item, "downloaded_bytes": downloaded_size, "total_bytes": total_size, "status": DownloadStatus.COMPLETED.value})
+
+    def pause(self):
+        self._paused = True
+        self.signals["status"].emit(self.download_id, DownloadStatus.PAUSED)
+
+    def resume(self):
+        self._paused = False
+        self.signals["status"].emit(self.download_id, DownloadStatus.RESUMED)
+
+    def cancel(self):
+        self._cancelled = True