Simplify models

src/ctp_slack_bot/models/__init__.py

@@ -1,2 +1,3 @@
 from ctp_slack_bot.models.base import Chunk, Content, VectorizedChunk, VectorQuery
 from ctp_slack_bot.models.slack import SlackEventPayload, SlackMessage, SlackReaction, SlackResponse, SlackUserTimestampPair
+from ctp_slack_bot.models.webvtt import WebVTTContent, WebVTTFrame

src/ctp_slack_bot/models/base.py

@@ -1,7 +1,6 @@
 from abc import ABC, abstractmethod
 from pydantic import BaseModel, ConfigDict, Field
-from types import MappingProxyType
-from typing import Any, Dict, final, Mapping, Self, Sequence, Optional
+from typing import Any, final, Mapping, Self, Sequence, Optional
 
 
 class Chunk(BaseModel):
@@ -14,6 +13,7 @@ class Chunk(BaseModel):
 
     model_config = ConfigDict(frozen=True)
 
+
 @final
 class VectorQuery(BaseModel):
     """Model for vector database similarity search queries.
@@ -30,12 +30,14 @@ class VectorQuery(BaseModel):
     score_threshold: float = Field(default=0.7)
     filter_metadata: Optional[Mapping[str, Any]] = None
 
+    model_config = ConfigDict(frozen=True)
+
     
 @final
 class VectorizedChunk(Chunk):
     """A class representing a vectorized chunk of content."""
 
-    embedding: Sequence[float]  # The vector representation
+    embedding: Sequence[float] # The vector representation
 
 
 class Content(ABC, BaseModel):
@@ -44,22 +46,13 @@ class Content(ABC, BaseModel):
     model_config = ConfigDict(frozen=True)
 
     @abstractmethod
-    def get_chunks(self: Self) -> Sequence[Chunk]:
+    def get_id(self: Self) -> str:
         pass
 
     @abstractmethod
-    def get_metadata(self: Self) -> Mapping[str, Any]:
-        pass
-
-    @abstractmethod
-    def get_text(self: Self) -> str:
-        pass
-
-    @abstractmethod
-    def get_bytes(self: Self) -> bytes:
+    def get_chunks(self: Self) -> Sequence[Chunk]:
         pass
 
-    @property
     @abstractmethod
-    def id(self: Self) -> str:
+    def get_metadata(self: Self) -> Mapping[str, Any]:
         pass

src/ctp_slack_bot/models/slack.py

@@ -63,31 +63,19 @@ class SlackMessage(Content):
     is_starred: Optional[bool] = None
     pinned_to: Optional[Sequence[str]] = None
     reactions: Optional[Sequence[SlackReaction]] = None
-    _canonical_json: PrivateAttr
 
-    def __init__(self: Self, **data: Dict[str, Any]) -> None:
-        super().__init__(**data)
-        self._canonical_json = PrivateAttr(default_factory=lambda: dumps(data, sort_keys=True).encode())
+    def get_id(self: Self) -> str:
+        """Unique identifier for this message."""
+        return f"slack-message:{self.channel}:{self.ts}"
 
     def get_chunks(self: Self) -> Sequence[Chunk]:
-        return (Chunk(text=self.text, parent_id=self.id, chunk_id="", metadata=self.get_metadata()), )
+        return (Chunk(text=self.text, parent_id=self.get_id(), chunk_id="", metadata=self.get_metadata()), )
 
     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
-
-    def get_bytes(self: Self) -> bytes:
-        return self._canonical_json
-
-    @property
-    def id(self: Self) -> str:
-        """Unique identifier for this message."""
-        return f"slack-message:{self.channel}:{self.ts}"
-
 class SlackResponse(BaseModel): # TODO: This should also be based on Content as it is a SlackMessage―just not one for which we know the identity yet.
     """Represents a response message to be sent to Slack."""
 

src/ctp_slack_bot/models/webvtt.py

@@ -1,15 +1,17 @@
 from datetime import datetime, timedelta
 from io import BytesIO
 from json import dumps
-from pydantic import BaseModel, ConfigDict, PositiveInt, PrivateAttr
-import re
+from more_itertools import windowed
+from pydantic import BaseModel, ConfigDict, Field, PositiveInt, PrivateAttr
 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('(?:([^:]+): )?(.*)')
+CHUNK_FRAMES_OVERLAP = 1
+CHUNK_FRAMES_WINDOW = 5
+SPEAKER_SPEECH_TEXT_SEPARATOR = ": "
 
 class WebVTTFrame(BaseModel):
     """Represents a WebVTT frame"""
@@ -27,50 +29,44 @@ class WebVTTFrame(BaseModel):
         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):
+        match caption.text.split(SPEAKER_SPEECH_TEXT_SEPARATOR, 1):
+            case [speaker, speech]:
                 return cls(identifier=identifier, start=start, end=end, speaker=speaker, speech=speech)
-            case _:
+            case [speech]:
                 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."""
+class WebVTTContent(Content):
+    """Represents parsed WebVTT content."""
 
-    filename: str
-    modification_time: datetime
-    bytes: bytes
+    id: str
+    metadata: Mapping[str, Any] = Field(default_factory=dict)
+    frames: Sequence[WebVTTFrame]
 
+    def get_id(self: Self) -> str:
+        return self.id
 
     def get_chunks(self: Self) -> Sequence[Chunk]:
-        return tuple(Chunk(text=frame.speech,
-                           parent_id=self.id,
-                           chunk_id=frame.identifier,
+        windows = (tuple(filter(None, window))
+                   for window
+                   in windowed(self.frames, CHUNK_FRAMES_WINDOW, step=CHUNK_FRAMES_WINDOW-CHUNK_FRAMES_OVERLAP))
+        return tuple(Chunk(text="\n\n".join(": ".join(filter(None, (frame.speaker, frame.speech)))
+                                            for frame
+                                            in frames),
+                           parent_id=self.get_id(),
+                           chunk_id=f"{frames[0].identifier}-{frames[-1].identifier}",
                            metadata={
-                               "filename": self.filename,
-                               "start": self.modification_time + frame.start,
-                               "end": self.modification_time + frame.end,
-                               "user": frame.speaker
+                               "start": frames[0].start,
+                               "end": frames[-1].end,
+                               "speakers": frozenset(frame.speaker for frame in frames)
                            })
-                     for frame
-                     in self.get_frames())
+                     for frames
+                     in windows)
 
     def get_metadata(self: Self) -> Mapping[str, Any]:
-        return MappingProxyType({
-            "filename": self.filename,
-            "modificationTime": self.modification_time
-        })
+        return MappingProxyType(self.metadata)
 
-    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}"
+    @classmethod
+    def from_bytes(cls: type["WebVTTContent"], id: str, metadata: Mapping[str, Any], buffer: bytes) -> Self:
+        frames = tuple(map(WebVTTFrame.from_webvtt_caption, WebVTT.from_buffer(BytesIO(buffer)).captions))
+        return WebVTTContent(id=id, metadata=MappingProxyType(metadata), frames=frames)

src/ctp_slack_bot/services/content_ingestion_service.py

@@ -30,8 +30,8 @@ class ContentIngestionService(BaseModel):
 
     async def process_incoming_content(self: Self, content: Content) -> None:
         logger.debug("Content ingestion service received content with metadata: {}", content.get_metadata())
-        # if self.vector_database_service.has_content(content.id) # TODO
-        #    logger.debug("Ignored content with ID {} because it already exists in the database.", content.id)
+        # if self.vector_database_service.has_content(content.get_id()) # TODO
+        #    logger.debug("Ignored content with ID {} because it already exists in the database.", content.get_id())
         #    return
         chunks = content.get_chunks()
         await self.__vectorize_and_store_chunks_in_database(chunks)