annif.backend.yake

Complexity

Total Complexity

37

Size/Duplication

Total Lines	189
Duplicated Lines	0 %

Importance

Changes

0

18 Methods

Rating	Name	Duplication	Size	Complexity
A	YakeBackend._create_index()	0	11	3
A	YakeBackend.initialize()	0	2	1
A	YakeBackend._lemmatize_phrase()	0	6	2
A	YakeBackend._sort_phrase()	0	3	1
A	YakeBackend.default_params()	0	4	1
A	YakeBackend._load_index()	0	7	3
A	YakeBackend._validate_label_types()	0	5	3
A	YakeBackend._save_index()	0	5	3
A	YakeBackend.label_types()	0	9	2
A	YakeBackend.is_trained()	0	3	1
A	YakeBackend._initialize_index()	0	12	3
A	YakeBackend._normalize_label()	0	6	2
A	YakeBackend._keyphrases2suggestions()	0	16	5
A	YakeBackend._keyphrase2uris()	0	4	1
A	YakeBackend._suggest()	0	24	1
A	YakeBackend._combine_scores()	0	6	1
A	YakeBackend._combine_suggestions()	0	10	3
A	YakeBackend._transform_score()	0	3	1

"""Annif backend using Yake keyword extraction"""
# TODO Mention GPLv3 license also here?

import yake
import os.path
import re
from collections import defaultdict
from rdflib.namespace import SKOS
import annif.util
from . import backend
from annif.suggestion import SubjectSuggestion, ListSuggestionResult
from annif.exception import ConfigurationException


class YakeBackend(backend.AnnifBackend):
    """Yake based backend for Annif"""
    name = "yake"
    needs_subject_index = False

    # defaults for uninitialized instances
    _index = None
    _graph = None
    INDEX_FILE = 'yake-index'

    DEFAULT_PARAMETERS = {
        'max_ngram_size': 4,
        'deduplication_threshold': 0.9,
        'deduplication_algo': 'levs',
        'window_size': 1,
        'num_keywords': 100,
        'features': None,
        'label_types': ['prefLabel', 'altLabel'],
        'remove_parentheses': False
    }

    def default_params(self):
        params = backend.AnnifBackend.DEFAULT_PARAMETERS.copy()
        params.update(self.DEFAULT_PARAMETERS)
        return params

    @property
    def is_trained(self):
        return True

    @property
    def label_types(self):
        if type(self.params['label_types']) == str:  # Label types set by user
            label_types = [lt.strip() for lt
                           in self.params['label_types'].split(',')]
            self._validate_label_types(label_types)
        else:
            label_types = self.params['label_types']  # The defaults
        return [getattr(SKOS, lt) for lt in label_types]

    def _validate_label_types(self, label_types):
        for lt in label_types:
            if lt not in ('prefLabel', 'altLabel', 'hiddenLabel'):
                raise ConfigurationException(
                    f'invalid label type {lt}', backend_id=self.backend_id)

    def initialize(self):
        self._initialize_index()

    def _initialize_index(self):
        if self._index is None:
            path = os.path.join(self.datadir, self.INDEX_FILE)
            if os.path.exists(path):
                self._index = self._load_index(path)
                self.info(
                    f'Loaded index from {path} with {len(self._index)} labels')
            else:
                self.info('Creating index')
                self._index = self._create_index()
                self._save_index(path)
                self.info(f'Created index with {len(self._index)} labels')

    def _save_index(self, path):
        with open(path, 'w', encoding='utf-8') as indexfile:
            for label, uris in self._index.items():
                line = label + '\t' + ' '.join(uris)
                print(line, file=indexfile)

    def _load_index(self, path):
        index = dict()
        with open(path, 'r', encoding='utf-8') as indexfile:
            for line in indexfile:
                label, uris = line.strip().split('\t')
                index[label] = uris.split()
        return index

    def _create_index(self):
        index = defaultdict(set)
        for concept in self.project.vocab.skos_concepts:
            uri = str(concept)
            labels = self.project.vocab.get_skos_concept_labels(
                concept, self.label_types, self.params['language'])
            for label in labels:
                label = self._normalize_label(label)
                index[label].add(uri)
        index.pop('', None)  # Remove possible empty string entry
        return dict(index)

    def _normalize_label(self, label):
        label = str(label)
        if annif.util.boolean(self.params['remove_parentheses']):
            label = re.sub(r' \(.*\)', '', label)
        lemmatized_label = self._lemmatize_phrase(label)
        return self._sort_phrase(lemmatized_label)

    def _lemmatize_phrase(self, phrase):
        normalized = []
        for word in phrase.split():
            normalized.append(
                self.project.analyzer.normalize_word(word).lower())
        return ' '.join(normalized)

    def _sort_phrase(self, phrase):
        words = phrase.split()
        return ' '.join(sorted(words))

    def _suggest(self, text, params):
        self.debug(
            f'Suggesting subjects for text "{text[:20]}..." (len={len(text)})')
        limit = int(params['limit'])

        self._kw_extractor = yake.KeywordExtractor(
            lan=params['language'],
            n=int(params['max_ngram_size']),
            dedupLim=float(params['deduplication_threshold']),
            dedupFunc=params['deduplication_algo'],
            windowsSize=int(params['window_size']),
            top=int(params['num_keywords']),
            features=self.params['features'])
        keyphrases = self._kw_extractor.extract_keywords(text)
        suggestions = self._keyphrases2suggestions(keyphrases)

        subject_suggestions = [SubjectSuggestion(
                uri=uri,
                label=None,
                notation=None,
                score=score)
                for uri, score in suggestions[:limit] if score > 0.0]
        return ListSuggestionResult.create_from_index(subject_suggestions,
                                                      self.project.subjects)

    def _keyphrases2suggestions(self, keyphrases):
        suggestions = []
        not_matched = []
        for kp, score in keyphrases:
            uris = self._keyphrase2uris(kp)
            for uri in uris:
                suggestions.append(
                    (uri, self._transform_score(score)))
            if not uris:
                not_matched.append((kp, self._transform_score(score)))
        # Remove duplicate uris, conflating the scores
        suggestions = self._combine_suggestions(suggestions)
        self.debug('Keyphrases not matched:\n' + '\t'.join(
            [kp[0] + ' ' + str(kp[1]) for kp
             in sorted(not_matched, reverse=True, key=lambda kp: kp[1])]))
        return suggestions

    def _keyphrase2uris(self, keyphrase):
        keyphrase = self._lemmatize_phrase(keyphrase)
        keyphrase = self._sort_phrase(keyphrase)
        return self._index.get(keyphrase, [])

    def _transform_score(self, score):
        score = max(score, 0)
        return 1.0 / (score + 1)

    def _combine_suggestions(self, suggestions):
        combined_suggestions = {}
        for uri, score in suggestions:
            if uri not in combined_suggestions:
                combined_suggestions[uri] = score
            else:
                old_score = combined_suggestions[uri]
                combined_suggestions[uri] = self._combine_scores(
                    score, old_score)
        return list(combined_suggestions.items())

    def _combine_scores(self, score1, score2):
        # The result is never smaller than the greater input
        score1 = score1/2 + 0.5
        score2 = score2/2 + 0.5
        confl = score1 * score2 / (score1 * score2 + (1-score1) * (1-score2))
        return (confl-0.5) * 2