v1.8.3 - Concurrency improvements again

Dockerfile

@@ -32,7 +32,7 @@ ENV DEBIAN_FRONTEND=noninteractive \
 
 LABEL maintainer="Parker Wahle <regulad@regulad.xyz>" \
       name="freebooter" \
-      version="1.8.2"
+      version="1.8.3"
 
 # Add curl for MariaDB script
 RUN apt update && apt upgrade -y && apt install -y curl

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "freebooter"
-version = "1.8.2"
+version = "1.8.3"
 description = "freebooter downloads photos & videos from the internet and uploads it onto your social media accounts."
 authors = ["Parker Wahle <regulad@regulad.xyz>"]
 license = "GPLv3"

src/freebooter/__init__.py

@@ -24,4 +24,4 @@
 from .uploaders import *
 from .watchers import *
 
-version = "1.8.2"
+version = "1.8.3"

src/freebooter/metadata.py

@@ -138,7 +138,7 @@ class MediaMetadata:
     def __init__(
         self,
         *,
-        media_id: str,
+        media_id: Any,
         platform: Platform = Platform.UNKNOWN,
         title: str | None = None,
         description: str | None = None,
@@ -206,7 +206,7 @@ def data(self) -> dict[str, Any]:
         return self._data
 
     @property
-    def id(self) -> str:
+    def id(self) -> Any:
         return self._id
 
     @property

src/freebooter/watchers/common.py

@@ -26,7 +26,7 @@
 from logging import getLogger
 from os import sep
 from pathlib import Path
-from threading import Event
+from threading import Event, Lock
 from threading import Thread
 from typing import Any, cast, ClassVar
 
@@ -212,6 +212,9 @@ def _affirm_result(done_future: Future[list[tuple[ScratchFile, MediaMetadata | N
             except TimeoutError:
                 result = None
 
+            for _, metadata in downloaded:
+                self.mark_handled(metadata.id)
+
             if result is None:
                 self.logger.error(f"{self.name} upload callback failed!")
             else:
@@ -241,21 +244,24 @@ def __init__(
         self._prepare_task: Task | None = None
         self._closing_task: Task | None = None
 
+        self._process_lock = asyncio.Lock()
+
     async def aprocess(self, medias: list[tuple[ScratchFile, MediaMetadata]]) -> list[MediaMetadata]:
         """
         A coroutine that preprocesses and executes the given medias.
         :param medias: The medias to preprocess and execute
         :return: A future that resolves to the processed medias
         """
-        assert self._loop is not None, "No event loop set!"
+        async with self._process_lock:
+            assert self._loop is not None, "No event loop set!"
 
-        asyncio_future_concurrent_future = self._loop.run_in_executor(None, self._preprocess_and_execute, medias)
-        concurrent_future = await asyncio_future_concurrent_future
-        asyncio_future = asyncio.wrap_future(concurrent_future, loop=self._loop)
+            asyncio_future_concurrent_future = self._loop.run_in_executor(None, self._preprocess_and_execute, medias)
+            concurrent_future = await asyncio_future_concurrent_future
+            asyncio_future = asyncio.wrap_future(concurrent_future, loop=self._loop)
 
-        list_of_medias = await asyncio_future
+            list_of_medias = await asyncio_future
 
-        return [metadata for _, metadata in list_of_medias if metadata is not None]
+            return [metadata for _, metadata in list_of_medias if metadata is not None]
 
     async def async_prepare(self) -> None:
         """
@@ -317,6 +323,8 @@ def __init__(
         Thread.__init__(self, name=self._name)
         Watcher.__init__(self, self.name, preprocessors, **config)
 
+        self._process_lock = Lock()
+
     def check_for_uploads(self) -> list[tuple[ScratchFile, MediaMetadata]]:
         """
         Checks for new uploads and downloads them.
@@ -331,8 +339,9 @@ def close(self) -> None:
             self.join()  # just wait for it to spin down
 
     def process(self, medias: list[tuple[ScratchFile, MediaMetadata]]) -> list[MediaMetadata]:
-        fut = self._preprocess_and_execute(medias)
-        return [metadata for _, metadata in fut.result() if metadata is not None]
+        with self._process_lock:
+            fut = self._preprocess_and_execute(medias)
+            return [metadata for _, metadata in fut.result() if metadata is not None]
 
     def run(self) -> None:
         assert self.ready, "Watcher is not ready, cannot run thread!"

src/freebooter/watchers/discord_dpy.py

@@ -98,7 +98,7 @@ async def medias_in_message(
             await attachment.save(scratch_file.path)  # discord.py messed up the typing on this
 
             media_metadata = MediaMetadata(
-                media_id=str(attachment.id),
+                media_id=attachment.id,
                 platform=Platform.DISCORD,
                 title=attachment.filename,
                 description=message.content,
@@ -108,8 +108,6 @@ async def medias_in_message(
 
             yield scratch_file, media_metadata
 
-            self.mark_handled(attachment.id)
-
     async def process_message(
         self, message: Message, *, handle_if_already_handled: bool = False
     ) -> list[MediaMetadata]:

src/freebooter/watchers/discord_selfcord.py

@@ -89,7 +89,7 @@ async def medias_in_message(
             await attachment.save(scratch_file.path)  # discord.py messed up the typing on this
 
             media_metadata = MediaMetadata(
-                media_id=str(attachment.id),
+                media_id=attachment.id,
                 platform=Platform.DISCORD,
                 title=attachment.filename,
                 description=message.content,
@@ -99,8 +99,6 @@ async def medias_in_message(
 
             yield scratch_file, media_metadata
 
-            self.mark_handled(attachment.id)
-
     async def process_message(
         self, message: Message, *, handle_if_already_handled: bool = False
     ) -> list[MediaMetadata]:

src/freebooter/watchers/instagram_instaloader.py

@@ -257,8 +257,6 @@ def _check_for_uploads_generator(
 
             yield file, metadata
 
-            self.mark_handled(post.shortcode)
-
     def check_for_uploads(self) -> list[tuple[ScratchFile, MediaMetadata]]:
         return list(self._check_for_uploads_generator())
 

src/freebooter/watchers/local.py

@@ -85,7 +85,7 @@ def _check_in_folder(
                     media_type = MediaType.from_file_path(file)
 
                 metadata = MediaMetadata(
-                    media_id=file.stem,
+                    media_id=file.name,
                     platform=Platform.OTHER,
                     title=file.stem,
                     description="",
@@ -95,8 +95,6 @@ def _check_in_folder(
                     data=data,
                 )
 
-                self.mark_handled(file.name)
-
                 yield scratch_file, metadata
 
     def check_for_uploads(self) -> list[tuple[ScratchFile, MediaMetadata]]:

src/freebooter/watchers/rss.py

@@ -141,7 +141,7 @@ def _parse_entry(
                             file.write(chunk)
 
                     metadata = MediaMetadata(
-                        media_id=entry_id,
+                        media_id=entry.id,
                         platform=Platform.from_url(entry["link"]),
                         title=entry["title"],
                         description=entry["summary"],
@@ -151,7 +151,6 @@ def _parse_entry(
                         data=dict(entry),
                     )
 
-            self.mark_handled(entry.id)
             return media, metadata
         except Exception as e:
             self.logger.error(f"Error while parsing entry {entry_id}: {e}")

src/freebooter/watchers/youtube_ytdl.py

@@ -92,7 +92,6 @@ def _prepare_video(
         else:
             try:
                 scratch_file, metadata = self._download(video_id)
-                self.mark_handled(video_id, True)
                 return scratch_file, metadata
             except Exception as e:
                 self.logger.exception(f"Error downloading video {video_id}: {e}")