Recording mixed file

Author: borbuevbeksultan

meeting-recording.json

@@ -10,6 +10,8 @@ data class AwsConfig(
     var secretKey: String? = null,
     var region: String? = null,
     var attachmentsBucket: String? = null,
+    var recordingsBucket: String? = null,
+    var transcriptsBucket: String? = null,
     var cognito: Cognito = Cognito()
 )
 

src/main/kotlin/io/openfuture/openmessenger/configuration/AwsConfig.kt

@@ -1,7 +1,12 @@
 package io.openfuture.openmessenger.kurento
 
+import io.openfuture.openmessenger.assistant.gemini.GeminiService
 import io.openfuture.openmessenger.kurento.groupcall.CallHandler
 import io.openfuture.openmessenger.kurento.groupcall.RoomManager
+import io.openfuture.openmessenger.kurento.recording.RecordingCallHandler
+import io.openfuture.openmessenger.repository.MeetingNoteRepository
+import io.openfuture.openmessenger.service.RecordingManagementService
+import io.openfuture.openmessenger.service.SpeechToTextService
 import org.kurento.client.KurentoClient
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.context.annotation.Bean
@@ -10,13 +15,16 @@ import org.springframework.web.socket.config.annotation.EnableWebSocket
 import org.springframework.web.socket.config.annotation.WebSocketConfigurer
 import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry
 import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean
-import io.openfuture.openmessenger.kurento.recording.RecordingCallHandler as RecordingCallHandler
 
 @Configuration
 @EnableWebSocket
 class KurentoWebsocketConfigurer(
     @Value("\${kms.url}")
-    val kurentoUrl: String
+    val kurentoUrl: String,
+    val recordingManagementService: RecordingManagementService,
+    private val speechToTextService: SpeechToTextService,
+    private val geminiService: GeminiService,
+    private val meetingNoteRepository: MeetingNoteRepository
 ) : WebSocketConfigurer {
     @Bean
     fun handler(): HelloWorldHandler {
@@ -63,7 +71,9 @@ class KurentoWebsocketConfigurer(
 
     @Bean
     fun callHandler(): RecordingCallHandler {
-        return RecordingCallHandler(kurentoClient(), recordingUserRegistry())
+        return RecordingCallHandler(
+            kurentoClient(), recordingUserRegistry(), recordingManagementService, speechToTextService, geminiService, meetingNoteRepository
+        )
     }
 
 }

src/main/kotlin/io/openfuture/openmessenger/kurento/KurentoWebsocketConfigurer.kt

@@ -15,7 +15,8 @@ class CallMediaPipeline(kurento: KurentoClient, from: String, to: String?) {
 
     val composite: Composite = Composite.Builder(pipeline).build()
     val out = HubPort.Builder(composite).build()
-    val recorder: RecorderEndpoint = RecorderEndpoint.Builder(pipeline, RECORDING_PATH + "combined" + RECORDING_EXT)
+    val recordedFileUri = RECORDING_PATH + "combined" + RECORDING_EXT
+    val recorder: RecorderEndpoint = RecorderEndpoint.Builder(pipeline, recordedFileUri)
         .build()
 
     init {
@@ -57,7 +58,7 @@ class CallMediaPipeline(kurento: KurentoClient, from: String, to: String?) {
 
     companion object {
         private val df = SimpleDateFormat("yyyy-MM-dd_HH-mm-ss-S")
-        val RECORDING_PATH: String = "file:///tmp/" + df.format(Date()) + "-"
+        val RECORDING_PATH: String = "file:///tmp/recordings/" + df.format(Date()) + "-"
         const val RECORDING_EXT: String = ".webm"
     }
 }

src/main/kotlin/io/openfuture/openmessenger/kurento/recording/CallMediaPipeline.kt

@@ -3,6 +3,11 @@ package io.openfuture.openmessenger.kurento.recording
 import com.google.gson.Gson
 import com.google.gson.GsonBuilder
 import com.google.gson.JsonObject
+import io.openfuture.openmessenger.assistant.gemini.GeminiService
+import io.openfuture.openmessenger.repository.MeetingNoteRepository
+import io.openfuture.openmessenger.repository.entity.MeetingNoteEntity
+import io.openfuture.openmessenger.service.RecordingManagementService
+import io.openfuture.openmessenger.service.SpeechToTextService
 import org.kurento.client.IceCandidate
 import org.kurento.client.KurentoClient
 import org.kurento.client.MediaPipeline
@@ -14,21 +19,27 @@ import org.springframework.web.socket.TextMessage
 import org.springframework.web.socket.WebSocketSession
 import org.springframework.web.socket.handler.TextWebSocketHandler
 import java.io.IOException
+import java.time.LocalDateTime
 import java.util.concurrent.ConcurrentHashMap
 
 class RecordingCallHandler(
     private val kurento: KurentoClient,
     private val registry: UserRegistry,
+    private val recordingManagementService: RecordingManagementService,
+    private val speechToTextService: SpeechToTextService,
+    private val geminiService: GeminiService,
+    private val meetingNoteRepository: MeetingNoteRepository
 ): TextWebSocketHandler() {
     val pipelines = ConcurrentHashMap<String, MediaPipeline?>()
+    val recordings = ConcurrentHashMap<String, String>()
 
     @Throws(Exception::class)
     public override fun handleTextMessage(session: WebSocketSession, message: TextMessage) {
         val jsonMessage = gson.fromJson(
             message.payload,
             JsonObject::class.java
         )
-        val user = registry!!.getBySession(session)
+        val user = registry.getBySession(session)
 
         if (user != null) {
             log.debug("Incoming message from user '{}': {}", user.name, jsonMessage)
@@ -113,16 +124,16 @@ class RecordingCallHandler(
     private fun incomingCallResponse(callee: UserSession, jsonMessage: JsonObject) {
         val callResponse = jsonMessage["callResponse"].asString
         val from = jsonMessage["from"].asString
-        val calleer = registry!!.getByName(from)
-        val to = calleer?.callingTo
+        val caller = registry.getByName(from)
+        val to = caller?.callingTo
 
         if ("accept" == callResponse) {
             log.debug("Accepted call from '{}' to '{}'", from, to)
 
-            val callMediaPipeline = CallMediaPipeline(
-                kurento!!, from, to
-            )
-            pipelines[calleer!!.sessionId] = callMediaPipeline.pipeline
+            val callMediaPipeline = CallMediaPipeline(kurento, from, to)
+            recordings[caller!!.sessionId] = callMediaPipeline.recordedFileUri
+            recordings[callee.sessionId] = callMediaPipeline.recordedFileUri
+            pipelines[caller.sessionId] = callMediaPipeline.pipeline
             pipelines[callee.sessionId] = callMediaPipeline.pipeline
 
             callee.setWebRtcEndpoint(callMediaPipeline.calleeWebRtcEp)
@@ -153,14 +164,14 @@ class RecordingCallHandler(
 
             val callerSdpOffer = registry.getByName(from)?.sdpOffer
 
-            calleer!!.setWebRtcEndpoint(callMediaPipeline.callerWebRtcEp)
+            caller.setWebRtcEndpoint(callMediaPipeline.callerWebRtcEp)
             callMediaPipeline.callerWebRtcEp.addIceCandidateFoundListener { event ->
                 val response = JsonObject()
                 response.addProperty("id", "iceCandidate")
                 response.add("candidate", JsonUtils.toJsonObject(event.candidate))
                 try {
-                    synchronized(calleer.session) {
-                        calleer.session.sendMessage(TextMessage(response.toString()))
+                    synchronized(caller.session) {
+                        caller.session.sendMessage(TextMessage(response.toString()))
                     }
                 } catch (e: IOException) {
                     log.debug(e.message)
@@ -174,8 +185,8 @@ class RecordingCallHandler(
             response.addProperty("response", "accepted")
             response.addProperty("sdpAnswer", callerSdpAnswer)
 
-            synchronized(calleer) {
-                calleer.sendMessage(response)
+            synchronized(caller) {
+                caller.sendMessage(response)
             }
 
             callMediaPipeline.callerWebRtcEp.gatherCandidates()
@@ -185,15 +196,14 @@ class RecordingCallHandler(
             val response = JsonObject()
             response.addProperty("id", "callResponse")
             response.addProperty("response", "rejected")
-            calleer!!.sendMessage(response)
+            caller!!.sendMessage(response)
         }
     }
 
     @Throws(IOException::class)
     fun stop(session: WebSocketSession) {
-        // Both users can stop the communication. A 'stopCommunication'
-        // message will be sent to the other peer.
-        val stopperUser = registry!!.getBySession(session)
+        val stopperUser = registry.getBySession(session)
+
         if (stopperUser != null) {
             val stoppedUser =
                 if ((stopperUser.callingFrom != null))
@@ -211,6 +221,36 @@ class RecordingCallHandler(
                 stoppedUser.clear()
             }
             stopperUser.clear()
+
+            val caller = stopperUser.callingFrom ?: stoppedUser?.callingFrom
+            val callee = stopperUser.callingTo ?: stoppedUser?.callingTo
+            log.info("Caller = $caller, callee = $callee")
+
+            pipelines[stoppedUser?.sessionId]
+
+            val fileUriString = recordings[stoppedUser?.sessionId]
+            fileUriString?.let { val uploadToS3 = recordingManagementService.uploadToS3(it)
+                if (uploadToS3 == -1) {
+                    log.warn("No file was uploaded for call session ${stoppedUser?.sessionId}")
+                    return
+                }
+                val transcript = speechToTextService.extractTranscript(uploadToS3)
+                val chat = geminiService.chat("Generate a summary from the following meeting record: {$transcript}")
+
+                val meetingNoteEntity = MeetingNoteEntity(
+                    caller,
+                    777,
+                    999,
+                    "[$caller, $callee]",
+                    callee,
+                    LocalDateTime.now(),
+                    1,
+                    LocalDateTime.now(),
+                    LocalDateTime.now(),
+                    chat
+                )
+                meetingNoteRepository.save(meetingNoteEntity)
+            }
         }
     }
 
@@ -294,11 +334,12 @@ class RecordingCallHandler(
     @Throws(Exception::class)
     override fun afterConnectionClosed(session: WebSocketSession, status: CloseStatus) {
         stop(session)
-        registry!!.removeBySession(session)
+        registry.removeBySession(session)
     }
 
     companion object {
         private val log: Logger = LoggerFactory.getLogger(RecordingCallHandler::class.java)
         private val gson: Gson = GsonBuilder().create()
     }
+
 }

src/main/kotlin/io/openfuture/openmessenger/kurento/recording/RecordingCallHandler.kt

@@ -0,0 +1,7 @@
+package io.openfuture.openmessenger.repository
+
+import io.openfuture.openmessenger.repository.entity.MeetingNoteEntity
+import org.springframework.data.jpa.repository.JpaRepository
+
+interface MeetingNoteRepository : JpaRepository<MeetingNoteEntity, Long> {
+}

src/main/kotlin/io/openfuture/openmessenger/repository/MeetingNoteRepository.kt

@@ -0,0 +1,48 @@
+package io.openfuture.openmessenger.repository.entity
+
+import jakarta.persistence.*
+import java.time.LocalDateTime
+import java.time.LocalDateTime.now
+
+
+@Entity
+@Table(name = "meeting_notes")
+class MeetingNoteEntity() {
+    constructor(
+        author: String?,
+        chatId: Int?,
+        groupChatId: Int?,
+        members: String?,
+        recipient: String?,
+        generatedAt: LocalDateTime = now(),
+        version: Int = 1,
+        startTime: LocalDateTime?,
+        endTime: LocalDateTime?,
+        notes: String?
+    ): this() {
+        this.author = author
+        this.chatId = chatId
+        this.groupChatId = groupChatId
+        this.members = members
+        this.recipient = recipient
+        this.generatedAt = generatedAt
+        this.version = version
+        this.startTime = startTime
+        this.endTime = endTime
+        this.notes = notes
+    }
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    var id: Long? = null
+    var author: String? = null
+    var chatId: Int? = null
+    var groupChatId: Int? = null
+    var members: String? = null
+    var recipient: String? = null
+    var generatedAt: LocalDateTime = now()
+    var version: Int = 1
+    var startTime: LocalDateTime? = null
+    var endTime: LocalDateTime? = null
+    var notes: String? = null
+}

src/main/kotlin/io/openfuture/openmessenger/repository/entity/MeetingNoteEntity.kt

@@ -15,7 +15,7 @@ interface AttachmentService {
     fun uploadAndReturnId(file: MultipartFile): Int
 
     @Throws(IOException::class)
-    fun download(fileName: String?): ByteArray?
+    fun download(fileName: String?, bucket: String): ByteArray?
 
     fun downloadById(id: Int): ByteArray?
 }

src/main/kotlin/io/openfuture/openmessenger/service/AttachmentService.kt

@@ -0,0 +1,6 @@
+package io.openfuture.openmessenger.service
+
+interface RecordingManagementService {
+    fun list()
+    fun uploadToS3(fileUri: String): Int
+}

src/main/kotlin/io/openfuture/openmessenger/service/RecordingManagementService.kt

@@ -2,4 +2,5 @@ package io.openfuture.openmessenger.service
 
 interface SpeechToTextService {
     fun extractTranscript(attachmentId: Int): String
+    fun getTranscript(attachmentId: String): String
 }