annif.cli_util.open_documents() - Code Metrics - Inspection of "Issue690 use python type hints" - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#708)

by Juho

created 2023-05-25 11:26 UTC

annif.cli_util.open_documents() B

↳ Parent: annif.cli_util

Complexity

Conditions

Size

Total Lines	31
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	20
nop	4
dl	0
loc	31
rs	8.9332
c	0
b	0
f	0

"""Utility functions for Annif CLI commands"""
from __future__ import annotations

import collections
import itertools
import os
import sys
from typing import TYPE_CHECKING, DefaultDict, Dict, List, Optional, Tuple, Union

import click
import click_log
from flask import current_app

import annif
from annif.exception import ConfigurationException
from annif.project import Access

if TYPE_CHECKING:
    from datetime import datetime
    from io import TextIOWrapper

    from click.core import Argument, Context, Option

    from annif.corpus.document import DocumentCorpus, DocumentList
    from annif.corpus.subject import SubjectIndex
    from annif.project import AnnifProject
    from annif.suggestion import SuggestionResult
    from annif.vocab import AnnifVocabulary

logger = annif.logger


def _set_project_config_file_path(
    ctx: Context, param: Option, value: Optional[str]
) -> None:
    """Override the default path or the path given in env by CLI option"""
    with ctx.obj.load_app().app_context():
        if value:
            current_app.config["PROJECTS_CONFIG_PATH"] = value


def common_options(f):
    """Decorator to add common options for all CLI commands"""
    f = click.option(
        "-p",
        "--projects",
        help="Set path to project configuration file or directory",
        type=click.Path(dir_okay=True, exists=True),
        callback=_set_project_config_file_path,
        expose_value=False,
        is_eager=True,
    )(f)
    return click_log.simple_verbosity_option(logger)(f)


def project_id(f):
    """Decorator to add a project ID parameter to a CLI command"""
    return click.argument("project_id", shell_complete=complete_param)(f)


def backend_param_option(f):
    """Decorator to add an option for CLI commands to override BE parameters"""
    return click.option(
        "--backend-param",
        "-b",
        multiple=True,
        help="Override backend parameter of the config file. "
        + "Syntax: `-b <backend>.<parameter>=<value>`.",
    )(f)


def docs_limit_option(f):
    """Decorator to add an option for CLI commands to limit the number of documents to
    use"""
    return click.option(
        "--docs-limit",
        "-d",
        default=None,
        type=click.IntRange(0, None),
        help="Maximum number of documents to use",
    )(f)


def get_project(project_id: str) -> AnnifProject:
    """
    Helper function to get a project by ID and bail out if it doesn't exist"""
    try:
        return annif.registry.get_project(project_id, min_access=Access.private)
    except ValueError:
        click.echo("No projects found with id '{0}'.".format(project_id), err=True)
        sys.exit(1)


def get_vocab(vocab_id: str) -> AnnifVocabulary:
    """
    Helper function to get a vocabulary by ID and bail out if it doesn't
    exist"""
    try:
        return annif.registry.get_vocab(vocab_id, min_access=Access.private)
    except ValueError:
        click.echo(f"No vocabularies found with the id '{vocab_id}'.", err=True)
        sys.exit(1)


def make_list_template(*rows) -> str:
    """Helper function to create a template for a list of entries with fields of
    variable width. The width of each field is determined by the longest item in the
    field in the given rows."""

    max_field_widths = collections.defaultdict(int)
    for row in rows:
        for field_ind, item in enumerate(row):
            max_field_widths[field_ind] = max(max_field_widths[field_ind], len(item))

    return "  ".join(
        [
            f"{{{field_ind}: <{field_width}}}"
            for field_ind, field_width in max_field_widths.items()
        ]
    )


def format_datetime(dt: Optional[datetime]) -> str:
    """Helper function to format a datetime object as a string in the local time."""
    if dt is None:
        return "-"
    return dt.astimezone().strftime("%Y-%m-%d %H:%M:%S")


def open_documents(
    paths: Union[Tuple[str, ...], Tuple[()]],
    subject_index: SubjectIndex,
    vocab_lang: str,
    docs_limit: Optional[int],
) -> DocumentCorpus:
    """Helper function to open a document corpus from a list of pathnames,
    each of which is either a TSV file or a directory of TXT files. For
    directories with subjects in TSV files, the given vocabulary language
    will be used to convert subject labels into URIs. The corpus will be
    returned as an instance of DocumentCorpus or LimitingDocumentCorpus."""

    def open_doc_path(path, subject_index):
        """open a single path and return it as a DocumentCorpus"""
        if os.path.isdir(path):
            return annif.corpus.DocumentDirectory(
                path, subject_index, vocab_lang, require_subjects=True
            )
        return annif.corpus.DocumentFile(path, subject_index)

    if len(paths) == 0:
        logger.warning("Reading empty file")
        docs = open_doc_path(os.path.devnull, subject_index)
    elif len(paths) == 1:
        docs = open_doc_path(paths[0], subject_index)
    else:
        corpora = [open_doc_path(path, subject_index) for path in paths]
        docs = annif.corpus.CombinedCorpus(corpora)
    if docs_limit is not None:
        docs = annif.corpus.LimitingDocumentCorpus(docs, docs_limit)
    return docs


def open_text_documents(
    paths: Tuple[str, ...], docs_limit: Optional[int]
) -> DocumentList:
    """
    Helper function to read text documents from the given file paths. Returns a
    DocumentList object with Documents having no subjects. If a path is "-", the
    document text is read from standard input. The maximum number of documents to read
    is set by docs_limit parameter.
    """

    def _docs(paths):
        for path in paths:
            if path == "-":
                doc = annif.corpus.Document(text=sys.stdin.read(), subject_set=None)
            else:
                with open(path, errors="replace", encoding="utf-8-sig") as docfile:
                    doc = annif.corpus.Document(text=docfile.read(), subject_set=None)
            yield doc

    return annif.corpus.DocumentList(_docs(paths[:docs_limit]))


def show_hits(
    hits: SuggestionResult,
    project: AnnifProject,
    lang: str,
    file: Optional[TextIOWrapper] = None,
) -> None:
    """
    Print subject suggestions to the console or a file. The suggestions are displayed as
    a table, with one row per hit. Each row contains the URI, label, possible notation,
    and score of the suggestion. The label is given in the specified language.
    """
    template = "<{}>\t{}\t{:.04f}"
    for hit in hits:
        subj = project.subjects[hit.subject_id]
        line = template.format(
            subj.uri,
            "\t".join(filter(None, (subj.labels[lang], subj.notation))),
            hit.score,
        )
        click.echo(line, file=file)


def parse_backend_params(
    backend_param: Union[Tuple[str, ...], Tuple[()]], project: AnnifProject
) -> DefaultDict[str, Dict[str, str]]:
    """Parse a list of backend parameters given with the --backend-param
    option into a nested dict structure"""
    backend_params = collections.defaultdict(dict)
    for beparam in backend_param:
        backend, param = beparam.split(".", 1)
        key, val = param.split("=", 1)
        _validate_backend_params(backend, beparam, project)
        backend_params[backend][key] = val
    return backend_params


def _validate_backend_params(backend: str, beparam: str, project: AnnifProject) -> None:
    if backend != project.config["backend"]:
        raise ConfigurationException(
            'The backend {} in CLI option "-b {}" not matching the project'
            " backend {}.".format(backend, beparam, project.config["backend"])
        )


def generate_filter_params(filter_batch_max_limit: int) -> List[Tuple[int, float]]:
    limits = range(1, filter_batch_max_limit + 1)
    thresholds = [i * 0.05 for i in range(20)]
    return list(itertools.product(limits, thresholds))


def _get_completion_choices(
    param: Argument,
) -> Dict[str, Union[AnnifVocabulary, AnnifProject]]:
    if param.name == "project_id":
        return annif.registry.get_projects()
    elif param.name == "vocab_id":
        return annif.registry.get_vocabs()
    else:
        return []


def complete_param(ctx: Context, param: Argument, incomplete: str) -> List[str]:
    with ctx.obj.load_app().app_context():
        return [
            choice
            for choice in _get_completion_choices(param)
            if choice.startswith(incomplete)
        ]


1			"""Utility functions for Annif CLI commands"""
2			from __future__ import annotations
3
4			import collections
5			import itertools
6			import os
7			import sys
8			from typing import TYPE_CHECKING, DefaultDict, Dict, List, Optional, Tuple, Union
9
10			import click
11			import click_log
12			from flask import current_app
13
14			import annif
15			from annif.exception import ConfigurationException
16			from annif.project import Access
17
18			if TYPE_CHECKING:
19			from datetime import datetime
20			from io import TextIOWrapper
21
22			from click.core import Argument, Context, Option
23
24			from annif.corpus.document import DocumentCorpus, DocumentList
25			from annif.corpus.subject import SubjectIndex
26			from annif.project import AnnifProject
27			from annif.suggestion import SuggestionResult
28			from annif.vocab import AnnifVocabulary
29
30			logger = annif.logger
31
32
33			def _set_project_config_file_path(
34			ctx: Context, param: Option, value: Optional[str]
35			) -> None:
36			"""Override the default path or the path given in env by CLI option"""
37			with ctx.obj.load_app().app_context():
38			if value:
39			current_app.config["PROJECTS_CONFIG_PATH"] = value
40
41
42			def common_options(f):
43			"""Decorator to add common options for all CLI commands"""
44			f = click.option(
45			"-p",
46			"--projects",
47			help="Set path to project configuration file or directory",
48			type=click.Path(dir_okay=True, exists=True),
49			callback=_set_project_config_file_path,
50			expose_value=False,
51			is_eager=True,
52			)(f)
53			return click_log.simple_verbosity_option(logger)(f)
54
55
56			def project_id(f):
57			"""Decorator to add a project ID parameter to a CLI command"""
58			return click.argument("project_id", shell_complete=complete_param)(f)
59
60
61			def backend_param_option(f):
62			"""Decorator to add an option for CLI commands to override BE parameters"""
63			return click.option(
64			"--backend-param",
65			"-b",
66			multiple=True,
67			help="Override backend parameter of the config file. "
68			+ "Syntax: `-b <backend>.<parameter>=<value>`.",
69			)(f)
70
71
72			def docs_limit_option(f):
73			"""Decorator to add an option for CLI commands to limit the number of documents to
74			use"""
75			return click.option(
76			"--docs-limit",
77			"-d",
78			default=None,
79			type=click.IntRange(0, None),
80			help="Maximum number of documents to use",
81			)(f)
82
83
84			def get_project(project_id: str) -> AnnifProject:
85			"""
86			Helper function to get a project by ID and bail out if it doesn't exist"""
87			try:
88			return annif.registry.get_project(project_id, min_access=Access.private)
89			except ValueError:
90			click.echo("No projects found with id '{0}'.".format(project_id), err=True)
91			sys.exit(1)
92
93
94			def get_vocab(vocab_id: str) -> AnnifVocabulary:
95			"""
96			Helper function to get a vocabulary by ID and bail out if it doesn't
97			exist"""
98			try:
99			return annif.registry.get_vocab(vocab_id, min_access=Access.private)
100			except ValueError:
101			click.echo(f"No vocabularies found with the id '{vocab_id}'.", err=True)
102			sys.exit(1)
103
104
105			def make_list_template(*rows) -> str:
106			"""Helper function to create a template for a list of entries with fields of
107			variable width. The width of each field is determined by the longest item in the
108			field in the given rows."""
109
110			max_field_widths = collections.defaultdict(int)
111			for row in rows:
112			for field_ind, item in enumerate(row):
113			max_field_widths[field_ind] = max(max_field_widths[field_ind], len(item))
114
115			return " ".join(
116			[
117			f"{{{field_ind}: <{field_width}}}"
118			for field_ind, field_width in max_field_widths.items()
119			]
120			)
121
122
123			def format_datetime(dt: Optional[datetime]) -> str:
124			"""Helper function to format a datetime object as a string in the local time."""
125			if dt is None:
126			return "-"
127			return dt.astimezone().strftime("%Y-%m-%d %H:%M:%S")
128
129
130			def open_documents(
131			paths: Union[Tuple[str, ...], Tuple[()]],
132			subject_index: SubjectIndex,
133			vocab_lang: str,
134			docs_limit: Optional[int],
135			) -> DocumentCorpus:
136			"""Helper function to open a document corpus from a list of pathnames,
137			each of which is either a TSV file or a directory of TXT files. For
138			directories with subjects in TSV files, the given vocabulary language
139			will be used to convert subject labels into URIs. The corpus will be
140			returned as an instance of DocumentCorpus or LimitingDocumentCorpus."""
141
142			def open_doc_path(path, subject_index):
143			"""open a single path and return it as a DocumentCorpus"""
144			if os.path.isdir(path):
145			return annif.corpus.DocumentDirectory(
146			path, subject_index, vocab_lang, require_subjects=True
147			)
148			return annif.corpus.DocumentFile(path, subject_index)
149
150			if len(paths) == 0:
151			logger.warning("Reading empty file")
152			docs = open_doc_path(os.path.devnull, subject_index)
153			elif len(paths) == 1:
154			docs = open_doc_path(paths[0], subject_index)
155			else:
156			corpora = [open_doc_path(path, subject_index) for path in paths]
157			docs = annif.corpus.CombinedCorpus(corpora)
158			if docs_limit is not None:
159			docs = annif.corpus.LimitingDocumentCorpus(docs, docs_limit)
160			return docs
161
162
163			def open_text_documents(
164			paths: Tuple[str, ...], docs_limit: Optional[int]
165			) -> DocumentList:
166			"""
167			Helper function to read text documents from the given file paths. Returns a
168			DocumentList object with Documents having no subjects. If a path is "-", the
169			document text is read from standard input. The maximum number of documents to read
170			is set by docs_limit parameter.
171			"""
172
173			def _docs(paths):
174			for path in paths:
175			if path == "-":
176			doc = annif.corpus.Document(text=sys.stdin.read(), subject_set=None)
177			else:
178			with open(path, errors="replace", encoding="utf-8-sig") as docfile:
179			doc = annif.corpus.Document(text=docfile.read(), subject_set=None)
180			yield doc
181
182			return annif.corpus.DocumentList(_docs(paths[:docs_limit]))
183
184
185			def show_hits(
186			hits: SuggestionResult,
187			project: AnnifProject,
188			lang: str,
189			file: Optional[TextIOWrapper] = None,
190			) -> None:
191			"""
192			Print subject suggestions to the console or a file. The suggestions are displayed as
193			a table, with one row per hit. Each row contains the URI, label, possible notation,
194			and score of the suggestion. The label is given in the specified language.
195			"""
196			template = "<{}>\t{}\t{:.04f}"
197			for hit in hits:
198			subj = project.subjects[hit.subject_id]
199			line = template.format(
200			subj.uri,
201			"\t".join(filter(None, (subj.labels[lang], subj.notation))),
202			hit.score,
203			)
204			click.echo(line, file=file)
205
206
207			def parse_backend_params(
208			backend_param: Union[Tuple[str, ...], Tuple[()]], project: AnnifProject
209			) -> DefaultDict[str, Dict[str, str]]:
210			"""Parse a list of backend parameters given with the --backend-param
211			option into a nested dict structure"""
212			backend_params = collections.defaultdict(dict)
213			for beparam in backend_param:
214			backend, param = beparam.split(".", 1)
215			key, val = param.split("=", 1)
216			_validate_backend_params(backend, beparam, project)
217			backend_params[backend][key] = val
218			return backend_params
219
220
221			def _validate_backend_params(backend: str, beparam: str, project: AnnifProject) -> None:
222			if backend != project.config["backend"]:
223			raise ConfigurationException(
224			'The backend {} in CLI option "-b {}" not matching the project'
225			" backend {}.".format(backend, beparam, project.config["backend"])
226			)
227
228
229			def generate_filter_params(filter_batch_max_limit: int) -> List[Tuple[int, float]]:
230			limits = range(1, filter_batch_max_limit + 1)
231			thresholds = [i * 0.05 for i in range(20)]
232			return list(itertools.product(limits, thresholds))
233
234
235			def _get_completion_choices(
236			param: Argument,
237			) -> Dict[str, Union[AnnifVocabulary, AnnifProject]]:
238			if param.name == "project_id":
239			return annif.registry.get_projects()
240			elif param.name == "vocab_id":
241			return annif.registry.get_vocabs()
242			else:
243			return []
244
245
246			def complete_param(ctx: Context, param: Argument, incomplete: str) -> List[str]:
247			with ctx.obj.load_app().app_context():
248			return [
249			choice
250			for choice in _get_completion_choices(param)
251			if choice.startswith(incomplete)
252			]
253

NatLibFi / Annif

Pull Request — main (#708)

annif.cli_util.open_documents() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like