WorldTeacher
bb3b922325
- Replace hardcoded CONFIG_PATH with appdirs AppDirs (use app.user_config_dir) and update SETTINGS_PATH - Rework dataclasses: add Library, QbitTorrent; convert Komga.libraries to List[Library]; use Path for media_path; add getLibraryByName and proper __post_init__ conversions - Extend KomGrabber to support downloader/downloader_settings (aria2/qbit), normalize Path handling and expanduser usage - Add type fixes, utility methods (getattr/_setattr) and API __post_init__ to convert nested dicts to dataclass instances - Add package CLI entrypoints (__main__.py, src package main) and simple runner (main.py) - Add tests for package __init__ CLI and config behavior (tests/*) and a small test script (test.py) - Add .vscode/settings.json for pytest integration - Tidy pyproject.toml: format dependencies, add dev/test dependency groups and fix trailing newline in bumpversion section
471 lines
12 KiB
Python
471 lines
12 KiB
Python
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import List, Optional, Union
|
|
from urllib.parse import quote_plus
|
|
|
|
from omegaconf import DictConfig, OmegaConf
|
|
|
|
from komconfig import app
|
|
|
|
SETTINGS_PATH = os.path.join(app.user_config_dir, "config.yaml")
|
|
|
|
|
|
@dataclass
|
|
class RemoteSettings:
|
|
url: str
|
|
port: int
|
|
user: str
|
|
password: str
|
|
database_name: str = "komcache"
|
|
|
|
|
|
@dataclass
|
|
class Cache:
|
|
mode: str
|
|
remote: dict[str, Union[str, int]] | RemoteSettings
|
|
local_path: str | Path
|
|
|
|
def __post__init__(self):
|
|
if self.mode == "remote":
|
|
self.remote = RemoteSettings(**self.remote)
|
|
else:
|
|
self.remote = None
|
|
self.local_path = Path(self.local_path).expanduser()
|
|
|
|
@property
|
|
def path(self):
|
|
if self.mode == "local":
|
|
return Path(self.local_path).expanduser()
|
|
else:
|
|
return f"{self.remote.url}:{self.remote.port}"
|
|
|
|
@property
|
|
def url(self):
|
|
password = quote_plus(self.remote.password) if self.remote else ""
|
|
return f"mysql+pymysql://{self.remote.user}:{password}@{self.remote.url}:{self.remote.port}/{self.remote.database_name}"
|
|
|
|
|
|
@dataclass
|
|
class Library:
|
|
"""Komga library settings."""
|
|
|
|
id: Optional[str] = None
|
|
name: Optional[str] = None
|
|
type: Optional[str] = None
|
|
media_path: Optional[str] = None
|
|
valid_extensions: Optional[List[str]] = None
|
|
|
|
def from_ddict(self, d: dict):
|
|
keys = d.keys()
|
|
for key in keys:
|
|
self.name = key
|
|
data = d[self.name]
|
|
for key, value in data.items():
|
|
setattr(self, key, value)
|
|
return self
|
|
|
|
|
|
@dataclass
|
|
class Komga:
|
|
"""Komga server settings."""
|
|
|
|
url: str
|
|
user: str
|
|
password: str
|
|
media_path: Path
|
|
api_key: str = None
|
|
libraries: List[Library] = None
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
def __post_init__(self):
|
|
if "~" in self.media_path:
|
|
self.media_path = Path(os.path.expanduser(self.media_path))
|
|
if self.libraries:
|
|
self.libraries = [Library().from_ddict(lib) for lib in self.libraries]
|
|
|
|
def getLibraryByName(self, name: str) -> Optional[Library]:
|
|
"""Get a library by its name.
|
|
|
|
Args:
|
|
name (str): The name of the library.
|
|
|
|
Returns:
|
|
Optional[Library]: The library object if found, None otherwise.
|
|
"""
|
|
if self.libraries:
|
|
for lib in self.libraries:
|
|
if lib.name == name:
|
|
return lib
|
|
return None
|
|
|
|
|
|
@dataclass
|
|
class General:
|
|
"""General application settings."""
|
|
|
|
log_level: str
|
|
log_file: str
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class Aria2:
|
|
"""Aria2 settings."""
|
|
|
|
host: str
|
|
port: int
|
|
secret: str
|
|
timeout: int
|
|
kill_after_completion: bool
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class QbitTorrent:
|
|
"""QbitTorrent settings."""
|
|
|
|
host: str
|
|
port: int
|
|
username: str
|
|
password: str
|
|
category: str
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class EbookSettings:
|
|
min_filesize: int
|
|
max_filesize: int
|
|
data_directory: str
|
|
skip_parameters: List[str]
|
|
valid_file_extensions: List[str]
|
|
|
|
def __post_init__(self):
|
|
self.skip_parameters = [
|
|
param.lower() for param in self.skip_parameters if param is not None
|
|
]
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class ComicSettings:
|
|
min_filesize: int
|
|
max_filesize: int
|
|
data_directory: str
|
|
valid_file_extensions: List[str]
|
|
|
|
skip_parameters: List[str]
|
|
|
|
def __post_init__(self):
|
|
self.skip_parameters = [
|
|
param.lower() for param in self.skip_parameters if param is not None
|
|
]
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class KomGrabber:
|
|
"""
|
|
Configs used by the KomGrabber.
|
|
|
|
Attributes
|
|
----------
|
|
download_location : str | Path
|
|
The location to download the files to.
|
|
get_chapters : bool
|
|
Whether to get chapters or not.
|
|
skip_parameters : List[str]
|
|
A list of parameters to skip.
|
|
aria2 : Aria2
|
|
The aria2 settings. see Aria2 class.
|
|
tag_interactive : bool
|
|
Whether to use interactive tagging or not.
|
|
ebook : EbookSettings
|
|
The ebook settings. see EbookSettings class.
|
|
manga : ComicSettings
|
|
The manga settings. see ComicSettings class.
|
|
check_interval : int
|
|
The interval to check for new files.
|
|
use_cache : bool
|
|
Whether to use cache or not.
|
|
cache_check_interval : int
|
|
The days an entry will not be checked based on the last change in the cache.
|
|
|
|
"""
|
|
|
|
download_location: Path
|
|
tag_location: Path
|
|
copy_location: Optional[Path]
|
|
copy_files: Optional[List[str]]
|
|
copy: bool
|
|
|
|
get_chapters: bool
|
|
skip_parameters: List[str]
|
|
tag_interactive: bool
|
|
ebook: EbookSettings
|
|
manga: ComicSettings
|
|
check_interval: int
|
|
use_cache: bool
|
|
cache_check_interval: int
|
|
downloader: str
|
|
downloader_settings: Union[Aria2, QbitTorrent]
|
|
|
|
def __post_init__(self):
|
|
self.skip_parameters = [param.lower() for param in self.skip_parameters]
|
|
# if "~" in self.download_location:
|
|
# self.download_location = os.path.expanduser(self.download_location)
|
|
if isinstance(self.download_location, str):
|
|
self.download_location = Path(self.download_location).expanduser()
|
|
if isinstance(self.tag_location, str):
|
|
self.tag_location = Path(self.tag_location).expanduser()
|
|
if isinstance(self.downloader_settings, dict):
|
|
if self.downloader == "aria2":
|
|
self.downloader_settings = Aria2(**self.downloader_settings)
|
|
elif self.downloader == "qbit":
|
|
self.downloader_settings = QbitTorrent(**self.downloader_settings)
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class KomTagger:
|
|
"""KomTagger settings."""
|
|
|
|
sanitize_description: bool
|
|
check_cache: bool
|
|
check_interval: int
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
def __post_init__(self):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class ComicVine:
|
|
"""ComicVine settings."""
|
|
|
|
url: str
|
|
api_key: str
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class MangaDex:
|
|
"""MangaDex settings."""
|
|
|
|
url: str
|
|
username: str
|
|
password: str
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
@dataclass
|
|
class ComicsOrg:
|
|
"""Comics.org settings."""
|
|
|
|
path: str
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
def __post_init__(self):
|
|
if "~" in self.path:
|
|
self.path = os.path.expanduser(self.path)
|
|
|
|
|
|
@dataclass
|
|
class API:
|
|
"""API settings."""
|
|
|
|
comicvine: ComicVine
|
|
mangadex: MangaDex
|
|
comicsorg: ComicsOrg
|
|
|
|
def __post_init__(self):
|
|
# Convert dictionaries to their respective dataclass objects
|
|
self.comicvine = ComicVine(**self.comicvine)
|
|
self.mangadex = MangaDex(**self.mangadex)
|
|
self.comicsorg = ComicsOrg(**self.comicsorg)
|
|
|
|
def getattr(self, name):
|
|
return getattr(self, name)
|
|
|
|
def _setattr(self, name, value):
|
|
setattr(self, name, value)
|
|
|
|
|
|
class Settings:
|
|
"""A class to handle the configuration of the application. After initializing, it will try to load the config file and store it for future access. Any changes made can be saved to the file using the .save() method. Changes are used in real time in the app, if a restart is required, the Application will show a window.
|
|
|
|
Raises:
|
|
RuntimeError: Configuration not loaded
|
|
|
|
KeyError: Invalid option
|
|
|
|
"""
|
|
|
|
_config: Optional[DictConfig] = None
|
|
|
|
def __init__(self, config_path: str = SETTINGS_PATH):
|
|
"""
|
|
Loads the configuration file and stores it for future access.
|
|
|
|
Args:
|
|
config_path (str): Path to the YAML configuration file.
|
|
|
|
Raises:
|
|
FileNotFoundError: Configuration file not found
|
|
"""
|
|
if "~" in config_path:
|
|
config_path = os.path.expanduser(config_path)
|
|
if not os.path.exists(config_path):
|
|
raise FileNotFoundError(f"Configuration file not found: {config_path}")
|
|
self._config = OmegaConf.load(os.path.expanduser(config_path))
|
|
self.config_path = config_path
|
|
|
|
def save(self):
|
|
"""
|
|
Saves the current configuration to the file.
|
|
|
|
Args:
|
|
config_path (str): Path to the YAML configuration file.
|
|
"""
|
|
OmegaConf.save(self._config, self.config_path)
|
|
|
|
@property
|
|
def general(self):
|
|
return General(**self._config.general)
|
|
|
|
@property
|
|
def komga(self):
|
|
return Komga(**self._config.komga)
|
|
|
|
@property
|
|
def komga_attr(self, name):
|
|
return getattr(self.komga, name)
|
|
|
|
@komga_attr.setter
|
|
def komga_attr(self, name, value):
|
|
self.komga._setattr(name, value)
|
|
|
|
@property
|
|
def general_attr(self, name):
|
|
return getattr(self.general, name)
|
|
|
|
@general_attr.setter
|
|
def general_attr(self, name, value):
|
|
self.general._setattr(name, value)
|
|
|
|
@property
|
|
def komgrabber(self):
|
|
return KomGrabber(**self._config.komgrabber)
|
|
|
|
@property
|
|
def komgrabber_attr(self, name):
|
|
return getattr(self.komgrabber, name)
|
|
|
|
@komgrabber_attr.setter
|
|
def komgrabber_attr(self, name, value):
|
|
self.komgrabber._setattr(name, value)
|
|
|
|
@property
|
|
def komtagger(self):
|
|
return KomTagger(**self._config.komtagger)
|
|
|
|
@property
|
|
def cache(self):
|
|
return Cache(**self._config.cache)
|
|
|
|
def komtagger_attr(self, name):
|
|
return getattr(self.komtagger, name)
|
|
|
|
def set_komtagger_attr(self, name, value):
|
|
OmegaConf.update(self._config, f"komtagger.{name}", value)
|
|
|
|
def set_komgrabber_attr(self, name, value):
|
|
OmegaConf.update(self._config, f"komgrabber.{name}", value)
|
|
|
|
def set_general_attr(self, name, value):
|
|
OmegaConf.update(self._config, f"general.{name}", value)
|
|
|
|
def set_api_attr(self, name, value):
|
|
OmegaConf.update(self._config, f"api.{name}", value)
|
|
|
|
def set_komga_attr(self, name, value):
|
|
OmegaConf.update(self._config, f"komga.{name}", value)
|
|
|
|
@property
|
|
def save_path(self):
|
|
return self._config.save_path
|
|
|
|
@save_path.setter
|
|
def save_path(self, value: str):
|
|
self._config.save_path = value
|
|
|
|
def load_config(self, path, filename):
|
|
return OmegaConf.load(os.path.join(path, filename))
|
|
|
|
@property
|
|
def api(self):
|
|
return API(**self._config.api)
|
|
|
|
def dict(self):
|
|
return OmegaConf.to_container(self._config)
|
|
|
|
@property
|
|
def komga_auth(self):
|
|
return (self.komga.user, self.komga.password)
|