Issue #SC-1847 feat: added enhancement to support content stage

content-api/content-actors/src/main/scala/org/sunbird/content/constant/ImportConstants.scala

@@ -13,5 +13,6 @@ object ImportConstants {
 	val IDENTIFIER: String = "identifier"
 	val OBJECT_TYPE: String = "objectType"
 	val REPOSITORY: String = "repository"
+	val STAGE: String = "stage"
 
 }

content-api/content-actors/src/main/scala/org/sunbird/content/error/ImportErrors.scala

@@ -8,6 +8,7 @@ object ImportErrors {
 	val ERR_READ_SOURCE: String = "ERR_READ_SOURCE"
 	val ERR_REQUIRED_PROPS_VALIDATION: String = "ERR_REQUIRED_PROPS_VALIDATION"
 	val BE_JOB_REQUEST_EXCEPTION: String = "BE_JOB_REQUEST_EXCEPTION"
+	val ERR_CONTENT_STAGE_VALIDATION: String = "ERR_CONTENT_STAGE_VALIDATION"
 
 	//Error Messages
 	val ERR_INVALID_IMPORT_REQUEST_MSG: String = "Invalid Request! Please Provide Valid Request."
@@ -15,5 +16,6 @@ object ImportErrors {
 	val ERR_READ_SOURCE_MSG: String = "Received Invalid Response While Reading Data From Source. Response Code is : "
 	val ERR_REQUIRED_PROPS_VALIDATION_MSG: String = "Validation Failed! Mandatory Properties Are "
 	val BE_JOB_REQUEST_EXCEPTION_MSG: String = "Kafka Event Is Not Generated Properly."
+	val ERR_CONTENT_STAGE_VALIDATION_MSG: String = "Content Stage Validation Failed! Valid Content Stages Are "
 
 }

content-api/content-actors/src/main/scala/org/sunbird/content/mgr/ImportManager.scala

@@ -26,6 +26,7 @@ object ImportManager {
 	val REQUEST_LIMIT = Platform.getInteger("content.import.request_size_limit", 200)
 	val AUTO_CREATE_TOPIC_NAME = Platform.config.getString("content.import.topic_name")
 	val REQUIRED_PROPS = Platform.getStringList("content.import.required_props", java.util.Arrays.asList("name", "code", "mimeType", "contentType", "artifactUrl", "framework"))
+	val VALID_CONTENT_STAGE = Platform.getStringList("content.import.valid_stages", java.util.Arrays.asList("Draft", "Review", "Live"))
 
 	def importContent(request: Request)(implicit oec: OntologyEngineContext, ec: ExecutionContext): Future[Response] = {
 		val graphId: String = request.getContext.get("graph_id").asInstanceOf[String]
@@ -34,11 +35,13 @@ object ImportManager {
 			throw new ClientException(ImportErrors.ERR_REQUEST_LIMIT_EXCEED, ImportErrors.ERR_REQUEST_LIMIT_EXCEED_MSG + REQUEST_LIMIT)
 		val processId: String = UUID.randomUUID().toString
 		val invalidCodes: util.List[String] = new util.ArrayList[String]()
-		validateAndGetRequest(reqList, processId, invalidCodes).map(contents => {
+		val invalidStage: util.List[String] = new util.ArrayList[String]()
+		validateAndGetRequest(reqList, processId, invalidCodes, invalidStage).map(contents => {
 			if (CollectionUtils.isNotEmpty(invalidCodes)) {
 				val msg = if (invalidCodes.asScala.filter(c => StringUtils.isNotBlank(c)).toList.size > 0) " | Required Property's Missing For " + invalidCodes else ""
 				throw new ClientException(ImportErrors.ERR_REQUIRED_PROPS_VALIDATION, ImportErrors.ERR_REQUIRED_PROPS_VALIDATION_MSG + REQUIRED_PROPS + msg)
-			} else {
+			} else if (CollectionUtils.isNotEmpty(invalidStage)) throw new ClientException(ImportErrors.ERR_CONTENT_STAGE_VALIDATION, ImportErrors.ERR_CONTENT_STAGE_VALIDATION_MSG + VALID_CONTENT_STAGE)
+			else {
 				contents.asScala.map(content => pushInstructionEvent(graphId, content))
 				val response = ResponseHandler.OK()
 				response.put(ImportConstants.PROCESS_ID, processId)
@@ -48,10 +51,11 @@ object ImportManager {
 
 	}
 
-	def validateAndGetRequest(contents: util.List[util.Map[String, AnyRef]], processId: String, invalidCodes: util.List[String])(implicit oec: OntologyEngineContext, ec: ExecutionContext): Future[util.List[util.Map[String, AnyRef]]] = {
+	def validateAndGetRequest(contents: util.List[util.Map[String, AnyRef]], processId: String, invalidCodes: util.List[String], invalidStages: util.List[String])(implicit oec: OntologyEngineContext, ec: ExecutionContext): Future[util.List[util.Map[String, AnyRef]]] = {
 		Future {
 			contents.asScala.map(content => {
 				val source: String = content.getOrDefault(ImportConstants.SOURCE, "").toString
+				val stage: String = content.getOrDefault(ImportConstants.STAGE, "").toString
 				val reqMetadata: util.Map[String, AnyRef] = content.getOrDefault(ImportConstants.METADATA, new util.HashMap[String, AnyRef]()).asInstanceOf[util.Map[String, AnyRef]]
 				val sourceMetadata: util.Map[String, AnyRef] = getMetadata(source)
 				val finalMetadata: util.Map[String, AnyRef] = if (MapUtils.isNotEmpty(sourceMetadata)) {
@@ -62,6 +66,7 @@ object ImportManager {
 				finalMetadata.put(ImportConstants.PROCESS_ID, processId)
 				if (!validateMetadata(finalMetadata))
 					invalidCodes.add(finalMetadata.getOrDefault(ImportConstants.CODE, "").asInstanceOf[String])
+				if(!validateStage(stage)) invalidStages.add(finalMetadata.getOrDefault(ImportConstants.CODE, "").asInstanceOf[String])
 				content.put(ImportConstants.METADATA, finalMetadata)
 				content
 			}).asJava
@@ -95,24 +100,27 @@ object ImportManager {
 		reqFields.isEmpty
 	}
 
-	def getInstructionEvent(identifier: String, source: String, metadata: util.Map[String, AnyRef], collection: util.List[util.Map[String, AnyRef]]): String = {
+	def validateStage(stage: String): Boolean = if(StringUtils.isNotBlank(stage)) VALID_CONTENT_STAGE.contains(stage) else true
+
+	def getInstructionEvent(identifier: String, source: String, metadata: util.Map[String, AnyRef], collection: util.List[util.Map[String, AnyRef]], stage: String): String = {
 		val actor = mapAsJavaMap[String, AnyRef](Map[String, AnyRef]("id" -> "Auto Creator", "type" -> "System"))
 		val context = mapAsJavaMap[String, AnyRef](Map[String, AnyRef]("pdata" -> mapAsJavaMap(Map[String, AnyRef]("id" -> "org.sunbird.platform", "ver" -> "1.0", "env" -> Platform.getString("cloud_storage.env", "dev"))), ImportConstants.CHANNEL -> metadata.getOrDefault(ImportConstants.CHANNEL, "")))
 		val objectData = mapAsJavaMap[String, AnyRef](Map[String, AnyRef]("id" -> identifier, "ver" -> metadata.get(ImportConstants.VERSION_KEY)))
 		val edata = mutable.Map[String, AnyRef]("action" -> "auto-create", "iteration" -> 1.asInstanceOf[AnyRef], ImportConstants.OBJECT_TYPE -> metadata.getOrDefault(ImportConstants.OBJECT_TYPE, "Content").asInstanceOf[String],
-			if (StringUtils.isNotBlank(source)) ImportConstants.REPOSITORY -> source else ImportConstants.IDENTIFIER -> identifier, ImportConstants.METADATA -> metadata, if (CollectionUtils.isNotEmpty(collection)) ImportConstants.COLLECTION -> collection else ImportConstants.COLLECTION -> List().asJava).asJava
+			if (StringUtils.isNotBlank(source)) ImportConstants.REPOSITORY -> source else ImportConstants.IDENTIFIER -> identifier, ImportConstants.METADATA -> metadata, if (CollectionUtils.isNotEmpty(collection)) ImportConstants.COLLECTION -> collection else ImportConstants.COLLECTION -> List().asJava, ImportConstants.STAGE -> stage).asJava
 		val kafkaEvent: String = LogTelemetryEventUtil.logInstructionEvent(actor, context, objectData, edata)
 		if (StringUtils.isBlank(kafkaEvent)) throw new ClientException(ImportErrors.BE_JOB_REQUEST_EXCEPTION, ImportErrors.ERR_INVALID_IMPORT_REQUEST_MSG)
 		kafkaEvent
 	}
 
 	def pushInstructionEvent(graphId: String, content: util.Map[String, AnyRef])(implicit oec: OntologyEngineContext): Unit = {
+		val stage = content.getOrDefault(ImportConstants.STAGE, "").toString
 		val source: String = content.getOrDefault(ImportConstants.SOURCE, "").toString
 		//TODO: Enhance identifier extraction logic for handling any query param, if present in source
 		val identifier = if (StringUtils.isNotBlank(source)) source.substring(source.lastIndexOf('/') + 1) else Identifier.getIdentifier(graphId, Identifier.getUniqueIdFromTimestamp)
 		val metadata = content.getOrDefault(ImportConstants.METADATA, new util.HashMap()).asInstanceOf[util.Map[String, AnyRef]]
 		val collection = content.getOrDefault(ImportConstants.COLLECTION, new util.ArrayList[util.Map[String, AnyRef]]()).asInstanceOf[util.ArrayList[util.Map[String, AnyRef]]]
-		val event = getInstructionEvent(identifier, source, metadata, collection)
+		val event = getInstructionEvent(identifier, source, metadata, collection, stage)
 		oec.kafkaClient.send(event, AUTO_CREATE_TOPIC_NAME)
 	}
 }

content-api/content-actors/src/test/scala/org/sunbird/content/mgr/ImportManagerTest.scala

@@ -3,7 +3,7 @@ package org.sunbird.content.mgr
 import java.util
 
 import org.apache.commons.collections4.{CollectionUtils, MapUtils}
-import org.apache.commons.lang3.StringUtils
+import org.apache.commons.lang3.{BooleanUtils, StringUtils}
 import org.scalamock.matchers.Matchers
 import org.scalamock.scalatest.AsyncMockFactory
 import org.scalatest.AsyncFlatSpec
@@ -111,7 +111,7 @@ class ImportManagerTest extends AsyncFlatSpec with Matchers with AsyncMockFactor
 				put("unitId", "do_3456")
 			}})
 		}}
-		val result = ImportManager.getInstructionEvent("do_11307822356267827219477", source, metadata, collection)
+		val result = ImportManager.getInstructionEvent("do_11307822356267827219477", source, metadata, collection, "")
 		assert(StringUtils.isNoneBlank(result))
 		val resultMap = JsonUtils.deserialize(result, classOf[util.Map[String, AnyRef]])
 		assert(MapUtils.isNotEmpty(resultMap))
@@ -124,6 +124,7 @@ class ImportManagerTest extends AsyncFlatSpec with Matchers with AsyncMockFactor
 		val request = getRequest()
 		request.putAll(new util.HashMap[String, AnyRef](){{
 			put("content", new util.HashMap[String, AnyRef](){{
+				put("stage", "Draft")
 				put("source","https://dock.sunbirded.org/api/content/v1/read/do_11307822356267827219477")
 				put("metadata", new util.HashMap[String, AnyRef](){{
 					put("name", "Test Content 2")
@@ -162,6 +163,16 @@ class ImportManagerTest extends AsyncFlatSpec with Matchers with AsyncMockFactor
 		})
 	}
 
+	"validateStage with invalid input" should "return false" in {
+		val result = ImportManager.validateStage("Flagged")
+		assert(BooleanUtils.isFalse(result))
+	}
+
+	"validateStage with valid input" should "return true" in {
+		val result = ImportManager.validateStage("Review")
+		assert(BooleanUtils.isTrue(result))
+	}
+
 	private def getRequest(): Request = {
 		val request = new Request()
 		request.setContext(new util.HashMap[String, AnyRef]() {