<commit_before>package se.kb.libris.whelks.component

import groovy.util.logging.Slf4j as Log

import java.util.concurrent.*

import org.codehaus.jackson.map.ObjectMapper

import se.kb.libris.whelks.exception.*
import se.kb.libris.whelks.plugin.*
import se.kb.libris.whelks.*

@Log
abstract class BasicComponent extends BasicPlugin implements Component {

    static final ObjectMapper mapper = new ObjectMapper()
    static final String LAST_UPDATED = "last_updated"

    static final String STATE_FILE_SUFFIX = ".state"

    private Map componentState
    boolean stateUpdated

    Whelk whelk
    List contentTypes

    int batchUpdateSize = 1000 // Default. May be overriden by whelk.json

    // Plugins
    Map<String,FormatConverter> formatConverters = new HashMap<String,FormatConverter>()
    Map<String,DocumentSplitter> documentSplitters = new HashMap<String,DocumentSplitter>()
    List<LinkExpander> linkExpanders = new ArrayList<LinkExpander>()
    Map<String,Component> components = new HashMap<String,Component>()

    Listener listener
    Thread listenerThread = null
    Thread stateThread = null

    void init(String whelkId) {
        File stateFile= new File("${global.WHELK_WORK_DIR}/${whelkId}_${this.id}${STATE_FILE_SUFFIX}")
        if (stateFile.exists()) {
            log.info("Loading persisted state for [${this.id}].")
            try {
                componentState = mapper.readValue(stateFile, Map).asSynchronized()
            } catch (Exception e) {
                log.error("[${this.id}] Failed to read statefile: ${e.message}", e)
                throw new WhelkRuntimeException("Couldn't read statefile for ${whelkId}/${this.id}", e)
            }
        } else {
            componentState = [:].asSynchronized()
            componentState[LAST_UPDATED] = 0L
        }
        componentState['componentId'] = this.id
        componentState['status'] = "started"

        log.debug("[${this.id}] Loading format converters")
        for (f in plugins.findAll { it instanceof FormatConverter }) {
            formatConverters.put(f.requiredContentType, f)
        }
        log.debug("[${this.id}] Loading document splitters")
        for (d in plugins.findAll { it instanceof DocumentSplitter }) {
            documentSplitters.put(d.requiredContentType, d)
        }
        // Look for components configured for whelk. Saves having to add component to all components in plugins.
        for (c in whelk.getComponents()) {
            components.put(c.id, c)
        }

        listener = plugins.find { it instanceof Listener }
        startStateThread(whelkId)
        startListenerThread()
    }


    @Override
    public URI add(Document document) {
        try {
            long updatetime = document.timestamp
            List<Document> docs = prepareDocs([document], document.contentType)
            batchLoad(docs)
            setState(LAST_UPDATED, updatetime)
            return new URI(document.identifier)
        } catch (Exception e) {
            log.error("[${this.id}] failed to add documents. (${e.message})", e)
            throw e
        }
    }

    public void bulkAdd(List<Document> docs, String contentType, long updatetime = -1)  {
        log.debug("[${this.id}] bulkAdd called with ${docs.size()} documents.")
        try {
            if (updatetime < 0) {
                updatetime = new Date().getTime()
            }
            log.debug("Now is ${new Date().getTime()}, updatetime is $updatetime")
            docs = prepareDocs(docs, contentType)
            log.trace("[${this.id}] Calling batchload on ${this.id} with batch of ${docs.size()}")
            batchLoad(docs)
            setState(LAST_UPDATED, updatetime)
        } catch (Exception e) {
            log.error("[${this.id}] failed to add documents. (${e.message})", e)
            throw e
        }
    }

    @Override
    boolean handlesContent(String ctype) {
        return (ctype == "*/*" || !this.contentTypes || this.contentTypes.contains(ctype))
    }

    protected abstract void batchLoad(List<Document> docs);

    List<Document> prepareDocs(final List<Document> documents, String contentType) {
        FormatConverter fc = formatConverters.get(contentType)
        DocumentSplitter preSplitter = documentSplitters.get(contentType)

        DocumentSplitter postSplitter = documentSplitters.get(contentType)
        log.trace("fc: $fc")
        log.trace("postSplitter: $postSplitter")
        if (!postSplitter && fc) {
            postSplitter = documentSplitters.get(fc.resultContentType)
        }
        List docs = []
        if (fc || preSplitter || postSplitter) {
            for (doc in documents) {
                log.trace("[${this.id}] Calling prepare on doc ${doc.identifier}");
                if (preSplitter) {
                    log.trace("[${this.id}] Running preSplitter")
                    for (d in preSplitter.split(doc)) {
                        if (fc) {
                            log.trace(" ... with conversion")
                            docs << fc.convert(d)
                        } else {
                            log.trace(" ... without conversion")
                            docs << d
                        }
                    }
                } else if (fc) {
                    log.trace("[${this.id}] Adding document after conversion.")
                    docs.add(fc.convert(doc))
                } else {
                    log.trace("[${this.id}] Adding document without conversion.")
                    docs.add(doc)
                }
                if (postSplitter) {
                    log.trace("[${this.id}] Running postSplitter")
                    if (docs.size() > 0) {
                        List<Document> convertedDocs = []
                        for (d in docs) {
                            convertedDocs.addAll(postSplitter.split(d))
                        }
                        docs = convertedDocs
                    }
                }
            }
            log.debug("[${this.id}] Returning document list of size ${docs.size()}.")
        }
        if (linkExpanders.size() > 0) {
            log.trace("Link expanders configured. Must loop through documents.")
            if (docs.size() == 0) {
                docs = documents
            }
            for (doc in docs) {
                LinkExpander expander = getLinkExpanderFor(doc)
                if (expander) {
                    log.debug("Expanding ${doc.identifier}")
                    doc = expander.expand(doc)
                }
            }
        }
        if (docs.size() > 0) {
            return docs
        }
        return documents
    }

    LinkExpander getLinkExpanderFor(Document doc) { return linkExpanders.find { it.valid(doc) } }

    public Map getComponentState() { return componentState.asImmutable() }

    void setState(key, value) {
        synchronized(componentState) {
            componentState.put(key, value)
            stateUpdated = true
        }
        if (key == LAST_UPDATED && listener) {
            log.debug("[${this.id}] Notifying listeners ..")
            listener.registerUpdate(this.id, value)
        }
    }

    void startStateThread(String whelkId) {
        log.info("[${this.id}] Starting thread to periodically save state.")
        if (!stateThread) {
            stateThread = Thread.start {
                boolean ok = true
                while(ok) {
                    try {
                        if (componentState && stateUpdated) {
                            log.trace("[${whelkId}-${this.id}] Saving state ...")
                            mapper.writeValue(stateFile, componentState)
                            stateUpdated = false
                        }
                        Thread.sleep(20000)
                    } catch (Exception e) {
                        ok = false
                        log.error("[${this.id}] Failed to write statefile: ${e.message}", e)
                        throw new WhelkRuntimeException("Couldn't write statefile for ${whelkId}/${this.id}", e)
                    }
                }
            }
        }
    }

    void startListenerThread() {
        if (listener && listener.hasQueue(this.id) && !listenerThread) {

            listenerThread = Thread.start {
                log.debug("[${this.id}] Starting notification listener.")
                boolean ok = true
                try {
                    while (ok) {
                        log.debug("[${this.id}] Waiting for update ...")<commit_msg>fixed bugu<commit_after>package se.kb.libris.whelks.component

import groovy.util.logging.Slf4j as Log

import java.util.concurrent.*

import org.codehaus.jackson.map.ObjectMapper

import se.kb.libris.whelks.exception.*
import se.kb.libris.whelks.plugin.*
import se.kb.libris.whelks.*

@Log
abstract class BasicComponent extends BasicPlugin implements Component {

    static final ObjectMapper mapper = new ObjectMapper()
    static final String LAST_UPDATED = "last_updated"

    static final String STATE_FILE_SUFFIX = ".state"

    private Map componentState
    boolean stateUpdated

    Whelk whelk
    List contentTypes

    int batchUpdateSize = 1000 // Default. May be overriden by whelk.json

    // Plugins
    Map<String,FormatConverter> formatConverters = new HashMap<String,FormatConverter>()
    Map<String,DocumentSplitter> documentSplitters = new HashMap<String,DocumentSplitter>()
    List<LinkExpander> linkExpanders = new ArrayList<LinkExpander>()
    Map<String,Component> components = new HashMap<String,Component>()

    Listener listener
    Thread listenerThread = null
    Thread stateThread = null

    void init(String whelkId) {
        File stateFile= new File("${global.WHELK_WORK_DIR}/${whelkId}_${this.id}${STATE_FILE_SUFFIX}")
        if (stateFile.exists()) {
            log.info("Loading persisted state for [${this.id}].")
            try {
                componentState = mapper.readValue(stateFile, Map).asSynchronized()
            } catch (Exception e) {
                log.error("[${this.id}] Failed to read statefile: ${e.message}", e)
                throw new WhelkRuntimeException("Couldn't read statefile for ${whelkId}/${this.id}", e)
            }
        } else {
            componentState = [:].asSynchronized()
            componentState[LAST_UPDATED] = 0L
        }
        componentState['componentId'] = this.id
        componentState['status'] = "started"

        log.debug("[${this.id}] Loading format converters")
        for (f in plugins.findAll { it instanceof FormatConverter }) {
            formatConverters.put(f.requiredContentType, f)
        }
        log.debug("[${this.id}] Loading document splitters")
        for (d in plugins.findAll { it instanceof DocumentSplitter }) {
            documentSplitters.put(d.requiredContentType, d)
        }
        // Look for components configured for whelk. Saves having to add component to all components in plugins.
        for (c in whelk.getComponents()) {
            components.put(c.id, c)
        }

        listener = plugins.find { it instanceof Listener }
        startStateThread(stateFile, whelkId)
        startListenerThread()
    }


    @Override
    public URI add(Document document) {
        try {
            long updatetime = document.timestamp
            List<Document> docs = prepareDocs([document], document.contentType)
            batchLoad(docs)
            setState(LAST_UPDATED, updatetime)
            return new URI(document.identifier)
        } catch (Exception e) {
            log.error("[${this.id}] failed to add documents. (${e.message})", e)
            throw e
        }
    }

    public void bulkAdd(List<Document> docs, String contentType, long updatetime = -1)  {
        log.debug("[${this.id}] bulkAdd called with ${docs.size()} documents.")
        try {
            if (updatetime < 0) {
                updatetime = new Date().getTime()
            }
            log.debug("Now is ${new Date().getTime()}, updatetime is $updatetime")
            docs = prepareDocs(docs, contentType)
            log.trace("[${this.id}] Calling batchload on ${this.id} with batch of ${docs.size()}")
            batchLoad(docs)
            setState(LAST_UPDATED, updatetime)
        } catch (Exception e) {
            log.error("[${this.id}] failed to add documents. (${e.message})", e)
            throw e
        }
    }

    @Override
    boolean handlesContent(String ctype) {
        return (ctype == "*/*" || !this.contentTypes || this.contentTypes.contains(ctype))
    }

    protected abstract void batchLoad(List<Document> docs);

    List<Document> prepareDocs(final List<Document> documents, String contentType) {
        FormatConverter fc = formatConverters.get(contentType)
        DocumentSplitter preSplitter = documentSplitters.get(contentType)

        DocumentSplitter postSplitter = documentSplitters.get(contentType)
        log.trace("fc: $fc")
        log.trace("postSplitter: $postSplitter")
        if (!postSplitter && fc) {
            postSplitter = documentSplitters.get(fc.resultContentType)
        }
        List docs = []
        if (fc || preSplitter || postSplitter) {
            for (doc in documents) {
                log.trace("[${this.id}] Calling prepare on doc ${doc.identifier}");
                if (preSplitter) {
                    log.trace("[${this.id}] Running preSplitter")
                    for (d in preSplitter.split(doc)) {
                        if (fc) {
                            log.trace(" ... with conversion")
                            docs << fc.convert(d)
                        } else {
                            log.trace(" ... without conversion")
                            docs << d
                        }
                    }
                } else if (fc) {
                    log.trace("[${this.id}] Adding document after conversion.")
                    docs.add(fc.convert(doc))
                } else {
                    log.trace("[${this.id}] Adding document without conversion.")
                    docs.add(doc)
                }
                if (postSplitter) {
                    log.trace("[${this.id}] Running postSplitter")
                    if (docs.size() > 0) {
                        List<Document> convertedDocs = []
                        for (d in docs) {
                            convertedDocs.addAll(postSplitter.split(d))
                        }
                        docs = convertedDocs
                    }
                }
            }
            log.debug("[${this.id}] Returning document list of size ${docs.size()}.")
        }
        if (linkExpanders.size() > 0) {
            log.trace("Link expanders configured. Must loop through documents.")
            if (docs.size() == 0) {
                docs = documents
            }
            for (doc in docs) {
                LinkExpander expander = getLinkExpanderFor(doc)
                if (expander) {
                    log.debug("Expanding ${doc.identifier}")
                    doc = expander.expand(doc)
                }
            }
        }
        if (docs.size() > 0) {
            return docs
        }
        return documents
    }

    LinkExpander getLinkExpanderFor(Document doc) { return linkExpanders.find { it.valid(doc) } }

    public Map getComponentState() { return componentState.asImmutable() }

    void setState(key, value) {
        synchronized(componentState) {
            componentState.put(key, value)
            stateUpdated = true
        }
        if (key == LAST_UPDATED && listener) {
            log.debug("[${this.id}] Notifying listeners ..")
            listener.registerUpdate(this.id, value)
        }
    }

    void startStateThread(File stateFile, String whelkId) {
        log.info("[${this.id}] Starting thread to periodically save state.")
        if (!stateThread) {
            stateThread = Thread.start {
                boolean ok = true
                while(ok) {
                    try {
                        if (componentState && stateUpdated) {
                            log.trace("[${whelkId}-${this.id}] Saving state ...")
                            mapper.writeValue(stateFile, componentState)
                            stateUpdated = false
                        }
                        Thread.sleep(20000)
                    } catch (Exception e) {
                        ok = false
                        log.error("[${this.id}] Failed to write statefile: ${e.message}", e)
                        throw new WhelkRuntimeException("Couldn't write statefile for ${whelkId}/${this.id}", e)
                    }
                }
            }
        }
    }

    void startListenerThread() {
        if (listener && listener.hasQueue(this.id) && !listenerThread) {

            listenerThread = Thread.start {
                log.debug("[${this.id}] Starting notification listener.")
                boolean ok = true
                try {
                    while (ok) {
                        log.debug("[${this.id}] Waiting for update ...")
{'fraction_non_alphanumeric': 0.11078683166005136, 'fraction_numerical': 0.001984590240485641, 'mean_word_length': 1.7087747035573122, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4760231', 'n_tokens_mistral': 4605, 'n_tokens_neox': 4404, 'n_words': 1044}