Fixed kaiscans and Comick. Changed comick.cc to comick.io

Changelog.md

@@ -2,6 +2,14 @@
 
 #### Consider supporting me on [Patreon](https://patreon.com/mooshi69) or [Ko-Fi](https://ko-fi.com/mooshi69)!
 
+## // March 31st 2024
+
+### Bug Fixes:
+
+- Fixed Kaiscans having incorrect selector for the front page cover for manhwa.
+- Changed comicks domain to .io from .cc and changed the bot to use their public API insteasd.
+- Attempted fix for update check on series from Asura that don't have an ID assigned yet.
+
 ## // March 29th 2024
 
 - Added new setting -> Set a bot manager role that will give you access to the track commands without requiring perms.

src/core/apis/comickAPI.py

@@ -15,7 +15,7 @@ def __init__(
             self,
             api_manager: APIManager
     ):
-        self.api_url: str = "https://api.comick.cc"
+        self.api_url: str = "https://api.comick.fun"
         self.manager = api_manager
         self.headers = {
             # "User-Agent": "github.com/MooshiMochi/ManhwaUpdatesBot",
@@ -63,12 +63,16 @@ async def __request(
             self.manager.session.logger.error("Server disconnected, retrying request...")
             return await self.__request(method, endpoint, params, data, headers, **kwargs)
 
-    async def get_manga(self, manga_id: str) -> Dict[str, Any]:
-        endpoint = f"comic/{manga_id}"
+    async def get_manga(self, url_name: str) -> Dict[str, Any]:
+        endpoint = f"comic/{url_name}"
         return await self.__request("GET", endpoint)
 
-    async def get_synopsis(self, manga_id: str) -> Optional[str]:
-        endpoint = f"comic/{manga_id}"
+    async def get_id(self, url_name: str) -> str:
+        data = await self.get_manga(url_name)
+        return data["comic"]["hid"]
+
+    async def get_synopsis(self, url_name: str) -> Optional[str]:
+        endpoint = f"comic/{url_name}"
         data = await self.__request("GET", endpoint)
         return data.get("comic", {}).get("desc", None)
 

src/core/scanlators/api_based.py

@@ -8,7 +8,6 @@
 from src.core.objects import Chapter, PartialManga
 from .classes import AbstractScanlator, scanlators
 from ...static import RegExpressions
-from ...utils import raise_and_report_for_status
 
 __all__ = (
     "scanlators",
@@ -17,8 +16,8 @@
 
 class _Comick(AbstractScanlator):
     rx: re.Pattern = RegExpressions.comick_url
-    icon_url = "https://comick.cc/static/icons/unicorn-256_maskable.png"
-    base_url = "https://comick.cc"
+    icon_url = "https://comick.io/static/icons/unicorn-256_maskable.png"
+    base_url = "https://comick.io"
     fmt_url = base_url + "/comic/{url_name}?lang=en"
     chp_url_fmt = base_url + "/comic/{url_name}/{chapter_id}"
 
@@ -48,14 +47,14 @@ def check_ownership(self, raw_url: str) -> bool:
         return self.rx.search(raw_url) is not None
 
     async def get_synopsis(self, raw_url: str) -> str:
-        manga_id = await self.get_id(raw_url)
-        return await self.bot.apis.comick.get_synopsis(manga_id)
+        url_name = await self._get_url_name(raw_url)
+        return await self.bot.apis.comick.get_synopsis(url_name)
 
     async def get_all_chapters(self, raw_url: str) -> list[Chapter] | None:
         manga_id = await self.get_id(raw_url)
         chapters = await self.bot.apis.comick.get_chapters_list(manga_id)
         if chapters:
-            url_name = self.rx.search(raw_url).groupdict()["url_name"]
+            url_name = await self._get_url_name(raw_url)
             return [
                 Chapter(
                     self.chp_url_fmt.format(
@@ -67,22 +66,27 @@ async def get_all_chapters(self, raw_url: str) -> list[Chapter] | None:
         else:
             return []
 
+    async def _get_url_name(self, raw_url: str) -> str:
+        try:
+            return self.rx.search(raw_url).groupdict().get("url_name")
+        except (AttributeError, TypeError) as e:
+            self.bot.logger.error(raw_url)
+            raise e
+
     async def get_title(self, raw_url: str) -> str | None:
-        manga_id = await self.get_id(raw_url)
-        manga = await self.bot.apis.comick.get_manga(manga_id)
+        url_name = await self._get_url_name(raw_url)
+        manga = await self.bot.apis.comick.get_manga(url_name)
         if manga.get("statusCode", 200) == 404:
             return None
         return manga["comic"]["title"]
 
     async def get_id(self, raw_url: str) -> str | None:
-        async with self.bot.session.get(raw_url) as resp:
-            await raise_and_report_for_status(self.bot, resp)
-            manga_id = re.search(r"\"hid\":\"([^\"]+)\"", await resp.text()).group(1)
-            return manga_id
+        url_name = await self._get_url_name(raw_url)
+        return await self.bot.apis.comick.get_id(url_name)
 
     async def get_status(self, raw_url: str) -> str:
-        manga_id = await self.get_id(raw_url)
-        manga = await self.bot.apis.comick.get_manga(manga_id)
+        url_name = await self._get_url_name(raw_url)
+        manga = await self.bot.apis.comick.get_manga(url_name)
         if manga.get("statusCode", 200) == 404:
             return "Unknown"
         status_map = {1: "Ongoing", 2: "Completed", 3: "Cancelled", 4: "Hiatus"}
@@ -92,12 +96,12 @@ async def format_manga_url(
             self, raw_url: Optional[str] = None, url_name: Optional[str] = None, _: Optional[str] = None
     ) -> str:
         if not url_name:
-            url_name = self.rx.search(raw_url).groupdict()["url_name"]
+            url_name = await self._get_url_name(raw_url)
         return self.fmt_url.format(url_name=url_name)
 
     async def get_cover(self, raw_url: str) -> str | None:
-        manga_id = await self.get_id(raw_url)
-        return await self.bot.apis.comick.get_cover(manga_id)
+        url_name = await self._get_url_name(raw_url)
+        return await self.bot.apis.comick.get_cover(url_name)
 
     async def search(self, query: str, as_em: bool = False) -> list[PartialManga] | list[discord.Embed]:
         json_resp: list[dict] = await self.bot.apis.comick.search(query=query)

src/core/scanlators/classes.py

@@ -72,8 +72,8 @@ async def _get_status_tag(self: BasicScanlator, raw_url: str) -> bs4.Tag | None:
             if not isinstance(self, DynamicURLScanlator):
                 request_url = await self.format_manga_url(raw_url, use_ajax_url=True)
             else:
-                requests_url = raw_url
-            text = await self._get_text(requests_url, method=method)  # noqa
+                request_url = raw_url
+            text = await self._get_text(request_url, method=method)  # noqa
         else:
 
             text = await self._get_text(
@@ -818,6 +818,45 @@ def _extract_dynamic_ids(self, partial_manhwas: list[PartialManga]):
             if (manga_id_found and chapter_id_found) is True:
                 break
 
+    async def _get_text(self, url: str, method: Literal["GET", "POST"] = "GET", **params) -> str:
+        # one thing to note:
+        # headers are only really required to let websites that gave us special access to identify us.
+        # so, realistically, we can ignore headers in the flare request, as it's intended to be used for websites that
+        # block us.
+        provided_headers = params.pop("headers", None)
+        if not provided_headers: provided_headers = {}  # noqa: Allow inline operation
+        headers = ((self.create_headers() or {}) | provided_headers) or None
+        if self.json_tree.request_method == "http":
+            async with self.bot.session.request(
+                    method, url, headers=headers, **self.get_extra_req_kwargs(), **params
+            ) as resp:
+                if resp.status == 404 and self.manga_id is not None:
+                    if self.manga_id in url:
+                        self.bot.logger.warning(f"404 error on {url}. Removing manga_id from URL and trying again.")
+                        url = url.replace(self.manga_id + self.json_tree.properties.missing_id_connector_char, "")
+                        return await self._get_text(url, method, **params)
+
+                await raise_and_report_for_status(self.bot, resp)
+                return await resp.text()
+        elif self.json_tree.request_method == "curl":
+            resp = await self.bot.curl_session.request(
+                method, url, headers=headers, **self.get_extra_req_kwargs(), **params
+            )
+            if resp.status_code == 404 and self.manga_id is not None:
+                if self.manga_id in url:
+                    self.bot.logger.warning(f"404 error on {url}. Removing manga_id from URL and trying again.")
+                    url = url.replace(self.manga_id + self.json_tree.properties.missing_id_connector_char, "")
+                    return await self._get_text(url, method, **params)
+
+            await raise_and_report_for_status(self.bot, resp)
+            return resp.text
+        elif self.json_tree.request_method == "flare":
+            resp = await self.bot.apis.flare.get(url, headers=headers, **params)
+            await raise_and_report_for_status(self.bot, resp)
+            return (await resp.json()).get("solution", {}).get("response")
+        else:
+            raise ValueError(f"Unknown {self.json_tree.request_method} request method.")
+
     async def format_manga_url(
             self, raw_url: Optional[str] = None, url_name: Optional[str] = None, _id: Optional[str] = None,
             *, use_ajax_url: bool = False

src/core/scanlators/lookup_map.json

@@ -1387,7 +1387,7 @@
           },
           "title": "div.tt > a",
           "url": "div.tt > a",
-          "cover": "img.ts-post-image, picture.ts-post-image > img"
+          "cover": "img"
         },
         "unwanted_tags": [],
         "search": {

src/static.py

@@ -201,7 +201,7 @@ class RegExpressions:
     )
 
     comick_url = re.compile(
-        r"(?:https?://)?(?:www\.)?comick\.(?:cc|app)/comic/(?P<url_name>[a-zA-Z0-9-]+)(?:\??/.*)?"
+        r"(?:https?://)?(?:www\.)?comick\.io/comic/(?P<url_name>[a-zA-Z0-9-]+)(?:\??/.*)?"
     )
     zeroscans_url = re.compile(
         r"(?:https?://)?(?:www\.)?zscans\.com/comics/(?P<url_name>\w[\w-]*?)(?:-chapter-[\d.-]+(-\w+)?)?/?(?:/.*)?$"

tests/test.py → tests/__main__.py

@@ -570,7 +570,7 @@ async def test_single_scanlator(scanlator: str):
         asyncio.run(main())
     else:
         # asyncio.run(test_single_method("show_front_page_results", "mangabat"))
-        asyncio.run(test_single_scanlator("epsilonscan"))
+        # asyncio.run(test_single_scanlator("epsilonscan"))
         # asyncio.run(sub_main())
         # asyncio.run(paused_test())
-        # asyncio.run(main())
+        asyncio.run(main())

tests/test_beta.py

@@ -0,0 +1,169 @@
+import asyncio
+import logging
+import os
+import sys
+from typing import Dict, Optional
+
+from src.core import (
+    CachedClientSession,
+    CachedCurlCffiSession,
+    Database,
+)
+from src.core.apis import APIManager, ComickAppAPI, MangaDexAPI
+from src.utils import setup_logging, silence_debug_loggers
+
+root_path = [x for x in sys.path if x.endswith("ManhwaUpdatesBot")][0]
+logger = logging.getLogger()
+
+
+class _ThirdProperties:
+    url = ""
+
+
+class User:
+    @property
+    def display_avatar(self):
+        return _ThirdProperties()
+
+    @property
+    def display_name(self):
+        return "Manhwa Updates"
+
+
+# noinspection PyTypeChecker
+class Bot:
+    config: dict = None
+    proxy_addr: Optional[str] = None
+
+    def __init__(self, proxy_url: Optional[str] = None, scanlaors: dict = None):
+        self.session = CachedClientSession(proxy=proxy_url)
+        self.db = Database(self, database_name=os.path.join(root_path, "tests", "database.db"))
+        self.mangadex_api = MangaDexAPI(self.session)
+        self.comick_api = ComickAppAPI(self.session)
+        self.curl_session = CachedCurlCffiSession(
+            impersonate="chrome101",
+            name="cache.curl_cffi",
+            proxies={"http": proxy_url, "https": proxy_url},
+        )
+        self.logger = logging.getLogger("bot")
+        self.user = User()
+        self.apis: APIManager | None = None
+        if scanlaors:
+            self.load_scanlators(scanlaors)
+
+    async def close(self):
+        await self.curl_session.close()
+        await self.session.close()
+
+    async def __aenter__(self):
+        await self.db.async_init()
+        self.apis = APIManager(self, CachedClientSession(proxy=self.proxy_addr, name="cache.apis", trust_env=True))
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.close()
+        await self.apis.session.close() if self.apis else None
+
+    def load_scanlators(self, scanlators: dict):
+        for scanlator in scanlators.values():
+            scanlator.bot = self
+
+    @staticmethod
+    async def log_to_discord(content, **kwargs):
+        print(content)
+        print(kwargs)
+
+
+class TestCog:
+    def __init__(self, bot):
+        self.bot = bot
+        self.rate_limiter = {}
+
+
+def load_config() -> Dict:
+    import yaml
+    path = os.path.join(root_path, "tests", "config.yml")
+    with open(path, "r") as f:
+        config = yaml.safe_load(f)
+    return config
+
+
+def toggle_logging(name: str = "__main__") -> logging.Logger:
+    logger = logging.getLogger(name)
+    logger.setLevel(logging.DEBUG)
+    logger.addHandler(logging.StreamHandler())
+    return logger
+
+
+def fmt_proxy(username: str, password: str, ip: str, port: str) -> str:
+    # noinspection HttpUrlsUsage
+    return f"http://{username}:{password}@{ip}:{port}"
+
+
+def init_scanlators(bot, scanlators: dict) -> None:
+    for _name in scanlators:
+        scanlators[_name].bot = bot
+
+
+async def main():
+    config = load_config()
+    proxy_url = fmt_proxy(
+        config["proxy"]["username"], config["proxy"]["password"], config["proxy"]["ip"], config["proxy"]["port"]
+    )
+
+    Bot.config = config
+    Bot.proxy_addr = proxy_url
+
+    # noinspection PyProtectedMember
+    from src.core.scanlators import scanlators
+
+    async with Bot(proxy_url=proxy_url) as bot:
+        init_scanlators(bot, scanlators)
+        key = "comick"
+        url = "https://comick.io/comic/the-main-characters-that-only-i-know"
+        query = "he"
+        scanlator = scanlators[key]
+        title = await scanlator.get_title(url)
+
+        _id = await scanlator.get_id(url)
+        all_chapters = await scanlator.get_all_chapters(url)
+        status = await scanlator.get_status(url)
+        synopsis = await scanlator.get_synopsis(url)
+        cover = await scanlator.get_cover(url)
+        fp_manga = await scanlator.get_fp_partial_manga()
+        search_result = await scanlator.search(query)
+        manga = await scanlator.make_manga_object(url)
+        updates_result = await scanlator.check_updates(manga)
+        #
+        results = (
+            f"Title: {title}", f"ID: {_id}", f"All Chapters: {all_chapters}", f"Status: {status}",
+            f"Synopsis: {synopsis}", f"Cover: {cover}", f"FP Manhwa: {fp_manga}", f"Search: {search_result}",
+            f"Manga obj: {manga}", f"Updates result: {updates_result}"
+        )
+        for result in results:
+            print(result, "\n")
+
+        res, = await scanlator.unload_manga([manga])
+        print("Unloaded manhwa:", res)
+        res2, = await scanlator.load_manga([res])
+        print("Loaded manhwa:", res2)
+        print(res2.last_chapter)
+
+
+if __name__ == "__main__":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+    setup_logging(level=logging.DEBUG)
+    silence_debug_loggers(
+        logger,
+        [
+            "websockets.client",
+            "aiosqlite",
+            "discord.gateway",
+            "discord.client",
+            "discord.http",
+            "discord.webhook.async_",
+            "asyncio",
+            "filelock"
+        ]
+    )
+    asyncio.run(main())