17425918180

Committed 03 Sep 2025 07:11AM UTC coverage: 95.796% (-0.6%) from 96.378%

Build # 17425918180

Build Type

push

github

Committed by

PascalRepond

Commit Message

translations: extract messages

Co-Authored-by: Pascal Repond <pascal.repond@rero.ch>

Run Details

7816 of 8159 relevant lines covered (95.8%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.02

/sonar/modules/cli/utils.py

# Swiss Open Access Repository
# Copyright (C) 2021 RERO
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Utils commands."""

import json
import pathlib
import shutil
import sys
from os.path import dirname, join

import click
from flask import current_app
from flask.cli import with_appcontext
from invenio_db import db
from invenio_files_rest.models import Location
from invenio_oauth2server.cli import process_scopes, process_user
from invenio_oauth2server.models import Client, Token
from invenio_records_rest.utils import obj_or_import_string
from invenio_search.cli import search_version_check
from invenio_search.proxies import current_search
from werkzeug.local import LocalProxy
from werkzeug.security import gen_salt

from sonar.modules.api import SonarRecord

_datastore = LocalProxy(lambda: current_app.extensions["security"].datastore)


@click.group()
def utils():
    """Utils commands."""


@utils.command()
@click.option("--force", is_flag=True, default=False)
@with_appcontext
@search_version_check
def es_init(force):
    """Initialize registered templates, aliases and mappings."""
    # TODO: to remove once it is fixed in invenio-search module
    click.secho("Putting templates...", fg="green", bold=True, file=sys.stderr)
    with click.progressbar(
        current_search.put_templates(ignore=[400] if force else None),
        length=len(current_search.templates),
    ) as item:
        for response in item:
            item.label = response
    click.secho("Creating indexes...", fg="green", bold=True, file=sys.stderr)
    with click.progressbar(
        current_search.create(ignore=[400] if force else None),
        length=len(current_search.mappings),
    ) as item:
        for name, response in item:
            item.label = name


@utils.command()
@with_appcontext
def clear_files():
    """Remove all files and delete directory from all locations."""
    for location in Location.query.all():
        try:
            shutil.rmtree(location.uri)
        except Exception:
            click.secho(f"Directory {location.uri} cannot be cleaned", fg="yellow")

    click.secho("Finished", fg="green")


@utils.command("export")
@click.option("-p", "--pid-type", "pid_type", default="doc")
@click.option("-s", "--serializer", "serializer_key", default="export")
@click.option("-o", "--output-dir", "output_dir", required=True, type=click.File("w"))
@with_appcontext
def export(pid_type, serializer_key, output_dir):
    """Export records for the given record type.

    :param pid_type: record type
    :param output_dir: Output directory
    """
    click.secho(f'Export "{pid_type}" records in {output_dir.name}')

    try:
        # Get the correct record class
        record_class = SonarRecord.get_record_class_by_pid_type(pid_type)

        if not record_class:
            raise Exception(f'No record class found for type "{pid_type}"')

        # Load the serializer
        serializer_class = current_app.config.get("SONAR_APP_EXPORT_SERIALIZERS", {}).get(pid_type)

        serializer = obj_or_import_string(serializer_class)() if serializer_class else None

        pids = record_class.get_all_pids()
        records = []

        # Create ouptut directory if not exists
        if pids:
            pathlib.Path(output_dir.name).mkdir(mode=0o755, parents=True, exist_ok=True)

        for pid in pids:
            record = record_class.get_record_by_pid(pid)

            record = serializer.dump(record) if serializer else record.dumps()

            for file in record.get("files", []):
                if file.get("uri"):
                    target_path = join(output_dir.name, pid, file["key"])
                    pathlib.Path(dirname(target_path)).mkdir(mode=0o755, parents=True, exist_ok=True)
                    shutil.copyfile(file["uri"], target_path)
                    file.pop("uri")
                    file["path"] = f"./{pid}/{file['key']}"

            records.append(record)

        if records:
            # Write data
            output_file = join(output_dir.name, "data.json")
            with open(output_file, "w") as f:
                f.write(json.dumps(records))

        click.secho("Finished", fg="green")

    except Exception as err:
        click.secho(f"An error occured during export: {err}", fg="red")


def create_personal(name, user_id, scopes=None, is_internal=False, access_token=None):
    """Create a personal access token.

    A token that is bound to a specific user and which doesn't expire, i.e.
    similar to the concept of an API key.

    :param name: Client name.
    :param user_id: User ID.
    :param scopes: The list of permitted scopes. (Default: ``None``)
    :param is_internal: If ``True`` it's a internal access token.
            (Default: ``False``)
    :param access_token: personalized access_token.
    :returns: A new access token.
    """
    with db.session.begin_nested():
        scopes = " ".join(scopes) if scopes else ""

        client = Client(
            name=name,
            user_id=user_id,
            is_internal=True,
            is_confidential=False,
            _default_scopes=scopes,
        )
        client.gen_salt()

        if not access_token:
            access_token = gen_salt(current_app.config.get("OAUTH2SERVER_TOKEN_PERSONAL_SALT_LEN"))
        token = Token(
            client_id=client.client_id,
            user_id=user_id,
            access_token=access_token,
            expires=None,
            _scopes=scopes,
            is_personal=True,
            is_internal=is_internal,
        )
        db.session.add(client)
        db.session.add(token)

    return token


@utils.command()
@click.option("-n", "--name", required=True)
@click.option("-u", "--user", required=True, callback=process_user, help="User ID or email.")
@click.option("-s", "--scope", "scopes", multiple=True, callback=process_scopes)
@click.option("-i", "--internal", is_flag=True)
@click.option(
    "-t",
    "--access_token",
    "access_token",
    required=False,
    help="personalized access_token.",
)
@with_appcontext
def token_create(name, user, scopes, internal, access_token):
    """Create a personal OAuth token."""
    if user:
        token = create_personal(
            name,
            user.id,
            scopes=scopes,
            is_internal=internal,
            access_token=access_token,
        )
        db.session.commit()
        click.secho(token.access_token, fg="blue")
    else:
        click.secho("No user found", fg="red")

1	# Swiss Open Access Repository
2	# Copyright (C) 2021 RERO
3	#
4	# This program is free software: you can redistribute it and/or modify
5	# it under the terms of the GNU Affero General Public License as published by
6	# the Free Software Foundation, version 3 of the License.
7	#
8	# This program is distributed in the hope that it will be useful,
9	# but WITHOUT ANY WARRANTY; without even the implied warranty of
10	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11	# GNU Affero General Public License for more details.
12	#
13	# You should have received a copy of the GNU Affero General Public License
14	# along with this program. If not, see <http://www.gnu.org/licenses/>.
15
16	"""Utils commands."""
17
18	import json	1✔
19	import pathlib	1✔
20	import shutil	1✔
21	import sys	1✔
22	from os.path import dirname, join	1✔
23
24	import click	1✔
25	from flask import current_app	1✔
26	from flask.cli import with_appcontext	1✔
27	from invenio_db import db	1✔
28	from invenio_files_rest.models import Location	1✔
29	from invenio_oauth2server.cli import process_scopes, process_user	1✔
30	from invenio_oauth2server.models import Client, Token	1✔
31	from invenio_records_rest.utils import obj_or_import_string	1✔
32	from invenio_search.cli import search_version_check	1✔
33	from invenio_search.proxies import current_search	1✔
34	from werkzeug.local import LocalProxy	1✔
35	from werkzeug.security import gen_salt	1✔
36
37	from sonar.modules.api import SonarRecord	1✔
38
39	_datastore = LocalProxy(lambda: current_app.extensions["security"].datastore)	1✔
40
41
42	@click.group()	1✔
43	def utils():	1✔
44	"""Utils commands."""
45
46
47	@utils.command()	1✔
48	@click.option("--force", is_flag=True, default=False)	1✔
49	@with_appcontext	1✔
50	@search_version_check	1✔
51	def es_init(force):	1✔
52	"""Initialize registered templates, aliases and mappings."""
53	# TODO: to remove once it is fixed in invenio-search module
54	click.secho("Putting templates...", fg="green", bold=True, file=sys.stderr)	1✔
55	with click.progressbar(	1✔
56	current_search.put_templates(ignore=[400] if force else None),
57	length=len(current_search.templates),
58	) as item:
59	for response in item:	1✔
60	item.label = response	1✔
61	click.secho("Creating indexes...", fg="green", bold=True, file=sys.stderr)	1✔
62	with click.progressbar(	1✔
63	current_search.create(ignore=[400] if force else None),
64	length=len(current_search.mappings),
65	) as item:
66	for name, response in item:	1✔
67	item.label = name	1✔
68
69
70	@utils.command()	1✔
71	@with_appcontext	1✔
72	def clear_files():	1✔
73	"""Remove all files and delete directory from all locations."""
74	for location in Location.query.all():	1✔
75	try:	1✔
76	shutil.rmtree(location.uri)	1✔
77	except Exception:	1✔
78	click.secho(f"Directory {location.uri} cannot be cleaned", fg="yellow")	1✔
79
80	click.secho("Finished", fg="green")	1✔
81
82
83	@utils.command("export")	1✔
84	@click.option("-p", "--pid-type", "pid_type", default="doc")	1✔
85	@click.option("-s", "--serializer", "serializer_key", default="export")	1✔
86	@click.option("-o", "--output-dir", "output_dir", required=True, type=click.File("w"))	1✔
87	@with_appcontext	1✔
88	def export(pid_type, serializer_key, output_dir):	1✔
89	"""Export records for the given record type.
90
91	:param pid_type: record type
92	:param output_dir: Output directory
93	"""
94	click.secho(f'Export "{pid_type}" records in {output_dir.name}')	1✔
95
96	try:	1✔
97	# Get the correct record class
98	record_class = SonarRecord.get_record_class_by_pid_type(pid_type)	1✔
99
100	if not record_class:	1✔
101	raise Exception(f'No record class found for type "{pid_type}"')	1✔
102
103	# Load the serializer
104	serializer_class = current_app.config.get("SONAR_APP_EXPORT_SERIALIZERS", {}).get(pid_type)	1✔
105
106	serializer = obj_or_import_string(serializer_class)() if serializer_class else None	1✔
107
108	pids = record_class.get_all_pids()	1✔
109	records = []	1✔
110
111	# Create ouptut directory if not exists
112	if pids:	1✔
113	pathlib.Path(output_dir.name).mkdir(mode=0o755, parents=True, exist_ok=True)	1✔
114
115	for pid in pids:	1✔
116	record = record_class.get_record_by_pid(pid)	1✔
117
118	record = serializer.dump(record) if serializer else record.dumps()	1✔
119
120	for file in record.get("files", []):	1✔
121	if file.get("uri"):	1✔
122	target_path = join(output_dir.name, pid, file["key"])	1✔
123	pathlib.Path(dirname(target_path)).mkdir(mode=0o755, parents=True, exist_ok=True)	1✔
124	shutil.copyfile(file["uri"], target_path)	1✔
125	file.pop("uri")	1✔
126	file["path"] = f"./{pid}/{file['key']}"	1✔
127
128	records.append(record)	1✔
129
130	if records:	1✔
131	# Write data
132	output_file = join(output_dir.name, "data.json")	1✔
133	with open(output_file, "w") as f:	1✔
134	f.write(json.dumps(records))	1✔
135
136	click.secho("Finished", fg="green")	1✔
137
138	except Exception as err:	1✔
139	click.secho(f"An error occured during export: {err}", fg="red")	1✔
140
141
142	def create_personal(name, user_id, scopes=None, is_internal=False, access_token=None):	1✔
143	"""Create a personal access token.
144
145	A token that is bound to a specific user and which doesn't expire, i.e.
146	similar to the concept of an API key.
147
148	:param name: Client name.
149	:param user_id: User ID.
150	:param scopes: The list of permitted scopes. (Default: ``None``)
151	:param is_internal: If ``True`` it's a internal access token.
152	(Default: ``False``)
153	:param access_token: personalized access_token.
154	:returns: A new access token.
155	"""
156	with db.session.begin_nested():	1✔
157	scopes = " ".join(scopes) if scopes else ""	1✔
158
159	client = Client(	1✔
160	name=name,
161	user_id=user_id,
162	is_internal=True,
163	is_confidential=False,
164	_default_scopes=scopes,
165	)
166	client.gen_salt()	1✔
167
168	if not access_token:	1✔
169	access_token = gen_salt(current_app.config.get("OAUTH2SERVER_TOKEN_PERSONAL_SALT_LEN"))	×
170	token = Token(	1✔
171	client_id=client.client_id,
172	user_id=user_id,
173	access_token=access_token,
174	expires=None,
175	_scopes=scopes,
176	is_personal=True,
177	is_internal=is_internal,
178	)
179	db.session.add(client)	1✔
180	db.session.add(token)	1✔
181
182	return token	1✔
183
184
185	@utils.command()	1✔
186	@click.option("-n", "--name", required=True)	1✔
187	@click.option("-u", "--user", required=True, callback=process_user, help="User ID or email.")	1✔
188	@click.option("-s", "--scope", "scopes", multiple=True, callback=process_scopes)	1✔
189	@click.option("-i", "--internal", is_flag=True)	1✔
190	@click.option(	1✔
191	"-t",
192	"--access_token",
193	"access_token",
194	required=False,
195	help="personalized access_token.",
196	)
197	@with_appcontext	1✔
198	def token_create(name, user, scopes, internal, access_token):	1✔
199	"""Create a personal OAuth token."""
200	if user:	1✔
201	token = create_personal(	1✔
202	name,
203	user.id,
204	scopes=scopes,
205	is_internal=internal,
206	access_token=access_token,
207	)
208	db.session.commit()	1✔
209	click.secho(token.access_token, fg="blue")	1✔
210	else:
211	click.secho("No user found", fg="red")	1✔

rero / sonar / 17425918180

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous