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

55.45

/sonar/modules/documents/cli/urn.py

# Swiss Open Access Repository
# Copyright (C) 2022 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/>.

"""URN specific CLI commands."""

import os

import click
from flask import current_app
from flask.cli import with_appcontext
from invenio_db import db
from invenio_pidstore.errors import PIDDoesNotExistError
from invenio_pidstore.models import PersistentIdentifier, PIDStatus

from sonar.modules.documents.api import DocumentRecord
from sonar.modules.documents.dnb import DnbServerError, DnbUrnService
from sonar.modules.documents.urn import Urn
from sonar.snl.ftp import SNLRepository


@click.group()
def urn():
    """URN specific commands."""


@urn.command()
@click.argument("urn")
@with_appcontext
def get(urn):
    """Get a URN information.

    :param urn: str - URN identifier
    """
    p = PersistentIdentifier.get("urn", urn)
    if not p.is_registered():
        click.secho(f"error: bad pid status ({p.status})", fg="red")
    dnb_urn = DnbUrnService.get(urn)
    dnb_urls = DnbUrnService.get_urls(urn)
    document = DocumentRecord.get_record(p.object_uuid)
    url = DnbUrnService.get_url(document)
    if dnb_urls["items"][0]["url"] != url:
        click.secho(
            f"error: DNB has the wrong url ({dnb_urls['items'][0]['url']} != {url})",
            fg="red",
        )
    if len(dnb_urls["items"]) > 1:
        urls = [item["url"] for item in dnb_urls["items"]]
        click.secho(f"error: DNB has more than one urls ({', '.join(urls)})", fg="red")

    click.echo(f"urn created: {dnb_urn['created']}")
    click.echo(f"urn modified: {dnb_urn['lastModified']}")
    dnb_url = dnb_urls["items"][0]
    click.echo(f"url created: {dnb_url['created']}")
    click.echo(f"url modified: {dnb_url['lastModified']}")
    click.echo(f"url: {url}")


@urn.command()
@with_appcontext
def create():
    """Create and register urns for loaded records."""
    idx = 0
    for idx, document in enumerate(Urn.get_documents_to_generate_urns(), 1):
        click.secho(f"\t{idx}: generate urn code for pid: {document['pid']}", fg="green")
        Urn.create_urn(document)
        document.commit()
        db.session.commit()
        document.reindex()
        Urn.register_urn_code_from_document(document)
    click.secho(f"{idx} URN created.", fg="green")


@urn.command()
@with_appcontext
def register():
    """Register urns for reserved URN pids."""
    query = PersistentIdentifier.query.filter_by(pid_type="urn").filter_by(status=PIDStatus.RESERVED)
    for idx, pid in enumerate(query.all()):
        doc = DocumentRecord.get_record(pid.object_uuid)
        click.secho(f"Registering document (pid: {doc['pid']})", fg="yellow")
        Urn.register_urn_code_from_document(doc)
    click.secho(f"{idx} URN registered.", fg="green")


@urn.command("snl-upload-file")
@click.argument("urn_code")
@with_appcontext
def snl_upload_file(urn_code):
    r"""Upload file on the SNL server via FTP for REGISTRED urn.

    Ex. invenio documents urn snl-upload-file 006-72 ~/my_dir/a.pdf
    :param urn_code: urn code of the document. Ex. 006-72
    :param filepath: local filepath of file to upload
    """
    try:
        pid = PersistentIdentifier.get(Urn.urn_pid_type, urn_code)
    except PIDDoesNotExistError:
        click.secho("Error: URN does not exists.")
        return

    if not pid.is_registered():
        click.secho("Error: the given urn is not registered.")
        return

    doc = DocumentRecord.get_record(pid.object_uuid)
    if not doc:
        click.secho("Error: the given urn is not linked to any document.")
        return

    files = [file for file in doc.files if file.get("type") == "file"]
    if not files:
        click.secho("Error: the document does not contains any files.")
        return

    snl_repository = SNLRepository(
        host=current_app.config.get("SONAR_APP_FTP_SNL_HOST"),
        user=current_app.config.get("SONAR_APP_FTP_SNL_USER"),
        password=current_app.config.get("SONAR_APP_FTP_SNL_PASSWORD"),
        directory=current_app.config.get("SONAR_APP_FTP_SNL_PATH"),
    )
    snl_repository.connect()

    dnb_base_urn = current_app.config.get("SONAR_APP_FTP_SNL_PATH")
    urn_dir = "/".join([dnb_base_urn, urn_code.split(":")[-1]])

    # create directory for urn
    snl_repository.make_dir(urn_dir)

    # upload file
    for _file in files:
        try:
            snl_repository.upload_file(_file.file.uri, os.path.join(urn_dir, _file.key))
            click.secho(f"Successfully uploaded file {os.path.basename(_file.key)}.", fg="green")
        except Exception as exception:
            click.secho(str(exception), fg="red")

    # print email template
    template_email_snl = current_app.config.get("SONAR_APP_SNL_EMAIL_TEMPLATE")

    click.secho("Template of email to send to SNL:", fg="green")
    with open(template_email_snl) as file:
        email_txt = file.read()
        email_txt = email_txt.replace("<URN>", urn_code)
        email_txt = email_txt.replace("<URL>", f"https://sonar.ch/global/documents/{doc.get('pid')}")
        click.secho(email_txt)


@urn.command("snl-list-files")
@with_appcontext
def snl_list_files():
    """List files uploaded on SNL server."""
    snl_repository = SNLRepository(
        host=current_app.config.get("SONAR_APP_FTP_SNL_HOST"),
        user=current_app.config.get("SONAR_APP_FTP_SNL_USER"),
        password=current_app.config.get("SONAR_APP_FTP_SNL_PASSWORD"),
        directory=current_app.config.get("SONAR_APP_FTP_SNL_PATH"),
    )
    snl_repository.connect()
    snl_repository.client.walktree(".", lambda x: click.secho(x), lambda x: click.secho(x), lambda x: click.secho(x))


@urn.command()
@click.argument("urn")
@click.argument("successor_urn")
@with_appcontext
def successor(urn, successor_urn):
    """Set a successor for a given run.

    :param urn: str - URN identifier
    :param successor_urn: str - Sucessor URN identifier
    """
    try:
        DnbUrnService().set_successor(urn, successor_urn)
        click.secho("Added successfully a successor.", fg="green")
    except DnbServerError as err:
        click.secho(str(err), fg="red")

1	# Swiss Open Access Repository
2	# Copyright (C) 2022 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	"""URN specific CLI commands."""
17
18	import os	1✔
19
20	import click	1✔
21	from flask import current_app	1✔
22	from flask.cli import with_appcontext	1✔
23	from invenio_db import db	1✔
24	from invenio_pidstore.errors import PIDDoesNotExistError	1✔
25	from invenio_pidstore.models import PersistentIdentifier, PIDStatus	1✔
26
27	from sonar.modules.documents.api import DocumentRecord	1✔
28	from sonar.modules.documents.dnb import DnbServerError, DnbUrnService	1✔
29	from sonar.modules.documents.urn import Urn	1✔
30	from sonar.snl.ftp import SNLRepository	1✔
31
32
33	@click.group()	1✔
34	def urn():	1✔
35	"""URN specific commands."""
36
37
38	@urn.command()	1✔
39	@click.argument("urn")	1✔
40	@with_appcontext	1✔
41	def get(urn):	1✔
42	"""Get a URN information.
43
44	:param urn: str - URN identifier
45	"""
46	p = PersistentIdentifier.get("urn", urn)	×
47	if not p.is_registered():	×
48	click.secho(f"error: bad pid status ({p.status})", fg="red")	×
49	dnb_urn = DnbUrnService.get(urn)	×
50	dnb_urls = DnbUrnService.get_urls(urn)	×
51	document = DocumentRecord.get_record(p.object_uuid)	×
52	url = DnbUrnService.get_url(document)	×
53	if dnb_urls["items"][0]["url"] != url:	×
54	click.secho(	×
55	f"error: DNB has the wrong url ({dnb_urls['items'][0]['url']} != {url})",
56	fg="red",
57	)
58	if len(dnb_urls["items"]) > 1:	×
59	urls = [item["url"] for item in dnb_urls["items"]]	×
60	click.secho(f"error: DNB has more than one urls ({', '.join(urls)})", fg="red")	×
61
62	click.echo(f"urn created: {dnb_urn['created']}")	×
63	click.echo(f"urn modified: {dnb_urn['lastModified']}")	×
64	dnb_url = dnb_urls["items"][0]	×
65	click.echo(f"url created: {dnb_url['created']}")	×
66	click.echo(f"url modified: {dnb_url['lastModified']}")	×
67	click.echo(f"url: {url}")	×
68
69
70	@urn.command()	1✔
71	@with_appcontext	1✔
72	def create():	1✔
73	"""Create and register urns for loaded records."""
74	idx = 0	×
75	for idx, document in enumerate(Urn.get_documents_to_generate_urns(), 1):	×
76	click.secho(f"\t{idx}: generate urn code for pid: {document['pid']}", fg="green")	×
77	Urn.create_urn(document)	×
78	document.commit()	×
79	db.session.commit()	×
80	document.reindex()	×
81	Urn.register_urn_code_from_document(document)	×
82	click.secho(f"{idx} URN created.", fg="green")	×
83
84
85	@urn.command()	1✔
86	@with_appcontext	1✔
87	def register():	1✔
88	"""Register urns for reserved URN pids."""
89	query = PersistentIdentifier.query.filter_by(pid_type="urn").filter_by(status=PIDStatus.RESERVED)	×
90	for idx, pid in enumerate(query.all()):	×
91	doc = DocumentRecord.get_record(pid.object_uuid)	×
92	click.secho(f"Registering document (pid: {doc['pid']})", fg="yellow")	×
93	Urn.register_urn_code_from_document(doc)	×
94	click.secho(f"{idx} URN registered.", fg="green")	×
95
96
97	@urn.command("snl-upload-file")	1✔
98	@click.argument("urn_code")	1✔
99	@with_appcontext	1✔
100	def snl_upload_file(urn_code):	1✔
101	r"""Upload file on the SNL server via FTP for REGISTRED urn.
102
103	Ex. invenio documents urn snl-upload-file 006-72 ~/my_dir/a.pdf
104	:param urn_code: urn code of the document. Ex. 006-72
105	:param filepath: local filepath of file to upload
106	"""
107	try:	1✔
108	pid = PersistentIdentifier.get(Urn.urn_pid_type, urn_code)	1✔
109	except PIDDoesNotExistError:	1✔
110	click.secho("Error: URN does not exists.")	1✔
111	return	1✔
112
113	if not pid.is_registered():	1✔
114	click.secho("Error: the given urn is not registered.")	×
115	return	×
116
117	doc = DocumentRecord.get_record(pid.object_uuid)	1✔
118	if not doc:	1✔
119	click.secho("Error: the given urn is not linked to any document.")	×
120	return	×
121
122	files = [file for file in doc.files if file.get("type") == "file"]	1✔
123	if not files:	1✔
124	click.secho("Error: the document does not contains any files.")	×
125	return	×
126
127	snl_repository = SNLRepository(	1✔
128	host=current_app.config.get("SONAR_APP_FTP_SNL_HOST"),
129	user=current_app.config.get("SONAR_APP_FTP_SNL_USER"),
130	password=current_app.config.get("SONAR_APP_FTP_SNL_PASSWORD"),
131	directory=current_app.config.get("SONAR_APP_FTP_SNL_PATH"),
132	)
133	snl_repository.connect()	1✔
134
135	dnb_base_urn = current_app.config.get("SONAR_APP_FTP_SNL_PATH")	1✔
136	urn_dir = "/".join([dnb_base_urn, urn_code.split(":")[-1]])	1✔
137
138	# create directory for urn
139	snl_repository.make_dir(urn_dir)	1✔
140
141	# upload file
142	for _file in files:	1✔
143	try:	1✔
144	snl_repository.upload_file(_file.file.uri, os.path.join(urn_dir, _file.key))	1✔
145	click.secho(f"Successfully uploaded file {os.path.basename(_file.key)}.", fg="green")	1✔
146	except Exception as exception:	×
147	click.secho(str(exception), fg="red")	×
148
149	# print email template
150	template_email_snl = current_app.config.get("SONAR_APP_SNL_EMAIL_TEMPLATE")	1✔
151
152	click.secho("Template of email to send to SNL:", fg="green")	1✔
153	with open(template_email_snl) as file:	1✔
154	email_txt = file.read()	1✔
155	email_txt = email_txt.replace("<URN>", urn_code)	1✔
156	email_txt = email_txt.replace("<URL>", f"https://sonar.ch/global/documents/{doc.get('pid')}")	1✔
157	click.secho(email_txt)	1✔
158
159
160	@urn.command("snl-list-files")	1✔
161	@with_appcontext	1✔
162	def snl_list_files():	1✔
163	"""List files uploaded on SNL server."""
164	snl_repository = SNLRepository(	×
165	host=current_app.config.get("SONAR_APP_FTP_SNL_HOST"),
166	user=current_app.config.get("SONAR_APP_FTP_SNL_USER"),
167	password=current_app.config.get("SONAR_APP_FTP_SNL_PASSWORD"),
168	directory=current_app.config.get("SONAR_APP_FTP_SNL_PATH"),
169	)
170	snl_repository.connect()	×
171	snl_repository.client.walktree(".", lambda x: click.secho(x), lambda x: click.secho(x), lambda x: click.secho(x))	×
172
173
174	@urn.command()	1✔
175	@click.argument("urn")	1✔
176	@click.argument("successor_urn")	1✔
177	@with_appcontext	1✔
178	def successor(urn, successor_urn):	1✔
179	"""Set a successor for a given run.
180
181	:param urn: str - URN identifier
182	:param successor_urn: str - Sucessor URN identifier
183	"""
184	try:	×
185	DnbUrnService().set_successor(urn, successor_urn)	×
186	click.secho("Added successfully a successor.", fg="green")	×
187	except DnbServerError as err:	×
188	click.secho(str(err), fg="red")	×

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