From e685c7b930051bca4ea5b80be91dd38cc4d6a5ed Mon Sep 17 00:00:00 2001 From: WorldTeacher Date: Mon, 17 Feb 2025 20:28:17 +0100 Subject: [PATCH] initial commit --- .bumpversion.toml | 26 + .gitignore | 234 ++++++++ .hatch.yaml | 14 + LICENSE | 21 + README.md | 0 pyproject.toml | 14 + src/komgapi/__init__.py | 6 + src/komgapi/endpoints/__init__.py | 12 + .../endpoints/announcement_controller.py | 21 + src/komgapi/endpoints/baseapi.py | 143 +++++ src/komgapi/endpoints/book_controller.py | 501 +++++++++++++++++ src/komgapi/endpoints/claim_controller.py | 18 + src/komgapi/endpoints/common_controller.py | 91 +++ src/komgapi/endpoints/library_controller.py | 58 ++ src/komgapi/endpoints/readlist_controller.py | 256 +++++++++ .../endpoints/referential_controller.py | 120 ++++ .../endpoints/series_collection_controller.py | 136 +++++ src/komgapi/endpoints/series_controller.py | 530 ++++++++++++++++++ src/komgapi/endpoints/settings_controller.py | 22 + src/komgapi/endpoints/user_controller.py | 92 +++ src/komgapi/errors.py | 10 + src/komgapi/komgapi.py | 90 +++ src/komgapi/py.typed | 0 src/komgapi/schemas/AlternateTitle.py | 10 + src/komgapi/schemas/Announcement.py | 26 + src/komgapi/schemas/Authentication.py | 86 +++ src/komgapi/schemas/Author.py | 38 ++ src/komgapi/schemas/Book.py | 31 + src/komgapi/schemas/BooksMetadata.py | 41 ++ src/komgapi/schemas/Collection.py | 15 + src/komgapi/schemas/Duplicate.py | 19 + src/komgapi/schemas/Latest.py | 23 + src/komgapi/schemas/Library.py | 67 +++ src/komgapi/schemas/Link.py | 10 + src/komgapi/schemas/Locations.py | 10 + src/komgapi/schemas/Manifest.py | 139 +++++ src/komgapi/schemas/Media.py | 14 + src/komgapi/schemas/Metadata.py | 41 ++ src/komgapi/schemas/Page.py | 15 + src/komgapi/schemas/Pageable.py | 16 + src/komgapi/schemas/Position.py | 22 + src/komgapi/schemas/Progress.py | 26 + src/komgapi/schemas/Readlist.py | 12 + src/komgapi/schemas/Series.py | 28 + src/komgapi/schemas/Sort.py | 11 + src/komgapi/schemas/Text.py | 8 + src/komgapi/schemas/Thumbnail.py | 27 + src/komgapi/schemas/User.py | 26 + src/komgapi/schemas/__init__.py | 30 + src/komgapi/schemas/apiSearch.py | 22 + src/komgapi/schemas/apikey.py | 12 + src/komgapi/schemas/settings.py | 34 ++ src/komgapi/schemas/violation.py | 11 + 53 files changed, 3285 insertions(+) create mode 100644 .bumpversion.toml create mode 100644 .gitignore create mode 100644 .hatch.yaml create mode 100644 LICENSE create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 src/komgapi/__init__.py create mode 100644 src/komgapi/endpoints/__init__.py create mode 100644 src/komgapi/endpoints/announcement_controller.py create mode 100644 src/komgapi/endpoints/baseapi.py create mode 100644 src/komgapi/endpoints/book_controller.py create mode 100644 src/komgapi/endpoints/claim_controller.py create mode 100644 src/komgapi/endpoints/common_controller.py create mode 100644 src/komgapi/endpoints/library_controller.py create mode 100644 src/komgapi/endpoints/readlist_controller.py create mode 100644 src/komgapi/endpoints/referential_controller.py create mode 100644 src/komgapi/endpoints/series_collection_controller.py create mode 100644 src/komgapi/endpoints/series_controller.py create mode 100644 src/komgapi/endpoints/settings_controller.py create mode 100644 src/komgapi/endpoints/user_controller.py create mode 100644 src/komgapi/errors.py create mode 100644 src/komgapi/komgapi.py create mode 100644 src/komgapi/py.typed create mode 100644 src/komgapi/schemas/AlternateTitle.py create mode 100644 src/komgapi/schemas/Announcement.py create mode 100644 src/komgapi/schemas/Authentication.py create mode 100644 src/komgapi/schemas/Author.py create mode 100644 src/komgapi/schemas/Book.py create mode 100644 src/komgapi/schemas/BooksMetadata.py create mode 100644 src/komgapi/schemas/Collection.py create mode 100644 src/komgapi/schemas/Duplicate.py create mode 100644 src/komgapi/schemas/Latest.py create mode 100644 src/komgapi/schemas/Library.py create mode 100644 src/komgapi/schemas/Link.py create mode 100644 src/komgapi/schemas/Locations.py create mode 100644 src/komgapi/schemas/Manifest.py create mode 100644 src/komgapi/schemas/Media.py create mode 100644 src/komgapi/schemas/Metadata.py create mode 100644 src/komgapi/schemas/Page.py create mode 100644 src/komgapi/schemas/Pageable.py create mode 100644 src/komgapi/schemas/Position.py create mode 100644 src/komgapi/schemas/Progress.py create mode 100644 src/komgapi/schemas/Readlist.py create mode 100644 src/komgapi/schemas/Series.py create mode 100644 src/komgapi/schemas/Sort.py create mode 100644 src/komgapi/schemas/Text.py create mode 100644 src/komgapi/schemas/Thumbnail.py create mode 100644 src/komgapi/schemas/User.py create mode 100644 src/komgapi/schemas/__init__.py create mode 100644 src/komgapi/schemas/apiSearch.py create mode 100644 src/komgapi/schemas/apikey.py create mode 100644 src/komgapi/schemas/settings.py create mode 100644 src/komgapi/schemas/violation.py diff --git a/.bumpversion.toml b/.bumpversion.toml new file mode 100644 index 0000000..a27ec9d --- /dev/null +++ b/.bumpversion.toml @@ -0,0 +1,26 @@ +[tool.bumpversion] +current_version = "0.5.2" +parse = "(?P\\d+)\\.(?P\\d+)\\.(?P\\d+)" +serialize = ["{major}.{minor}.{patch}"] +search = "{current_version}" +replace = "{new_version}" +regex = false +ignore_missing_version = false +ignore_missing_files = false +tag = false +sign_tags = false +tag_name = "v{new_version}" +tag_message = "Bump version: {current_version} → {new_version}" +allow_dirty = false +commit = false +message = "Bump version: {current_version} → {new_version}" +moveable_tags = [] +commit_args = "" +setup_hooks = [] +pre_commit_hooks = [] +post_commit_hooks = [] + +[[tool.bumpversion.files]] +filename = "src/komgapi/__init__.py" +[[tool.bumpversion.files]] +filename = "pyproject.toml" diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ed468fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,234 @@ +# ---> Python +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +# ---> Qt +# C++ objects and libs +*.slo +*.lo +*.o +*.a +*.la +*.lai +*.so +*.so.* +*.dll +*.dylib + +# Qt-es +object_script.*.Release +object_script.*.Debug +*_plugin_import.cpp +/.qmake.cache +/.qmake.stash +*.pro.user +*.pro.user.* +*.qbs.user +*.qbs.user.* +*.moc +moc_*.cpp +moc_*.h +qrc_*.cpp +ui_*.h +*.qmlc +*.jsc +Makefile* +*build-* +*.qm +*.prl + +# Qt unit tests +target_wrapper.* + +# QtCreator +*.autosave + +# QtCreator Qml +*.qmlproject.user +*.qmlproject.user.* + +# QtCreator CMake +CMakeLists.txt.user* + +# QtCreator 4.8< compilation database +compile_commands.json + +# QtCreator local machine specific files for imported projects +*creator.user* + +*_qmlcache.qrc + + +.history +depend +output/output/LOGtoJSON.exe + +.pytest_cache +output +docs/ +config.yaml +**/tempCodeRunnerFile.py + +uv.lock +.history +.venv +venv +*.log diff --git a/.hatch.yaml b/.hatch.yaml new file mode 100644 index 0000000..a66fc54 --- /dev/null +++ b/.hatch.yaml @@ -0,0 +1,14 @@ +build: + exclude: + - .git + - .gitignore + - .hatch.yaml + - .travis.yml + - .coveragerc + - .editorconfig + - .gitattributes + - .readthedocs.yml + - .readthedocs + - .trunk + - .venv + - .vscode \ No newline at end of file diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..aec92ac --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 WorldTeacher + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6a35078 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,14 @@ +[project] +name = "komgapi" +version = "0.5.2" +description = "Add your description here" +readme = "README.md" +authors = [ + { name = "WorldTeacher", email = "coding_contact@pm.me" } +] +requires-python = ">=3.13" +dependencies = [] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" diff --git a/src/komgapi/__init__.py b/src/komgapi/__init__.py new file mode 100644 index 0000000..d1c16bc --- /dev/null +++ b/src/komgapi/__init__.py @@ -0,0 +1,6 @@ +__version__ = "0.5.2" +__all__ = ["komgAPI"] + +from .komgapi import KOMGAPI_REST as komgapi +from .endpoints import * +from .schemas import * diff --git a/src/komgapi/endpoints/__init__.py b/src/komgapi/endpoints/__init__.py new file mode 100644 index 0000000..e0bd071 --- /dev/null +++ b/src/komgapi/endpoints/__init__.py @@ -0,0 +1,12 @@ +from .common_controller import CommonBookController +from .library_controller import LibraryController +from .series_controller import SeriesController +from .readlist_controller import ReadListController +from .series_collection_controller import SeriesCollectionController +from .book_controller import BookController +from .announcement_controller import AnnouncementController +from .user_controller import UserController +from .settings_controller import SettingsController +from .claim_controller import ClaimController +from .referential_controller import ReferentialController + diff --git a/src/komgapi/endpoints/announcement_controller.py b/src/komgapi/endpoints/announcement_controller.py new file mode 100644 index 0000000..ffa684c --- /dev/null +++ b/src/komgapi/endpoints/announcement_controller.py @@ -0,0 +1,21 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.errors import KomgaError, LoginError, ResultErrror +from komgapi.schemas import * # Progress, Series + + +class AnnouncementController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getAnnouncements(self) -> List[Announcement]: + url = self.url + "announcements" + data = self.getRequest(url) + ret = [] + for announcement in data: + ret.append(Announcement(**announcement)) + return ret diff --git a/src/komgapi/endpoints/baseapi.py b/src/komgapi/endpoints/baseapi.py new file mode 100644 index 0000000..1d1692b --- /dev/null +++ b/src/komgapi/endpoints/baseapi.py @@ -0,0 +1,143 @@ +import requests +from komgapi.errors import KomgaError, LoginError, ResultErrror +from typing import Any, Union +from limit import limit + + +class BaseAPI: + def __init__(self, username, password, url, timeout=20, api_version=1) -> None: + self._username = username + self._password = password + self.url = url + f"api/v{api_version}/" + self.timeout = timeout + + def setParams(self, locals: dict) -> dict: + return { + param_name: param + for param_name, param in locals.items() + if param is not None and param_name not in ["self", "series_idurl"] + } + + def test_connection(self): + """Test the connection to the server. + + Returns: + bool: True if the connection is successful, False otherwise. + """ + try: + requests.get(self.url, timeout=self.timeout) + return True + except requests.exceptions.RequestException: + return False + + def overwriteVersion(self, version: int): + self.url = self.url.replace("api/v1/", f"api/v{version}/") + return self + + @limit(1, 1) + def getRequest(self, url, params: Union[dict, None] = None) -> Any: + if params is None: + params = {} + try: + # ic(url, params) + response = requests.get( + url, + auth=(self._username, self._password), + params=params, + timeout=self.timeout, + ) + response.raise_for_status() + # print(response.content) + return response.json() + except ConnectionError as e: + message = f"Connection Error: {e}" + raise KomgaError(message) from e + except requests.exceptions.Timeout as e: + raise KomgaError(f"Timeout Error: {e}") from e + + def postRequest(self, url, data: Union[dict, None] = None): + if data is None: + data = {} + try: + response = requests.post( + url, + auth=(self._username, self._password), + json=data, + timeout=self.timeout, + ) + response.raise_for_status() + status_code = response.status_code + if status_code != 202: + raise ResultErrror(f"Result Error: {response.json()}") + elif status_code == 200: + return response.json() + except ConnectionError as e: + message = f"Connection Error: {e}" + raise KomgaError(message) from e + except requests.exceptions.Timeout as e: + raise KomgaError(f"Timeout Error: {e}") from e + + def patchRequest(self, url, data: Union[dict, None] = None): + if data is None: + data = {} + try: + response = requests.patch( + url, + auth=(self._username, self._password), + json=data, + timeout=self.timeout, + ) + response.raise_for_status() + if response.status_code != 204: + raise ResultErrror(f"Result Error: {response.json()}") + except ConnectionError as e: + message = f"Connection Error: {e}" + raise KomgaError(message) from e + except requests.exceptions.Timeout as e: + raise KomgaError(f"Timeout Error: {e}") from e + + def putRequest(self, url, data: Union[dict, None] = None): + if data is None: + data = {} + try: + response = requests.put( + url, + auth=(self._username, self._password), + json=data, + timeout=self.timeout, + ) + response.raise_for_status() + return response.json() + except ConnectionError as e: + message = f"Connection Error: {e}" + raise KomgaError(message) from e + except requests.exceptions.Timeout as e: + raise KomgaError(f"Timeout Error: {e}") from e + + def deleteRequest(self, url): + try: + response = requests.delete( + url, auth=(self._username, self._password), timeout=self.timeout + ) + response.raise_for_status() + return response.json() + except ConnectionError as e: + message = f"Connection Error: {e}" + raise KomgaError(message) from e + except requests.exceptions.Timeout as e: + raise KomgaError(f"Timeout Error: {e}") from e + + @classmethod + def from_env(cls): + """Create a KOMGA API object from environment variables. + + Returns: + KOMGAPI_REST: The KOMGA API object. + """ + import os + + return cls( + os.environ["KOMGA_USERNAME"], + os.environ["KOMGA_PASSWORD"], + os.environ["KOMGA_URL"], + ) diff --git a/src/komgapi/endpoints/book_controller.py b/src/komgapi/endpoints/book_controller.py new file mode 100644 index 0000000..c0f69a8 --- /dev/null +++ b/src/komgapi/endpoints/book_controller.py @@ -0,0 +1,501 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.schemas import * # Progress, Series + + +class BookController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getBooks( + self, + search_string: str = None, + library_id: List[str] = None, + media_status: List[str] = None, + read_status: List[str] = None, + released_after: List[str] = None, + tag: List[str] = None, + unpaged: bool = True, + page: int = None, + size: int = None, + sort: List[str] = None, + ) -> List[Book]: + """Get all Books that match the given query. + + Args: + ---- + - search_string (str, optional): An optional filter limiting the search to only return results containing the string. Defaults to None. + - library_id (List[str], optional): An optional filter limiting the searched libraries. Defaults to None. + - media_status (List[str], optional): An optional filter limiting the status of the books. Available values : UNKNOWN, ERROR, READY, UNSUPPORTED, OUTDATED. Defaults to None. + - read_status (List[str], optional): An optional filter limiting the read status of the books. Available values : UNREAD, READ, IN_PROGRESS. Defaults to None. + - released_after (List[str], optional): An optional filter limiting the search to books released after the set date. Defaults to None. + - tag (List[str], optional): Optional tags limiting the search. Defaults to None. + - unpaged (bool, optional): Set to False if a single Page of results should be returned. By default, a page contains 20 entries. Defaults to True. + - page (int, optional): An integer to get a set page. Defaults to None. + - size (int, optional): An integer setting the size of the page. Defaults to None. + - sort (List[str], optional): An option to sort the result. Sorting criteria in the format: property(,asc|desc). Default sort order is ascending. Multiple sort criteria are supported. Defaults to None. + + Returns: + ------- + List[Book]: A list of all books that match the query. Each book is represented as a Book object. + """ + + params = locals() + params = self.setParams(params) + url = self.url + "books" + data = self.getRequest(url, params) + ret = [] + for book in data["content"]: + ret.append(Book(**book)) + return ret + + def getBook(self, book_id: str) -> Book: + """Get a specific book. + + Args: + ---- + - book_id (str): the ID of the book to get. + + Returns: + ------- + - Book: the book wrapped in a Book class. + """ + url = self.url + f"books/{book_id}" + data = self.getRequest(url) + return Book(**data) + + def analyzeBook(self, book_id: str) -> Optional[Dict[str, Any]]: + """Analyze a specific book. + + Args: + ---- + - book_id (str): the ID of the book to analyze. + + Returns: + ------- + - Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"books/{book_id}/analyze" + data = self.postRequest(url) + return data + + @typing_extensions.deprecated("This function is not implemented yet.") + def deleteBookFile(self, book_id: str) -> Optional[Dict[str, Any]]: + """Not implemented yet.""" + # url = self.url + f"books/{book_id}/file" + # data = self.deleteRequest(url) + raise NotImplementedError + + # TODO: add getBookFile (with /*) if this gets a better description + def getManifest(self, book_id: str) -> Manifest: + """Get the manifest of a book. + + Args: + ---- + - book_id (str): the ID of the book to get the manifest from. + + Returns: + ------- + - Manifest: the manifest wrapped in a Manifest class. + """ + url = self.url + f"books/{book_id}/manifest" + data = self.getRequest(url) + return Manifest(**data) + + def getSpecificManifest(self, book_id: str, manifest: str) -> Manifest: + """Get the specified Manifest. 3 options are available. + + Args: + ---- + - book_id (str): The ID of the book to get the manifest from. + - manifest (str): the manifest to get. Options are epub, pdf, divina + + Returns: + ------- + - Manifest: the manifest wrapped in a Manifest class. + """ + url = self.url + f"books/{book_id}/manifest/{manifest}" + data = self.getRequest(url) + return Manifest(**data) + + def patchBookMetadata( + self, book_id: str, metadata: BookMetadata + ) -> Optional[Dict[str, Any]]: + """Get the metadata of a book. + + Args: + ---- + - book_id (str): the ID of the book to get the metadata from. + - metadata (BookMetadata): the metadata to set. All fields need to be set. + + Returns: + ------- + - Metadata: the metadata wrapped in a Metadata class. + """ + url = self.url + f"books/{book_id}/metadata" + data = self.patchRequest(url, metadata.model_dump()) + return data + + def refreshBookMetadata(self, book_id: str) -> Optional[Dict[str, Any]]: + """Refresh the metadata of a book. + + Args: + ---- + - book_id (str): the ID of the book to refresh the metadata from. + + Returns: + ------- + - Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"books/{book_id}/metadata/refresh" + data = self.postRequest(url) + return data + + def getNextBook(self, book_id: str) -> Book: + """Get the next Book in the series. If this is not possible, a HTTPError is raised. + + Args: + ---- + - book_id (str): the ID of the book to get the next book from. + + Returns: + ------- + - Book: the next book wrapped in a Book class. + """ + + url = self.url + f"books/{book_id}/next" + data = self.getRequest(url) + return Book(**data) + + def getBookPages(self, book_id: str) -> List[Page]: + """Get the pages of a book. + + Args: + ---- + - book_id (str): the ID of the book to get the pages from. + + Returns: + ------- + - List[Page]: A list of all pages that match the query. Each page is represented as a Page object. + """ + url = self.url + f"books/{book_id}/pages" + data = self.getRequest(url) + ret = [] + for page in data: + ret.append(Page(**page)) + return ret + + def getBookPage( + self, + book_id: str, + page: int, + convert: str = None, + zero_based: bool = False, + Accept: List[str] = None, + ) -> Page: + """Get a specific page of a book. + + Args: + ---- + - book_id (str): the ID of the book to get the page from. + - page (int): the page to get. + - convert (str, optional): The format to convert the page to. Available values : JPEG, PNG. Defaults to None. + - zero_based (bool, optional): If set to true, pages will start at zero. Defaults to False. + - Accept (List[str]): "Some very limited server driven content negotiation is handled. If a book is a PDF book, and the Accept header contains 'application/pdf' as a more specific type than other 'image/' types, a raw PDF page will be returned." + + Returns: + ------- + - Page: the page wrapped in a Page class. + """ + url = self.url + f"books/{book_id}/pages/{page}" + data = self.getRequest(url) + return Page(**data) + + def getBookPositions(self, book_id: str) -> List[Position]: + """Get the positions in an ebook. USE UNCLEAR. + + Args: + ---- + - book_id (str): the id of the book to get the positions from. + + Returns: + ------- + - List[Position]: A list of all positions that match the query. Each position is represented as a Position object. + """ + url = self.url + f"books/{book_id}/positions" + data = self.getRequest(url) + ret = [] + for position in data: + ret.append(Position(**position)) + return ret + + def getPreviousBook(self, book_id: str) -> Book: + """Get the previous Book in the series. If this is not possible, a HTTPError is raised. + + Args: + ---- + - book_id (str): the ID of the book to get the previous book from. + + Returns: + ------- + - Book: the previous book wrapped in a Book class. + """ + url = self.url + f"books/{book_id}/previous" + data = self.getRequest(url) + return Book(**data) + + def setProgress(self, book_id: str, progress: Progress) -> Optional[Dict[str, Any]]: + """Set the progress of a book. + + Args: + ---- + - book_id (str): the ID of the book to set the progress to. + - progress (Progress): the progress to set. + + Returns: + ------- + - Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"books/{book_id}/progression" + data = self.putRequest(url, progress.model_dump()) + return data + + def deleteProgress(self, book_id: str) -> Optional[Dict[str, Any]]: + """Set the book as unread. + + Args: + ---- + - book_id (str): the ID of the book to delete the progress from. + + Returns: + ------- + - Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"books/{book_id}/read-progress" + + def setProgress( + self, book_id: str, progress: Dict[str, Any] + ) -> Optional[Dict[str, Any]]: + """Set the progress of a book. + + Args: + ---- + - book_id (str): the ID of the book to set the progress to. + - progress (Progress): the progress to set. Needs to be a dict in the format: {"page": int, "completed": bool} + + Returns: + ------- + - Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"books/{book_id}/read-progress" + data = self.putRequest(url, progress) + return data + + def getBookReadlists(self, book_id: str) -> List[Readlist]: + """Get the readlists of a book. + + Args: + ---- + - book_id (str): the ID of the book to get the readlists from. + + Returns: + ------- + - List[Readlist]: A list of all readlists that match the query. Each readlist is represented as a Readlist object. + """ + url = self.url + f"books/{book_id}/readlists" + data = self.getRequest(url) + ret = [] + for readlist in data: + ret.append(Readlist(**readlist)) + return ret + + def getBookThumbnail(self, book_id: str) -> bytes: + """Get the thumbnail of a book. + + Args: + ---- + - book_id (str): the ID of the book to get the thumbnail from. + + Returns: + ------- + - bytes: the thumbnail as a bytestring. + """ + url = self.url + f"books/{book_id}/thumbnail" + data = requests.get(url, auth=(self._username, self._password)) + return data.content + + def getBookThumbnails(self, book_id: str) -> List[Thumbnail]: + """Get the metadata of all available thumbnails for the book. + + Args: + ---- + - book_id (str): the ID of the book to get the thumbnails from. + + Returns: + ------- + - List[Thumbnail]: A list of all possible thumbnails wrapped in a Thumbnail class. + """ + url = self.url + f"books/{book_id}/thumbnails" + data = self.getRequest(url) + ret = [] + for thumb in data: + ret.append(Thumbnail(**thumb)) + return ret + + def setBookThumbnail( + self, book_id: str, selected: bool, file: bytes + ) -> Union[Optional[Dict[str, Any]], Thumbnail]: + """Set a thumbnail based on a bytestring sent through this function. + + Args: + ---- + - book_id (str): the ID of the book to set the thumbnail to. + - selected (bool): Whether the thumbnail should be selected.Set to True if the thumbnail should be selected. + - file (bytes): the thumbnail to set as a bytestring. + + Returns: + ------- + - Union[Optional[Dict[str, Any]],Thumbnail]: If successful, the server will return the thumbnail data wrapped in a Thumbnail class. If a violation is found, the server will return a dictionary with the violation. + """ + url = self.url + f"books/{book_id}/thumbnails" + params = {"selected": selected, "thumbnail": file} + data = self.postRequest(url, params) + return Thumbnail(**data) + + def getSpecificBookThumbnail(self, book_id: str, thumbnail_id: str) -> bytes: + """Get a specific thumbnail based on the book id and the thumbnail id. + + Args: + ---- + - book_id (str): the ID of the book to get the thumbnail from. + - thumbnail_id (str): the ID of the thumbnail to get.1 + + Returns: + ------- + - bytes: the thumbnail as a bytestring. + """ + url = self.url + f"books/{book_id}/thumbnails/{thumbnail_id}" + data = requests.get(url, auth=(self._username, self._password)) + return data.content + + def deleteSpecificBookThumbnail( + self, book_id: str, thumbnail_id: str + ) -> Optional[Dict[str, Any]]: + """Delete a specific Thumbnail based on book id and thumbnail id. + + Args: + ---- + - book_id (str): the ID of the book to delete the thumbnail from. + - thumbnail_id (str): the ID of the thumbnail to delete. + + Returns: + ------- + - Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"books/{book_id}/thumbnails/{thumbnail_id}" + data = self.deleteRequest(url) + return data + + def setSpecificBookThumbnail( + self, book_id: str, thumbnail_id: str + ) -> Optional[Dict[str, Any]]: + """Set the thumbnail to be selected. + + Args: + ---- + - book_id (str): the ID of the book to set the thumbnail to. + - thumbnail_id (str): the ID of the thumbnail to set. + + Returns: + ------- + - Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"books/{book_id}/thumbnail/{thumbnail_id}/selected" + data = self.putRequest(url) + return data + + def getDuplicates( + self, + unpaged: bool = True, + page: int = None, + size: int = None, + sort: List[str] = None, + ) -> List[Duplicate]: + """Get a list of duplicates. + + Args: + ---- + - unpaged (bool, optional): If the result should be unpaged. Defaults to True. + - page (int, optional): the page to request. Defaults to None. + - size (int, optional): the size of a page. Defaults to None. + - sort (List[str], optional): Sorting criteria in the format: property(,asc|desc). Default sort order is ascending. Multiple sort criteria are supported.. Defaults to None. + + Returns: + ------- + - List[Duplicate]: A list of all duplicates that match the query. Each duplicate is represented as a Duplicate object. + """ + params = locals() + params = self.setParams(params) + url = self.url + "books/duplicates" + data = self.getRequest(url, params) + ret = [] + for duplicate in data["content"]: + ret.append(Duplicate(**duplicate)) + return ret + + @typing_extensions.deprecated("This function is not implemented yet.") + def importBook(self): + raise NotImplementedError + + def getLatestBooks( + self, unpaged: str = False, page: int = None, size: int = None + ) -> List[Book]: + """Get a list of the newly added or updated books. By default, it returns up to 20 entries. + Setting unpaged to True will return all entries. + + Args: + ---- + - unpaged (str, optional): If the result should be paged. Defaults to False. + - page (int, optional): The page to request. Defaults to None. + - size (int, optional): The size of the page. Defaults to None. + + Returns: + ------- + - List[Book]: A list of all books that match the query. Each book is represented as a Book object. + """ + params = locals() + params = self.setParams(params) + url = self.url + "books/latest" + data = self.getRequest(url, params) + ret = [] + for book in data["content"]: + ret.append(Book(**book)) + return ret + + def getOnDeckBooks( + self, library_id: List[str] = None, page: int = None, size: int = None + ) -> List[Book]: + """Return first unread book of series with at least one book read and no books in progress.. + + Args: + ---- + - library_id (List[str], optional): Optional Filter to set specific libraries. Defaults to None. + - page (int, optional): _description_. Defaults to None. + - size (int, optional): _description_. Defaults to None. + + Returns: + ------- + - List[Book]: A list of all books that match the query. Each book is represented as a Book object. + """ + params = locals() + params = self.setParams(params) + url = self.url + "books/ondeck" + data = self.getRequest(url, params) + ret = [] + for book in data["content"]: + ret.append(Book(**book)) + return ret diff --git a/src/komgapi/endpoints/claim_controller.py b/src/komgapi/endpoints/claim_controller.py new file mode 100644 index 0000000..e2a2360 --- /dev/null +++ b/src/komgapi/endpoints/claim_controller.py @@ -0,0 +1,18 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.errors import KomgaError, LoginError, ResultErrror +from komgapi.schemas import * # Progress, Series + + +class ClaimController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getClaim(self) -> dict: + url = self.url + "claim" + data = self.getRequest(url) + return data diff --git a/src/komgapi/endpoints/common_controller.py b/src/komgapi/endpoints/common_controller.py new file mode 100644 index 0000000..08474fb --- /dev/null +++ b/src/komgapi/endpoints/common_controller.py @@ -0,0 +1,91 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from komgapi.schemas import * + + +class CommonBookController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getBookFile( + self, book_id: str, download_path: str = "~/Downloads" + ) -> pathlib.Path: + """Download the book file. + + Args: + ---- + - book_id (str): the ID of the book to download. + - download_path (str, optional): The path to download the file to. Defaults to "~/Downloads". + + Returns: + ------- + - pathlib.Path: The path to the downloaded file. + """ + url = self.url + f"books/{book_id}/file" + download_path = pathlib.Path(download_path).expanduser() + subprocess.run( + [ + "curl", + "-u", + f"{self._username}:{self._password}", + "-o", + f"{download_path}/{book_id}.zip", + url, + ] + ) + return pathlib.Path(f"{download_path}/{book_id}.zip") + + def getRawPage(self, book_id: str, page: int) -> bytes: + """Get the raw page of a book. + + Args: + ---- + - book_id (str): the ID of the book to get the page from. + - page (int): the page to get. + + Returns: + ------- + - bytes: the page as a bytestring. + """ + url = self.url + f"books/{book_id}/pages/{page}/raw" + data = requests.get(url, auth=(self._username, self._password)) + return data.content + + def getPageThumbnail(self, book_id: str, page: int) -> bytes: + """Get the thumbnail of a page. + + Args: + ---- + - book_id (str): the ID of the book to get the page from. + - page (int): the page to get. + + Returns: + ------- + - bytes: the thumbnail as a bytestring. + """ + url = self.url + f"books/{book_id}/pages/{page}/thumbnail" + ic(url) + data = requests.get(url, auth=(self._username, self._password)) + return data.content + + def getProgress(self, book_id: str) -> Progress: + """Get the progress of a book. The progress contains a timestamp of the latest modification, the device where the modification was made (if the data is present), and a Locator object. + + Args: + ---- + - book_id (str): the ID of the book to get the progress from. + + Returns: + ------- + - Progress: the progress wrapped in a Progress class. + """ + url = self.url + f"books/{book_id}/progression" + data = self.getRequest(url) + return Progress(**data) + + @typing_extensions.deprecated("This function is not implemented yet.") + def getbookRessource(self, book_id: str, ressource: str) -> bytes: + raise NotImplementedError diff --git a/src/komgapi/endpoints/library_controller.py b/src/komgapi/endpoints/library_controller.py new file mode 100644 index 0000000..db27cf0 --- /dev/null +++ b/src/komgapi/endpoints/library_controller.py @@ -0,0 +1,58 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.errors import KomgaError, LoginError, ResultErrror +from komgapi.schemas import * # Progress, Series + + +class LibraryController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getLibraries(self) -> List[Library]: + url = self.url + "libraries" + data = self.getRequest(url) + ret = [] + for library in data: + ret.append(Library(**library)) + return ret + + def createLibrary( + self, library: CreateLibrary + ) -> Union[Library, Optional[Dict[str, Any]]]: + url = self.url + "libraries" + data = self.postRequest(url, library.model_dump()) + return Library(**data) if data else None + + def getLibrary(self, library_id: str) -> Library: + url = self.url + f"libraries/{library_id}" + data = self.getRequest(url) + return Library(**data) + + def patchLibrary(self, library_id: str, changed_library: Dict[str, Any]) -> None: + url = self.url + f"libraries/{library_id}" + data = self.patchRequest(url, changed_library) + return data + + def analyzeLibrary(self, library_id: str) -> Optional[Dict[str, Any]]: + url = self.url + f"libraries/{library_id}/analyze" + data = self.postRequest(url) + return data + + def emptyTrash(self, library_id: str) -> None: + url = self.url + f"libraries/{library_id}/empty-trash" + data = self.postRequest(url) + return data + + def refreshMetadata(self, library_id: str) -> None: + url = self.url + f"libraries/{library_id}/metadata/refresh" + data = self.postRequest(url) + return data + + def scanLibrary(self, library_id: str) -> Optional[Dict[str, Any]]: + url = self.url + f"libraries/{library_id}/scan" + data = self.postRequest(url) + return data diff --git a/src/komgapi/endpoints/readlist_controller.py b/src/komgapi/endpoints/readlist_controller.py new file mode 100644 index 0000000..4f05492 --- /dev/null +++ b/src/komgapi/endpoints/readlist_controller.py @@ -0,0 +1,256 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.errors import KomgaError, LoginError, ResultErrror +from komgapi.schemas import * # Progress, Series + + +class ReadListController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getReadlists( + self, + search_string: str = None, + library_id: List[str] = None, + unpaged: bool = True, + page: int = None, + size: int = None, + ) -> List[Series]: + """Get a list of all readlists. + + Args: + search_string (str, optional): The string to search for. Defaults to None. + library_id (List[str], optional): The library to search in. Defaults to None. + unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True. + page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None. + size (int, optional): Defines the size of the page. Defaults to None. + + Returns: + List[dict]: A list of all readlists. + """ + params = locals() + params = self.setParams(params) + url = self.url + "readlists" + data = self.getRequest(url, params) + ret = [] + for readlist in data["content"]: + ic(readlist) + ret.append(Series(**readlist)) + return ret + + def setReadlist( + self, readlist: Readlist + ) -> Union[Readlist, Optional[Dict[str, Any]]]: + """Create a new readlist. + + Args: + readlist (Readlist): The readlist to create. + + Returns: + Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + "readlists" + data = self.postRequest(url, readlist.model_dump()) + return Readlist(**data) if data else None + + def getReadlist(self, readlist_id: str) -> Series: + """Get a specific readlist. + + Args: + readlist_id (str): The ID of the readlist to get. + + Returns: + Series: The readlist wrapped in a Series class. + """ + url = self.url + f"readlists/{readlist_id}" + data = self.getRequest(url) + return Series(**data) + + def deleteReadlist(self, readlist_id: str) -> Optional[Dict[str, Any]]: + """Delete a specific readlist. + + Args: + readlist_id (str): The ID of the readlist to delete. + + Returns: + Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"readlists/{readlist_id}" + data = self.deleteRequest(url) + return data + + def patchReadlist( + self, readlist_id: str, changed_readlist: Dict[str, Any] + ) -> Optional[Dict[str, Any]]: + """Change the metadata of a readlist. + + Args: + readlist_id (str): The ID of the readlist to change. + changed_readlist (Dict[str,Any]): The changed readlist. This should be a dictionary with the changed fields. Only the fields that are changed need to be included. + + Returns: + Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"readlists/{readlist_id}" + data = self.patchRequest(url, changed_readlist) + return data + + def getReadlistBooks( + self, + readlist_id: str, + library_id: List[str] = None, + read_status: List[str] = None, + media_status: List[str] = None, + deleted: bool = None, + unpaged: bool = True, + page: int = None, + size: int = None, + author: List[str] = None, + ) -> List[Book]: + url = self.url + f"readlists/{readlist_id}/books" + params = locals() + params = self.setParams(params) + data = self.getRequest(url, params) + ret = [] + for book in data["content"]: + ret.append(Book(**book)) + return ret + + def getNextReadlistBook(self, readlist_id: str, book_id: str) -> Book: + """Get the next book in a readlist. + + Args: + readlist_id (str): The ID of the readlist to get the next book from. + book_id (str): The ID of the current book. + + Returns: + Book: The next book wrapped in a Book class. + """ + url = self.url + f"readlists/{readlist_id}/books/{book_id}/next" + data = self.getRequest(url) + return Book(**data) + + def getPreviousReadlistBook(self, readlist_id: str, book_id: str) -> Book: + """Get the previous book in a readlist. + + Args: + readlist_id (str): The ID of the readlist to get the previous book from. + book_id (str): The ID of the current book. + + Returns: + Book: The previous book wrapped in a Book class. + """ + url = self.url + f"readlists/{readlist_id}/books/{book_id}/previous" + data = self.getRequest(url) + return Book(**data) + + def getReadlistFile( + self, readlist_id: str, download_path: str = "~/Downloads" + ) -> pathlib.Path: + """Download the complete readlist as a giant zip file. + Unlike the other functions, this one has no timeout. It will wait until the file is downloaded. + WARNING: THIS WILL TAKE A VERY LONG TIME FOR LARGE READLISTS. USE WITH CAUTION. + My test: 800mb took ~7min to download. + Args: + readlist_id (str): ID of the readlist to download the file from. + download_path (str, optional): The path to download the file to. Defaults to "~/Downloads". + + Returns: + pathlib.Path: The path to the downloaded file. + """ + url = self.url + f"readlists/{readlist_id}/file" + download_path = pathlib.Path("~/Downloads").expanduser() + subprocess.run( + [ + "curl", + "-u", + f"{self._username}:{self._password}", + "-o", + f"{download_path}/{readlist_id}.zip", + url, + ] + ) + return pathlib.Path(f"{download_path}/{readlist_id}.zip") + + def getReadlistThumbnail(self, readlist_id: str) -> str: + """Get the current active thumbnail of a readlist. + + Args: + readlist_id (str): the ID of the readlist to get the thumbnail from. + + Returns: + str: the thumbnail of the readlist formatted as bytestring. + """ + url = self.url + f"readlists/{readlist_id}/thumbnail" + data = requests.get(url, auth=(self._username, self._password)) + return data.content + + def getReadlistThumbnails(self, readlist_id: str) -> List[ReadlistThumbnail]: + """Get the metadata of all available thumbnails for the readlist. + + Args: + readlist_id (str): the ID of the readlist to get the thumbnails from. + + Returns: + List[Thumbnail]: A list of all possible thumbnails wrapped in a Thumbnail class. + """ + url = self.url + f"readlists/{readlist_id}/thumbnails" + data = self.getRequest(url) + ret = [] + for thumb in data: + ret.append(ReadlistThumbnail(**thumb)) + return ret + + def setReadlistThumbnail( + self, readlist_id: str, selected: bool, thumbnail: bytes + ) -> Union[Thumbnail, Optional[Dict[str, Any]]]: + """Set a new thumbnail for the readlist. + + Args: + readlist_id (str): The ID of the readlist to set the thumbnail to. + selected (bool): If the thumbnail should be selected. + thumbnail (bytes): The thumbnail to set. + + Returns: + Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"readlists/{readlist_id}/thumbnail" + data = self.postRequest(url, {"selected": selected, "thumbnail": thumbnail}) + return Thumbnail(**data) if data else None + + def getReadlistSpecificThumbnail( + self, readlist_id: str, thumbnail_id: str + ) -> bytes: + url = self.url + f"readlists/{readlist_id}/thumbnails/{thumbnail_id}" + try: + data = requests.get( + url, auth=(self._username, self._password), timeout=self.timeout + ) + return data.content + except requests.exceptions.Timeout as e: + raise KomgaError(f"Timeout Error: {e}") from e + + def deleteReadlistSpecificThumbnail( + self, readlist_id: str, thumbnail_id: str + ) -> Optional[Dict[str, Any]]: + """Delete a specific thumbnail of a readlist. + + Args: + readlist_id (str): The ID of the readlist to delete the thumbnail from. + thumbnail_id (str): The ID of the thumbnail to delete. + + Returns: + Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"readlists/{readlist_id}/thumbnails/{thumbnail_id}" + data = self.deleteRequest(url) + return data + + def setReadlistSpecificThumbnail(self, readlist_id: str, thumbnail_id: str): + url = self.url + f"readlists/{readlist_id}/thumbnail/{thumbnail_id}/selected" + response = self.putRequest(url) + return response diff --git a/src/komgapi/endpoints/referential_controller.py b/src/komgapi/endpoints/referential_controller.py new file mode 100644 index 0000000..dc7e81d --- /dev/null +++ b/src/komgapi/endpoints/referential_controller.py @@ -0,0 +1,120 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.errors import KomgaError, LoginError, ResultErrror +from komgapi.schemas import * # Progress, Series + + +class ReferentialController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getAgeRatings(self, library_id: str, collection_id: str) -> str: + url = self.url + f"age-ratings" + data = self.getRequest(url) + return data + + def getAuthors( + self, + search: str = None, + library_id: str = None, + collection_id: str = None, + series_id: str = None, + ) -> List[Author]: + url = self.url + f"authors" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return [Author(**author) for author in data] + + def getAuthorNames(self, search: str = None) -> List[str]: + url = self.url + f"authors/names" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getAuthorRoles(self) -> List[str]: + url = self.url + f"authors/roles" + data = self.getRequest(url) + return data + + def getGenres(self, library_id: str = None, collection_id: str = None) -> List[str]: + url = self.url + f"genres" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getLanguages( + self, library_id: str = None, collection_id: str = None + ) -> List[str]: + url = self.url + f"languages" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getPublishers( + self, search: str = None, library_id: str = None, collection_id: str = None + ) -> List[str]: + url = self.url + f"publishers" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getReleaseDates( + self, library_id: str = None, collection_id: str = None + ) -> List[str]: + url = self.url + f"release-dates" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getSharingLabels( + self, library_id: str = None, collection_id: str = None + ) -> List[str]: + url = self.url + f"sharing-labels" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getTags(self, library_id: str = None, collection_id: str = None) -> List[str]: + url = self.url + f"tags" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getBookTags(self, library_id: str = None, readlist_id: str = None) -> List[str]: + url = self.url + f"book-tags" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getSeriesTags( + self, library_id: str = None, collection_id: str = None + ) -> List[str]: + url = self.url + f"series-tags" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return data + + def getAuthorsV2( + self, + search: str = None, + role: str = None, + library_id: str = None, + collection_id: str = None, + series_id: str = None, + readlist_id: str = None, + unpaged: bool = None, + page: int = None, + size: int = None, + ) -> List[Author]: + url = self.url + f"authors" + params = self.setParams(locals()) + print(params) + import requests + + data: requests.Response = self.overwriteVersion(2).getRequest(url, params) + content = data.json() + return [Author(**author) for author in content] diff --git a/src/komgapi/endpoints/series_collection_controller.py b/src/komgapi/endpoints/series_collection_controller.py new file mode 100644 index 0000000..f9745a4 --- /dev/null +++ b/src/komgapi/endpoints/series_collection_controller.py @@ -0,0 +1,136 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any + +from komgapi.schemas import * # Progress, Series + + +class SeriesCollectionController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getCollections( + self, + search_string: str = None, + library_id: List[str] = None, + unpaged: bool = True, + page: int = None, + size: int = None, + ) -> List[Collection]: + url = self.url + "collections" + params = locals() + params = self.setParams(params) + data = self.getRequest(url, params) + ret = [] + for collection in data["content"]: + ret.append(Collection(**collection)) + return ret + + def createCollection(self, collection: dict) -> Optional[Dict[str, Any]]: + """Create a new collection. + + Args: + collection (dict): A dictionary with the collection data.Needs to contain this: + { + "name": string, + "ordered": bool, + "seriesIds": [ + string + ] + } + + Returns: + Optional[Dict[str, Any]]: An optional dictionary with the created collection. + """ + url = self.url + "collections" + data = self.postRequest(url, collection) + return data + + def getCollection(self, collection_id: str) -> Collection: + url = self.url + f"collections/{collection_id}" + data = self.getRequest(url) + return Collection(**data) + + def deleteCollection(self, collection_id: str) -> Optional[Dict[str, Any]]: + url = self.url + f"collections/{collection_id}" + data = self.deleteRequest(url) + return data + + def patchCollection( + self, collection_id: str, changed_collection: dict + ) -> Optional[Dict[str, Any]]: + url = self.url + f"collections/{collection_id}" + data = self.patchRequest(url, changed_collection) + return data + + def getCollectionSeries( + self, + collection_id: str, + library_id: List[str] = None, + read_status: List[str] = None, + publisher: List[str] = None, + language: List[str] = None, + genre: List[str] = None, + tag: List[str] = None, + age_rating: List[str] = None, + deleted: bool = None, + complete: bool = None, + unpaged: bool = True, + page: int = None, + size: int = None, + author: List[str] = None, + ) -> List[Series]: + params = locals() + params = self.setParams(params) + url = self.url + f"collections/{collection_id}/series" + data = self.getRequest(url, params) + ret = [] + for series in data["content"]: + ret.append(Series(**series)) + return ret + + def getCollectionThumbnail(self, collection_id: str) -> str: + url = self.url + f"collections/{collection_id}/thumbnail" + data = requests.get(url, auth=(self._username, self._password)) + return data.content + + def getCollectionThumbnails(self, collection_id: str) -> List[Thumbnail]: + url = self.url + f"collections/{collection_id}/thumbnails" + data = self.getRequest(url) + ret = [] + for thumb in data: + ret.append(Thumbnail(**thumb)) + return ret + + def setCollectionThumbnail( + self, collection_id: str, thumbnail_id: str, selected: bool + ) -> Optional[Dict[str, Any]]: + url = self.url + f"collections/{collection_id}/thumbnail/{thumbnail_id}" + data = self.putRequest(url, {"selected": selected}) + return data + + def getSpecificCollectionThumbnail( + self, collection_id: str, thumbnail_id: str + ) -> bytes: + url = self.url + f"collections/{collection_id}/thumbnails/{thumbnail_id}" + data = requests.get(url, auth=(self._username, self._password)) + return data.content + + def deleteSpecificCollectionThumbnail( + self, collection_id: str, thumbnail_id: str + ) -> Optional[Dict[str, Any]]: + url = self.url + f"collections/{collection_id}/thumbnails/{thumbnail_id}" + data = self.deleteRequest(url) + return data + + def setSpecificCollectionThumbnail( + self, collection_id: str, thumbnail_id: str + ) -> Optional[Dict[str, Any]]: + url = ( + self.url + f"collections/{collection_id}/thumbnail/{thumbnail_id}/selected" + ) + data = self.putRequest(url) + return data diff --git a/src/komgapi/endpoints/series_controller.py b/src/komgapi/endpoints/series_controller.py new file mode 100644 index 0000000..f45c816 --- /dev/null +++ b/src/komgapi/endpoints/series_controller.py @@ -0,0 +1,530 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union + +from komgapi.schemas import * # Progress, Series + + +class SeriesController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getAllSeries( + self, + library_id: List[str] = None, + collection_id: List[str] = None, + status: List[str] = None, + publisher: List[str] = None, + lang: List[str] = None, + genre: List[str] = None, + tag: List[str] = None, + age_rating: List[str] = None, + release_year: List[str] = None, + deleted: bool = None, + complete: bool = None, + unpaged: bool = True, + sort: List[str] = None, + author: List[str] = None, + oneshot: bool = None, + size: int = None, + page: int = None, + ) -> list[Series]: + """Get all series from the server. + By default, this will return all series in the server. You can filter the results by using the parameters. + + Args: + ---- + - library_id (List[str], optional): The library to be queried. If None, all available libraries will be used. Defaults to None. + - collection_id (List[str], optional): The collection to be queried. Defaults to None. + - status (List[str], optional): The status of the series. Can be: ENDED,ONGOING,ABANDONED,HIATUS. Defaults to None. + - publisher (List[str], optional): Publisher(s) to be searched for. Defaults to None. + - lang (List[str], optional): Language to query for. Uses two-letter codec. Defaults to None. + - genre (List[str], optional): Genre(s) to query for. Defaults to None. + - tag (List[str], optional): Tag(s) to query. Defaults to None. + - age_rating (List[str], optional): A custom age-rating to search for. Needs to be configured manually. Defaults to None. + - release_year (List[str], optional): When the series were released. Defaults to None. + - deleted (bool, optional): If the series is deleted. Defaults to None. + - complete (bool, optional): Turn to true to only search series that are complete. Complete requires TotalBookCount to be set and equal to BookCount. Defaults to None. + - unpaged (bool, optional): Set to False if a single Page of results should be returned. By default, a page contains 20 entries. Defaults to True. + - sort (List[str], optional): Sorting of the returned data. Sort using asc|desc. Multiple sort criteria are supported. Defaults to None. + - author (List[str], optional): Author(s) to include in the query. Defaults to None. + - oneshot (bool, optional): If the series is categorized as oneshot. Defaults to None. + - size (int, optional): The size of the page. Defaults to None. + - page (int, optional): The page to be returned. Defaults to None. + + Returns: + ------- + - list[Series]: a list of all series that match the query. Each series is represented as a Series object. + """ + + params = locals() + params = self.setParams(params) + url = self.url + "series" + data = self.getRequest(url, params) + ret = [] + for series in data["content"]: + ret.append(Series(**series)) + return ret + + def getSeries(self, series_id: str) -> Series: + """Get a single series from the server. + + Args: + series_id (str): The ID of the series to get. + + Returns: + Series: the series that matches the ID wrapped in a Series class. + """ + url = self.url + f"series/{series_id}" + data = self.getRequest(url) + return Series(**data) + + def analyzeSeries(self, series_id: str) -> Optional[Dict[str, Any]]: + """Instruct the server to analyze a series. This will update the metadata of the series. + + Args: + series_id (str): the ID of the series to analyze. + + Returns: + Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"series/{series_id}/analyze" + data = self.postRequest(url) + return data + + def getSeriesBooks( + self, + series_id: str, + media_status: List[str] = None, + read_status: List[str] = None, + tag: List[str] = None, + deleted: bool = None, + unpaged: bool = True, + page: int = None, + size: int = None, + sort: List[str] = None, + author: List[str] = None, + ) -> List[Book]: + """Get all books in a series. + + Args: + series_id (str): The ID of the series to get books from. + media_status (List[str], optional): The status of the media. Can be: UNKNOWN,ERROR,READY,UNSUPPORTED,OUTDATED. Defaults to None. + read_status (List[str], optional): The read status of the book. Can be: UNREAD,READ,IN_PROGRESS. Defaults to None. + tag (List[str], optional): Tag(s) to query. Defaults to None. + deleted (bool, optional): If the series is deleted. Defaults to None. + unpaged (bool, optional): Set to False if a single Page of results should be returned. By default, a page contains 20 entries. Defaults to True. + page (int, optional): The page to be returned. Defaults to None. + size (int, optional): The size of the page. Defaults to None. + sort (List[str], optional): Sorting of the returned data. Sort using asc|desc. Multiple sort criteria are supported. Defaults to None. + author (List[str], optional): Author(s) to include in the query. Defaults to None. + + Returns: + List[Book]: A list of all books that match the query. Each book is represented as a Book object. + """ + url = self.url + f"series/{series_id}/books" + params = locals() + params = self.setParams(params) + data = self.getRequest(url, params) + ret = [] + for book in data["content"]: + ret.append(Book(**book)) + return ret + + def getSeriesCollections(self, series_id: str) -> Optional[List[Collection]]: + """Get all collections assigned to a series. + + Args: + series_id (str): the ID of the series to get collections from. + + Returns: + Optional[List[Collection]]: A list of all collections that are assigned to the series. Each collection is represented as a Collection object. If no collections are assigned, None is returned. + """ + url = self.url + f"series/{series_id}/collections" + data = self.getRequest(url) + if data is None: + return None + ret = [] + for collection in data: + ret.append(Collection(**collection)) + return ret + + def getFile( + self, series_id: str, filename: str, download_path: str = "~/Downloads" + ) -> pathlib.Path: + """Download the complete series as a giant zip file. + Unlike the other functions, this one has no timeout. It will wait until the file is downloaded. + WARNING: THIS WILL TAKE A VERY LONG TIME FOR LARGE SERIES. USE WITH CAUTION. + My test: 800mb took ~7min to download. + Args: + series_id (str): ID of the series to download the file from. + filename (str): The name of the file to download. + download_path (str, optional): The path to download the file to. Defaults to "~/Downloads". + + Returns: + pathlib.Path: The path to the downloaded file. + """ + url = self.url + f"series/{series_id}/file" + download_path = pathlib.Path(download_path).expanduser() + subprocess.run( + [ + "curl", + "-u", + f"{self._username}:{self._password}", + "-o", + f"{download_path}/{filename}", + url, + ] + ) + return pathlib.Path(f"{download_path}/{filename}") + + @typing_extensions.deprecated("This function is not implemented yet.") + def deleteFile(self, series_id: str) -> None: + """Not implemented yet.""" + # url = self.url + f"series/{series_id}/file" + # data = self.deleteRequest(url) + raise NotImplementedError + + def patchMetadata(self, series_id: str, changed_metadata: Dict[str, Any]) -> None: + """Change the metadata of a series. + The changed metadata should be a dictionary with the changed fields. Only the fields that are changed need to be included. + Args: + series_id (str): The ID of the series to change. + changed_metadata (Dict[str,Any]): The changed metadata. This should be a dictionary with the changed fields. Only the fields that are changed need to be included. + + Returns: + None + Example for full metadata: + + { + "status": "ENDED", + "statusLock": true, + "title": "string", + "titleLock": true, + "titleSort": "string", + "titleSortLock": true, + "summary": "string", + "summaryLock": true, + "publisher": "string", + "publisherLock": true, + "readingDirectionLock": true, + "ageRatingLock": true, + "language": "string", + "languageLock": true, + "genresLock": true, + "tagsLock": true, + "totalBookCountLock": true, + "sharingLabelsLock": true, + "linksLock": true, + "alternateTitlesLock": true, + "tags": [ + "string" + ], + "links": [ + { + "label": "string", + "url": "string" + } + ], + "readingDirection": "LEFT_TO_RIGHT", + "ageRating": 0, + "genres": [ + "string" + ], + "totalBookCount": 0, + "sharingLabels": [ + "string" + ], + "alternateTitles": [ + { + "label": "string", + "title": "string" + } + ] + } + + + """ + url = self.url + f"series/{series_id}/metadata" + data = self.patchRequest(url, changed_metadata) + return data + + def refreshMetadata(self, series_id: str) -> Optional[Dict[str, Any]]: + """Refresh the metadata of a series. + + Args: + series_id (str): The ID of the series to refresh. + + Returns: + Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"series/{series_id}/metadata/refresh" + data = self.postRequest(url) + return data + + def setReadProgress(self, series_id: str) -> Optional[Dict[str, Any]]: + """Mark all books in a series as read. + + Args: + series_id (str): The ID of the series to get the read progress from. + + Returns: + Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"series/{series_id}/read-progress" + data = self.postRequest(url) + return data + + def deleteReadProgress(self, series_id: str) -> Optional[Dict[str, Any]]: + """Delete the read progress of a series. + + Args: + series_id (str): The ID of the series to delete the read progress from. + + Returns: + Optional[Dict[str,Any]] : If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"series/{series_id}/read-progress" + data = self.deleteRequest(url) + return data + + def getSeriesThumbnail(self, series_id: str) -> str: + """Get the current active thumbnail of a series. + + Args: + series_id (str): the ID of the series to get the thumbnail from. + + Returns: + str: the thumbnail of the series formatted as bytestring. + """ + url = self.url + f"series/{series_id}/thumbnail" + data = requests.get(url, auth=(self._username, self._password)) + data = data.content + return data + + def getSeriesThumbnails(self, series_id: str) -> List[Thumbnail]: + """Get the metadata of all available thumbnails for the series. + + Args: + series_id (str): the ID of the series to get the thumbnails from. + + Returns: + List[Thumbnail]: A list of all possible thumbnails wrapped in a Thumbnail class. + """ + url = self.url + f"series/{series_id}/thumbnails" + data = self.getRequest(url) + ret = [] + for thumb in data: + ret.append(Thumbnail(**thumb)) + return ret + + def postThumbnail( + self, series_id: str, thumbnail: bytes, selected: bool = None + ) -> Union[Thumbnail, Dict[str, Any]]: + """Set a new thumbnail for the series. + + Args: + thumbnail (bytes): The thumbnail to set. This should be a bytestring. + selected (bool, optional): If the thumbnail should be set as the active thumbnail. Defaults to None. + + Returns: + None + """ + url = self.url + f"series/{series_id}/thumbnail" + data = self.postRequest(url, {"selected": selected, "thumbnail": thumbnail}) + return Thumbnail(**data) if data else None + + def getSpecificThumbnail(self, series_id: str, thumbnail_id: str) -> Thumbnail: + """Get a specific thumbnail of a series. + + Args: + series_id (str): The ID of the series to get the thumbnail from. + thumbnail_id (str): The ID of the thumbnail to get. + + Returns: + Thumbnail: The thumbnail wrapped in a Thumbnail class. + """ + url = self.url + f"series/{series_id}/thumbnails/{thumbnail_id}" + data = self.getRequest(url) + return Thumbnail(**data) + + def deleteSpecificThumbnail( + self, series_id: str, thumbnail_id: str + ) -> Optional[Dict[str, Any]]: + """Delete a specific thumbnail of a series. + + Args: + series_id (str): The ID of the series to delete the thumbnail from. + thumbnail_id (str): The ID of the thumbnail to delete. + + Returns: + Optional[Dict[str,Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"series/{series_id}/thumbnails/{thumbnail_id}" + data = self.deleteRequest(url) + return data + + def setThumbnail( + self, series_id: str, thumbnail_id: str + ) -> Optional[Dict[str, Any]]: + """Set a specific thumbnail as the active thumbnail of a series. + + Args: + series_id (str): The ID of the series to set the thumbnail to. + thumbnail_id (str): The ID of the thumbnail to set. + + Returns: + Optional[Dict[str, Any]]: If a violation is found, the server will return a dictionary with the violation. If no violation is found, None is returned. + """ + url = self.url + f"series/{series_id}/thumbnail/{thumbnail_id}/selected" + data = self.putRequest(url) + return data + + def getAlphabeticalGroups( + self, + search_string: str = None, + library_id: List[str] = None, + collection_id: List[str] = None, + status: List[str] = None, + read_status: List[str] = None, + publisher: List[str] = None, + language: List[str] = None, + genre: List[str] = None, + tag: List[str] = None, + age_rating: List[str] = None, + release_year: List[str] = None, + sharing_label: List[str] = None, + deleted: bool = None, + complete: bool = None, + oneshot: bool = None, + search_regex: str = None, + author: List[str] = None, + ) -> List[dict]: + """Get a list of all series sorted by the starting letter of the series name. + + Args: + search_string (str): The string to search for. Defaults to None.1 + library_id (List[str], optional): The library to search in. Defaults to None. + collection_id (List[str], optional): The collection to search in. Defaults to None. + status (List[str], optional): The status of the series. Available values : ENDED, ONGOING, ABANDONED, HIATUS. Defaults to None. + read_status (List[str], optional): The read status of the series. Available values : UNREAD, READ, IN_PROGRESS. Defaults to None. + publisher (List[str], optional): The publisher(s) to search for. Defaults to None. + language (List[str], optional): The language(s) to search for. Defaults to None. + genre (List[str], optional): The genre(s) to search for. Defaults to None. + tag (List[str], optional): The tag(s) to search for. Defaults to None. + age_rating (List[str], optional): The age rating to filter for. Needs to be set manually. Defaults to None. + release_year (List[str], optional): The release year(s) to limit the search. Defaults to None. + sharing_label (List[str], optional): The sharing label(s) to . Defaults to None. + deleted (bool, optional): If deleted entries should be shown as well. Defaults to None. + complete (bool, optional): If only complete entries should be shown. Defaults to None. + oneshot (bool, optional): _description_. Defaults to None. + search_regex (str, optional): _description_. Defaults to None. + author (List[str], optional): _description_. Defaults to None. + + Returns: + List[dict]: _description_ + """ + params = locals() + params = self.setParams(params) + url = self.url + "series/alphabetical-groups" + data = self.getRequest(url, params) + return data + + def getLatestSeries( + self, + library_id: List[str] = None, + deleted: bool = None, + oneshot: bool = None, + unpaged: bool = True, + page: int = None, + size: int = None, + ) -> List[Series]: + """Get the latest changed series from the server. + + Args: + library_id (List[str], optional): Limits the request to the specified libraries. Defaults to None. + deleted (bool, optional): Set to true to include the deleted series. Defaults to None. + oneshot (bool, optional): Set to true to limit search to oneshot series only. Defaults to None. + unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True. + page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None. + size (int, optional): Defines the size of the page. Defaults to None. + + Returns: + List[Series]: A list of all series that match the query. Each series is represented as a Series object. + """ + + url = self.url + "series/latest" + params = locals() + params = self.setParams(params) + data = self.getRequest(url, params) + ret = [] + for series in data["content"]: + ret.append(Series(**series)) + return ret + + def getNewSeries( + self, + library_id: List[str] = None, + deleted: bool = None, + oneshot: bool = None, + unpaged: bool = True, + page: int = None, + size: int = None, + ) -> List[Series]: + """Get the newest series from the server. + + Args: + library_id (List[str], optional): Limits the request to the specified libraries. Defaults to None. + deleted (bool, optional): Set to true to include the deleted series. Defaults to None. + oneshot (bool, optional): Set to true to limit search to oneshot series only. Defaults to None. + unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True. + page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None. + size (int, optional): Defines the size of the page. Defaults to None. + + Returns: + List[Series]: A list of all series that match the query. Each series is represented as a Series object. + """ + + url = self.url + "series/new" + params = locals() + params = self.setParams(params) + data = self.getRequest(url, params) + ret = [] + for series in data["content"]: + ret.append(Series(**series)) + return ret + + def getUpdatedSeries( + self, + library_id: List[str] = None, + deleted: bool = None, + oneshot: bool = None, + unpaged: bool = True, + page: int = None, + size: int = None, + ) -> List["Series"]: + """Get the latest updated series from the server. + + Args: + library_id (List[str], optional): Limits the request to the specified libraries. Defaults to None. + deleted (bool, optional): Set to true to include the deleted series. Defaults to None. + oneshot (bool, optional): Set to true to limit search to oneshot series only. Defaults to None. + unpaged (bool, optional): Set to False to request the result in N pages. Defaults to True. + page (int, optional): Used in conjunction with unpaged. Sets the requested page. Defaults to None. + size (int, optional): Defines the size of the page. Defaults to None. + + Returns: + List[Series]: A list of all series that match the query. Each series is represented as a Series object. + """ + + params = locals() + params = self.setParams(params) + url = self.url + "series/updated" + data = self.getRequest(url, params) + ret = [] + for series in data["content"]: + ret.append(Series(**series)) + return ret diff --git a/src/komgapi/endpoints/settings_controller.py b/src/komgapi/endpoints/settings_controller.py new file mode 100644 index 0000000..03ad4cd --- /dev/null +++ b/src/komgapi/endpoints/settings_controller.py @@ -0,0 +1,22 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.errors import KomgaError, LoginError, ResultErrror +from komgapi.schemas import * # Progress, Series + + +class SettingsController(BaseAPI): + def __init__(self, username, password, url, timeout=20) -> None: + super().__init__(username, password, url, timeout) + + def getSettings(self) -> Settings: + url = self.url + "settings" + data = self.getRequest(url) + return Settings(**data) + + def updateSettings(self, settings: Settings) -> None: + url = self.url + "settings" + data = self.patchRequest(url, settings.model_dump()) diff --git a/src/komgapi/endpoints/user_controller.py b/src/komgapi/endpoints/user_controller.py new file mode 100644 index 0000000..40ad26c --- /dev/null +++ b/src/komgapi/endpoints/user_controller.py @@ -0,0 +1,92 @@ +from .baseapi import BaseAPI +import pathlib +import subprocess +import requests +import typing_extensions +from typing import List, Optional, Dict, Any, Union +from komgapi.errors import KomgaError, LoginError, ResultErrror +from komgapi.schemas import * # Progress, Series + + +class UserController(BaseAPI): + def __init__(self, username, password, url, timeout=20, api_version="2") -> None: + super().__init__(username, password, url, timeout, api_version="2") + + def getUsers(self) -> List[User]: + url = self.url + "users" + data = self.getRequest(url) + return [User(**user) for user in data] + + def createUser(self, user: CreateUser) -> User: + url = self.url + "users" + data = self.postRequest(url, user.model_dump()) + return User(**data) + + def deleteUser(self, user_id: str) -> None: + url = self.url + f"users/{user_id}" + self.deleteRequest(url) + + def updateUser(self, user_id: str, user: User) -> None: + userData = user.model_dump() + # remove id, email and password + userData.pop("id") + userData.pop("email") + userData.pop("password") + url = self.url + f"users/{user_id}" + data = self.patchRequest(url, user) + + def getLatestAuthenticationActivity(self, user_id, api_key): + raise NotImplementedError + + def changePassword(self, user_id: str, password: str) -> None: + url = self.url + f"users/{user_id}/password" + self.patchRequest(url, {"password": password}) + + def getUserAuthenticationActivity( + self, + unpaged: bool = None, + page: int = 0, + size: int = 20, + sort: List[str] = None, + ) -> Authentication: + url = self.url + "users/authentification-activity" + params = self.setParams(locals()) + data = self.getRequest(url, params) + return Authentication(**data) + + def getMe(self) -> User | Violation: + url = self.url + "users/me" + data = self.getRequest(url) + return User(**data) if "id" in data else Violation(**data) + + def getMyAPIKeys(self) -> List[APIKey] | Violation: + url = self.url + "users/me/api-keys" + data = self.getRequest(url) + return [APIKey(**data) for key in data] if "id" in data else [] + + def createAPIKey(self, key: str) -> APIKey: + url = self.url + "users/me/api-keys" + data = self.postRequest(url, f'"comment":{key}') + return APIKey(**data) + + def deleteAPIKey(self, key_id: str) -> None: + url = self.url + f"users/me/api-keys/{key_id}" + self.deleteRequest(url) + + def getMyAuthenticationActivity( + self, + unpaged: bool = None, + page: int = 0, + size: int = 20, + sort: List[str] = None, + ) -> Authentication: + raise NotImplementedError + url = self.url + "users/me/authentication-activity" + params = self.setParams(locals()) + data = self.getRequest(url, params) + print(data) + return UserAuthActivity(**data) + + def changeMyPassword(self, password: str) -> None: + url = self.url + "users/me/password" + self.patchRequest(url, {"password": password}) diff --git a/src/komgapi/errors.py b/src/komgapi/errors.py new file mode 100644 index 0000000..08aea7f --- /dev/null +++ b/src/komgapi/errors.py @@ -0,0 +1,10 @@ +class KomgaError(Exception): + pass + + +class LoginError(KomgaError): + pass + + +class ResultErrror(KomgaError): + pass diff --git a/src/komgapi/komgapi.py b/src/komgapi/komgapi.py new file mode 100644 index 0000000..b9232a2 --- /dev/null +++ b/src/komgapi/komgapi.py @@ -0,0 +1,90 @@ + +"""Generic API for KOMGA""" + +import requests +from .endpoints import * + + +class KOMGAPI_REST: + """The REST API interface for KOMGA. This class is used to interact with the KOMGA server. It provides methods to get series, books, and more. + + + Args: + ---- + username (str): The username to use for the API. + password (str): The password to use for the API. + url (str): The URL of the KOMGA server. This should be the base URL of the server, without any paths. + timeout (int): The timeout for the requests. Defaults to 20 seconds. + + Example: + ------- + data= KOMGAPI_REST('username', 'password', 'http://localhost:8080/') + """ + + def __init__(self, username, password, url, timeout=20) -> None: + self._username = username + self._password = password + self.url = url + self.timeout = timeout + if not url.endswith("/"): + url += "/" + self.common_book_controller = CommonBookController(username, password, url, timeout) + self.series_controller = SeriesController(username, password, url, timeout) + self.readlist_controller = ReadListController(username, password, url, timeout) + self.page_hash_controller = None + self.library_controller = LibraryController(username, password, url, timeout) + self.series_collection_controller = SeriesCollectionController(username, password, url, timeout) + self.book_controler = BookController(username, password, url, timeout) + self.announcement_controller = AnnouncementController(username, password, url, timeout) + self.user_controller = UserController(username, password, url, timeout) + self.transient_books_controller = None + self.file_system_controller = None + self.claim_controller = ClaimController(username, password, url, timeout) + self.settings_controller = SettingsController(username, password, url, timeout) + self.referential_controller = ReferentialController(username, password, url, timeout) + self.login_controller = None + self.historical_events_controller = None + self.task_controller = None + self.sync_point_controller = None + self.oauth2_controller = None + + def notImplemented(self): + raise NotImplementedError("Not implemented yet") + + + def from_env(self): + """Create a KOMGA API object from environment variables. + + Returns: + KOMGAPI_REST: The KOMGA API object. + """ + import os + + return self( + os.environ["KOMGA_USERNAME"], + os.environ["KOMGA_PASSWORD"], + os.environ["KOMGA_URL"], + ) + + def test_connection(self): + """Test the connection to the KOMGA server. + + Returns: + bool: True if the connection is successful, False otherwise. + """ + try: + requests.get(self.url, auth=(self._username, self._password), timeout=self.timeout) + return True + except Exception as e: + return False + + + + + + + # Book controller + + + + \ No newline at end of file diff --git a/src/komgapi/py.typed b/src/komgapi/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/komgapi/schemas/AlternateTitle.py b/src/komgapi/schemas/AlternateTitle.py new file mode 100644 index 0000000..f756e98 --- /dev/null +++ b/src/komgapi/schemas/AlternateTitle.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class AlternateTitle(BaseModel): + label: str + title: str diff --git a/src/komgapi/schemas/Announcement.py b/src/komgapi/schemas/Announcement.py new file mode 100644 index 0000000..0016435 --- /dev/null +++ b/src/komgapi/schemas/Announcement.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from typing import List, Dict + +from pydantic import BaseModel +from komgapi.schemas.Author import Author + + +class AnnouncementItem(BaseModel): + id: str + url: str + title: str + summary: str + content_html: str + date_modified: str + author: Author + + +class Announcement(BaseModel): + version: str + title: str + home_page_url: str + description: str + items: List[AnnouncementItem] + tags: List[str] + _komga: Dict[str, str] diff --git a/src/komgapi/schemas/Authentication.py b/src/komgapi/schemas/Authentication.py new file mode 100644 index 0000000..154bc02 --- /dev/null +++ b/src/komgapi/schemas/Authentication.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class UserAccess(BaseModel): + userId: str + email: str + apiKeyId: str + apiKeyComment: str + ip: str + userAgent: str + success: bool + error: str + dateTime: str + source: str + + + +class Sort(BaseModel): + empty: bool + sorted: bool + unsorted: bool + +class Pageable(BaseModel): + offset: int + sort: Sort + pageNumber: int + pageSize: int + paged: bool + unpaged: bool + + +class Authentication(BaseModel): + totalElements: int + totalPages: int + size: int + content: List[UserAccess] + number: int + sort: Sort + first: bool + last: bool + numberOfElements: int + pageable: Pageable + empty: bool + + + + + + +class Sort(BaseModel): + empty: bool + sorted: bool + unsorted: bool + + +class Sort1(BaseModel): + empty: bool + sorted: bool + unsorted: bool + + +class Pageable(BaseModel): + offset: int + sort: Sort1 + pageNumber: int + pageSize: int + paged: bool + unpaged: bool + + +class UserAuthActivity(BaseModel): + totalElements: int + totalPages: int + size: int + content: List[UserAccess] + number: int + sort: Sort + first: bool + last: bool + numberOfElements: int + pageable: Pageable + empty: bool \ No newline at end of file diff --git a/src/komgapi/schemas/Author.py b/src/komgapi/schemas/Author.py new file mode 100644 index 0000000..b7ca2e6 --- /dev/null +++ b/src/komgapi/schemas/Author.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class Author(BaseModel): + name: str + role: str + + +class Sort(BaseModel): + empty: bool + sorted: bool + unsorted: bool + +class Pageable(BaseModel): + offset: int + sort: Sort + pageNumber: int + pageSize: int + paged: bool + unpaged: bool + + +class AuthorResponse(BaseModel): + totalElements: int + totalPages: int + size: int + content: List[Author] + number: int + sort: Sort + first: bool + last: bool + numberOfElements: int + pageable: Pageable + empty: bool \ No newline at end of file diff --git a/src/komgapi/schemas/Book.py b/src/komgapi/schemas/Book.py new file mode 100644 index 0000000..56b52f3 --- /dev/null +++ b/src/komgapi/schemas/Book.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from typing import Any, List, Optional + +from pydantic import BaseModel + +from .Author import Author +from .BooksMetadata import BookMetadata +from .Link import Link +from .Media import Media + + +class Book(BaseModel): + id: Optional[str] = None + seriesId: Optional[str] = None + seriesTitle: Optional[str] = None + libraryId: Optional[str] = None + name: Optional[str] = None + url: Optional[str] = None + number: Optional[int] = None + created: Optional[str] = None + lastModified: Optional[str] = None + fileLastModified: Optional[str] = None + sizeBytes: Optional[int] = None + size: Optional[str] = None + media: Optional[Media] = None + metadata: Optional[BookMetadata] = None + readProgress: Optional[Any] = None + deleted: Optional[bool] = None + fileHash: Optional[str] = None + oneshot: Optional[bool] = None diff --git a/src/komgapi/schemas/BooksMetadata.py b/src/komgapi/schemas/BooksMetadata.py new file mode 100644 index 0000000..742a99a --- /dev/null +++ b/src/komgapi/schemas/BooksMetadata.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel + +from .Author import Author +from .Link import Link + + +class BooksMetadata(BaseModel): + authors: Optional[List[Author]] = None + tags: Optional[List[str]] = None + releaseDate: Optional[str] = None + summary: Optional[str] = None + summaryNumber: Optional[str] = None + created: Optional[str] = None + lastModified: Optional[str] = None + + +class BookMetadata(BaseModel): + title: Optional[str] = None + titleLock: Optional[bool] = None + summary: Optional[str] = None + summaryLock: Optional[int] = None + number: Optional[str] = None + numberLock: Optional[bool] = None + numberSort: Optional[float] = None + numberSortLock: Optional[bool] = None + releaseDate: Optional[str] = None + releaseDateLock: Optional[bool] = None + authors: List[Author] + authorsLock: Optional[bool] = None + tags: List + tagsLock: Optional[bool] = None + isbn: Optional[str] = None + isbnLock: Optional[bool] = None + links: List[Link] + linksLock: Optional[bool] = None + created: Optional[str] = None + lastModified: Optional[str] = None diff --git a/src/komgapi/schemas/Collection.py b/src/komgapi/schemas/Collection.py new file mode 100644 index 0000000..28088b6 --- /dev/null +++ b/src/komgapi/schemas/Collection.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class Collection(BaseModel): + id: str + name: str + ordered: bool + seriesIds: List[str] + createdDate: str + lastModifiedDate: str + filtered: bool diff --git a/src/komgapi/schemas/Duplicate.py b/src/komgapi/schemas/Duplicate.py new file mode 100644 index 0000000..a7275db --- /dev/null +++ b/src/komgapi/schemas/Duplicate.py @@ -0,0 +1,19 @@ +from __future__ import annotations +from typing import List, Optional +from pydantic import BaseModel +from .Sort import Sort +from .Pageable import Pageable +from .Book import Book + +class Duplicate(BaseModel): + totalElements: Optional[int] = None + totalPages: Optional[int] = None + size: Optional[int] = None + content: Optional[List[Book]] = None + number: Optional[int] = None + sort: Optional[Sort] = None + first: Optional[bool] = None + last: Optional[bool] = None + numberOfElements: Optional[int] = None + pageable: Optional[Pageable] = None + empty: Optional[bool] = None \ No newline at end of file diff --git a/src/komgapi/schemas/Latest.py b/src/komgapi/schemas/Latest.py new file mode 100644 index 0000000..a7216f1 --- /dev/null +++ b/src/komgapi/schemas/Latest.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel + +from .Pageable import Pageable +from .Series import Series +from .Sort import Sort + + +class LatestSeriesData(BaseModel): + totalElements: Optional[int] = None + totalPages: Optional[int] = None + size: Optional[int] = None + content: Optional[List[Series]] = None + number: Optional[int] = None + sort: Optional[Sort] = None + first: Optional[bool] = None + last: Optional[bool] = None + numberOfElements: Optional[int] = None + pageable: Optional[Pageable] = None + empty: Optional[bool] = None diff --git a/src/komgapi/schemas/Library.py b/src/komgapi/schemas/Library.py new file mode 100644 index 0000000..bc0a18a --- /dev/null +++ b/src/komgapi/schemas/Library.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel + + +class Library(BaseModel): + id: Optional[str] = None + name: Optional[str] = None + root: Optional[str] = None + importComicInfoBook: Optional[bool] = None + importComicInfoSeries: Optional[bool] = None + importComicInfoCollection: Optional[bool] = None + importComicInfoReadList: Optional[bool] = None + importComicInfoSeriesAppendVolume: Optional[bool] = None + importEpubBook: Optional[bool] = None + importEpubSeries: Optional[bool] = None + importMylarSeries: Optional[bool] = None + importLocalArtwork: Optional[bool] = None + importBarcodeIsbn: Optional[bool] = None + scanForceModifiedTime: Optional[bool] = None + scanInterval: Optional[str] = None + scanOnStartup: Optional[bool] = None + scanCbx: Optional[bool] = None + scanPdf: Optional[bool] = None + scanEpub: Optional[bool] = None + scanDirectoryExclusions: Optional[List[str]] = None + repairExtensions: Optional[bool] = None + convertToCbz: Optional[bool] = None + emptyTrashAfterScan: Optional[bool] = None + seriesCover: Optional[str] = None + hashFiles: Optional[bool] = None + hashPages: Optional[bool] = None + analyzeDimensions: Optional[bool] = None + oneshotsDirectory: Optional[str] = None + unavailable: Optional[bool] = None + +class CreateLibrary(BaseModel): + name: str + root: str + importComicInfoBook: bool + importComicInfoSeries: bool + importComicInfoCollection: bool + importComicInfoReadList: bool + importComicInfoSeriesAppendVolume: bool + importEpubBook: bool + importEpubSeries: bool + importMylarSeries: bool + importLocalArtwork: bool + importBarcodeIsbn: bool + scanForceModifiedTime: bool + scanInterval: str + scanOnStartup: bool + scanCbx: bool + scanPdf: bool + scanEpub: bool + scanDirectoryExclusions: List[str] + repairExtensions: bool + convertToCbz: bool + emptyTrashAfterScan: bool + seriesCover: str + hashFiles: bool + hashPages: bool + analyzeDimensions: bool + oneshotsDirectory: str + diff --git a/src/komgapi/schemas/Link.py b/src/komgapi/schemas/Link.py new file mode 100644 index 0000000..74b20fb --- /dev/null +++ b/src/komgapi/schemas/Link.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class Link(BaseModel): + label: str + url: str diff --git a/src/komgapi/schemas/Locations.py b/src/komgapi/schemas/Locations.py new file mode 100644 index 0000000..7929e5f --- /dev/null +++ b/src/komgapi/schemas/Locations.py @@ -0,0 +1,10 @@ +from __future__ import annotations +from typing import List, Optional +from pydantic import BaseModel + +class Locations(BaseModel): + fragments: Optional[List[str]] = None + progression: Optional[int] = None + position: Optional[int] = None + totalProgression: Optional[int] = None + \ No newline at end of file diff --git a/src/komgapi/schemas/Manifest.py b/src/komgapi/schemas/Manifest.py new file mode 100644 index 0000000..ab5ea2c --- /dev/null +++ b/src/komgapi/schemas/Manifest.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel + + +class Link(BaseModel): + title: Optional[str] = None + rel: Optional[str] = None + href: Optional[str] = None + type: Optional[str] = None + templated: Optional[bool] = None + width: Optional[int] = None + height: Optional[int] = None + + +class Series(BaseModel): + name: Optional[str] = None + position: Optional[int] = None + links: Optional[List[Link]] = None + + +class CollectionItem(BaseModel): + name: Optional[str] = None + position: Optional[int] = None + links: Optional[List[Link]] = None + + +class BelongsTo(BaseModel): + series: Optional[List[Series]] = None + collection: Optional[List[CollectionItem]] = None + + +class Rendition(BaseModel): + additionalProp1: Dict[str, Any] + additionalProp2: Dict[str, Any] + additionalProp3: Dict[str, Any] + + +class Metadata(BaseModel): + title: Optional[str] = None + identifier: Optional[str] = None + type: Optional[str] = None + conformsTo: Optional[str] = None + sortAs: Optional[str] = None + subtitle: Optional[str] = None + modified: Optional[str] = None + published: Optional[str] = None + language: Optional[str] = None + author: Optional[List[str]] = None + translator: Optional[List[str]] = None + editor: Optional[List[str]] = None + artist: Optional[List[str]] = None + illustrator: Optional[List[str]] = None + letterer: Optional[List[str]] = None + penciler: Optional[List[str]] = None + colorist: Optional[List[str]] = None + inker: Optional[List[str]] = None + contributor: Optional[List[str]] = None + publisher: Optional[List[str]] = None + subject: Optional[List[str]] = None + readingProgression: Optional[str] = None + description: Optional[str] = None + numberOfPages: Optional[int] = None + belongsTo: Optional[BelongsTo] = None + rendition: Optional[Rendition] = None + + +class Image(BaseModel): + title: Optional[str] = None + rel: Optional[str] = None + href: Optional[str] = None + type: Optional[str] = None + templated: Optional[bool] = None + width: Optional[int] = None + height: Optional[int] = None + + +class ReadingOrderItem(BaseModel): + title: Optional[str] = None + rel: Optional[str] = None + href: Optional[str] = None + type: Optional[str] = None + templated: Optional[bool] = None + width: Optional[int] = None + height: Optional[int] = None + + +class Resource(BaseModel): + title: Optional[str] = None + rel: Optional[str] = None + href: Optional[str] = None + type: Optional[str] = None + templated: Optional[bool] = None + width: Optional[int] = None + height: Optional[int] = None + + +class TocItem(BaseModel): + title: Optional[str] = None + rel: Optional[str] = None + href: Optional[str] = None + type: Optional[str] = None + templated: Optional[bool] = None + width: Optional[int] = None + height: Optional[int] = None + + +class Landmark(BaseModel): + title: Optional[str] = None + rel: Optional[str] = None + href: Optional[str] = None + type: Optional[str] = None + templated: Optional[bool] = None + width: Optional[int] = None + height: Optional[int] = None + + +class PageListItem(BaseModel): + title: Optional[str] = None + rel: Optional[str] = None + href: Optional[str] = None + type: Optional[str] = None + templated: Optional[bool] = None + width: Optional[int] = None + height: Optional[int] = None + + +class Manifest(BaseModel): + context: Optional[str] = None + metadata: Optional[Metadata] = None + links: Optional[List[Link]] = None + images: Optional[List[Image]] = None + readingOrder: Optional[List[ReadingOrderItem]] = None + resources: Optional[List[Resource]] = None + toc: Optional[List[TocItem]] = None + landmarks: Optional[List[Landmark]] = None + pageList: Optional[List[PageListItem]] = None diff --git a/src/komgapi/schemas/Media.py b/src/komgapi/schemas/Media.py new file mode 100644 index 0000000..6cd9ab7 --- /dev/null +++ b/src/komgapi/schemas/Media.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class Media(BaseModel): + status: str + mediaType: str + pagesCount: int + comment: str + epubDivinaCompatible: bool + mediaProfile: str diff --git a/src/komgapi/schemas/Metadata.py b/src/komgapi/schemas/Metadata.py new file mode 100644 index 0000000..deb9788 --- /dev/null +++ b/src/komgapi/schemas/Metadata.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel + +from .AlternateTitle import AlternateTitle +from .Link import Link + + +class Metadata(BaseModel): + status: Optional[str] | None = None + statusLock: bool + title: Optional[str] | None = None + titleLock: bool + titleSort: Optional[str] | None = None + titleSortLock: bool + summary: Optional[str] | None = None + summaryLock: bool + readingDirection: Optional[str] | None = None + readingDirectionLock: bool + publisher: Optional[str] | None = None + publisherLock: bool + ageRating: Optional[int] | None = None + ageRatingLock: bool + language: Optional[str] | None = None + languageLock: bool + genres: List[str] | None + genresLock: bool + tags: List[str] | None = None + tagsLock: bool + totalBookCount: Optional[int] | None = None + totalBookCountLock: bool + sharingLabels: List[str | None] = None + sharingLabelsLock: bool + links: List[Link] | None = None + linksLock: bool + alternateTitles: List[AlternateTitle] | None = None + alternateTitlesLock: bool + created: Optional[str] | None = None + lastModified: Optional[str] | None = None diff --git a/src/komgapi/schemas/Page.py b/src/komgapi/schemas/Page.py new file mode 100644 index 0000000..656fa65 --- /dev/null +++ b/src/komgapi/schemas/Page.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel + + +class Page(BaseModel): + number: Optional[int] = None + fileName: Optional[str] = None + mediaType: Optional[str] = None + width: Optional[int] = None + height: Optional[int] = None + sizeBytes: Optional[int] = None + size: Optional[str] = None diff --git a/src/komgapi/schemas/Pageable.py b/src/komgapi/schemas/Pageable.py new file mode 100644 index 0000000..2edf594 --- /dev/null +++ b/src/komgapi/schemas/Pageable.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + +from .Sort import Sort + + +class Pageable(BaseModel): + offset: int + sort: Sort + pageNumber: int + pageSize: int + paged: bool + unpaged: bool \ No newline at end of file diff --git a/src/komgapi/schemas/Position.py b/src/komgapi/schemas/Position.py new file mode 100644 index 0000000..22f4c99 --- /dev/null +++ b/src/komgapi/schemas/Position.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel +from .Text import Text +from .Locations import Locations + + + + +class Position(BaseModel): + href: Optional[str] = None + type: Optional[str] = None + title: Optional[str] = None + locations: Optional[Locations] = None + text: Optional[Text] = None + + +class Position(BaseModel): + total: Optional[int] = None + positions: Optional[List[Position]] = None diff --git a/src/komgapi/schemas/Progress.py b/src/komgapi/schemas/Progress.py new file mode 100644 index 0000000..bed151f --- /dev/null +++ b/src/komgapi/schemas/Progress.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel +from .Text import Text +from .Locations import Locations + + +class Device(BaseModel): + id: Optional[str] = None + name: Optional[str] = None + + +class Locator(BaseModel): + href: Optional[str] = None + type: Optional[str] = None + title: Optional[str] = None + locations: Optional[Locations] = None + text: Optional[Text] = None + + +class Progress(BaseModel): + modified: Optional[str] = None + device: Optional[Device] = None + locator: Optional[Locator] = None diff --git a/src/komgapi/schemas/Readlist.py b/src/komgapi/schemas/Readlist.py new file mode 100644 index 0000000..c432f8d --- /dev/null +++ b/src/komgapi/schemas/Readlist.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class Readlist(BaseModel): + name: str + summary: str + ordered: bool + bookIds: List[str] diff --git a/src/komgapi/schemas/Series.py b/src/komgapi/schemas/Series.py new file mode 100644 index 0000000..5c0765b --- /dev/null +++ b/src/komgapi/schemas/Series.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel + +from .AlternateTitle import AlternateTitle +from .BooksMetadata import BooksMetadata +from .Link import Link +from .Metadata import Metadata + + +class Series(BaseModel): + id: Optional[str] = None + libraryId: Optional[str] = None + name: Optional[str] = None + url: Optional[str] = None + created: Optional[str] = None + lastModified: Optional[str] = None + fileLastModified: Optional[str] = None + booksCount: Optional[int] = None + booksReadCount: Optional[int] = None + booksUnreadCount: Optional[int] = None + booksInProgressCount: Optional[int] = None + metadata: Optional[Metadata] = None + booksMetadata: Optional[BooksMetadata] = None + deleted: Optional[bool] = None + oneshot: Optional[bool] = None diff --git a/src/komgapi/schemas/Sort.py b/src/komgapi/schemas/Sort.py new file mode 100644 index 0000000..0aa8bdb --- /dev/null +++ b/src/komgapi/schemas/Sort.py @@ -0,0 +1,11 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class Sort(BaseModel): + empty: bool + sorted: bool + unsorted: bool diff --git a/src/komgapi/schemas/Text.py b/src/komgapi/schemas/Text.py new file mode 100644 index 0000000..14f6c17 --- /dev/null +++ b/src/komgapi/schemas/Text.py @@ -0,0 +1,8 @@ +from __future__ import annotations +from typing import List, Optional +from pydantic import BaseModel + +class Text(BaseModel): + after: Optional[str] = None + before: Optional[str] = None + highlight: Optional[str] = None diff --git a/src/komgapi/schemas/Thumbnail.py b/src/komgapi/schemas/Thumbnail.py new file mode 100644 index 0000000..aadf845 --- /dev/null +++ b/src/komgapi/schemas/Thumbnail.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel + + +class Thumbnail(BaseModel): + id: str + seriesId: str + type: str + selected: bool + mediaType: str + fileSize: int + width: int + height: int + + +class ReadlistThumbnail(BaseModel): + id: str + readlistId: str + type: str + selected: bool + mediaType: str + fileSize: int + width: int + height: int diff --git a/src/komgapi/schemas/User.py b/src/komgapi/schemas/User.py new file mode 100644 index 0000000..53bdb17 --- /dev/null +++ b/src/komgapi/schemas/User.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from typing import List, Dict + +from pydantic import BaseModel + +class User(BaseModel): + """Object representing a User + + Args: + BaseModel (Model): Pydantic BaseModel + """ + id:str + email:str + roles:List[str] + sharedAllLibraries:bool + sharedLibrariesIds:List[str] + labelsAllow:List[str] + labelsExclude:List[str] + ageRestriction:Dict[str, int|str] | None + + +class CreateUser(BaseModel): + email:str + password:str + roles:List[str] \ No newline at end of file diff --git a/src/komgapi/schemas/__init__.py b/src/komgapi/schemas/__init__.py new file mode 100644 index 0000000..3dc44b3 --- /dev/null +++ b/src/komgapi/schemas/__init__.py @@ -0,0 +1,30 @@ +from .AlternateTitle import AlternateTitle +from .Author import Author, AuthorResponse +from .Book import Book +from .BooksMetadata import BooksMetadata, BookMetadata +from .Collection import Collection +from .Duplicate import Duplicate +from .Latest import LatestSeriesData +from .Library import Library, CreateLibrary +from .Link import Link +from .Locations import Locations +from .Manifest import Manifest +from .Media import Media +from .Metadata import Metadata +from .Page import Page +from .Pageable import Pageable +from .Position import Position +from .Progress import Progress +from .Readlist import Readlist +from .Series import Series +from .Sort import Sort +from .Text import Text +from .Thumbnail import Thumbnail, ReadlistThumbnail +from .Announcement import Announcement +from .User import User, CreateUser +from .Authentication import Authentication, UserAuthActivity +from .apikey import APIKey +from .settings import Settings + + +from .violation import Violation \ No newline at end of file diff --git a/src/komgapi/schemas/apiSearch.py b/src/komgapi/schemas/apiSearch.py new file mode 100644 index 0000000..b9a8b54 --- /dev/null +++ b/src/komgapi/schemas/apiSearch.py @@ -0,0 +1,22 @@ +ALLSERIESPARAMS = { + "search": str | None, + "library_id": list[str] | None, + "collection_id": list[str] | None, + "status": list[str] | None, + "read_status": list[str] | None, + "publisher": list[str] | None, + "genre": list[str] | None, + "language": list[str] | None, + "tag": list[str] | None, + "age_rating": list[str] | None, + "sharing_label": list[str] | None, + "deleted": bool | None, + "complete": bool | None, + "oneshot": bool | None, + "unpaged": bool | None, + "search_regex": str | None, + "page": int | None, + "size": int | None, + "sort": list[str] | None, + "authors": list[str] | None, +} diff --git a/src/komgapi/schemas/apikey.py b/src/komgapi/schemas/apikey.py new file mode 100644 index 0000000..dacaa88 --- /dev/null +++ b/src/komgapi/schemas/apikey.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +from pydantic import BaseModel + + +class APIKey(BaseModel): + id: str + userId: str + key: str + comment: str + createdDate: str + lastModifiedDate: str \ No newline at end of file diff --git a/src/komgapi/schemas/settings.py b/src/komgapi/schemas/settings.py new file mode 100644 index 0000000..3a7806d --- /dev/null +++ b/src/komgapi/schemas/settings.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from pydantic import BaseModel + + +class ServerPort(BaseModel): + configurationSource: int + databaseSource: int + effectiveValue: int + + +class ServerContextPath(BaseModel): + configurationSource: str + databaseSource: str + effectiveValue: str + + +class KepubifyPath(BaseModel): + configurationSource: str + databaseSource: str + effectiveValue: str + + +class Settings(BaseModel): + deleteEmptyCollections: bool + deleteEmptyReadLists: bool + rememberMeDurationDays: int + thumbnailSize: str + taskPoolSize: int + serverPort: ServerPort + serverContextPath: ServerContextPath + koboProxy: bool + koboPort: int + kepubifyPath: KepubifyPath diff --git a/src/komgapi/schemas/violation.py b/src/komgapi/schemas/violation.py new file mode 100644 index 0000000..be711cb --- /dev/null +++ b/src/komgapi/schemas/violation.py @@ -0,0 +1,11 @@ +from __future__ import annotations + + +from pydantic import BaseModel + + +class Violation(BaseModel): + fieldName: str + message: str + +