WebVTT models and changes to support it

src/ctp_slack_bot/models/base.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 from pydantic import BaseModel, ConfigDict, Field
-from typing import Any, Dict, final, Self, Sequence, Optional
+from types import MappingProxyType
+from typing import Any, Dict, final, Mapping, Self, Sequence, Optional
 
 
 class Chunk(BaseModel):
@@ -9,7 +10,7 @@ class Chunk(BaseModel):
     text: str                   # The text representation
     parent_id: str              # The source content’s identity
     chunk_id: str               # This chunk’s identity—unique within the source content
-    metadata: Dict[str, Any]
+    metadata: Mapping[str, Any]
 
     model_config = ConfigDict(frozen=True)
 
@@ -27,7 +28,7 @@ class VectorQuery(BaseModel):
     query_embeddings: Sequence[float]
     k: int
     score_threshold: float = Field(default=0.7)
-    filter_metadata: Optional[Dict[str, Any]] = None
+    filter_metadata: Optional[Mapping[str, Any]] = None
 
     
 @final
@@ -47,17 +48,15 @@ class Content(ABC, BaseModel):
         pass
 
     @abstractmethod
-    def get_metadata(self: Self) -> Dict[str, Any]:
+    def get_metadata(self: Self) -> Mapping[str, Any]:
         pass
 
-    @property
     @abstractmethod
     def get_text(self: Self) -> str:
         pass
 
-    @property
     @abstractmethod
-    def bytes(self: Self) -> bytes:
+    def get_bytes(self: Self) -> bytes:
         pass
 
     @property

src/ctp_slack_bot/models/slack.py

@@ -1,7 +1,8 @@
 from datetime import datetime
 from json import dumps
 from pydantic import BaseModel, ConfigDict, PositiveInt, PrivateAttr
-from typing import Any, Dict, Literal, Optional, Self, Sequence
+from types import MappingProxyType
+from typing import Any, Dict, Literal, Mapping, Optional, Self, Sequence
 
 from ctp_slack_bot.models.base import Chunk, Content
 
@@ -71,16 +72,15 @@ class SlackMessage(Content):
     def get_chunks(self: Self) -> Sequence[Chunk]:
         return (Chunk(text=self.text, parent_id=self.id, chunk_id="", metadata=self.get_metadata()), )
 
-    def get_metadata(self: Self) -> Dict[str, Any]:
-        return {
+    def get_metadata(self: Self) -> Mapping[str, Any]:
+        return MappingProxyType({
             "modificationTime": datetime.fromtimestamp(float(self.ts))
-        }
+        })
 
     def get_text(self: Self) -> str:
         return self.text
 
-    @property
-    def bytes(self: Self) -> bytes:
+    def get_bytes(self: Self) -> bytes:
         return self._canonical_json
 
     @property

src/ctp_slack_bot/models/webvtt.py

@@ -0,0 +1,76 @@
+from datetime import datetime, timedelta
+from io import BytesIO
+from json import dumps
+from pydantic import BaseModel, ConfigDict, PositiveInt, PrivateAttr
+import re
+from types import MappingProxyType
+from typing import Any, Dict, Literal, Mapping, Optional, Self, Sequence
+from webvtt import Caption, WebVTT
+
+from ctp_slack_bot.models.base import Chunk, Content
+
+SPEAKER_SPEECH_CAPTION_TEXT_PATTERN = re.compile('(?:([^:]+): )?(.*)')
+
+class WebVTTFrame(BaseModel):
+    """Represents a WebVTT frame"""
+
+    identifier: str
+    start: timedelta
+    end: timedelta
+    speaker: Optional[str] = None
+    speech: str
+
+    model_config = ConfigDict(frozen=True)
+
+    @classmethod
+    def from_webvtt_caption(cls: type["WebVTTFrame"], caption: Caption) -> Self:
+        identifier = caption.identifier
+        start = timedelta(**caption.start_time.__dict__)
+        end = timedelta(**caption.end_time.__dict__)
+        speech = caption.text
+        match SPEAKER_SPEECH_CAPTION_TEXT_PATTERN.search(speech).groups():
+            case (speaker, speech):
+                return cls(identifier=identifier, start=start, end=end, speaker=speaker, speech=speech)
+            case _:
+                return cls(identifier=identifier, start=start, end=end, speech=speech)
+
+
+class WebVTTFile(Content): # TODO: insert a FileContent class in the object inheritance hierarchy.
+    """Represents a message from Slack after adaptation."""
+
+    filename: str
+    modification_time: datetime
+    bytes: bytes
+
+
+    def get_chunks(self: Self) -> Sequence[Chunk]:
+        return tuple(Chunk(text=frame.speech,
+                           parent_id=self.id,
+                           chunk_id=frame.identifier,
+                           metadata={
+                               "filename": self.filename,
+                               "start": self.modification_time + frame.start,
+                               "end": self.modification_time + frame.end,
+                               "user": frame.speaker
+                           })
+                     for frame
+                     in self.get_frames())
+
+    def get_metadata(self: Self) -> Mapping[str, Any]:
+        return MappingProxyType({
+            "filename": self.filename,
+            "modificationTime": self.modification_time
+        })
+
+    def get_text(self: Self) -> str: # TODO
+        raise NotImplemented()
+
+    def get_bytes(self: Self) -> bytes:
+        return self.bytes
+
+    def get_frames(self: Self) -> Sequence[WebVTTFrame]:
+        return tuple(map(WebVTTFrame.from_webvtt_caption, WebVTT.from_buffer(BytesIO(buffer)).captions))
+
+    @property
+    def id(self: Self) -> str:
+        return f"file:{self.filename}"

src/ctp_slack_bot/services/context_retrieval_service.py

@@ -53,8 +53,8 @@ class ContextRetrievalService(BaseModel):
         
         # Perform similarity search
         try:
-            results = await self.vector_db_service.search_by_similarity(query)
-            #logger.info(f"Retrieved {len(results)} context chunks for query")
+            results = await self.vector_database_service.search_by_similarity(query)
+            logger.info(f"Retrieved {len(results)} context chunks for query")
             return results
         except Exception as e:
             logger.error(f"Error retrieving context: {str(e)}")