Merge pull request #50 from mensch272/wb@mensch

Fixed discord thread access errors

Author: m-haisham

CHANGELOG.md

@@ -6,13 +6,19 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased] - yyyy-mm-dd
 
+## [0.8.1] - 2021-12-13
+
 ## Changed
 
 - Removed url lazy loading from db in favour of sql statement
+- Changed discord session `call` to `get` and it now returns the method
+  instead of calling it
 
 ## Fixed
 
 - Fixed where provided webnovel urls not added to db #44
+- Fixed discord dm message
+- Fixed discord thread access errors
 
 ## [0.8.0] - 2021-11-25
 

README.md

@@ -42,7 +42,7 @@ The default environmental variables are shown below. Modify them to your liking
 ```shell
 DISCORD_TOKEN=  # Required: discord bot token
 DISCORD_SESSION_TIMEOUT=10 # Minutes
-DISCORD_DOWNLOAD_THREADS=4
+DISCORD_SESSION_THREADS=5
 DISCORD_SEARCH_LIMIT=20  # Maximum results to show
 DISCORD_SEARCH_DISABLED=no  # Disable search functionality
 ```

novelsave/client/bots/discord/config.py

@@ -5,21 +5,21 @@
 
 import dotenv
 from loguru import logger
+import copy
 
 from novelsave.settings import config, console_formatter
 
 
-@functools.lru_cache()
 def app() -> dict:
     """Initialize and return the configuration used by the base application"""
-    return config.copy()
+    return copy.deepcopy(config)
 
 
 def logger_config() -> dict:
     return {
         "handlers": [
             {
-                "sink": sys.stderr,
+                "sink": sys.stdout,
                 "level": "TRACE",
                 "format": console_formatter,
                 "backtrace": True,
@@ -58,9 +58,7 @@ def discord() -> dict:
         "key": discord_token,
         "session": {
             "retain": timedelta(minutes=intenv("DISCORD_SESSION_TIMEOUT", 10)),
-        },
-        "download": {
-            "threads": intenv("DISCORD_DOWNLOAD_THREADS", 4),
+            "threads": intenv("DISCORD_SESSION_THREADS", 5),
         },
         "search": {
             "limit": intenv("DISCORD_SEARCH_LIMIT", 20),

novelsave/client/bots/discord/decorators.py

@@ -5,28 +5,32 @@
 from .session import Session
 
 
-def ensure_close(func):
-    """Ensures that when this method ends the session will be closed"""
+def session_task(close_on_exit=True):
+    def inner(func):
+        """Ensures that when this method ends the session will be closed"""
 
-    @functools.wraps(func)
-    def wrapped(*args, **kwargs):
-        session: Session = args[0].session
+        @functools.wraps(func)
+        def wrapped(*args, **kwargs):
+            session: Session = args[0].session
 
-        result = None
-        try:
-            result = func(*args, **kwargs)
-        except Exception as e:
-            if not session.is_closed:
-                session.send_sync(f"`❗ {str(e).strip()}`")
+            result = None
+            try:
+                result = func(*args, **kwargs)
+            except Exception as e:
+                if not session.is_closed:
+                    session.send_sync(f"`❗ {str(e).strip()}`")
 
-            logger.exception(e)
+                logger.exception(e)
 
-        if not session.is_closed:
-            session.sync(session.close_and_inform)
+            session.close_session()
+            if close_on_exit and not session.is_closed:
+                session.sync(session.close_and_inform)
 
-        return result
+            return result
 
-    return wrapped
+        return wrapped
+
+    return inner
 
 
 def log_error(func):

novelsave/client/bots/discord/endpoints/commands.py

@@ -13,7 +13,7 @@ async def dm(ctx: commands.Context):
     """Send a direct message to you"""
     await ctx.author.send(
         f"Hello, {ctx.author.name}.\n"
-        f" Send `{ctx.clean_prefix}help` to get usage instructions."
+        f"Send `{ctx.clean_prefix}help` to get usage instructions."
     )
 
 

novelsave/client/bots/discord/endpoints/download.py

@@ -16,13 +16,11 @@
 from novelsave.exceptions import SourceNotFoundException
 from novelsave.utils.helpers import url_helper, string_helper
 from .. import checks, mfmt
-from ..decorators import ensure_close
+from ..decorators import session_task
 from ..session import SessionFragment, SessionHandler
 
 
 class DownloadHandler(SessionFragment):
-    download_threads: int = Provide["discord_config.download.threads"]
-
     def __init__(self, *args, **kwargs):
         super(DownloadHandler, self).__init__(*args, **kwargs)
 
@@ -46,12 +44,12 @@ async def download_state(self, ctx: commands.Context):
     async def packaging_state(self, ctx: commands.Context):
         await ctx.send("I'm currently packaging the novel.")
 
-    @ensure_close
+    @session_task()
     def download(self, url: str, targets: List[str]):
         self.session.state = self.info_state
 
         try:
-            source_gateway = self.session.source_service.source_from_url(url)
+            source_gateway = self.session.source_service().source_from_url(url)
         except SourceNotFoundException:
             self.session.send_sync(mfmt.error("This website is not yet supported."))
             self.session.send_sync(
@@ -61,7 +59,7 @@ def download(self, url: str, targets: List[str]):
             return
 
         try:
-            packagers = self.session.packager_provider.filter_packagers(targets)
+            packagers = self.session.packager_provider().filter_packagers(targets)
         except ValueError as e:
             self.session.send_sync(mfmt.error(str(e)))
             return
@@ -77,9 +75,11 @@ def download(self, url: str, targets: List[str]):
             f"volumes of {chapter_count} chapters."
         )
 
-        novel = self.session.novel_service.insert_novel(novel_dto)
-        self.session.novel_service.insert_chapters(novel, novel_dto.volumes)
-        self.session.novel_service.insert_metadata(novel, novel_dto.metadata)
+        novel_service = self.session.novel_service()
+
+        novel = novel_service.insert_novel(novel_dto)
+        novel_service.insert_chapters(novel, novel_dto.volumes)
+        novel_service.insert_metadata(novel, novel_dto.metadata)
 
         self.session.state = self.download_state
         self.download_thumbnail(novel)
@@ -114,57 +114,61 @@ def download_thumbnail(self, novel: Novel):
             )
             return
 
-        thumbnail_path = self.session.path_service.thumbnail_path(novel)
-        self.session.novel_service.set_thumbnail_asset(
-            novel, self.session.path_service.relative_to_data_dir(thumbnail_path)
+        path_service = self.session.path_service()
+        novel_service = self.session.novel_service()
+
+        thumbnail_path = path_service.thumbnail_path(novel)
+        novel_service.set_thumbnail_asset(
+            novel, path_service.relative_to_data_dir(thumbnail_path)
         )
 
         thumbnail_path.parent.mkdir(parents=True, exist_ok=True)
-        self.session.file_service.write_bytes(thumbnail_path, response.content)
+        self.session.file_service().write_bytes(thumbnail_path, response.content)
 
         size = string_helper.format_bytes(len(response.content))
         self.session.send_sync(f"Downloaded and saved thumbnail image ({size}).")
 
     def download_chapters(self, novel: Novel, source_gateway: BaseSourceGateway):
-        chapters = self.session.novel_service.get_pending_chapters(novel, -1)
+        novel_service = self.session.novel_service()
+
+        chapters = novel_service.get_pending_chapters(novel, -1)
         if not chapters:
             logger.info("Skipped chapter download as none are pending.")
             return
 
         self.session.send_sync(
-            f"Downloading {len(chapters)} chapters using {self.download_threads} threads…"
+            f"Downloading {len(chapters)} chapters using {self.session.thread_count - 1} threads…"
         )
+
         self.total = len(chapters)
         self.value = 1
 
-        with futures.ThreadPoolExecutor(
-            max_workers=self.download_threads
-        ) as download_executor:
-            download_futures = [
-                download_executor.submit(
-                    source_gateway.update_chapter_content,
-                    self.session.dto_adapter.chapter_to_dto(c),
-                )
-                for c in chapters
-            ]
-
-            for chapter in futures.as_completed(download_futures):
-                try:
-                    chapter_dto = chapter.result()
-                except Exception as e:
-                    logger.exception(e)
-                    continue
-
-                chapter_dto.content = self.session.asset_service.collect_assets(
-                    novel, chapter_dto
-                )
-                self.session.novel_service.update_content(chapter_dto)
+        dto_adapter = self.session.dto_adapter()
+        asset_service = self.session.asset_service()
 
-                logger.debug(
-                    f"Chapter content downloaded: '{chapter_dto.title}' ({chapter_dto.index})"
-                )
+        download_futures = [
+            self.session.executor.submit(
+                source_gateway.update_chapter_content,
+                dto_adapter.chapter_to_dto(c),
+            )
+            for c in chapters
+        ]
+
+        for chapter in futures.as_completed(download_futures):
+            try:
+                chapter_dto = chapter.result()
+            except Exception as e:
+                logger.exception(e)
+                continue
+
+            chapter_dto.content = asset_service.collect_assets(novel, chapter_dto)
+            novel_service.update_content(chapter_dto)
+
+            logger.debug(
+                f"Chapter content downloaded: '{chapter_dto.title}' ({chapter_dto.index})"
+            )
 
-                self.value += 1
+            self.value += 1
 
     def package(self, novel: Novel, packagers: Iterable[BasePackager]):
         formats = ", ".join(p.keywords()[0] for p in packagers)
@@ -179,6 +183,7 @@ def package(self, novel: Novel, packagers: Iterable[BasePackager]):
             else:
                 self.session.send_sync(f"Uploading {output.name}…")
 
+            # TODO: ability upload larger than 8 Mb
             if output.stat().st_size > 7.99 * 1024 * 1024:
                 self.session.send_sync(
                     mfmt.error(

novelsave/client/bots/discord/endpoints/search.py

@@ -8,7 +8,7 @@
 from novelsave.core.dtos import NovelDTO
 from novelsave.core.services.source import BaseSourceService
 from .. import checks, mfmt
-from ..decorators import log_error
+from ..decorators import log_error, session_task
 from ..session import SessionHandler, SessionFragment, Session
 
 
@@ -49,6 +49,7 @@ async def _state_source_select(self, ctx: commands.Context):
         await ctx.send(self._source_list())
 
     @log_error
+    @session_task(False)
     def search(self, word: str):
         self.session.state = self._state_searching
         search_capable = [
@@ -178,13 +179,13 @@ async def select(self, ctx: commands.Context, num: int):
             await ctx.send(self.unsupported)
             return
 
-        if not await session.call(SearchHandler.is_select):
+        if not session.get(SearchHandler.is_select)():
             await ctx.send("Session does not require selection.")
             return
 
-        if await session.call(SearchHandler.is_novel_select):
+        if session.get(SearchHandler.is_novel_select)():
             await session.run(ctx, SearchHandler.select_novel, num - 1)
         else:
-            url = await session.call(SearchHandler.select_source, num - 1)
+            url = session.get(SearchHandler.select_source)(num - 1)
             await ctx.send(f"{url} selected.")
             await ctx.invoke(session.bot.get_command("download"), url)

novelsave/client/bots/discord/main.py

@@ -24,9 +24,9 @@ def wire(packages):
 def main():
     """Start the discord bot"""
     from .bot import bot
-    from . import endpoints
+    from . import endpoints, session
 
-    application, discord_application = wire([endpoints])
+    application, discord_application = wire([endpoints, session])
 
     # cogs
     bot.add_cog(endpoints.SessionCog())

novelsave/client/bots/discord/mixins.py

@@ -1,4 +1,7 @@
 import shutil
+import threading
+
+from loguru import logger
 
 from novelsave import migrations
 from novelsave.containers import Application
@@ -62,19 +65,33 @@ def setup_container(self, id_: str):
         self.application = Application()
         self.application.config.from_dict(self._make_unique_config(id_))
 
-        # acquire services
-        self.source_service = self.application.services.source_service()
-        self.novel_service = self.application.services.novel_service()
-        self.path_service = self.application.services.path_service()
-        self.dto_adapter = self.application.adapters.dto_adapter()
-        self.asset_service = self.application.services.asset_service()
-        self.file_service = self.application.services.file_service()
-        self.packager_provider = self.application.packagers.packager_provider()
-
         # migrate database to latest schema
         migrations.migrate(self.application.config.get("infrastructure.database.url"))
 
+    def source_service(self):
+        return self.application.services.source_service()
+
+    def novel_service(self):
+        return self.application.services.novel_service()
+
+    def path_service(self):
+        return self.application.services.path_service()
+
+    def dto_adapter(self):
+        return self.application.adapters.dto_adapter()
+
+    def asset_service(self):
+        return self.application.services.asset_service()
+
+    def file_service(self):
+        return self.application.services.file_service()
+
+    def packager_provider(self):
+        return self.application.packagers.packager_provider()
+
     def close_session(self):
+        logger.debug(f"Session closed; thread id: {threading.current_thread().ident}")
         self.application.infrastructure.session().close()
-        self.application.infrastructure.session_factory().close_all()
+
+    def close_engine(self):
         self.application.infrastructure.engine().dispose()

novelsave/client/bots/discord/session/handler.py

@@ -28,7 +28,7 @@ def get(self, ctx: commands.Context) -> Session:
     def get_or_create(self, ctx: commands.Context):
         """Create or return already existing session"""
         try:
-            return self.get(ctx).renew_context(ctx)
+            return self.get(ctx)
         except KeyError:
             session = self.session_factory(bot, ctx)
             self.sessions[session_key(ctx)] = session