Convert WebVTT parser to a dynamically registered MIME type handler

src/ctp_slack_bot/containers.py

@@ -1,11 +1,14 @@
 from dependency_injector.containers import DeclarativeContainer
-from dependency_injector.providers import Resource, Singleton
+from dependency_injector.providers import Callable, Resource, Singleton
+from importlib import import_module
+from pathlib import Path
 from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
 from slack_bolt.async_app import AsyncApp
 
 from ctp_slack_bot.core.config import Settings
 from ctp_slack_bot.db.mongo_db import MongoDBResource
-from ctp_slack_bot.db.repositories import MongoVectorizedChunkRepository
+from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepository
+from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandlerMeta
 from ctp_slack_bot.services.answer_retrieval_service import AnswerRetrievalService
 from ctp_slack_bot.services.content_ingestion_service import ContentIngestionService
 from ctp_slack_bot.services.context_retrieval_service import ContextRetrievalService
@@ -20,6 +23,17 @@ from ctp_slack_bot.services.vector_database_service import VectorDatabaseService
 from ctp_slack_bot.services.vectorization_service import VectorizationService
 
 
+def __load_plugins(plugin_dir) -> None:
+    for path in Path(plugin_dir).glob("*.py"):
+        if path.stem == "__init__":
+            continue  # Skip __init__.py files
+        module_name = f"{plugin_dir.replace('/', '.')}.{path.stem}"
+        import_module(module_name)
+
+
+__load_plugins("ctp_slack_bot/mime_type_handlers")
+
+
 class Container(DeclarativeContainer):
     settings = Singleton(Settings)
     event_brokerage_service = Singleton(EventBrokerageService)
@@ -37,4 +51,6 @@ class Container(DeclarativeContainer):
     slack_bolt_app = Singleton(AsyncApp, token=settings.provided.SLACK_BOT_TOKEN().get_secret_value())
     slack_service = Resource(SlackServiceResource, event_brokerage_service=event_brokerage_service, slack_bolt_app=slack_bolt_app)
     socket_mode_handler = Singleton(lambda _, app, app_token: AsyncSocketModeHandler(app, app_token), slack_service, slack_bolt_app, settings.provided.SLACK_APP_TOKEN().get_secret_value())
+    mime_type_handler_factory = Callable(lambda mime_type: MimeTypeHandlerMeta._registry[mime_type]())
     google_drive_service = Singleton(GoogleDriveService, settings=settings)
+    # file_monitor_service = Singleton(FileMonitorService, settings=settings, google_drive_service=google_drive_service, mime_type_handler_factory=mime_type_handler_factory)

src/ctp_slack_bot/mime_type_handlers/__init__.py

@@ -0,0 +1,2 @@
+from ctp_slack_bot.mime_type_handlers.base import BaseMimeTypeHandler, MimeTypeHandlerMeta
+from ctp_slack_bot.mime_type_handlers.text.vtt import WebVTTMimeTypeHandler

src/ctp_slack_bot/mime_type_handlers/base.py

@@ -0,0 +1,29 @@
+from abc import ABCMeta, abstractmethod
+from typing import Any, ClassVar, Dict, Mapping, Optional
+
+from ctp_slack_bot.models import Content
+
+
+class MimeTypeHandlerMeta(type):
+
+    _registry: ClassVar[dict[str, type["BaseMimeTypeHandler"]]] = {}
+
+    def __init__(cls, name: str, bases: tuple[type, ...], dict: Dict[str, Any]) -> None:
+        super().__init__(name, bases, dict)
+        if hasattr(cls, "MIME_TYPE"):
+            MimeTypeHandlerMeta._registry[cls.MIME_TYPE] = cls
+
+
+class MimeTypeHandlerABCMeta(MimeTypeHandlerMeta, ABCMeta):
+    pass
+
+
+class BaseMimeTypeHandler(metaclass=MimeTypeHandlerABCMeta):
+
+    @classmethod
+    def for_mime_type(cls, mime_type: str) -> Optional[type["BaseMimeHandler"]]:
+        return cls._registry.get(mime_type)
+
+    @abstractmethod
+    def from_bytes(cls, id: str, metadata: Mapping[str, Any], buffer: bytes) -> Content:
+        pass

src/ctp_slack_bot/mime_type_handlers/text/__init__.py

@@ -0,0 +1 @@
+from ctp_slack_bot.mime_type_handlers.text.vtt import WebVTTMimeTypeHandler 

src/ctp_slack_bot/mime_type_handlers/text/vtt.py

@@ -0,0 +1,39 @@
+from datetime import datetime
+from io import BytesIO
+from re import compile as compile_re
+from typing import Any, Mapping, Optional, Self
+from webvtt import WebVTT
+
+from ctp_slack_bot.mime_type_handlers import BaseMimeTypeHandler
+from ctp_slack_bot.models import Content, WebVTTContent, WebVTTFrame
+
+
+ISO_DATE_TIME_PATTERN = compile_re(r"Start time: (\d{4}-\d{2}-\d{2}(?: \d{2}:\d{2}:\d{2}(?:Z|[+-]\d{2}:\d{2})?)?)")
+
+class WebVTTMimeTypeHandler(BaseMimeTypeHandler):
+
+    MIME_TYPE = "text/vtt"
+
+    def from_buffer(self: Self, id: str, metadata: Mapping[str, Any], buffer: bytes) -> WebVTTContent:
+        web_vtt = WebVTT.from_buffer(BytesIO(buffer))
+        frames = tuple(WebVTTFrame.from_webvtt_caption(caption, index)
+                       for index, caption
+                       in enumerate(web_vtt.captions, 1))
+        return WebVTTContent(id=id, metadata=MappingProxyType(metadata), start_time=cls.__get_start_time(web_vtt), frames=frames)
+
+    @classmethod
+    def __get_start_time(cls, web_vtt: WebVTT) -> Optional[datetime]:
+        try:
+            return next(datetime.fromisoformat(result[0])
+                        for result
+                        in map(ISO_DATE_TIME_PATTERN.findall, web_vtt.header_comments)
+                        if result)
+        except ValueError:
+            return None
+
+    def from_bytes(self: Self, id: str, metadata: Mapping[str, Any], buffer: bytes) -> WebVTTContent:
+        web_vtt = WebVTT.from_buffer(BytesIO(buffer))
+        frames = tuple(WebVTTFrame.from_webvtt_caption(caption, index)
+                       for index, caption
+                       in enumerate(web_vtt.captions, 1))
+        return WebVTTContent(id=id, metadata=metadata, start_time=WebVTTMimeTypeHandler.__get_start_time(web_vtt), frames=frames)

src/ctp_slack_bot/models/webvtt.py

@@ -4,7 +4,6 @@ from itertools import starmap
 from json import dumps
 from more_itertools import windowed
 from pydantic import BaseModel, ConfigDict, Field, PositiveInt, PrivateAttr
-from re import compile as compile_re
 from types import MappingProxyType
 from typing import Any, Dict, Literal, Mapping, Optional, Self, Sequence
 from webvtt import Caption, WebVTT
@@ -15,7 +14,6 @@ from ctp_slack_bot.models.base import Chunk, Content
 CHUNK_FRAMES_OVERLAP = 1
 CHUNK_FRAMES_WINDOW = 5
 SPEAKER_SPEECH_TEXT_SEPARATOR = ": "
-ISO_DATE_TIME_PATTERN = compile_re(r"Start time: (\d{4}-\d{2}-\d{2}(?: \d{2}:\d{2}:\d{2}(?:Z|[+-]\d{2}:\d{2})?)?)")
 
 
 class WebVTTFrame(BaseModel):
@@ -71,21 +69,3 @@ class WebVTTContent(Content):
 
     def get_metadata(self: Self) -> Mapping[str, Any]:
         return MappingProxyType(self.metadata)
-
-    @classmethod
-    def __get_start_time(cls, web_vtt: WebVTT) -> Optional[datetime]:
-        try:
-            return next(datetime.fromisoformat(result[0])
-                        for result
-                        in map(ISO_DATE_TIME_PATTERN.findall, web_vtt.header_comments)
-                        if result)
-        except ValueError:
-            return None
-
-    @classmethod
-    def from_bytes(cls, id: str, metadata: Mapping[str, Any], buffer: bytes) -> Self:
-        web_vtt = WebVTT.from_buffer(BytesIO(buffer))
-        frames = tuple(WebVTTFrame.from_webvtt_caption(caption, index)
-                       for index, caption
-                       in enumerate(web_vtt.captions, 1))
-        return WebVTTContent(id=id, metadata=MappingProxyType(metadata), start_time=cls.__get_start_time(web_vtt), frames=frames)