feat(server): Fake qbitorrent server

server/holerr/api/__init__.py

@@ -1,13 +1,15 @@
 from holerr.core import config
 from .server import Server
 from .routers import api_router
+from .fakes import fakes_router
 
 from fastapi import HTTPException
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from fastapi.staticfiles import StaticFiles
 
 server = Server()
 server.app.include_router(api_router)
+server.app.include_router(fakes_router)
 
 # Serve the frontend
 # https://stackoverflow.com/a/73552966

server/holerr/api/fakes/__init__.py

@@ -0,0 +1,6 @@
+from . import qbittorrent
+
+from fastapi import APIRouter
+
+fakes_router = APIRouter(prefix="/fake")
+fakes_router.include_router(qbittorrent.router, prefix="/qbittorrent")

server/holerr/api/fakes/qbittorrent.py

@@ -0,0 +1,67 @@
+from holerr.core import config
+from holerr.core.db import db
+from holerr.core.config_repositories import PresetRepository
+from holerr.database.repositories import DownloadRepository
+from holerr.database.models import DownloadStatus
+from .qbittorrent_models import Torrent
+from .qbittorrent_repositories import QBittorrentTorrentRepository
+
+import random
+import string
+from datetime import datetime
+
+from fastapi import APIRouter, Request, Response, status, HTTPException
+from fastapi.responses import PlainTextResponse
+
+router = APIRouter(prefix="/api/v2")
+
+
+# Faked endpoints: https://github.com/Radarr/Radarr/blob/develop/src/NzbDrone.Core/Download/Clients/QBittorrent/QBittorrentProxyV2.cs
+# API documentation: https://github.com/qbittorrent/qBittorrent/wiki/WebUI-API-(qBittorrent-4.1)
+
+@router.get("/auth/login", tags=["QBittorrent"])
+async def auth_login(response: Response):
+    s = string.ascii_uppercase+string.ascii_lowercase+string.digits
+    response.set_cookie(key="SID", value=f"{''.join(random.sample(s, 35))}", path="/")
+    response.status_code = status.HTTP_200_OK
+    return response
+
+@router.get("/auth/logout", tags=["QBittorrent"])
+async def auth_logout():
+    return Response(status_code=status.HTTP_200_OK)
+
+
+@router.get("/app/version", tags=["QBittorrent"])
+async def app_version():
+    return PlainTextResponse("v4.1.3")
+
+@router.get("/app/webapiVersion", tags=["QBittorrent"])
+async def app_webapiVersion():
+    return PlainTextResponse("2.8.3")
+
+
+@router.get("/app/preferences", tags=["QBittorrent"])
+async def app_preferences():
+    return {}
+
+@router.get("/torrents/categories", tags=["QBittorrent"])
+async def torrents_categories():
+    categories = {}
+    for preset in config.presets:
+        categories[preset.name] = {"name":preset.name, "savePath":preset.output_dir}
+    return categories
+
+@router.get("/torrents/info", response_model=list[Torrent], tags=["QBittorrent"])
+async def torrents_info(request:Request):
+    preset_name = request.query_params.get("category")
+    preset = PresetRepository().get_preset(preset_name)
+    if preset is None:
+        raise HTTPException(status_code=404, detail=f"Preset {preset_name} not found")
+    session = db.new_session()
+    downloads = DownloadRepository(session).get_all_for_preset(preset_name)
+
+    response_torrents = []
+    for download in downloads:
+        torrent = QBittorrentTorrentRepository.torrent_from_download(download)
+        response_torrents.append(torrent)
+    return response_torrents

server/holerr/api/fakes/qbittorrent_models.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, SecretStr
+
+from typing import Optional
+from datetime import datetime
+
+TorrentState = {
+    "ERROR" : "error",
+    "DOWNLOADING": "downloading",
+    "UPLOADING": "uploading"
+}
+
+class Torrent(BaseModel):
+    hash: str
+    name: str
+    size: int
+    progress: float
+    eta: int
+    state: str
+    category: Optional[str] = None
+    save_path: str
+    content_path: str
+    ratio: float
+    last_activity: int

server/holerr/api/fakes/qbittorrent_repositories.py

@@ -0,0 +1,35 @@
+from .qbittorrent_models import Torrent, TorrentState
+
+from holerr.database.models import Download, DownloadStatus
+from holerr.core.config_repositories import PresetRepository
+
+from datetime import datetime
+
+class QBittorrentTorrentRepository:
+    @staticmethod
+    def torrent_from_download(download:Download) -> Torrent:
+        time_elapsed = datetime.now().timestamp() - download.created_at.timestamp()
+        time_per_percent = time_elapsed / download.total_progress
+        estimated_time = (100-download.total_progress) * time_per_percent
+        eta = int(round(datetime.now().timestamp() + estimated_time))
+        state = TorrentState["DOWNLOADING"]
+        if download.status >= DownloadStatus["ERROR_NO_FILES_FOUND"]:
+            state = TorrentState["ERROR"]
+        elif download.status == DownloadStatus["DOWNLOADED"]:
+            state = TorrentState["UPLOADING"]
+
+        preset = PresetRepository.get_preset(download.preset)
+
+        return Torrent(
+            hash=download.id,
+            name=download.title,
+            size=download.total_bytes,
+            progress=download.total_progress,
+            eta=eta,
+            state=state,
+            category=download.preset,
+            save_path=preset.output_dir,
+            content_path=preset.output_dir,
+            ratio=1,
+            last_activity=int(round(download.updated_at.timestamp())),
+        )

server/holerr/api/routers/__init__.py

@@ -1,3 +1,4 @@
+from ..fakes import qbittorrent
 from . import actions, constants, config, downloads, presets, status, websocket
 
 from fastapi import APIRouter, Depends
@@ -10,3 +11,6 @@
 api_router.include_router(presets.router)
 api_router.include_router(status.router)
 api_router.include_router(websocket.router)
+
+qbittorrent_router = APIRouter(prefix="/qbittorrent")
+qbittorrent_router.include_router(qbittorrent.router)

server/holerr/database/repositories.py

@@ -136,6 +136,9 @@ def get_all_to_delete(self) -> list[Download]:
             Download.to_delete,
         )
 
+    def get_all_for_preset(self, preset_name: str) -> list[Download]:
+        return self.get_all_models(Download.preset == preset_name)
+
     def clean_downloaded(self) -> list[str]:
         deleted_ids = []
         downloads = self.get_all_models(Download.status == DownloadStatus["DOWNLOADED"])