Ingestion status callback update

app/domain/data/lecture_unit_dto.py

@@ -2,8 +2,6 @@
 
 
 class LectureUnitDTO(BaseModel):
-    to_update: bool = Field(alias="toUpdate")
-    base_url: str = Field(alias="artemisBaseUrl")
     pdf_file_base64: str = Field(default="", alias="pdfFile")
     lecture_unit_id: int = Field(alias="lectureUnitId")
     lecture_unit_name: str = Field(default="", alias="lectureUnitName")

app/domain/ingestion/deletionPipelineExecutionDto.py

@@ -0,0 +1,15 @@
+from typing import List, Optional
+
+from pydantic import Field
+
+from app.domain import PipelineExecutionDTO, PipelineExecutionSettingsDTO
+from app.domain.data.lecture_unit_dto import LectureUnitDTO
+from app.domain.status.stage_dto import StageDTO
+
+
+class LecturesDeletionExecutionDto(PipelineExecutionDTO):
+    lecture_units: List[LectureUnitDTO] = Field(..., alias="pyrisLectureUnits")
+    settings: Optional[PipelineExecutionSettingsDTO]
+    initial_stages: Optional[List[StageDTO]] = Field(
+        default=None, alias="initialStages"
+    )

app/domain/ingestion/ingestion_pipeline_execution_dto.py

@@ -8,9 +8,7 @@
 
 
 class IngestionPipelineExecutionDto(PipelineExecutionDTO):
-    lecture_units: List[LectureUnitDTO] = Field(
-        ..., alias="pyrisLectureUnitWebhookDTOS"
-    )
+    lecture_unit: LectureUnitDTO = Field(..., alias="pyrisLectureUnit")
     settings: Optional[PipelineExecutionSettingsDTO]
     initial_stages: Optional[List[StageDTO]] = Field(
         default=None, alias="initialStages"

app/domain/ingestion/ingestion_status_update_dto.py

@@ -5,3 +5,4 @@
 
 class IngestionStatusUpdateDTO(StatusUpdateDTO):
     result: Optional[str] = None
+    id: Optional[int] = None

app/pipeline/lecture_ingestion_pipeline.py

@@ -3,6 +3,8 @@
 import tempfile
 import threading
 from asyncio.log import logger
+from typing import Optional
+
 import fitz
 from langchain_core.output_parsers import StrOutputParser
 from langchain_core.prompts import ChatPromptTemplate
@@ -90,7 +92,7 @@ class LectureIngestionPipeline(AbstractIngestion, Pipeline):
     def __init__(
         self,
         client: WeaviateClient,
-        dto: IngestionPipelineExecutionDto,
+        dto: Optional[IngestionPipelineExecutionDto],
         callback: IngestionStatusCallback,
     ):
         super().__init__()
@@ -116,33 +118,31 @@ def __init__(
     def __call__(self) -> bool:
         try:
             self.callback.in_progress("Deleting old slides from database...")
-            self.delete_old_lectures()
+            self.delete_lecture_unit(
+                self.dto.lecture_unit.course_id,
+                self.dto.lecture_unit.lecture_id,
+                self.dto.lecture_unit.lecture_unit_id,
+                self.dto.settings.artemis_base_url,
+            )
             self.callback.done("Old slides removed")
-            # Here we check if the operation is for updating or for deleting,
-            # we only check the first file because all the files will have the same operation
-            if not self.dto.lecture_units[0].to_update:
-                self.callback.skip("Lecture Chunking and interpretation Skipped")
-                self.callback.skip("No new slides to update")
-                return True
             self.callback.in_progress("Chunking and interpreting lecture...")
             chunks = []
-            for i, lecture_unit in enumerate(self.dto.lecture_units):
-                pdf_path = save_pdf(lecture_unit.pdf_file_base64)
-                chunks.extend(
-                    self.chunk_data(
-                        lecture_pdf=pdf_path,
-                        lecture_unit_dto=lecture_unit,
-                        base_url=self.dto.settings.artemis_base_url,
-                    )
+            pdf_path = save_pdf(self.dto.lecture_unit.pdf_file_base64)
+            chunks.extend(
+                self.chunk_data(
+                    lecture_pdf=pdf_path,
+                    lecture_unit_dto=self.dto.lecture_unit,
+                    base_url=self.dto.settings.artemis_base_url,
                 )
-                cleanup_temporary_file(pdf_path)
+            )
+            cleanup_temporary_file(pdf_path)
             self.callback.done("Lecture Chunking and interpretation Finished")
             self.callback.in_progress("Ingesting lecture chunks into database...")
             self.batch_update(chunks)
             self.callback.done("Lecture Ingestion Finished")
             logger.info(
                 f"Lecture ingestion pipeline finished Successfully for course "
-                f"{self.dto.lecture_units[0].course_name}"
+                f"{self.dto.lecture_unit.course_name}"
             )
             return True
         except Exception as e:
@@ -294,23 +294,27 @@ def get_course_language(self, page_content: str) -> str:
         )
         return response.contents[0].text_content
 
-    def delete_old_lectures(self):
+    def delete_old_lectures(
+        self, lecture_units: list[LectureUnitDTO], artemis_base_url: str
+    ):
         """
         Delete the lecture unit from the database
         """
         try:
-            for lecture_unit in self.dto.lecture_units:
+            for lecture_unit in lecture_units:
                 if self.delete_lecture_unit(
                     lecture_unit.course_id,
                     lecture_unit.lecture_id,
                     lecture_unit.lecture_unit_id,
-                    self.dto.settings.artemis_base_url,
+                    artemis_base_url,
                 ):
                     logger.info("Lecture deleted successfully")
                 else:
                     logger.error("Failed to delete lecture")
+            self.callback.done("Old slides removed")
         except Exception as e:
             logger.error(f"Error deleting lecture unit: {e}")
+            self.callback.error("Error while removing old slides")
             return False
 
     def delete_lecture_unit(self, course_id, lecture_id, lecture_unit_id, base_url):

app/web/routers/webhooks.py

@@ -10,12 +10,15 @@
     IngestionPipelineExecutionDto,
 )
 from ..status.IngestionStatusCallback import IngestionStatusCallback
+from ..status.LecturesDeletionStatusCallback import LecturesDeletionStatusCallback
+from ...domain.ingestion.deletionPipelineExecutionDto import (
+    LecturesDeletionExecutionDto,
+)
 from ...pipeline.lecture_ingestion_pipeline import LectureIngestionPipeline
 from ...vector_database.database import VectorDatabase
 
 router = APIRouter(prefix="/api/v1/webhooks", tags=["webhooks"])
 
-
 semaphore = Semaphore(5)
 
 
@@ -29,6 +32,7 @@ def run_lecture_update_pipeline_worker(dto: IngestionPipelineExecutionDto):
                 run_id=dto.settings.authentication_token,
                 base_url=dto.settings.artemis_base_url,
                 initial_stages=dto.initial_stages,
+                lecture_unit_id=dto.lecture_unit.lecture_unit_id,
             )
             db = VectorDatabase()
             client = db.get_client()
@@ -44,14 +48,46 @@ def run_lecture_update_pipeline_worker(dto: IngestionPipelineExecutionDto):
             semaphore.release()
 
 
+def run_lecture_deletion_pipeline_worker(dto: LecturesDeletionExecutionDto):
+    """
+    Run the exercise chat pipeline in a separate thread
+    """
+    try:
+        callback = LecturesDeletionStatusCallback(
+            run_id=dto.settings.authentication_token,
+            base_url=dto.settings.artemis_base_url,
+            initial_stages=dto.initial_stages,
+        )
+        db = VectorDatabase()
+        client = db.get_client()
+        pipeline = LectureIngestionPipeline(client=client, dto=None, callback=callback)
+        pipeline.delete_old_lectures(dto.lecture_units)
+    except Exception as e:
+        logger.error(f"Error while deleting lectures: {e}")
+        logger.error(traceback.format_exc())
+
+
 @router.post(
     "/lectures/fullIngestion",
     status_code=status.HTTP_202_ACCEPTED,
     dependencies=[Depends(TokenValidator())],
 )
-def lecture_webhook(dto: IngestionPipelineExecutionDto):
+def lecture_ingestion_webhook(dto: IngestionPipelineExecutionDto):
     """
     Webhook endpoint to trigger the exercise chat pipeline
     """
     thread = Thread(target=run_lecture_update_pipeline_worker, args=(dto,))
     thread.start()
+
+
+@router.post(
+    "/lectures/delete",
+    status_code=status.HTTP_202_ACCEPTED,
+    dependencies=[Depends(TokenValidator())],
+)
+def lecture_deletion_webhook(dto: LecturesDeletionExecutionDto):
+    """
+    Webhook endpoint to trigger the exercise chat pipeline
+    """
+    thread = Thread(target=run_lecture_deletion_pipeline_worker, args=(dto,))
+    thread.start()

app/web/status/IngestionStatusCallback.py

@@ -15,7 +15,11 @@ class IngestionStatusCallback(StatusCallback):
     """
 
     def __init__(
-        self, run_id: str, base_url: str, initial_stages: List[StageDTO] = None
+        self,
+        run_id: str,
+        base_url: str,
+        initial_stages: List[StageDTO] = None,
+        lecture_unit_id: int = None,
     ):
         url = f"{base_url}/api/public/pyris/webhooks/ingestion/runs/{run_id}/status"
 
@@ -36,6 +40,6 @@ def __init__(
                 name="Slides ingestion",
             ),
         ]
-        status = IngestionStatusUpdateDTO(stages=stages)
+        status = IngestionStatusUpdateDTO(stages=stages, id=lecture_unit_id)
         stage = stages[current_stage_index]
         super().__init__(url, run_id, status, stage, current_stage_index)

app/web/status/LecturesDeletionStatusCallback.py

@@ -0,0 +1,31 @@
+from typing import List
+
+from .status_update import StatusCallback
+from ...domain.ingestion.ingestion_status_update_dto import IngestionStatusUpdateDTO
+from ...domain.status.stage_state_dto import StageStateEnum
+from ...domain.status.stage_dto import StageDTO
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class LecturesDeletionStatusCallback(StatusCallback):
+    """
+    Callback class for updating the status of a Tutor Chat pipeline run.
+    """
+
+    def __init__(
+        self, run_id: str, base_url: str, initial_stages: List[StageDTO] = None
+    ):
+        url = f"{base_url}/api/public/pyris/webhooks/ingestion/runs/{run_id}/status"
+
+        current_stage_index = len(initial_stages) if initial_stages else 0
+        stages = initial_stages or []
+        stages += [
+            StageDTO(
+                weight=100, state=StageStateEnum.NOT_STARTED, name="Slides removal"
+            ),
+        ]
+        status = IngestionStatusUpdateDTO(stages=stages)
+        stage = stages[current_stage_index]
+        super().__init__(url, run_id, status, stage, current_stage_index)