fix: AppriseURL parsing (#408)

* fix: simplify `AppriseModel`

- Do not convert to `yarl.URL` to keep the model simple
- Handle simplified cases for URLs that have specific limitations. ex: OS notification URL have a character limit

* refactor: move apprise utils to its own module

* refactor: define simplify URLs as a function

* refactor: capture log messages + def as async (apprise)

* fix: do not use `asyncio.gather` to notify

log messages were being written out of order

* refactor: simplify notifications

* refactor: better logs

* refactor: add more simplified cases

* chore: add `pywin32` as depency on windows

* refactor: log webhook results

* refactor: enable windows notifications by default

* refactor: clean up code

* refactor: handle possible errors while getting the main log file

* refactor: only log spacer if apprise failed

* docs: update notifications details

* docs: update notifications details x2

* fix: last line not being log

* test: add tests for the `apprise` module

* refactor: return result and logs in `send_apprise_notifications`

* refactor: move apprise URLs validation to config startup

Fail immediately if the file has invalid URLs instead of waiting for CDL to finish the run.

* fix: make sure every parsed URL has at least 1 valid tag

* test: add tests for the `send_apprise_notifications`

* chore: bump `pytest` min version

* refactor: split `file` and `url` into 2 diferent parameters

* docs: add docstrings

* fix: type checking errors

* refactor: make some functions private

* refactor: make validation error exit

* refactor: make `get_apprise_urls` always return a list

* test: update `invalid_multiple_urls.txt`

* refactor: move `set_apprise_fixed` to config manager

* refactor: move `set_pydantic_config` to `load_configs`

* fix: file reference

* refactor: add `ValueError` test case

* test: increase coverage

* refactor: allow passing multiple urls to `get_apprise_urls`

* refactor: move file parsing back to `get_apprise_urls`

* fix: handle files that do not exists

* test: get 100% coverage

* refactor: remove duplicate case

* test: split test cases into individual functions

* refactor: remove unused manager parameter

.gitignore

@@ -5,6 +5,7 @@
 /Downloads/
 /Old Files/
 cyberdrop_dl_debug.log
+.coverage
 
 # Python cache
 __pycache__

cyberdrop_dl/config_definitions/config_settings.py

@@ -8,7 +8,7 @@
 from cyberdrop_dl.utils.data_enums_classes.hash import Hashing
 from cyberdrop_dl.utils.data_enums_classes.supported_domains import SUPPORTED_SITES_DOMAINS
 
-from .custom_types import AliasModel, HttpAppriseURLModel, NonEmptyStr
+from .custom_types import AliasModel, HttpAppriseURL, NonEmptyStr
 
 
 class DownloadOptions(BaseModel):
@@ -41,7 +41,7 @@ class Files(AliasModel):
 
 class Logs(AliasModel):
     log_folder: Path = APP_STORAGE / "Configs" / "{config}" / "Logs"
-    webhook: HttpAppriseURLModel | None = Field(validation_alias="webhook_url", default=None)
+    webhook: HttpAppriseURL | None = Field(validation_alias="webhook_url", default=None)
     main_log: Path = Field(Path("downloader.log"), validation_alias="main_log_filename")
     last_forum_post: Path = Field(Path("Last_Scraped_Forum_Posts.csv"), validation_alias="last_forum_post_filename")
     unsupported_urls: Path = Field(Path("Unsupported_URLs.csv"), validation_alias="unsupported_urls_filename")

cyberdrop_dl/config_definitions/custom_types.py

@@ -37,74 +37,35 @@ class FrozenModel(BaseModel):
 
 
 class AppriseURLModel(FrozenModel):
-    url: SecretAnyURL
+    url: Secret[AnyUrl]
     tags: set[str]
 
     @model_serializer()
     def serialize(self, info: SerializationInfo):
         dump_secret = info.mode != "json"
         url = self.url.get_secret_value() if dump_secret else self.url
         tags = self.tags - set("no_logs")
+        tags = sorted(tags)
         return f"{','.join(tags)}{'=' if tags else ''}{url}"
 
     @model_validator(mode="before")
     @staticmethod
     def parse_input(value: dict | URL | str):
-        url_obj = value
-        tags = None
-        if isinstance(url_obj, dict):
-            tags = url_obj.get("tags")
-            url_obj = url_obj.get("url")
-        if isinstance(value, URL):
-            url_obj = str(value)
-        url = AppriseURL(url_obj, validate=False)
-        return {"url": url._url, "tags": tags or url.tags or set("no_logs")}
-
-
-class AppriseURL:
-    _validator = AppriseURLModel
-
-    def __init__(self, url: URL | str, tags: set | None = None, *, validate: bool = True):
-        self._actual_url = None
-        self._url = str(url)
-        if validate:
-            self._validate()
-        else:
-            self.parse_str(url, tags)
-
-    @property
-    def tags(self) -> set[str]:
-        return self._tags
-
-    @property
-    def url(self) -> URL:
-        self._validate()
-        return self._actual_url
-
-    def parse_str(self, url: URL | str, tags: set | None = None):
-        self._tags = tags or set("no_logs")
-        self._url = str(url)
-        self._actual_url = url if isinstance(url, URL) else None
-        parts = self._url.split("://", 1)[0].split("=", 1)
-        if len(parts) == 2 and not self._actual_url:
-            self._tags = set(parts[0].split(","))
-            self._url: str = url.split("=", 1)[-1]
-
-    def _validate(self):
-        if not self._actual_url:
-            apprise_model = self._validator(url=self._url)
-            self._actual_url = apprise_model.url
-
-    def __repr__(self):
-        return f"AppriseURL({self._url}, tags={self.tags})"
-
-    def __str__(self):
-        return f"{','.join(self.tags)}{'=' if self.tags else ''}{self.url}"
-
-
-class HttpAppriseURLModel(AppriseURLModel):
-    url: SecretHttpURL
-
-
-class HttpAppriseURL(AppriseURL):
-    _validator = HttpAppriseURLModel
+        url = value
+        tags = set()
+        if isinstance(value, dict):
+            tags = value.get("tags") or tags
+            url = value.get("url", "")
+
+        if isinstance(url, URL):
+            url = str(url)
+        parts = url.split("://", 1)[0].split("=", 1)
+        if len(parts) == 2:
+            tags = set(parts[0].split(","))
+            url: str = url.split("=", 1)[-1]
+
+        return {"url": url, "tags": tags}
+
+
+class HttpAppriseURL(AppriseURLModel):
+    url: Secret[HttpURL]

cyberdrop_dl/main.py

@@ -18,14 +18,10 @@
 from cyberdrop_dl.managers.manager import Manager
 from cyberdrop_dl.scraper.scraper import ScrapeMapper
 from cyberdrop_dl.ui.program_ui import ProgramUI
+from cyberdrop_dl.utils.apprise import send_apprise_notifications
 from cyberdrop_dl.utils.logger import RedactedConsole, log, log_spacer, log_with_color, print_to_console
 from cyberdrop_dl.utils.sorting import Sorter
-from cyberdrop_dl.utils.utilities import (
-    check_latest_pypi,
-    check_partials_and_empty_folders,
-    send_webhook_message,
-    sent_apprise_notifications,
-)
+from cyberdrop_dl.utils.utilities import check_latest_pypi, check_partials_and_empty_folders, send_webhook_message
 from cyberdrop_dl.utils.yaml import handle_validation_error
 
 if TYPE_CHECKING:
@@ -58,7 +54,6 @@ def startup() -> Manager:
             "AuthSettings": manager.config_manager.authentication_settings,
         }
         handle_validation_error(e, sources=sources)
-        sys.exit(1)
 
     except KeyboardInterrupt:
         print_to_console("Exiting...")
@@ -226,7 +221,7 @@ async def director(manager: Manager) -> None:
             log_with_color("Finished downloading. Enjoy :)", "green", 20, show_in_stats=False)
 
         await send_webhook_message(manager)
-        sent_apprise_notifications(manager)
+        await send_apprise_notifications(manager)
         start_time = perf_counter()
 
 

cyberdrop_dl/managers/config_manager.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import shutil
 from dataclasses import field
 from time import sleep
@@ -8,13 +9,15 @@
 from cyberdrop_dl.config_definitions import AuthSettings, ConfigSettings, GlobalSettings
 from cyberdrop_dl.managers.log_manager import LogManager
 from cyberdrop_dl.utils import yaml
+from cyberdrop_dl.utils.apprise import get_apprise_urls
 
 if TYPE_CHECKING:
     from pathlib import Path
 
     from pydantic import BaseModel
 
     from cyberdrop_dl.managers.manager import Manager
+    from cyberdrop_dl.utils.apprise import AppriseURL
 
 
 class ConfigManager:
@@ -26,6 +29,7 @@ def __init__(self, manager: Manager) -> None:
         self.settings: Path = field(init=False)
         self.global_settings: Path = field(init=False)
         self.deep_scrape: bool = False
+        self.apprise_urls: list[AppriseURL] = []
 
         self.authentication_data: AuthSettings = field(init=False)
         self.settings_data: ConfigSettings = field(init=False)
@@ -51,15 +55,16 @@ def startup(self) -> None:
         self.settings.parent.mkdir(parents=True, exist_ok=True)
         self.pydantic_config = self.manager.cache_manager.get("pydantic_config")
         self.load_configs()
-        if not self.pydantic_config:
-            self.pydantic_config = True
-            self.manager.cache_manager.save("pydantic_config", True)
 
     def load_configs(self) -> None:
         """Loads all the configs."""
         self._load_authentication_config()
         self._load_global_settings_config()
         self._load_settings_config()
+        self.apprise_file = self.manager.path_manager.config_folder / self.loaded_config / "apprise.txt"
+        self.apprise_urls = get_apprise_urls(file=self.apprise_file)
+        self._set_apprise_fixed()
+        self._set_pydantic_config()
 
     @staticmethod
     def get_model_fields(model: type[BaseModel], *, exclude_unset: bool = True) -> set[str]:
@@ -172,3 +177,18 @@ def change_config(self, config_name: str) -> None:
         sleep(1)
         self.manager.log_manager = LogManager(self.manager)
         sleep(1)
+
+    def _set_apprise_fixed(self):
+        apprise_fixed = self.manager.cache_manager.get("apprise_fixed")
+        if apprise_fixed:
+            return
+        if os.name == "nt":
+            with self.apprise_file.open("a", encoding="utf8") as f:
+                f.write("windows://\n")
+        self.manager.cache_manager.save("apprise_fixed", True)
+
+    def _set_pydantic_config(self):
+        if self.pydantic_config:
+            return
+        self.manager.cache_manager.save("pydantic_config", True)
+        self.pydantic_config = True