initial commit

.gitignore

@@ -0,0 +1 @@
+__pycache__

pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "komtagger"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "anilistapi",
+    "komgapi",
+    "limit>=0.2.3",
+    "loguru>=0.7.3",
+    "python-dotenv>=1.0.1",
+    "requests>=2.32.3",
+]
+
+[dependency-groups]
+dev = [
+    "bump-my-version>=0.32.1",
+]
+
+[tool.uv.sources]
+komgapi = { workspace = true }
+anilistapi = { workspace = true }

src/logger.py

@@ -0,0 +1,7 @@
+from loguru import logger
+import sys
+logger.remove()
+log = logger
+log.add("logs/tagger.log",rotation="1 week", compression="zip")
+#logs with severity of DEBUG or higher to terminal
+log.add(sys.stderr, level="DEBUG")

src/similarity.py

@@ -0,0 +1,10 @@
+# calculate similarity between two strings
+from difflib import SequenceMatcher
+
+
+def similarity(a, b):
+    return SequenceMatcher(None, a, b).ratio()
+
+
+if __name__ == "__main__":
+    print(similarity("Dr. Stone", "Dr. STONE"))

src/tagger.py

@@ -0,0 +1,244 @@
+from komgAPI import KOMGAPI_REST as komga
+from komgAPI.schemas import Metadata
+from .api import constants
+from .api.mangadex import MangadexAPI
+from .api.anilistapi import anilistAPI
+from typing import Union
+from dotenv import load_dotenv
+from .logger import log
+import os
+from src.logic.metadataHandlers import handleComicVine, handleAnilist
+from .similarity import similarity
+
+load_dotenv()
+
+
+def translate(input: str):
+    # replace  & with and
+    title = input.replace("&", "and")
+    title = title.replace(" ", "%20")
+    return title
+
+
+def write(name, file):
+    with open(f"{file}.txt", "a") as f:
+        # add a new line
+        f.write(name + "\n")
+
+
+SKIPS = ["Fairy Tail Side Stories"]
+
+
+class Tagger:
+    def __init__(self, fallbackAPI=None):
+        self.komga = komga(
+            url=os.getenv("KOMGA_URL"),
+            username=os.getenv("KOMGA_USERNAME"),
+            password=os.getenv("KOMGA_PASSWORD"),
+            timeout=60,
+        )
+        self.mangadex = MangadexAPI()
+        self.mode = None
+
+        self.api: anilistAPI = anilistAPI()
+
+        log.info("Tagger initialized")
+
+    def get_manga(self, manga_id):
+        manga = self.komga.series_controller.getSeries(manga_id)
+        return manga
+
+    @property
+    def overwrite(self):
+        self.mode = "overwrite"
+        log.info("Mode set to overwrite")
+
+    @property
+    def merge(self):
+        self.mode = "merge"
+        log.info("Mode set to merge")
+
+    def get_and_update_manga(self, komga_id, mangadex_id):
+        api_metadata = self.api.get_metadata(mangadex_id)
+        komga_metadata = self.komga.series_controller.getSeries(komga_id)
+        converted_api_metadata = handleAnilist(api_metadata)
+
+        newMetadata = {
+            "status": converted_api_metadata.status,
+            "statusLock": True,
+            "title": api_metadata.title.english
+            if api_metadata.title.english
+            else komga_metadata.name,
+            "titleLock": True,
+            "titleSort": api_metadata.title.english
+            if api_metadata.title.english
+            else komga_metadata.name,
+            "titleSortLock": True,
+            "summary": api_metadata.description.split("<br></br>")[0],
+            "summaryLock": True,
+            "readingDirectionLock": True,
+            "ageRating": converted_api_metadata.ageRating,
+            "ageRatingLock": True,
+            "language": "en",
+            "languageLock": True,
+            "genresLock": True,
+            "tagsLock": True,
+            "totalBookCountLock": False,
+            "sharingLabelsLock": False,
+            "linksLock": True,
+            "alternateTitlesLock": True,
+            "tags": converted_api_metadata.tags,
+            "links": converted_api_metadata.links,
+            "readingDirection": converted_api_metadata.readingDirection,
+            "genres": api_metadata.genres,
+            "totalBookCount": converted_api_metadata.totalBookCount,
+            "alternateTitles": converted_api_metadata.alternateTitles,
+        }
+        if self.mode == "overwrite":
+            log.debug(f"Overwriting metadata for {komga_metadata.name}")
+            log.debug(newMetadata)
+            self.komga.series_controller.patchMetadata(komga_id, newMetadata)
+        elif self.mode == "merge":
+            komga_tags = komga_metadata.metadata.tags
+            newMetadata["tags"] = list(set(komga_tags + newMetadata["tags"]))
+            komga_genres = komga_metadata.metadata.genres
+            newMetadata["genres"] = list(set(komga_genres + newMetadata["genres"]))
+            komga_links = komga_metadata.metadata.links
+            newMetadata["links"] = list(set(komga_links + newMetadata["links"]))
+
+            self.komga.series_controller.patchMetadata(komga_id, newMetadata)
+
+        write(
+            api_metadata.title.english
+            if api_metadata.title.english
+            else komga_metadata.name,
+            "success",
+        )
+
+    def tag(self):
+        komgaseries = self.komga.series_controller.getAllSeries()
+        if not os.path.exists("success.txt"):
+            with open("success.txt", "w") as f:
+                f.write("")
+        with open("success.txt", "r") as f:
+            done = f.read().splitlines()
+        for series in komgaseries:
+            seriesTitle = series.name.strip()
+            if "Omnibus" in seriesTitle:
+                seriesTitle = seriesTitle.replace("Omnibus", "").strip()
+
+            if seriesTitle in done:
+                log.success(f"{seriesTitle} already tagged, skipping")
+                continue
+            if seriesTitle in SKIPS:
+                log.info(f"{seriesTitle} is in the skip list, skipping")
+                continue
+            if "&" in seriesTitle:
+                log.info(f"{seriesTitle} contains an ampersand, might cause issues")
+                # continue
+            log.info(f"Tagging {seriesTitle}")
+            result = self.get_metadata(seriesTitle, series.id)
+            if result == 1:
+                log.success(f"Tagged {seriesTitle} successfully")
+                write(seriesTitle, "success")
+            else:
+                log.info("Found no titles using main title, using alternate titles")
+                mangadex_titles = self.api.getSeries(series.metadata.get("title"))
+                if mangadex_titles is None:
+                    log.error(f"No titles found for {seriesTitle}")
+                    write(seriesTitle, "failed")
+                    continue
+                for result in mangadex_titles:
+                    title = result.title.english
+                    alt_titles = result.synonyms
+                    result = self.get_metadata(series.metadata.get("title"), series.id)
+                    if result == 1:
+                        log.success(f"Tagged {seriesTitle} successfully")
+                        write(seriesTitle, "success")
+            #     for alt in alt_titles:
+            #         # alt is a dict, lang: title
+            #         title = alt.get(list(alt.keys())[0])
+            #         if series.name.strip() == title.strip():
+            #             self.get_and_update_manga(series.id, result.series_id)
+            #             log.success(f"Tagged {series.name} successfully")
+            #             tagged = True
+            #             break
+            #     if not tagged:
+            #         log.error(f"No match found for {series.name}")
+            #         write(series.name, "failed")
+            #     else:
+            #         break
+
+    def get_metadata(self, mtitle: str, id=None):
+        mangadex_titles = self.api.getSeries(mtitle)
+        if mangadex_titles is None or len(mangadex_titles) == 0:
+            log.error(f"No api titles found for {mtitle}")
+            write(mtitle, "failed")
+            return 0
+        tagged = False
+        for result in mangadex_titles:
+            if result.title.english is None:
+                log.error(f"No title found for {mtitle}")
+                return 0
+            if similarity(mtitle.lower(), result.title.english.lower()) > 0.8:
+                title = result.title.english
+                log.info(f"Found API result for {mtitle}")
+                log.debug(result)
+                if title is None:
+                    log.warning(f"No title found for {mtitle}")
+                    write(mtitle, "failed")
+                    return 0
+                alt_titles = result.synonyms
+                if title.strip() == title.strip():
+                    self.get_and_update_manga(id, result.id)
+                    log.success(f"Tagged {mtitle} successfully")
+                    return 1
+                else:
+                    log.info(f"No direct match, searching in alternate Titles {mtitle}")
+                    if similarity(mtitle, title) > 0.8:
+                        self.get_and_update_manga(id, result.id)
+                        log.success(f"Tagged {mtitle} successfully")
+                        return 1
+
+                    for alt in alt_titles:
+                        if similarity(mtitle, alt) > 0.8:
+                            self.get_and_update_manga(id, result.id)
+                            log.success(f"Tagged {mtitle} successfully")
+                            tagged = True
+                            break
+                    if not tagged:
+                        log.error(f"No match found for {mtitle}")
+                        return 0
+                    else:
+                        return 1
+
+    def tag_series(self, series_id):
+        series = self.komga.series_controller.getSeries(series_id)
+        log.info(f"Tagging {series.name}")
+        title = translate(series.name)
+        mangadex_titles = self.mangadex.get_series(title)
+        if len(mangadex_titles) == 0:
+            log.warning(f"No mangadex titles found for {series.name}")
+        print(len(mangadex_titles))
+        for result in mangadex_titles:
+            if similarity(series.name, result.name) > 0.8:
+                title = result.name
+                alt_titles = result.alternate_names
+                if series.name.strip() == title.strip():
+                    self.get_and_update_manga(series.id, result.series_id)
+                    log.info(f"Tagged {series.name} successfully")
+                else:
+                    log.info(
+                        f"No direct match, searching in alternate Titles {series.name}"
+                    )
+                    found = False
+                    for alt in alt_titles:
+                        # alt is a dict, lang: title
+                        title = alt.get(list(alt.keys())[0])
+                        if series.name.strip() == title.strip():
+                            self.get_and_update_manga(series.id, result.series_id)
+                            log.info(f"Tagged {series.name} successfully")
+                            found = True
+                            break
+                    if found:
+                        break