annif.cli_util.show_hits() - Code Metrics - Inspection of "Implement `annif upload` and `annif download` comm..." - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#762)

by Juho

created 2024-03-08 14:56 UTC

annif.cli_util.show_hits() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	20
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	13
nop	4
dl	0
loc	20
rs	9.75
c	0
b	0
f	0

"""Utility functions for Annif CLI commands"""

from __future__ import annotations

import binascii
import collections
import configparser
import importlib
import io
import itertools
import os
import pathlib
import shutil
import sys
import tempfile
import time
import zipfile
from fnmatch import fnmatch
from typing import TYPE_CHECKING

import click
import click_log
from flask import current_app

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

if TYPE_CHECKING:
    from datetime import datetime

    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: str | None
) -> 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: datetime | None) -> 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: tuple[str, ...],
    subject_index: SubjectIndex,
    vocab_lang: str,
    docs_limit: int | None,
) -> 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: int | None) -> 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: io.TextIOWrapper | None = 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: tuple[str, ...] | tuple[()], project: AnnifProject
) -> collections.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 _is_train_file(fname):
    train_file_patterns = ("-train", "tmp-")
    for pat in train_file_patterns:
        if pat in fname:
            return True
    return False


def archive_dir(data_dir):
    fp = tempfile.TemporaryFile()
    path = pathlib.Path(data_dir)
    fpaths = [fpath for fpath in path.glob("**/*") if not _is_train_file(fpath.name)]
    with zipfile.ZipFile(fp, mode="w") as zfile:
        zfile.comment = bytes(
            f"Archived by Annif {importlib.metadata.version('annif')}",
            encoding="utf-8",
        )
        for fpath in fpaths:
            logger.debug(f"Adding {fpath}")
            arcname = os.path.join(*fpath.parts[1:])
            zfile.write(fpath, arcname=arcname)
    fp.seek(0)
    return fp


def write_config(project):
    fp = tempfile.TemporaryFile(mode="w+t")
    config = configparser.ConfigParser()
    config[project.project_id] = project.config
    config.write(fp)  # This needs tempfile in text mode
    fp.seek(0)
    # But for upload fobj needs to be in binary mode
    return io.BytesIO(fp.read().encode("utf8"))


def upload_to_hf_hub(fileobj, filename, repo_id, token, commit_message):
    from huggingface_hub import HfApi
    from huggingface_hub.utils import HfHubHTTPError, HFValidationError

    api = HfApi()
    try:
        api.upload_file(
            path_or_fileobj=fileobj,
            path_in_repo=filename,
            repo_id=repo_id,
            token=token,
            commit_message=commit_message,
        )
    except (HfHubHTTPError, HFValidationError) as err:
        raise OperationFailedException(str(err))


def get_selected_project_ids_from_hf_hub(project_ids_pattern, repo_id, token, revision):
    all_repo_file_paths = _list_files_in_hf_hub(repo_id, token, revision)
    return [
        path.rsplit(".zip")[0].split("projects/")[1]
        for path in all_repo_file_paths
        if fnmatch(path, f"projects/{project_ids_pattern}.zip")
    ]


def _list_files_in_hf_hub(repo_id, token, revision):
    from huggingface_hub import list_repo_files
    from huggingface_hub.utils import HfHubHTTPError, HFValidationError

    try:
        return [
            repofile
            for repofile in list_repo_files(
                repo_id=repo_id, token=token, revision=revision
            )
        ]
    except (HfHubHTTPError, HFValidationError) as err:
        raise OperationFailedException(str(err))


def download_from_hf_hub(filename, repo_id, token, revision):
    from huggingface_hub import hf_hub_download
    from huggingface_hub.utils import HfHubHTTPError, HFValidationError

    try:
        return hf_hub_download(
            repo_id=repo_id,
            filename=filename,
            token=token,
            revision=revision,
        )
    except (HfHubHTTPError, HFValidationError) as err:
        raise OperationFailedException(str(err))


def unzip(src_path, force):
    datadir = current_app.config["DATADIR"]
    with zipfile.ZipFile(src_path, "r") as zfile:
        logger.debug(
            f"Extracting archive {src_path}; archive comment: "
            f"\"{str(zfile.comment, encoding='utf-8')}\""
        )
        for member in zfile.infolist():
            dest_path = os.path.join(datadir, member.filename)
            if os.path.exists(dest_path) and not force:
                if _are_identical(member, dest_path):
                    logger.debug(f"Skipping unzip to {dest_path}; already in place")
                else:
                    click.echo(f"Not overwriting {dest_path} (use --force to override)")
            else:
                logger.debug(f"Unzipping to {dest_path}")
                zfile.extract(member, path=datadir)
                date_time = time.mktime(member.date_time + (0, 0, -1))
                os.utime(dest_path, (date_time, date_time))


def move_project_config(src_path, force):
    dest_path = os.path.join("projects.d", os.path.basename(src_path))
    if os.path.exists(dest_path) and not force:
        if _compute_crc32(dest_path) == _compute_crc32(src_path):
            logger.debug(
                f"Skipping move of {os.path.basename(src_path)}; already in place"
            )
        else:
            click.echo(f"Not overwriting {dest_path} (use --force to override)")
    else:
        logger.debug(f"Moving to {dest_path}")
        shutil.copy(src_path, dest_path)


def _are_identical(member, dest_path):
    path_crc = _compute_crc32(dest_path)
    return path_crc == member.CRC


def _compute_crc32(path):
    if os.path.isdir(path):
        return 0

    size = 1024 * 1024 * 10  # 10 MiB chunks
    with open(path, "rb") as fp:
        crcval = 0
        while chunk := fp.read(size):
            crcval = binascii.crc32(chunk, crcval)
    return crcval


def get_vocab_id(config_path):
    config = configparser.ConfigParser()
    config.read(config_path)
    section = config.sections()[0]
    return config[section]["vocab"]


def _get_completion_choices(
    param: Argument,
) -> dict[str, AnnifVocabulary] | dict[str, AnnifProject] | list:
    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
3			from __future__ import annotations
4
5			import binascii
6			import collections
7			import configparser
8			import importlib
9			import io
10			import itertools
11			import os
12			import pathlib
13			import shutil
14			import sys
15			import tempfile
16			import time
17			import zipfile
18			from fnmatch import fnmatch
19			from typing import TYPE_CHECKING
20
21			import click
22			import click_log
23			from flask import current_app
24
25			import annif
26			from annif.exception import ConfigurationException, OperationFailedException
27			from annif.project import Access
28
29			if TYPE_CHECKING:
30			from datetime import datetime
31
32			from click.core import Argument, Context, Option
33
34			from annif.corpus.document import DocumentCorpus, DocumentList
35			from annif.corpus.subject import SubjectIndex
36			from annif.project import AnnifProject
37			from annif.suggestion import SuggestionResult
38			from annif.vocab import AnnifVocabulary
39
40			logger = annif.logger
41
42
43			def _set_project_config_file_path(
44			ctx: Context, param: Option, value: str \| None
45			) -> None:
46			"""Override the default path or the path given in env by CLI option"""
47			with ctx.obj.load_app().app_context():
48			if value:
49			current_app.config["PROJECTS_CONFIG_PATH"] = value
50
51
52			def common_options(f):
53			"""Decorator to add common options for all CLI commands"""
54			f = click.option(
55			"-p",
56			"--projects",
57			help="Set path to project configuration file or directory",
58			type=click.Path(dir_okay=True, exists=True),
59			callback=_set_project_config_file_path,
60			expose_value=False,
61			is_eager=True,
62			)(f)
63			return click_log.simple_verbosity_option(logger)(f)
64
65
66			def project_id(f):
67			"""Decorator to add a project ID parameter to a CLI command"""
68			return click.argument("project_id", shell_complete=complete_param)(f)
69
70
71			def backend_param_option(f):
72			"""Decorator to add an option for CLI commands to override BE parameters"""
73			return click.option(
74			"--backend-param",
75			"-b",
76			multiple=True,
77			help="Override backend parameter of the config file. "
78			+ "Syntax: `-b <backend>.<parameter>=<value>`.",
79			)(f)
80
81
82			def docs_limit_option(f):
83			"""Decorator to add an option for CLI commands to limit the number of documents to
84			use"""
85			return click.option(
86			"--docs-limit",
87			"-d",
88			default=None,
89			type=click.IntRange(0, None),
90			help="Maximum number of documents to use",
91			)(f)
92
93
94			def get_project(project_id: str) -> AnnifProject:
95			"""
96			Helper function to get a project by ID and bail out if it doesn't exist"""
97			try:
98			return annif.registry.get_project(project_id, min_access=Access.private)
99			except ValueError:
100			click.echo("No projects found with id '{0}'.".format(project_id), err=True)
101			sys.exit(1)
102
103
104			def get_vocab(vocab_id: str) -> AnnifVocabulary:
105			"""
106			Helper function to get a vocabulary by ID and bail out if it doesn't
107			exist"""
108			try:
109			return annif.registry.get_vocab(vocab_id, min_access=Access.private)
110			except ValueError:
111			click.echo(f"No vocabularies found with the id '{vocab_id}'.", err=True)
112			sys.exit(1)
113
114
115			def make_list_template(*rows) -> str:
116			"""Helper function to create a template for a list of entries with fields of
117			variable width. The width of each field is determined by the longest item in the
118			field in the given rows."""
119
120			max_field_widths = collections.defaultdict(int)
121			for row in rows:
122			for field_ind, item in enumerate(row):
123			max_field_widths[field_ind] = max(max_field_widths[field_ind], len(item))
124
125			return " ".join(
126			[
127			f"{{{field_ind}: <{field_width}}}"
128			for field_ind, field_width in max_field_widths.items()
129			]
130			)
131
132
133			def format_datetime(dt: datetime \| None) -> str:
134			"""Helper function to format a datetime object as a string in the local time."""
135			if dt is None:
136			return "-"
137			return dt.astimezone().strftime("%Y-%m-%d %H:%M:%S")
138
139
140			def open_documents(
141			paths: tuple[str, ...],
142			subject_index: SubjectIndex,
143			vocab_lang: str,
144			docs_limit: int \| None,
145			) -> DocumentCorpus:
146			"""Helper function to open a document corpus from a list of pathnames,
147			each of which is either a TSV file or a directory of TXT files. For
148			directories with subjects in TSV files, the given vocabulary language
149			will be used to convert subject labels into URIs. The corpus will be
150			returned as an instance of DocumentCorpus or LimitingDocumentCorpus."""
151
152			def open_doc_path(path, subject_index):
153			"""open a single path and return it as a DocumentCorpus"""
154			if os.path.isdir(path):
155			return annif.corpus.DocumentDirectory(
156			path, subject_index, vocab_lang, require_subjects=True
157			)
158			return annif.corpus.DocumentFile(path, subject_index)
159
160			if len(paths) == 0:
161			logger.warning("Reading empty file")
162			docs = open_doc_path(os.path.devnull, subject_index)
163			elif len(paths) == 1:
164			docs = open_doc_path(paths[0], subject_index)
165			else:
166			corpora = [open_doc_path(path, subject_index) for path in paths]
167			docs = annif.corpus.CombinedCorpus(corpora)
168			if docs_limit is not None:
169			docs = annif.corpus.LimitingDocumentCorpus(docs, docs_limit)
170			return docs
171
172
173			def open_text_documents(paths: tuple[str, ...], docs_limit: int \| None) -> DocumentList:
174			"""
175			Helper function to read text documents from the given file paths. Returns a
176			DocumentList object with Documents having no subjects. If a path is "-", the
177			document text is read from standard input. The maximum number of documents to read
178			is set by docs_limit parameter.
179			"""
180
181			def _docs(paths):
182			for path in paths:
183			if path == "-":
184			doc = annif.corpus.Document(text=sys.stdin.read(), subject_set=None)
185			else:
186			with open(path, errors="replace", encoding="utf-8-sig") as docfile:
187			doc = annif.corpus.Document(text=docfile.read(), subject_set=None)
188			yield doc
189
190			return annif.corpus.DocumentList(_docs(paths[:docs_limit]))
191
192
193			def show_hits(
194			hits: SuggestionResult,
195			project: AnnifProject,
196			lang: str,
197			file: io.TextIOWrapper \| None = None,
198			) -> None:
199			"""
200			Print subject suggestions to the console or a file. The suggestions are displayed as
201			a table, with one row per hit. Each row contains the URI, label, possible notation,
202			and score of the suggestion. The label is given in the specified language.
203			"""
204			template = "<{}>\t{}\t{:.04f}"
205			for hit in hits:
206			subj = project.subjects[hit.subject_id]
207			line = template.format(
208			subj.uri,
209			"\t".join(filter(None, (subj.labels[lang], subj.notation))),
210			hit.score,
211			)
212			click.echo(line, file=file)
213
214
215			def parse_backend_params(
216			backend_param: tuple[str, ...] \| tuple[()], project: AnnifProject
217			) -> collections.defaultdict[str, dict[str, str]]:
218			"""Parse a list of backend parameters given with the --backend-param
219			option into a nested dict structure"""
220			backend_params = collections.defaultdict(dict)
221			for beparam in backend_param:
222			backend, param = beparam.split(".", 1)
223			key, val = param.split("=", 1)
224			_validate_backend_params(backend, beparam, project)
225			backend_params[backend][key] = val
226			return backend_params
227
228
229			def _validate_backend_params(backend: str, beparam: str, project: AnnifProject) -> None:
230			if backend != project.config["backend"]:
231			raise ConfigurationException(
232			'The backend {} in CLI option "-b {}" not matching the project'
233			" backend {}.".format(backend, beparam, project.config["backend"])
234			)
235
236
237			def generate_filter_params(filter_batch_max_limit: int) -> list[tuple[int, float]]:
238			limits = range(1, filter_batch_max_limit + 1)
239			thresholds = [i * 0.05 for i in range(20)]
240			return list(itertools.product(limits, thresholds))
241
242
243			def _is_train_file(fname):
244			train_file_patterns = ("-train", "tmp-")
245			for pat in train_file_patterns:
246			if pat in fname:
247			return True
248			return False
249
250
251			def archive_dir(data_dir):
252			fp = tempfile.TemporaryFile()
253			path = pathlib.Path(data_dir)
254			fpaths = [fpath for fpath in path.glob("*/") if not _is_train_file(fpath.name)]
255			with zipfile.ZipFile(fp, mode="w") as zfile:
256			zfile.comment = bytes(
257			f"Archived by Annif {importlib.metadata.version('annif')}",
258			encoding="utf-8",
259			)
260			for fpath in fpaths:
261			logger.debug(f"Adding {fpath}")
262			arcname = os.path.join(*fpath.parts[1:])
263			zfile.write(fpath, arcname=arcname)
264			fp.seek(0)
265			return fp
266
267
268			def write_config(project):
269			fp = tempfile.TemporaryFile(mode="w+t")
270			config = configparser.ConfigParser()
271			config[project.project_id] = project.config
272			config.write(fp) # This needs tempfile in text mode
273			fp.seek(0)
274			# But for upload fobj needs to be in binary mode
275			return io.BytesIO(fp.read().encode("utf8"))
276
277
278			def upload_to_hf_hub(fileobj, filename, repo_id, token, commit_message):
279			from huggingface_hub import HfApi
280			from huggingface_hub.utils import HfHubHTTPError, HFValidationError
281
282			api = HfApi()
283			try:
284			api.upload_file(
285			path_or_fileobj=fileobj,
286			path_in_repo=filename,
287			repo_id=repo_id,
288			token=token,
289			commit_message=commit_message,
290			)
291			except (HfHubHTTPError, HFValidationError) as err:
292			raise OperationFailedException(str(err))
293
294
295			def get_selected_project_ids_from_hf_hub(project_ids_pattern, repo_id, token, revision):
296			all_repo_file_paths = _list_files_in_hf_hub(repo_id, token, revision)
297			return [
298			path.rsplit(".zip")[0].split("projects/")[1]
299			for path in all_repo_file_paths
300			if fnmatch(path, f"projects/{project_ids_pattern}.zip")
301			]
302
303
304			def _list_files_in_hf_hub(repo_id, token, revision):
305			from huggingface_hub import list_repo_files
306			from huggingface_hub.utils import HfHubHTTPError, HFValidationError
307
308			try:
309			return [
310			repofile
311			for repofile in list_repo_files(
312			repo_id=repo_id, token=token, revision=revision
313			)
314			]
315			except (HfHubHTTPError, HFValidationError) as err:
316			raise OperationFailedException(str(err))
317
318
319			def download_from_hf_hub(filename, repo_id, token, revision):
320			from huggingface_hub import hf_hub_download
321			from huggingface_hub.utils import HfHubHTTPError, HFValidationError
322
323			try:
324			return hf_hub_download(
325			repo_id=repo_id,
326			filename=filename,
327			token=token,
328			revision=revision,
329			)
330			except (HfHubHTTPError, HFValidationError) as err:
331			raise OperationFailedException(str(err))
332
333
334			def unzip(src_path, force):
335			datadir = current_app.config["DATADIR"]
336			with zipfile.ZipFile(src_path, "r") as zfile:
337			logger.debug(
338			f"Extracting archive {src_path}; archive comment: "
339			f"\"{str(zfile.comment, encoding='utf-8')}\""
340			)
341			for member in zfile.infolist():
342			dest_path = os.path.join(datadir, member.filename)
343			if os.path.exists(dest_path) and not force:
344			if _are_identical(member, dest_path):
345			logger.debug(f"Skipping unzip to {dest_path}; already in place")
346			else:
347			click.echo(f"Not overwriting {dest_path} (use --force to override)")
348			else:
349			logger.debug(f"Unzipping to {dest_path}")
350			zfile.extract(member, path=datadir)
351			date_time = time.mktime(member.date_time + (0, 0, -1))
352			os.utime(dest_path, (date_time, date_time))
353
354
355			def move_project_config(src_path, force):
356			dest_path = os.path.join("projects.d", os.path.basename(src_path))
357			if os.path.exists(dest_path) and not force:
358			if _compute_crc32(dest_path) == _compute_crc32(src_path):
359			logger.debug(
360			f"Skipping move of {os.path.basename(src_path)}; already in place"
361			)
362			else:
363			click.echo(f"Not overwriting {dest_path} (use --force to override)")
364			else:
365			logger.debug(f"Moving to {dest_path}")
366			shutil.copy(src_path, dest_path)
367
368
369			def _are_identical(member, dest_path):
370			path_crc = _compute_crc32(dest_path)
371			return path_crc == member.CRC
372
373
374			def _compute_crc32(path):
375			if os.path.isdir(path):
376			return 0
377
378			size = 1024 * 1024 * 10 # 10 MiB chunks
379			with open(path, "rb") as fp:
380			crcval = 0
381			while chunk := fp.read(size):
382			crcval = binascii.crc32(chunk, crcval)
383			return crcval
384
385
386			def get_vocab_id(config_path):
387			config = configparser.ConfigParser()
388			config.read(config_path)
389			section = config.sections()[0]
390			return config[section]["vocab"]
391
392
393			def _get_completion_choices(
394			param: Argument,
395			) -> dict[str, AnnifVocabulary] \| dict[str, AnnifProject] \| list:
396			if param.name == "project_id":
397			return annif.registry.get_projects()
398			elif param.name == "vocab_id":
399			return annif.registry.get_vocabs()
400			else:
401			return []
402
403
404			def complete_param(ctx: Context, param: Argument, incomplete: str) -> list[str]:
405			with ctx.obj.load_app().app_context():
406			return [
407			choice
408			for choice in _get_completion_choices(param)
409			if choice.startswith(incomplete)
410			]
411

NatLibFi / Annif

Pull Request — main (#762)

annif.cli_util.show_hits() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like