Transform.kt

package org.memobase

import org.apache.jena.rdf.model.Literal
import org.apache.jena.rdf.model.ModelFactory
import org.apache.jena.rdf.model.Resource
import org.apache.jena.riot.RDFDataMgr
import org.apache.jena.riot.RDFFormat
import org.apache.kafka.streams.KeyValue
import org.apache.logging.log4j.LogManager
import org.memobase.model.MergedAddress
import org.memobase.model.MergedInstitution
import org.memobase.model.Municipality
import org.memobase.rdf.NS
import org.memobase.rdf.RDF
import org.memobase.rdf.RICO
import org.memobase.rdf.WD
import java.io.StringWriter

class Transform(private val municipalities: Map<String, Municipality>) {
    private val log = LogManager.getLogger("Transform")

    private val model = ModelFactory.createDefaultModel()
    private var resource: Resource? = null
    private var uri: String = ""

    fun createInstitution(input: MergedInstitution): Transform {
        resource = model.createResource(NS.memint + input.id)

        resource.let { valResource ->
            if (valResource != null) {
                uri = valResource.uri

                val identifier = model.createResource()
                identifier.addProperty(RDF.type, RICO.Identifier)
                identifier.addProperty(RICO.type, literal("main"))
                identifier.addProperty(RICO.identifier, literal(input.id))
                valResource.addProperty(RICO.identifiedBy, identifier)

                // TODO: proper multi language integration!
                valResource.addProperty(RICO.name, langLiteral(input.name.de, "de"))
                valResource.addProperty(RICO.name, langLiteral(input.name.fr, "fr"))
                valResource.addProperty(RICO.name, langLiteral(input.name.it, "it"))
                valResource.addProperty(RICO.descriptiveNote, langLiteral(input.description.de, "de"))
                valResource.addProperty(RICO.descriptiveNote, langLiteral(input.description.fr, "fr"))
                valResource.addProperty(RICO.descriptiveNote, langLiteral(input.description.it, "it"))

                input.addresses.forEach {
                    extractAddressField(valResource, it)
                }

                valResource.addProperty(WD.isil, literal(input.isil))
                valResource.addProperty(WD.emailAddress, literal(input.contactEmail))
                valResource.addProperty(WD.website, literal(input.website))
                valResource.addProperty(WD.onlineArchive, literal(input.onlineCatalogueLink))
            }
        }


        return this
    }

    fun write(): KeyValue<String, String> {
        return StringWriter().use { writer ->
            RDFDataMgr.write(writer, model, RDFFormat.NTRIPLES_UTF8)
            return@use KeyValue(uri, writer.toString().trim())
        }
    }

    private fun extractAddressField(resource: Resource, mergedAddress: MergedAddress) {
        val location = model.createResource()

        listOf("de", "fr", "it").forEach {
            val streetAddress = mergedAddress.addressLine1.get(it)
            val secondAddressLine = mergedAddress.addressLine2.get(it)
            val combinedStreetAddress = if (secondAddressLine.isNotEmpty()) {
                streetAddress + "\n" + secondAddressLine
            } else {
                streetAddress
            }
            val streetNumber = streetAddress.substringAfterLast(" ")
            val street = streetAddress.replace(streetNumber, "").trim()
            location.addProperty(WD.street, langLiteral(street, it))
            location.addProperty(WD.streetNumber, literal(streetNumber))
            location.addProperty(WD.streetAddress, langLiteral(combinedStreetAddress, it))
        }

        val postalCode = mergedAddress.postalCode.trim()
        val municipality = if (municipalities.containsKey(postalCode)) {
            municipalities[postalCode]
        } else {
            // the input validation in drupal should ensure that this never happens.
            log.error("Invalid postal code: $postalCode")
            null
        }
        location.addProperty(RDF.type, RICO.Place)
        location.addProperty(WD.postalCode, literal(postalCode))
        // does not enrich city, canton or cantons, if the postal code is not in the list.
        if (municipality != null) {
            // canton
            location.addProperty(WD.adminUnit, model.createResource(municipality.canton))
            // city
            location.addProperty(WD.adminUnit, model.createResource(municipality.id))
            // coordinates
            municipality.coordinates.forEach { coordinate ->
                location.addProperty(WD.coordinates, literal(coordinate))
            }
        }
        // country is currently hard coded to switzerland!
        location.addProperty(WD.country, WD.switzerland)
        resource.addProperty(RICO.hasLocation, location)
    }

    private fun langLiteral(text: String, language: String): Literal = model.createLiteral(text.trim(), language)
    private fun literal(text: String): Literal = model.createLiteral(text.trim())
}