generate content at delivery time

src/controllers/_bot_controller.py

@@ -45,7 +45,7 @@ class BotController:
                                 logger.info(f"file id is {file_id}")
                                 async with SlackBotService() as slack_bot_service:
                                     uploadedFile = await slack_bot_service.save_file(
-                                        event["team_id"], file
+                                        team, file
                                     )
                                     async with FileService() as file_service:
                                         file_service.process_file(uploadedFile, user.id)

src/controllers/_file_controller.py

@@ -14,11 +14,13 @@ class FileController:
         logger.info(f"Parsing file: {file.filename}")
         try:
             async with self.file_service as service:
-                result = await service.parse_document(file)
+                file_path = await service.save_uploaded_file(file)
+                file_name = file.filename
+                result = await service.parse_document(file_name, file_path)
             return result
         except ValueError as e:
             logger.warning(f"Validation error: {str(e)}")
             raise HTTPException(status_code=400, detail=str(e))
         except Exception as e:
             logger.error(f"Error parsing document: {str(e)}")
-            raise HTTPException(status_code=500, detail=str(e))
+            raise HTTPException(status_code=500, detail=str(e))

src/models/_user.py

@@ -21,7 +21,7 @@ class User(Base):
     content_delivery_frequency = Column(
         Enum(ContentDeliveryFrequency),
         nullable=False,
-        default=ContentDeliveryFrequency.WEEKLY,
+        default=ContentDeliveryFrequency.DAILY,
     )
     slack_id = Column(String, nullable=True)
     slack_team_id = Column(

src/repositories/_personalized_content_repository.py

@@ -45,3 +45,13 @@ class PersonalizedContentRepository(BaseRepository):
         if content:
             content.delivered_at = datetime.now()
             await self.update(content)
+
+    async def get_users_last_message(self, user_id: int) -> PersonalizedContent:
+        """Get the last message sent to the user"""
+        query = (
+            select(PersonalizedContent)
+            .where(PersonalizedContent.user_id == user_id)
+            .order_by(PersonalizedContent.delivered_at.desc())
+        )
+        result = await self.execute(query)
+        return result.scalar_one_or_none()

src/services/_content_delivery_service.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 from collections import defaultdict
 from typing import List
 
@@ -19,7 +19,9 @@ class ContentDeliveryService:
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         pass
 
-    async def search_and_generate_content(self, user_id: str = None) -> List[str]:
+    async def search_and_generate_content(
+        self, user_id: str = None, last_message: str = None
+    ) -> List[str]:
 
         query = """
         - Focus on the key concepts like B.E.A.M., B.L.I.P., C.O.R.E., and Situational Leadership.
@@ -56,6 +58,14 @@ class ContentDeliveryService:
                 "role": "system",
                 "content": "Here is an example of a personalized content for a participant in the Resilient Leadership program:- '🍁 *Gratitude Practice* With Thanksgiving on the horizon, let's cultivate gratitude. Start each meeting this week by sharing one thing you're grateful for.'",
             },
+            {
+                "role": "system",
+                "content": (
+                    "This was the last message sent to the user: " + last_message
+                    if last_message
+                    else "No previous message sent to the user"
+                ),
+            },
         ]
 
         async with OpenAIClient() as openai_client:
@@ -91,31 +101,53 @@ class ContentDeliveryService:
     async def send_messages(self):
         try:
 
-            results = await self.content_repository.get_due_content_for_delivery()
+            start_of_day = datetime.now().replace(
+                hour=0, minute=0, second=0, microsecond=0
+            )
+            end_of_day = start_of_day + timedelta(days=1)
 
-            if not results:
+            users = await self.user_repository.get_all(
+                filter_by={
+                    "next_content_delivery_date__gte": start_of_day,
+                    "next_content_delivery_date__lt": end_of_day,
+                }
+            )
+
+            if not users:
                 logger.info("No content due for delivery")
                 return
 
             # Group results by team for efficient Slack client usage
             team_grouped_data = defaultdict(list)
-            for user, content, team in results:
-                team_grouped_data[team].append((user, content))
+            for user in users:
+                team_grouped_data[user.slack_team_id].append(user)
 
             # Process each team's users
-            for team, user_contents in team_grouped_data.items():
+            for team in team_grouped_data.items():
                 try:
                     # Initialize Slack client with team token
                     async with SlackClient(token=team.token) as slack:
-                        for user, content in user_contents:
+                        for user in team:
                             try:
+                                last_messages = await self.content_repository.get_users_last_message(
+                                    user.id
+                                )
+                                content = await self.search_and_generate_content(
+                                    str(user.id),
+                                    last_messages[0].content if last_messages else None,
+                                )
 
-                                await slack.send_message_to_user(
-                                    user_id=user.slack_id,
-                                    message=content.content
+                                await slack.send_message(
+                                    user_id=user.slack_id, message=content[0]
                                 )
 
-                                await self.content_repository.mark_as_delivered(content.id)
+                                await self.content_repository.create(
+                                    {
+                                        "user_id": user.id,
+                                        "content": content[0],
+                                        "delivered_at": datetime.now(),
+                                    }
+                                )
 
                                 user.last_message_sent_at = datetime.now()
                                 user.next_content_delivery_date = self.calculate_next_delivery_date(

src/services/_file_service.py

@@ -1,3 +1,5 @@
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from typing import List, Tuple, Dict
 from fastapi import UploadFile
@@ -32,9 +34,13 @@ class FileService:
         Returns:
             Path: Path where file was saved
         """
+        loop = asyncio.get_event_loop()
         file_path = self.upload_dir / file.filename
         with open(file_path, "wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
+            with ThreadPoolExecutor() as pool:
+                await loop.run_in_executor(
+                    pool, lambda: shutil.copyfileobj(file.file, buffer)
+                )
         return file_path
 
     async def filter_ppt_content(self, parsed_content: Tuple[Tuple[int, str], ...]) -> Tuple[Tuple[int, str], ...]:
@@ -70,24 +76,8 @@ class FileService:
         return tuple(filtered_content)
 
     async def parse_document(
-        self, file: UploadFile, user_id: int = None, slack_file_id: str = None
+        self, file_name: str, file_path: str, user_id: int = None
     ) -> Dict:
-        file_path = await self.save_uploaded_file(file)
-        file_name = file.filename
-
-        if user_id is not None:
-            # Create file record in database only if user_id exists
-            mime_type = file.content_type or f"application/{file_name.split('.')[-1]}"
-            db_file = await self.file_repository.create_file(
-                    file_name=file_name,
-                    user_id=user_id,
-                    mime_type=mime_type,
-                    slack_file_id=slack_file_id
-                )
-            identifier = db_file.id
-        else:
-            # Skip database operations if user_id is None
-            identifier = file_name
 
         async with DocumentParser() as document_parser:
             parsed_content = await document_parser.parse_file_content(str(file_path))
@@ -96,30 +86,35 @@ class FileService:
             parsed_content = await self.filter_ppt_content(parsed_content)
 
         async with Helper() as helper:
-            documents = helper.create_documents(parsed_content, identifier, user_id)
+            documents = helper.create_documents(parsed_content, file_name, user_id)
 
         # Store documents in vector store and get record IDs
         async with VectorEmbedding() as vector_embedding:
             record_ids = await vector_embedding.store_documents(documents)
 
-        if user_id is not None:
-            if record_ids:
-                await self.vector_repository.create_records(db_file.id, record_ids)
-            await self.file_repository.mark_as_processed(db_file.id)
-            return_key = "file_id"
-        else:
-            return_key = "file_name"
-
-        return {
-                "message": "Document's embeddings created successfully",
-                return_key: identifier
-            }
+        return record_ids
 
     async def process_file(
         self, file: UploadFile, user_id: int = None, slack_file_id: str = None
     ) -> Dict:
         try:
-            await self.parse_document(file, user_id, slack_file_id)
+            file_path = await self.save_uploaded_file(file)
+            file_name = file.filename
+
+            if user_id is not None:
+                # Create file record in database only if user_id exists
+                mime_type = (
+                    file.content_type or f"application/{file_name.split('.')[-1]}"
+                )
+                db_file = await self.file_repository.create_file(
+                    file_name=file_name,
+                    user_id=user_id,
+                    mime_type=mime_type,
+                    slack_file_id=slack_file_id,
+                )
+            record_ids = await self.parse_document(file_name, file_path, user_id)
+            await self.vector_repository.create_records(db_file.id, record_ids)
+            await self.file_repository.mark_as_processed(db_file.id)
 
             async with UserService() as user_service:
                 await user_service.send_message(

src/services/_slack_bot_service.py

@@ -64,13 +64,7 @@ class SlackBotService:
             response = await slack_client.get_users_list()
             return response
 
-    async def save_file(self, team_id: str, file: dict):
-        logger.info(f"team_id is {team_id}")
-        team = await self.get_team_by_slack_team_id(team_id)
-
-        if not team:
-            logger.error(f"Team not found for slack_team_id: {team_id}")
-            return {"message": "Team not found"}
+    async def save_file(self, team: any, file: dict):
 
         file_url = file.get("url_private")
         file_name = file.get("name", "downloaded_file")

src/services/_user_service.py

@@ -25,6 +25,10 @@ class UserService:
         return await self.user_repository.get_user_by_email(email=email)
 
     async def update_user_info(self, user_id, info):
+        next_delivery_date = self.calculate_next_delivery_date(
+            datetime.now(), ContentDeliveryFrequency.DAILY
+        )
+        info["next_content_delivery_date"] = next_delivery_date
         return await self.user_repository.patch(user_id, info)
 
     async def get_users_list(self):
@@ -53,7 +57,7 @@ class UserService:
     def calculate_next_delivery_date(self, last_sent_at: Optional[datetime], frequency: ContentDeliveryFrequency) -> datetime:
         """Calculate next delivery date based on last message sent date"""
         base_date = last_sent_at if last_sent_at else datetime.now()
-        
+
         if frequency == ContentDeliveryFrequency.DAILY:
             return base_date + timedelta(days=1)
         elif frequency == ContentDeliveryFrequency.WEEKLY:
@@ -62,7 +66,7 @@ class UserService:
             return base_date + timedelta(weeks=2)
         elif frequency == ContentDeliveryFrequency.MONTHLY:
             return base_date + timedelta(days=30)
-        
+
         return base_date + timedelta(weeks=1)  # Default to weekly
 
     async def update_user_frequency(self, user_id: int, frequency: ContentDeliveryFrequency) -> Optional[User]:
@@ -85,7 +89,6 @@ class UserService:
 
             return await self.user_repository.patch(user_id, update_data)
 
-
     async def send_message(self, user_id: int, message: str):
         """Send message to user"""
         user = await self.user_repository.get(user_id)

src/utils/_helpers.py

@@ -15,7 +15,7 @@ class Helper:
     async def create_documents(
         page_texts: List[Dict],
         file_name: str,
-        user_id: Optional[str] = None,
+        user_id: int = None,
     ) -> List[Document]:
         """
         Create Document objects from parsed page texts using either file_id or file_name.
@@ -34,7 +34,7 @@ class Helper:
             metadata = {
                 "file_name": file_name,
                 "page_number": page_num,
-                "user_id": user_id,
+                "user_id": str(user_id) if user_id else "public",
             }
             loop = asyncio.get_event_loop()
             with ThreadPoolExecutor() as pool: