18634744198

Committed 19 Oct 2025 07:00PM UTC coverage: 31.618% (-29.9%) from 61.514%

Build # 18634744198

Build Type

push

github

Committed by

mpasternak

Commit Message

Merge branch 'release/v202510.1270'

Coverage Stats

657 of 9430 branches covered (6.97%)

Branch coverage included in aggregate %.

229 of 523 new or added lines in 42 files covered. (43.79%)

11303 existing lines in 316 files now uncovered.

14765 of 39346 relevant lines covered (37.53%)

0.38 hits per line

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

12.61

src/pbn_api/admin/scientist.py

from datetime import datetime

from django.contrib import admin
from django.urls import reverse
from django.utils.http import urlencode

from bpp.admin.filters import OrcidObecnyFilter
from pbn_api.admin.base import BaseMongoDBAdmin
from pbn_api.admin.filters import (
    OdpowiednikAutoraWBPPFilter,
    PbnIdObecnyFilter,
    PolonUidObecnyFilter,
)
from pbn_api.models import Scientist


@admin.register(Scientist)
class ScientistAdmin(BaseMongoDBAdmin):
    show_full_result_count = False
    list_display = [
        "lastName",
        "name",
        "qualifications",
        "polonUid",
        "orcid",
        "currentEmploymentsInstitutionDisplayName",
        "mongoId",
        "from_institution_api",
        "rekord_w_bpp",
    ]

    search_fields = [
        "mongoId",
        "lastName",
        "name",
        "orcid",
    ]

    fields = BaseMongoDBAdmin.fields + [
        "from_institution_api",
    ]
    readonly_fields = BaseMongoDBAdmin.readonly_fields + ["from_institution_api"]
    list_filter = [
        OdpowiednikAutoraWBPPFilter,
        OrcidObecnyFilter,
        PolonUidObecnyFilter,
        PbnIdObecnyFilter,
        "from_institution_api",
        "qualifications",
    ] + BaseMongoDBAdmin.list_filter

    def change_view(self, request, object_id, form_url="", extra_context=None):
        """Dodaje URL do utworzenia autora w BPP do kontekstu."""
        extra_context = extra_context or {}

        obj = self.get_object(request, object_id)
        if obj and not obj.rekord_w_bpp:
            # Przygotuj dane do utworzenia autora
            params = self._prepare_autor_params(obj)
            url = reverse("admin:bpp_autor_add") + "?" + urlencode(params)
            extra_context["create_in_bpp_url"] = url

        return super().change_view(request, object_id, form_url, extra_context)

    def _add_basic_autor_params(self, params, obj):
        """Add basic autor parameters (name, lastName, orcid, pbn_uid)."""
        if obj.name:
            params["imiona"] = obj.name
        if obj.lastName:
            params["nazwisko"] = obj.lastName
        if obj.orcid:
            params["orcid"] = obj.orcid
        params["pbn_uid"] = obj.pk

    def _add_tytul_param(self, params, obj):
        """Add tytul parameter if available."""
        tytul_id = self._get_tytul_id(obj.qualifications)
        if tytul_id:
            params["tytul"] = tytul_id

    def _add_employment_date_param(self, params, prefix, date_str):
        """Add employment date parameter with format conversion."""
        if not date_str:
            return
        try:
            date_obj = datetime.strptime(date_str, "%Y-%m-%d")
            params[prefix] = date_obj.strftime("%Y-%m-%d")
        except BaseException:
            pass

    def _add_employment_params(self, params, employment_data):
        """Add employment data parameters."""
        if not employment_data or not employment_data.get("jednostka_id"):
            return

        params["autor_jednostka_set-0-jednostka"] = employment_data["jednostka_id"]
        self._add_employment_date_param(
            params, "autor_jednostka_set-0-rozpoczal_prace", employment_data.get("od")
        )
        self._add_employment_date_param(
            params, "autor_jednostka_set-0-zakonczyl_prace", employment_data.get("do")
        )
        params["autor_jednostka_set-TOTAL_FORMS"] = "1"
        params["autor_jednostka_set-INITIAL_FORMS"] = "0"
        params["autor_jednostka_set-MIN_NUM_FORMS"] = "0"
        params["autor_jednostka_set-MAX_NUM_FORMS"] = "1000"

    def _add_disciplines_params(self, params, disciplines):
        """Add disciplines parameters."""
        if not disciplines:
            return

        for i, disc_data in enumerate(disciplines[:2]):  # Maksymalnie 2 dyscypliny
            year = disc_data.get("year", datetime.now().year)
            params[f"autor_dyscyplina_set-{i}-rok"] = year
            if disc_data.get("dyscyplina_id"):
                params[f"autor_dyscyplina_set-{i}-dyscyplina_naukowa"] = disc_data[
                    "dyscyplina_id"
                ]
            if disc_data.get("procent"):
                params[f"autor_dyscyplina_set-{i}-procent_dyscypliny"] = disc_data[
                    "procent"
                ]

        params["autor_dyscyplina_set-TOTAL_FORMS"] = str(len(disciplines))
        params["autor_dyscyplina_set-INITIAL_FORMS"] = "0"
        params["autor_dyscyplina_set-MIN_NUM_FORMS"] = "0"
        params["autor_dyscyplina_set-MAX_NUM_FORMS"] = "1000"

    def _prepare_autor_params(self, obj):
        """Przygotowuje parametry do utworzenia autora w BPP."""
        params = {}

        self._add_basic_autor_params(params, obj)
        self._add_tytul_param(params, obj)

        employment_data = self._get_employment_data(obj)
        self._add_employment_params(params, employment_data)

        disciplines = self._get_disciplines(obj)
        self._add_disciplines_params(params, disciplines)

        return params

    def _get_tytul_id(self, qualifications):
        """Znajdź ID tytułu naukowego na podstawie kwalifikacji z PBN."""
        if not qualifications:
            return None

        from bpp.models import Tytul

        # Mapowanie popularnych tytułów
        mapping = {
            "prof. dr hab.": "prof. dr hab.",
            "prof.": "prof.",
            "dr hab.": "dr hab.",
            "dr": "dr",
            "mgr": "mgr",
            "mgr inż.": "mgr inż.",
            "inż.": "inż.",
            "lek.": "lek.",
        }

        qualifications_lower = qualifications.lower().strip()
        for key, value in mapping.items():
            if key in qualifications_lower:
                try:
                    tytul = Tytul.objects.get(skrot=value)
                    return tytul.pk
                except Tytul.DoesNotExist:
                    pass

        return None

    def _get_jednostka_by_institution_id(self, institution_id):
        """Get Jednostka by PBN institution ID."""
        from bpp.models import Jednostka

        if not institution_id:
            return None
        try:
            jednostka = Jednostka.objects.get(pbn_uid_id=institution_id)
            return jednostka.pk
        except Jednostka.DoesNotExist:
            return None

    def _process_current_employment(self, data, employment):
        """Process current employment data."""
        if "from" in employment:
            data["od"] = employment["from"]

        jednostka_id = self._get_jednostka_by_institution_id(
            employment.get("institutionId")
        )
        if jednostka_id:
            data["jednostka_id"] = jednostka_id

    def _process_archival_employment(self, data, latest_employment):
        """Process archival employment data."""
        if "from" in latest_employment and "od" not in data:
            data["od"] = latest_employment["from"]
        if "to" in latest_employment:
            data["do"] = latest_employment["to"]

        if "jednostka_id" not in data:
            jednostka_id = self._get_jednostka_by_institution_id(
                latest_employment.get("institutionId")
            )
            if jednostka_id:
                data["jednostka_id"] = jednostka_id

    def _get_employment_data(self, obj):
        """Pobierz dane o zatrudnieniu z current_version."""
        data = {}

        # Sprawdź currentEmployments
        current_employments = obj.value_or_none("object", "currentEmployments")
        if current_employments and len(current_employments) > 0:
            self._process_current_employment(data, current_employments[0])

        # Sprawdź archivalEmployments dla daty zakończenia
        archival_employments = obj.value_or_none("object", "archivalEmployments")
        if archival_employments and len(archival_employments) > 0:
            latest = sorted(
                archival_employments, key=lambda x: x.get("to", ""), reverse=True
            )[0]
            self._process_archival_employment(data, latest)

        return data

    def _get_disciplines(self, obj):
        """Pobierz informacje o dyscyplinach z PBN i zamień na BPP."""

        disciplines = []

        # Sprawdź currentEmployments
        current_employments = obj.value_or_none("object", "currentEmployments")
        if current_employments and len(current_employments) > 0:
            employment = current_employments[0]
            pbn_disciplines = employment.get("disciplines", [])

            for pbn_disc in pbn_disciplines:
                discipline_uuid = pbn_disc.get("disciplineUuid")
                start_date = pbn_disc.get("startDate")

                if discipline_uuid and start_date:
                    # Wyciągnij rok z startDate
                    try:
                        year = int(start_date.split("-")[0])
                    except (ValueError, IndexError):
                        year = datetime.now().year

                    # Znajdź PBN dyscyplinę na podstawie UUID
                    try:
                        from pbn_api.models import Discipline

                        found = False
                        for pbn_discipline in Discipline.objects.filter(
                            uuid=discipline_uuid
                        ):
                            if pbn_discipline.parent_group.is_current:
                                found = True
                                break

                        if not found:
                            # Jeżeli nie ma tego kodu dyscypliny w aktualnym słowniku, to pomiń
                            continue

                        # Użyj TlumaczDyscyplin aby znaleźć odpowiednią BPP dyscyplinę
                        bpp_dyscyplina = self._find_bpp_discipline_by_pbn_discipline(
                            pbn_discipline, year
                        )
                        if bpp_dyscyplina:
                            disciplines.append(
                                {
                                    "year": year,
                                    "dyscyplina_id": bpp_dyscyplina.pk,
                                    "procent": 100.0,  # Domyślnie 100%
                                }
                            )
                    except Exception:
                        # Jeśli nic nie zadziała, pomiń tę dyscyplinę
                        pass

        return disciplines

    def _find_bpp_discipline_by_pbn_discipline(self, pbn_discipline, year):
        """Znajdź BPP dyscyplinę na podstawie obiektu PBN Discipline używając TlumaczDyscyplin."""
        from pbn_api.models import TlumaczDyscyplin

        # Znajdź TlumaczDyscyplin który mapuje na tę PBN dyscyplinę
        field_name = None
        if year >= 2024:
            field_name = "pbn_2024_now"
        elif year >= 2022:
            field_name = "pbn_2022_2023"
        elif year >= 2017:
            field_name = "pbn_2017_2021"

        if field_name:
            try:
                tlumacz = TlumaczDyscyplin.objects.get(**{field_name: pbn_discipline})
                return tlumacz.dyscyplina_w_bpp
            except TlumaczDyscyplin.DoesNotExist:
                pass

        return None

1	from datetime import datetime	1✔
2
3	from django.contrib import admin	1✔
4	from django.urls import reverse	1✔
5	from django.utils.http import urlencode	1✔
6
7	from bpp.admin.filters import OrcidObecnyFilter	1✔
8	from pbn_api.admin.base import BaseMongoDBAdmin	1✔
9	from pbn_api.admin.filters import (	1✔
10	OdpowiednikAutoraWBPPFilter,
11	PbnIdObecnyFilter,
12	PolonUidObecnyFilter,
13	)
14	from pbn_api.models import Scientist	1✔
15
16
17	@admin.register(Scientist)	1✔
18	class ScientistAdmin(BaseMongoDBAdmin):	1✔
19	show_full_result_count = False	1✔
20	list_display = [	1✔
21	"lastName",
22	"name",
23	"qualifications",
24	"polonUid",
25	"orcid",
26	"currentEmploymentsInstitutionDisplayName",
27	"mongoId",
28	"from_institution_api",
29	"rekord_w_bpp",
30	]
31
32	search_fields = [	1✔
33	"mongoId",
34	"lastName",
35	"name",
36	"orcid",
37	]
38
39	fields = BaseMongoDBAdmin.fields + [	1✔
40	"from_institution_api",
41	]
42	readonly_fields = BaseMongoDBAdmin.readonly_fields + ["from_institution_api"]	1✔
43	list_filter = [	1✔
44	OdpowiednikAutoraWBPPFilter,
45	OrcidObecnyFilter,
46	PolonUidObecnyFilter,
47	PbnIdObecnyFilter,
48	"from_institution_api",
49	"qualifications",
50	] + BaseMongoDBAdmin.list_filter
51
52	def change_view(self, request, object_id, form_url="", extra_context=None):	1✔
53	"""Dodaje URL do utworzenia autora w BPP do kontekstu."""
54	extra_context = extra_context or {}	×
55
56	obj = self.get_object(request, object_id)	×
57	if obj and not obj.rekord_w_bpp:	×
58	# Przygotuj dane do utworzenia autora
59	params = self._prepare_autor_params(obj)	×
60	url = reverse("admin:bpp_autor_add") + "?" + urlencode(params)	×
61	extra_context["create_in_bpp_url"] = url	×
62
63	return super().change_view(request, object_id, form_url, extra_context)	×
64
65	def _add_basic_autor_params(self, params, obj):	1✔
66	"""Add basic autor parameters (name, lastName, orcid, pbn_uid)."""
67	if obj.name:	×
68	params["imiona"] = obj.name	×
69	if obj.lastName:	×
70	params["nazwisko"] = obj.lastName	×
71	if obj.orcid:	×
72	params["orcid"] = obj.orcid	×
UNCOV 73	params["pbn_uid"] = obj.pk	×
74
75	def _add_tytul_param(self, params, obj):	1✔
76	"""Add tytul parameter if available."""
77	tytul_id = self._get_tytul_id(obj.qualifications)	×
78	if tytul_id:	×
79	params["tytul"] = tytul_id	×
80
81	def _add_employment_date_param(self, params, prefix, date_str):	1✔
82	"""Add employment date parameter with format conversion."""
NEW 83	if not date_str:	×
NEW 84	return	×
NEW 85	try:	×
NEW 86	date_obj = datetime.strptime(date_str, "%Y-%m-%d")	×
NEW 87	params[prefix] = date_obj.strftime("%Y-%m-%d")	×
NEW 88	except BaseException:	×
NEW 89	pass	×
90
91	def _add_employment_params(self, params, employment_data):	1✔
92	"""Add employment data parameters."""
NEW 93	if not employment_data or not employment_data.get("jednostka_id"):	×
NEW 94	return	×
95
NEW 96	params["autor_jednostka_set-0-jednostka"] = employment_data["jednostka_id"]	×
NEW 97	self._add_employment_date_param(	×
98	params, "autor_jednostka_set-0-rozpoczal_prace", employment_data.get("od")
99	)
NEW 100	self._add_employment_date_param(	×
101	params, "autor_jednostka_set-0-zakonczyl_prace", employment_data.get("do")
102	)
NEW 103	params["autor_jednostka_set-TOTAL_FORMS"] = "1"	×
NEW 104	params["autor_jednostka_set-INITIAL_FORMS"] = "0"	×
NEW 105	params["autor_jednostka_set-MIN_NUM_FORMS"] = "0"	×
NEW 106	params["autor_jednostka_set-MAX_NUM_FORMS"] = "1000"	×
107
108	def _add_disciplines_params(self, params, disciplines):	1✔
109	"""Add disciplines parameters."""
NEW 110	if not disciplines:	×
NEW 111	return	×
112
NEW 113	for i, disc_data in enumerate(disciplines[:2]): # Maksymalnie 2 dyscypliny	×
NEW 114	year = disc_data.get("year", datetime.now().year)	×
NEW 115	params[f"autor_dyscyplina_set-{i}-rok"] = year	×
NEW 116	if disc_data.get("dyscyplina_id"):	×
NEW 117	params[f"autor_dyscyplina_set-{i}-dyscyplina_naukowa"] = disc_data[	×
118	"dyscyplina_id"
119	]
NEW 120	if disc_data.get("procent"):	×
NEW 121	params[f"autor_dyscyplina_set-{i}-procent_dyscypliny"] = disc_data[	×
122	"procent"
123	]
124
NEW 125	params["autor_dyscyplina_set-TOTAL_FORMS"] = str(len(disciplines))	×
NEW 126	params["autor_dyscyplina_set-INITIAL_FORMS"] = "0"	×
NEW 127	params["autor_dyscyplina_set-MIN_NUM_FORMS"] = "0"	×
NEW 128	params["autor_dyscyplina_set-MAX_NUM_FORMS"] = "1000"	×
129
130	def _prepare_autor_params(self, obj):	1✔
131	"""Przygotowuje parametry do utworzenia autora w BPP."""
NEW 132	params = {}	×
133
NEW 134	self._add_basic_autor_params(params, obj)	×
NEW 135	self._add_tytul_param(params, obj)	×
136
NEW 137	employment_data = self._get_employment_data(obj)	×
NEW 138	self._add_employment_params(params, employment_data)	×
139
UNCOV 140	disciplines = self._get_disciplines(obj)	×
NEW 141	self._add_disciplines_params(params, disciplines)	×
142
143	return params	×
144
145	def _get_tytul_id(self, qualifications):	1✔
146	"""Znajdź ID tytułu naukowego na podstawie kwalifikacji z PBN."""
147	if not qualifications:	×
148	return None	×
149
150	from bpp.models import Tytul	×
151
152	# Mapowanie popularnych tytułów
153	mapping = {	×
154	"prof. dr hab.": "prof. dr hab.",
155	"prof.": "prof.",
156	"dr hab.": "dr hab.",
157	"dr": "dr",
158	"mgr": "mgr",
159	"mgr inż.": "mgr inż.",
160	"inż.": "inż.",
161	"lek.": "lek.",
162	}
163
164	qualifications_lower = qualifications.lower().strip()	×
165	for key, value in mapping.items():	×
166	if key in qualifications_lower:	×
167	try:	×
168	tytul = Tytul.objects.get(skrot=value)	×
169	return tytul.pk	×
170	except Tytul.DoesNotExist:	×
171	pass	×
172
173	return None	×
174
175	def _get_jednostka_by_institution_id(self, institution_id):	1✔
176	"""Get Jednostka by PBN institution ID."""
UNCOV 177	from bpp.models import Jednostka	×
178
NEW 179	if not institution_id:	×
NEW 180	return None	×
NEW 181	try:	×
NEW 182	jednostka = Jednostka.objects.get(pbn_uid_id=institution_id)	×
NEW 183	return jednostka.pk	×
NEW 184	except Jednostka.DoesNotExist:	×
NEW 185	return None	×
186
187	def _process_current_employment(self, data, employment):	1✔
188	"""Process current employment data."""
NEW 189	if "from" in employment:	×
NEW 190	data["od"] = employment["from"]	×
191
NEW 192	jednostka_id = self._get_jednostka_by_institution_id(	×
193	employment.get("institutionId")
194	)
NEW 195	if jednostka_id:	×
NEW 196	data["jednostka_id"] = jednostka_id	×
197
198	def _process_archival_employment(self, data, latest_employment):	1✔
199	"""Process archival employment data."""
NEW 200	if "from" in latest_employment and "od" not in data:	×
NEW 201	data["od"] = latest_employment["from"]	×
NEW 202	if "to" in latest_employment:	×
NEW 203	data["do"] = latest_employment["to"]	×
204
NEW 205	if "jednostka_id" not in data:	×
NEW 206	jednostka_id = self._get_jednostka_by_institution_id(	×
207	latest_employment.get("institutionId")
208	)
NEW 209	if jednostka_id:	×
NEW 210	data["jednostka_id"] = jednostka_id	×
211
212	def _get_employment_data(self, obj):	1✔
213	"""Pobierz dane o zatrudnieniu z current_version."""
UNCOV 214	data = {}	×
215
216	# Sprawdź currentEmployments
217	current_employments = obj.value_or_none("object", "currentEmployments")	×
218	if current_employments and len(current_employments) > 0:	×
NEW 219	self._process_current_employment(data, current_employments[0])	×
220
221	# Sprawdź archivalEmployments dla daty zakończenia
222	archival_employments = obj.value_or_none("object", "archivalEmployments")	×
223	if archival_employments and len(archival_employments) > 0:	×
UNCOV 224	latest = sorted(	×
225	archival_employments, key=lambda x: x.get("to", ""), reverse=True
226	)[0]
NEW 227	self._process_archival_employment(data, latest)	×
228
229	return data	×
230
231	def _get_disciplines(self, obj):	1✔
232	"""Pobierz informacje o dyscyplinach z PBN i zamień na BPP."""
233
234	disciplines = []	×
235
236	# Sprawdź currentEmployments
237	current_employments = obj.value_or_none("object", "currentEmployments")	×
238	if current_employments and len(current_employments) > 0:	×
239	employment = current_employments[0]	×
240	pbn_disciplines = employment.get("disciplines", [])	×
241
242	for pbn_disc in pbn_disciplines:	×
243	discipline_uuid = pbn_disc.get("disciplineUuid")	×
244	start_date = pbn_disc.get("startDate")	×
245
246	if discipline_uuid and start_date:	×
247	# Wyciągnij rok z startDate
248	try:	×
249	year = int(start_date.split("-")[0])	×
250	except (ValueError, IndexError):	×
251	year = datetime.now().year	×
252
253	# Znajdź PBN dyscyplinę na podstawie UUID
254	try:	×
255	from pbn_api.models import Discipline	×
256
257	found = False	×
258	for pbn_discipline in Discipline.objects.filter(	×
259	uuid=discipline_uuid
260	):
261	if pbn_discipline.parent_group.is_current:	×
262	found = True	×
263	break	×
264
265	if not found:	×
266	# Jeżeli nie ma tego kodu dyscypliny w aktualnym słowniku, to pomiń
267	continue	×
268
269	# Użyj TlumaczDyscyplin aby znaleźć odpowiednią BPP dyscyplinę
270	bpp_dyscyplina = self._find_bpp_discipline_by_pbn_discipline(	×
271	pbn_discipline, year
272	)
273	if bpp_dyscyplina:	×
274	disciplines.append(	×
275	{
276	"year": year,
277	"dyscyplina_id": bpp_dyscyplina.pk,
278	"procent": 100.0, # Domyślnie 100%
279	}
280	)
281	except Exception:	×
282	# Jeśli nic nie zadziała, pomiń tę dyscyplinę
283	pass	×
284
285	return disciplines	×
286
287	def _find_bpp_discipline_by_pbn_discipline(self, pbn_discipline, year):	1✔
288	"""Znajdź BPP dyscyplinę na podstawie obiektu PBN Discipline używając TlumaczDyscyplin."""
289	from pbn_api.models import TlumaczDyscyplin	×
290
291	# Znajdź TlumaczDyscyplin który mapuje na tę PBN dyscyplinę
292	field_name = None	×
293	if year >= 2024:	×
294	field_name = "pbn_2024_now"	×
295	elif year >= 2022:	×
296	field_name = "pbn_2022_2023"	×
297	elif year >= 2017:	×
298	field_name = "pbn_2017_2021"	×
299
300	if field_name:	×
301	try:	×
302	tlumacz = TlumaczDyscyplin.objects.get(**{field_name: pbn_discipline})	×
303	return tlumacz.dyscyplina_w_bpp	×
304	except TlumaczDyscyplin.DoesNotExist:	×
305	pass	×
306
307	return None	×

iplweb / bpp / 18634744198

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