annif.backend.yake.YakeBackend._suggest() - Code Metrics - Inspection of "Yake integration" - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#461)

unknown

created 2021-02-11 15:25 UTC

annif.backend.yake.YakeBackend._suggest() A

↳ Parent: annif.backend.yake

Complexity

Conditions

Size

Total Lines	28
Code Lines	25

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	25
nop	3
dl	0
loc	28
rs	9.28
c	0
b	0
f	0

"""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, RDF, OWL
import rdflib
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(',')]
            valid_types = ('prefLabel', 'altLabel', 'hiddenLabel')
            for lt in label_types:
                if lt not in valid_types:
                    raise ConfigurationException(
                        f'invalid label type {lt}', backend_id=self.backend_id)
        else:
            label_types = self.params['label_types']  # The defaults
        return [getattr(SKOS, lt) for lt in label_types]

    @property
    def graph(self):
        if self._graph is None:
            self.info('Loading graph')
            self._graph = self.project.vocab.as_graph()
        return self._graph

    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.graph.subjects(RDF.type, SKOS.Concept):
            if (concept, OWL.deprecated, rdflib.Literal(True)) in self.graph:
                continue
            uri = str(concept)
            labels = self._get_concept_labels(concept, self.label_types)
            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 _get_concept_labels(self, concept, label_types):
        labels = []
        for label_type in label_types:
            for label in self.graph.objects(concept, label_type):
                if label.language == self.params['language']:
                    labels.append(label)
        return labels

    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'])

        alphanum = re.compile('[^a-zA-Z0-9]')
        if len(re.sub(alphanum, '', text)) == 0:
            return ListSuggestionResult([])

        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):
        if score < 0:
            self.debug(f'Replacing negative YAKE score {score} with zero')
            return 1.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


1			"""Annif backend using Yake keyword extraction"""
2			# TODO Mention GPLv3 license also here?
3
4			import yake
5			import os.path
6			import re
7			from collections import defaultdict
8			from rdflib.namespace import SKOS, RDF, OWL
9			import rdflib
10			import annif.util
11			from . import backend
12			from annif.suggestion import SubjectSuggestion, ListSuggestionResult
13			from annif.exception import ConfigurationException
14
15
16			class YakeBackend(backend.AnnifBackend):
17			"""Yake based backend for Annif"""
18			name = "yake"
19			needs_subject_index = False
20
21			# defaults for uninitialized instances
22			_index = None
23			_graph = None
24			INDEX_FILE = 'yake-index'
25
26			DEFAULT_PARAMETERS = {
27			'max_ngram_size': 4,
28			'deduplication_threshold': 0.9,
29			'deduplication_algo': 'levs',
30			'window_size': 1,
31			'num_keywords': 100,
32			'features': None,
33			'label_types': ['prefLabel', 'altLabel'],
34			'remove_parentheses': False
35			}
36
37			def default_params(self):
38			params = backend.AnnifBackend.DEFAULT_PARAMETERS.copy()
39			params.update(self.DEFAULT_PARAMETERS)
40			return params
41
42			@property
43			def is_trained(self):
44			return True
45
46			@property
47			def label_types(self):
48			if type(self.params['label_types']) == str: # Label types set by user
49			label_types = [lt.strip() for lt
50			in self.params['label_types'].split(',')]
51			valid_types = ('prefLabel', 'altLabel', 'hiddenLabel')
52			for lt in label_types:
53			if lt not in valid_types:
54			raise ConfigurationException(
55			f'invalid label type {lt}', backend_id=self.backend_id)
56			else:
57			label_types = self.params['label_types'] # The defaults
58			return [getattr(SKOS, lt) for lt in label_types]
59
60			@property
61			def graph(self):
62			if self._graph is None:
63			self.info('Loading graph')
64			self._graph = self.project.vocab.as_graph()
65			return self._graph
66
67			def initialize(self):
68			self._initialize_index()
69
70			def _initialize_index(self):
71			if self._index is None:
72			path = os.path.join(self.datadir, self.INDEX_FILE)
73			if os.path.exists(path):
74			self._index = self._load_index(path)
75			self.info(
76			f'Loaded index from {path} with {len(self._index)} labels')
77			else:
78			self.info('Creating index')
79			self._index = self._create_index()
80			self._save_index(path)
81			self.info(f'Created index with {len(self._index)} labels')
82
83			def _save_index(self, path):
84			with open(path, 'w', encoding='utf-8') as indexfile:
85			for label, uris in self._index.items():
86			line = label + '\t' + ' '.join(uris)
87			print(line, file=indexfile)
88
89			def _load_index(self, path):
90			index = dict()
91			with open(path, 'r', encoding='utf-8') as indexfile:
92			for line in indexfile:
93			label, uris = line.strip().split('\t')
94			index[label] = uris.split()
95			return index
96
97			def _create_index(self):
98			index = defaultdict(set)
99			for concept in self.graph.subjects(RDF.type, SKOS.Concept):
100			if (concept, OWL.deprecated, rdflib.Literal(True)) in self.graph:
101			continue
102			uri = str(concept)
103			labels = self._get_concept_labels(concept, self.label_types)
104			for label in labels:
105			label = self._normalize_label(label)
106			index[label].add(uri)
107			index.pop('', None) # Remove possible empty string entry
108			return dict(index)
109
110			def _get_concept_labels(self, concept, label_types):
111			labels = []
112			for label_type in label_types:
113			for label in self.graph.objects(concept, label_type):
114			if label.language == self.params['language']:
115			labels.append(label)
116			return labels
117
118			def _normalize_label(self, label):
119			label = str(label)
120			if annif.util.boolean(self.params['remove_parentheses']):
121			label = re.sub(r' \(.*\)', '', label)
122			lemmatized_label = self._lemmatize_phrase(label)
123			return self._sort_phrase(lemmatized_label)
124
125			def _lemmatize_phrase(self, phrase):
126			normalized = []
127			for word in phrase.split():
128			normalized.append(
129			self.project.analyzer.normalize_word(word).lower())
130			return ' '.join(normalized)
131
132			def _sort_phrase(self, phrase):
133			words = phrase.split()
134			return ' '.join(sorted(words))
135
136			def _suggest(self, text, params):
137			self.debug(
138			f'Suggesting subjects for text "{text[:20]}..." (len={len(text)})')
139			limit = int(params['limit'])
140
141			alphanum = re.compile('[^a-zA-Z0-9]')
142			if len(re.sub(alphanum, '', text)) == 0:
143			return ListSuggestionResult([])
144
145			self._kw_extractor = yake.KeywordExtractor(
146			lan=params['language'],
147			n=int(params['max_ngram_size']),
148			dedupLim=float(params['deduplication_threshold']),
149			dedupFunc=params['deduplication_algo'],
150			windowsSize=int(params['window_size']),
151			top=int(params['num_keywords']),
152			features=self.params['features'])
153			keyphrases = self._kw_extractor.extract_keywords(text)
154			suggestions = self._keyphrases2suggestions(keyphrases)
155
156			subject_suggestions = [SubjectSuggestion(
157			uri=uri,
158			label=None,
159			notation=None,
160			score=score)
161			for uri, score in suggestions[:limit] if score > 0.0]
162			return ListSuggestionResult.create_from_index(subject_suggestions,
163			self.project.subjects)
164
165			def _keyphrases2suggestions(self, keyphrases):
166			suggestions = []
167			not_matched = []
168			for kp, score in keyphrases:
169			uris = self._keyphrase2uris(kp)
170			for uri in uris:
171			suggestions.append(
172			(uri, self._transform_score(score)))
173			if not uris:
174			not_matched.append((kp, self._transform_score(score)))
175			# Remove duplicate uris, conflating the scores
176			suggestions = self._combine_suggestions(suggestions)
177			self.debug('Keyphrases not matched:\n' + '\t'.join(
178			[kp[0] + ' ' + str(kp[1]) for kp
179			in sorted(not_matched, reverse=True, key=lambda kp: kp[1])]))
180			return suggestions
181
182			def _keyphrase2uris(self, keyphrase):
183			keyphrase = self._lemmatize_phrase(keyphrase)
184			keyphrase = self._sort_phrase(keyphrase)
185			return self._index.get(keyphrase, [])
186
187			def _transform_score(self, score):
188			if score < 0:
189			self.debug(f'Replacing negative YAKE score {score} with zero')
190			return 1.0
191			return 1.0 / (score + 1)
192
193			def _combine_suggestions(self, suggestions):
194			combined_suggestions = {}
195			for uri, score in suggestions:
196			if uri not in combined_suggestions:
197			combined_suggestions[uri] = score
198			else:
199			old_score = combined_suggestions[uri]
200			combined_suggestions[uri] = self._combine_scores(
201			score, old_score)
202			return list(combined_suggestions.items())
203
204			def _combine_scores(self, score1, score2):
205			# The result is never smaller than the greater input
206			score1 = score1/2 + 0.5
207			score2 = score2/2 + 0.5
208			confl = score1 * score2 / (score1 * score2 + (1-score1) * (1-score2))
209			return (confl-0.5) * 2
210

NatLibFi / Annif

Pull Request — master (#461)

annif.backend.yake.YakeBackend._suggest() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like