Commit ad271354 authored by Jonas Waeber's avatar Jonas Waeber
Browse files

Upgraded lib dependency to 2.0.0

parent 8e6ce5bb
......@@ -34,7 +34,7 @@ ext {
dependencies {
compile group: 'org.elasticsearch.client', name: 'elasticsearch-rest-high-level-client', version: '7.6.1'
implementation 'org.memobase:memobase-service-utilities:0.14.2'
implementation 'org.memobase:memobase-service-utilities:2.0.0'
// Logging Framework
implementation "org.apache.logging.log4j:log4j-api:${log4jV}"
......
......@@ -18,6 +18,7 @@
package org.memobase
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonObject
import org.apache.logging.log4j.LogManager
import org.memobase.builders.*
......@@ -26,7 +27,6 @@ import org.memobase.model.DocumentsSearchDoc
import org.memobase.model.EnrichedDigitalMetadata
import org.memobase.model.LanguageContainer
import org.memobase.model.Schema
import org.memobase.rdf.NS
class DocumentsSearchDocBuilder(private val mediaUrl: String) {
private val log = LogManager.getLogger("SearchDocTransform")
......
......@@ -18,6 +18,7 @@
package org.memobase
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonArray
import com.beust.klaxon.JsonObject
import java.util.Properties
......@@ -31,7 +32,7 @@ import org.memobase.model.FacetContainer
import org.memobase.model.InstitutionSearchDoc
import org.memobase.model.LanguageContainer
import org.memobase.model.Schema
import org.memobase.rdf.NS
class InstitutionSearchDocBuilder(path: String, appSettings: Properties) {
private val log = LogManager.getLogger("InstitutionSearchDocBuilder")
......
......@@ -18,8 +18,12 @@
package org.memobase
import ch.memobase.reporting.Report
import ch.memobase.reporting.ReportStatus
import ch.memobase.settings.SettingsLoader
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.kotlin.registerKotlinModule
import java.io.StringWriter
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.Topology
import org.apache.kafka.streams.kstream.KStream
......@@ -28,12 +32,8 @@ import org.apache.logging.log4j.LogManager
import org.memobase.helpers.Default
import org.memobase.helpers.JSON
import org.memobase.helpers.KEYS
import org.memobase.helpers.ReportStatus
import org.memobase.model.DocumentsSearchDoc
import org.memobase.model.Report
import org.memobase.model.Schema
import org.memobase.settings.SettingsLoader
import java.io.StringWriter
class KafkaTopology(private val settings: SettingsLoader) {
private val log = LogManager.getLogger("SearchDocService")
......@@ -43,7 +43,7 @@ class KafkaTopology(private val settings: SettingsLoader) {
private val searchDocTransform = DocumentsSearchDocBuilder(appSettings.getProperty(KEYS.SettingsProps.mediaUrl))
private val institutionSearchDoc =
InstitutionSearchDocBuilder(appSettings.getProperty(KEYS.SettingsProps.institutionTypeLabelsPath), appSettings)
InstitutionSearchDocBuilder(appSettings.getProperty(KEYS.SettingsProps.institutionTypeLabelsPath), appSettings)
private val jsonWriter = ObjectMapper().registerKotlinModule().writer()
......@@ -51,68 +51,73 @@ class KafkaTopology(private val settings: SettingsLoader) {
val builder = StreamsBuilder()
val stream = builder.stream<String, String>(settings.inputTopic)
val branchedStream = stream
.mapValues { value -> JSON.parse(value) }
.filter { _, value -> value.isNotEmpty() }
.mapValues { value -> JSON.unpack(value) }
.branch(
Predicate { _, value -> value.containsKey(JSON.record) },
Predicate { _, value -> value.containsKey(JSON.institution) },
Predicate { _, value -> value.containsKey(JSON.recordSet) },
Predicate { _, _ -> true }
)
.mapValues { value -> JSON.parse(value) }
.filter { _, value -> value.isNotEmpty() }
.mapValues { value -> JSON.unpack(value) }
.branch(
Predicate { _, value -> value.containsKey(JSON.record) },
Predicate { _, value -> value.containsKey(JSON.institution) },
Predicate { _, value -> value.containsKey(JSON.recordSet) },
Predicate { _, _ -> true }
)
val recordStream = branchedStream[0]
.mapValues { readOnlyKey, value ->
try {
Pair(
searchDocTransform.transform(value),
Report(readOnlyKey, "SUCCESS", "Transformed message into search doc.")
)
} catch (ex: InvalidInputException) {
Pair(DocumentsSearchDoc.DEFAULT, Report(readOnlyKey, "FAILURE", ex.localizedMessage))
.mapValues { readOnlyKey, value ->
try {
Pair(
searchDocTransform.transform(value),
Report(readOnlyKey, ReportStatus.success, "Transformed message into search doc.", Service.name)
)
} catch (ex: InvalidInputException) {
Pair(DocumentsSearchDoc.DEFAULT, Report(readOnlyKey, ReportStatus.warning, ex.localizedMessage, Service.name))
}
}
}
outputStreams(recordStream)
val institutionStream = branchedStream[1]
.mapValues { readOnlyKey, value ->
try {
Pair(
institutionSearchDoc.transform(readOnlyKey, value),
Report(readOnlyKey, "SUCCESS", "Transformed message into search doc.")
)
} catch (ex: InvalidInputException) {
Pair(Default.institutionSearchDoc, Report(readOnlyKey, "FAILURE", ex.localizedMessage))
.mapValues { readOnlyKey, value ->
try {
Pair(
institutionSearchDoc.transform(readOnlyKey, value),
Report(readOnlyKey, ReportStatus.success, "Transformed message into search doc.", Service.name)
)
} catch (ex: InvalidInputException) {
Pair(Default.institutionSearchDoc, Report(
readOnlyKey,
ReportStatus.warning,
ex.localizedMessage,
Service.name))
}
}
}
outputStreams(institutionStream)
branchedStream[3]
.mapValues { readOnlyKey, value ->
Report(
readOnlyKey,
ReportStatus.fatal,
"No record, memobase institution or record set present in input data: $value."
)
}
.to(reportTopic)
.mapValues { readOnlyKey, value ->
Report(
readOnlyKey,
ReportStatus.fatal,
"No record, memobase institution or record set present in input data: $value.",
Service.name
)
}
.to(reportTopic)
return builder.build()
}
private fun outputStreams(stream: KStream<String, Pair<Schema, Report>>) {
stream
.mapValues { value -> value.second.toJson() }
.to(reportTopic)
.mapValues { value -> value.second.toJson() }
.to(reportTopic)
stream
.filterNot { _, value -> value.second.status == "FAILURE" }
.mapValues { value -> value.first }
.mapValues { value ->
val out = StringWriter()
jsonWriter.writeValue(out, value)
out.toString()
}
.to(settings.outputTopic)
.filterNot { _, value -> value.second.status == "FAILURE" }
.mapValues { value -> value.first }
.mapValues { value ->
val out = StringWriter()
jsonWriter.writeValue(out, value)
out.toString()
}
.to(settings.outputTopic)
}
}
......@@ -18,13 +18,16 @@
package org.memobase
import ch.memobase.settings.SettingsLoader
import kotlin.system.exitProcess
import org.apache.kafka.streams.KafkaStreams
import org.apache.logging.log4j.LogManager
import org.memobase.helpers.KEYS
import org.memobase.settings.SettingsLoader
class Service(file: String = "app.yml") {
companion object {
const val name = "search-doc-service"
}
private val log = LogManager.getLogger("SearchDocService")
val settings = SettingsLoader(
......
......@@ -18,20 +18,20 @@
package org.memobase.builders
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonObject
import org.apache.logging.log4j.LogManager
import org.memobase.helpers.KEYS
import org.memobase.helpers.Extract
import org.memobase.helpers.FacetBuildHelpers
import org.memobase.helpers.KEYS
import org.memobase.model.AgentWithRelationContainer
import org.memobase.model.LanguageContainer
import org.memobase.rdf.NS
class AgentContainerBuilder(
private val identifiers: List<String>,
private val agentType: String,
private val creationRelationType: String?,
private val inputMap: Map<String, JsonObject>
private val identifiers: List<String>,
private val agentType: String,
private val creationRelationType: String?,
private val inputMap: Map<String, JsonObject>
) : IFieldBuilder {
private val log = LogManager.getLogger("AgentContainerBuilder")
private val containers = mutableListOf<AgentWithRelationContainer>()
......@@ -55,7 +55,7 @@ class AgentContainerBuilder(
override fun append(key: String, jsonObject: JsonObject): String {
val names = Extract.languageContainer("agent-name", jsonObject[KEYS.name])
val name = if (names.isNotEmpty())
names.reduce { acc, languageContainer -> acc.merge(languageContainer) }
names.reduce { acc, languageContainer -> acc.merge(languageContainer) }
else
LanguageContainer(emptyList(), emptyList(), emptyList(), listOf("NoNameFound"))
log.warn("Creation relation without names in agent ${jsonObject["@id"]}")
......@@ -82,12 +82,12 @@ class AgentContainerBuilder(
Pair(null, null)
}
containers.add(
AgentWithRelationContainer(
name,
relation,
filter = filterAndFacet.first,
facet = filterAndFacet.second
)
AgentWithRelationContainer(
name,
relation,
filter = filterAndFacet.first,
facet = filterAndFacet.second
)
)
return "Created person container for person ${jsonObject[KEYS.entityId]}"
}
......
package org.memobase.builders
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonArray
import com.beust.klaxon.JsonObject
import org.apache.logging.log4j.LogManager
import org.memobase.helpers.KEYS
import org.memobase.helpers.DateFacetBuildHelpers
import org.memobase.helpers.KEYS
import org.memobase.model.DateContainer
import org.memobase.rdf.NS
import java.lang.NumberFormatException
class DateContainerBuilder(private val containedIds: List<String>) : IFieldBuilder {
private val log = LogManager.getLogger("DateContainerBuilder")
......@@ -62,13 +61,13 @@ class DateContainerBuilder(private val containedIds: List<String>) : IFieldBuild
else -> emptyList()
}
dateContainers.add(
DateContainer(
date = date,
sort = sort,
qualifier = qualifier,
certainty = certainty,
facet = facetList
)
DateContainer(
date = date,
sort = sort,
qualifier = qualifier,
certainty = certainty,
facet = facetList
)
)
return "Transformed date to date container."
}
......
......@@ -19,20 +19,18 @@
package org.memobase.builders
import com.beust.klaxon.JsonObject
import com.fasterxml.jackson.module.kotlin.jacksonTypeRef
import org.memobase.helpers.Extract
import org.memobase.helpers.KEYS
import org.memobase.model.EnrichedFacetContainer
import org.memobase.model.LanguageContainer
import org.memobase.rdf.NS
/**
*
*/
class EnrichedFacetContainerBuilder(
private val identifiers: List<String>,
private val targetType: String,
private val nameProperty: String
private val identifiers: List<String>,
private val targetType: String,
private val nameProperty: String
) : IFieldBuilder {
private val containers = mutableListOf<EnrichedFacetContainer>()
......@@ -46,10 +44,10 @@ class EnrichedFacetContainerBuilder(
override fun append(key: String, jsonObject: JsonObject): String {
val names = Extract.languageContainer(targetType, jsonObject[nameProperty])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.fillInEmpty()
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.fillInEmpty()
val ricoType = jsonObject[KEYS.ricoType] as String?
containers.add(
EnrichedFacetContainer(LanguageContainer.EMPTY, names, ricoType)
EnrichedFacetContainer(LanguageContainer.EMPTY, names, ricoType)
)
return ""
}
......
......@@ -18,20 +18,20 @@
package org.memobase.builders
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonObject
import org.memobase.helpers.KEYS
import org.memobase.helpers.Extract
import org.memobase.helpers.KEYS
import org.memobase.model.FacetContainer
import org.memobase.rdf.NS
/**
* @param identifiers: A list of identifiers to filter on.
*/
class FacettedContainerBuilder(
private val identifiers: List<String>,
private val targetType: String,
private val nameProperty: String,
private val facetFunction: (jsonObject: JsonObject) -> Pair<String?, List<String>>
private val identifiers: List<String>,
private val targetType: String,
private val nameProperty: String,
private val facetFunction: (jsonObject: JsonObject) -> Pair<String?, List<String>>
) : IFieldBuilder {
private val containers = mutableListOf<FacetContainer>()
......@@ -44,12 +44,12 @@ class FacettedContainerBuilder(
override fun append(key: String, jsonObject: JsonObject): String {
val filterAndFacet = facetFunction(jsonObject)
containers.add(
FacetContainer(
Extract.languageContainer("", jsonObject[nameProperty])
.reduce { acc, languageContainer -> acc.merge(languageContainer) },
filter = filterAndFacet.first,
facet = filterAndFacet.second
)
FacetContainer(
Extract.languageContainer("", jsonObject[nameProperty])
.reduce { acc, languageContainer -> acc.merge(languageContainer) },
filter = filterAndFacet.first,
facet = filterAndFacet.second
)
)
return ""
}
......
......@@ -15,7 +15,6 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package org.memobase.builders
import com.beust.klaxon.JsonObject
......
......@@ -18,11 +18,11 @@
package org.memobase.builders
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonObject
import org.apache.logging.log4j.LogManager
import org.memobase.helpers.KEYS
import org.memobase.helpers.FacetBuildHelpers
import org.memobase.rdf.NS
import org.memobase.helpers.KEYS
class PersonFacetBuilder : IFieldBuilder {
private val log = LogManager.getLogger("PersonFacetBuilder")
......
......@@ -18,11 +18,11 @@
package org.memobase.builders
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonObject
import org.apache.logging.log4j.LogManager
import org.memobase.helpers.KEYS
import org.memobase.helpers.FacetBuildHelpers
import org.memobase.rdf.NS
import org.memobase.helpers.KEYS
class PlaceFacetBuilder : IFieldBuilder {
private val log = LogManager.getLogger("PersonFacetBuilder")
......
......@@ -18,11 +18,11 @@
package org.memobase.builders
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonObject
import org.memobase.helpers.KEYS
import org.memobase.helpers.Extract
import org.memobase.helpers.KEYS
import org.memobase.model.SuggestContainer
import org.memobase.rdf.NS
class SuggestContainerBuilder(private val hasSubjectIds: List<String>) : IFieldBuilder {
......@@ -53,22 +53,22 @@ class SuggestContainerBuilder(private val hasSubjectIds: List<String>) : IFieldB
return when (jsonObject[KEYS.ricoType]) {
KEYS.TitleTypes.main -> {
titles.addAll(
Extract.languageContainer("main-title", jsonObject[KEYS.title])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
Extract.languageContainer("main-title", jsonObject[KEYS.title])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
)
""
}
KEYS.TitleTypes.series -> {
seriesTitles.addAll(
Extract.languageContainer("series-title", jsonObject[KEYS.title])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
Extract.languageContainer("series-title", jsonObject[KEYS.title])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
)
""
}
KEYS.TitleTypes.broadcast -> {
broadcastTitles.addAll(
Extract.languageContainer("broadcast-title", jsonObject[KEYS.title])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
Extract.languageContainer("broadcast-title", jsonObject[KEYS.title])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
)
""
}
......@@ -78,20 +78,20 @@ class SuggestContainerBuilder(private val hasSubjectIds: List<String>) : IFieldB
private fun parseKeywords(jsonObject: JsonObject): String {
keywords.addAll(
Extract.languageContainer("keywords-concept", jsonObject[KEYS.prefLabel])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
Extract.languageContainer("keywords-concept", jsonObject[KEYS.prefLabel])
.reduce { acc, languageContainer -> acc.merge(languageContainer) }.toList()
)
return ""
}
override fun build(): List<SuggestContainer> {
return listOf(
SuggestContainer(
title = titles,
seriesTitle = seriesTitles,
broadcastTitle = broadcastTitles,
keywords = keywords
)
SuggestContainer(
title = titles,
seriesTitle = seriesTitles,
broadcastTitle = broadcastTitles,
keywords = keywords
)
)
}
}
package org.memobase.helpers
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonArray
import com.beust.klaxon.JsonObject
import com.beust.klaxon.Klaxon
import com.beust.klaxon.KlaxonException
import org.apache.logging.log4j.LogManager
import org.memobase.rdf.NS
import java.io.StringReader
import org.apache.logging.log4j.LogManager
object JSON {
......
package org.memobase.helpers
import ch.memobase.rdf.NS
import com.beust.klaxon.JsonObject
import org.apache.logging.log4j.LogManager
import org.memobase.model.FacetContainer
import org.memobase.model.LanguageContainer
import org.memobase.rdf.NS
/**
Extraction helpers for institutions and record sets.
......@@ -29,17 +29,17 @@ object Meta {
return record[key].let { items ->
when (items) {
is String -> listOf(
FacetContainer(
LanguageContainer(emptyList(), emptyList(), emptyList(), emptyList()),
null,
listOf(removeNamespace(key, items))
)
FacetContainer(
LanguageContainer(emptyList(), emptyList(), emptyList(), emptyList()),
null,
listOf(removeNamespace(key, items))
)
)
is List<*> -> items.map { item ->
FacetContainer(
LanguageContainer(emptyList(), emptyList(), emptyList(), emptyList()),
null,
listOf(removeNamespace(key, item as String))
LanguageContainer(emptyList(), emptyList(), emptyList(), emptyList()),
null,
listOf(removeNamespace(key, item as String))
)
}
else -> {
......
package org.memobase.helpers
object ReportStatus {
const val success = "SUCCESS"
const val ignored = "IGNORED"
const val warning = "WARNING"
const val fatal = "FATAL"
}
\ No newline at end of file
/*
* Media Linker
* Copyright (C) 2020 Memoriav
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package org.memobase.model
import com.beust.klaxon.Klaxon
import java.time.LocalDateTime
data class Report(
val id: String,
val status: String,
val message: String,
val step: String = "search-doc-service",
val timestamp: String = LocalDateTime.now().toString()
) {
fun toJson(): String {
return Klaxon().toJsonString(this)
}