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

40.54

/sonar/modules/cli/imports.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/>.

"""SONAR CLI commands."""

import csv
import os
from datetime import datetime

import click
from flask.cli import with_appcontext

from sonar.modules.documents.tasks import import_records
from sonar.modules.utils import chunks

DOCUMENT_TYPE_MAPPING = {
    "master thesis": "coar:c_bdcc",
    "bachelor thesis": "coar:c_7a1f",
    "thesis": "coar:c_46ec",
}


@click.group()
def imports():
    """Import commands."""


@imports.command()
@with_appcontext
@click.argument("data_file", type=click.File("r"), required=True)
@click.argument("pdf_directory", required=True)
def hepbejune(data_file, pdf_directory):
    """Import record from HEP BEJUNE.

    :param data_file: CSV File.
    :param pdf_directory: Directory containing the PDF files.
    """
    click.secho("Import records from HEPBEJUNE")

    records = []

    with open(data_file.name) as file:
        reader = csv.reader(file, delimiter=",")

        for row in reader:
            date = datetime.strptime(row[9], "%d/%m/%Y")
            if row[1] == "SKIP":
                continue

            degree = row[14]
            if degree == "bachelor thesis":
                degree = "Mémoire de bachelor"
            elif degree == "master thesis":
                degree = "Mémoire de master"
            else:
                degree = "Mémoire"

            data = {
                "title": [
                    {
                        "type": "bf:Title",
                        "mainTitle": [{"value": row[3], "language": "fre"}],
                    }
                ],
                "identifiedBy": [{"type": "bf:Local", "source": "hepbejune", "value": row[0]}],
                "language": [{"type": "bf:Language", "value": "fre"}],
                "contribution": [
                    {
                        "agent": {"type": "bf:Person", "preferred_name": row[8]},
                        "role": ["cre"],
                    }
                ],
                "dissertation": {
                    "degree": degree,
                    "grantingInstitution": "Haute école pédagogique BEJUNE",
                    "date": date.strftime("%Y-%m-%d"),
                },
                "provisionActivity": [{"type": "bf:Publication", "startDate": date.strftime("%Y")}],
                "customField1": [row[12]],
                "customField2": [row[13]],
                "documentType": DOCUMENT_TYPE_MAPPING.get(row[14], "coar:c_1843"),
                "usageAndAccessPolicy": {"license": "CC BY-NC-ND"},
                "organisation": [{"$ref": "https://sonar.ch/api/organisations/hepbejune"}],
                "harvested": True,
                "masked": "masked_for_external_ips",
            }

            file_path = os.path.join(pdf_directory, row[16])
            if os.path.isfile(file_path):
                data["files"] = [
                    {
                        "key": "fulltext.pdf",
                        "url": file_path,
                        "label": "Full-text",
                        "type": "file",
                        "order": 0,
                    }
                ]

            records.append(data)

    # Chunk record list and send celery task
    for chunk in list(chunks(records, 10)):
        import_records.delay(chunk)

    click.secho("Finished, records are imported in background", fg="green")

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	"""SONAR CLI commands."""
17
18	import csv	1✔
19	import os	1✔
20	from datetime import datetime	1✔
21
22	import click	1✔
23	from flask.cli import with_appcontext	1✔
24
25	from sonar.modules.documents.tasks import import_records	1✔
26	from sonar.modules.utils import chunks	1✔
27
28	DOCUMENT_TYPE_MAPPING = {	1✔
29	"master thesis": "coar:c_bdcc",
30	"bachelor thesis": "coar:c_7a1f",
31	"thesis": "coar:c_46ec",
32	}
33
34
35	@click.group()	1✔
36	def imports():	1✔
37	"""Import commands."""
38
39
40	@imports.command()	1✔
41	@with_appcontext	1✔
42	@click.argument("data_file", type=click.File("r"), required=True)	1✔
43	@click.argument("pdf_directory", required=True)	1✔
44	def hepbejune(data_file, pdf_directory):	1✔
45	"""Import record from HEP BEJUNE.
46
47	:param data_file: CSV File.
48	:param pdf_directory: Directory containing the PDF files.
49	"""
50	click.secho("Import records from HEPBEJUNE")	×
51
52	records = []	×
53
54	with open(data_file.name) as file:	×
55	reader = csv.reader(file, delimiter=",")	×
56
57	for row in reader:	×
58	date = datetime.strptime(row[9], "%d/%m/%Y")	×
59	if row[1] == "SKIP":	×
60	continue	×
61
62	degree = row[14]	×
63	if degree == "bachelor thesis":	×
64	degree = "Mémoire de bachelor"	×
65	elif degree == "master thesis":	×
66	degree = "Mémoire de master"	×
67	else:
68	degree = "Mémoire"	×
69
70	data = {	×
71	"title": [
72	{
73	"type": "bf:Title",
74	"mainTitle": [{"value": row[3], "language": "fre"}],
75	}
76	],
77	"identifiedBy": [{"type": "bf:Local", "source": "hepbejune", "value": row[0]}],
78	"language": [{"type": "bf:Language", "value": "fre"}],
79	"contribution": [
80	{
81	"agent": {"type": "bf:Person", "preferred_name": row[8]},
82	"role": ["cre"],
83	}
84	],
85	"dissertation": {
86	"degree": degree,
87	"grantingInstitution": "Haute école pédagogique BEJUNE",
88	"date": date.strftime("%Y-%m-%d"),
89	},
90	"provisionActivity": [{"type": "bf:Publication", "startDate": date.strftime("%Y")}],
91	"customField1": [row[12]],
92	"customField2": [row[13]],
93	"documentType": DOCUMENT_TYPE_MAPPING.get(row[14], "coar:c_1843"),
94	"usageAndAccessPolicy": {"license": "CC BY-NC-ND"},
95	"organisation": [{"$ref": "https://sonar.ch/api/organisations/hepbejune"}],
96	"harvested": True,
97	"masked": "masked_for_external_ips",
98	}
99
100	file_path = os.path.join(pdf_directory, row[16])	×
101	if os.path.isfile(file_path):	×
102	data["files"] = [	×
103	{
104	"key": "fulltext.pdf",
105	"url": file_path,
106	"label": "Full-text",
107	"type": "file",
108	"order": 0,
109	}
110	]
111
112	records.append(data)	×
113
114	# Chunk record list and send celery task
115	for chunk in list(chunks(records, 10)):	×
116	import_records.delay(chunk)	×
117
118	click.secho("Finished, records are imported in background", fg="green")	×

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