58b9a630-8512-44e6-b730-daac96d1c4d7

Committed 29 Aug 2025 07:21AM UTC coverage: 47.493% (+2.5%) from 45.008%

Build # 58b9a630-8512-44e6-b730-daac96d1c4d7

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Fix tests

Run Details

6 of 27 new or added lines in 2 files covered. (22.22%)

1342 existing lines in 64 files now uncovered.

19323 of 40686 relevant lines covered (47.49%)

1.51 hits per line

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

23.44

src/import_common/core.py

from typing import Union

import dateutil
from django.db.models import Q, Value
from django.db.models.functions import Lower, Replace, Trim

from .normalization import (
    normalize_doi,
    normalize_funkcja_autora,
    normalize_grupa_pracownicza,
    normalize_isbn,
    normalize_kod_dyscypliny,
    normalize_nazwa_dyscypliny,
    normalize_nazwa_jednostki,
    normalize_nazwa_wydawcy,
    normalize_public_uri,
    normalize_tytul_naukowy,
    normalize_tytul_publikacji,
    normalize_tytul_zrodla,
    normalize_wymiar_etatu,
)

from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.search import TrigramSimilarity

from bpp.models import (
    Autor,
    Autor_Jednostka,
    Dyscyplina_Naukowa,
    Funkcja_Autora,
    Grupa_Pracownicza,
    Jednostka,
    Rekord,
    Tytul,
    Wydawca,
    Wydawnictwo_Ciagle,
    Wydawnictwo_Zwarte,
    Wydzial,
    Wymiar_Etatu,
    Zrodlo,
)
from bpp.util import fail_if_seq_scan


def matchuj_wydzial(nazwa: str | None):
    if nazwa is None:
        return

    try:
        return Wydzial.objects.get(nazwa__iexact=nazwa.strip())
    except Wydzial.DoesNotExist:
        pass


def matchuj_tytul(tytul: str, create_if_not_exist=False) -> Tytul:
    """
    Dostaje tytuł: pełną nazwę albo skrót
    """

    try:
        return Tytul.objects.get(nazwa__iexact=tytul)
    except (Tytul.DoesNotExist, Tytul.MultipleObjectsReturned):
        return Tytul.objects.get(skrot=normalize_tytul_naukowy(tytul))


def matchuj_funkcja_autora(funkcja_autora: str) -> Funkcja_Autora:
    funkcja_autora = normalize_funkcja_autora(funkcja_autora)
    return Funkcja_Autora.objects.get(
        Q(nazwa__iexact=funkcja_autora) | Q(skrot__iexact=funkcja_autora)
    )


def matchuj_grupa_pracownicza(grupa_pracownicza: str) -> Grupa_Pracownicza:
    grupa_pracownicza = normalize_grupa_pracownicza(grupa_pracownicza)
    return Grupa_Pracownicza.objects.get(nazwa__iexact=grupa_pracownicza)


def matchuj_wymiar_etatu(wymiar_etatu: str) -> Wymiar_Etatu:
    wymiar_etatu = normalize_wymiar_etatu(wymiar_etatu)
    return Wymiar_Etatu.objects.get(nazwa__iexact=wymiar_etatu)


def wytnij_skrot(jednostka):
    if jednostka.find("(") >= 0 and jednostka.find(")") >= 0:
        jednostka, skrot = jednostka.split("(", 2)
        jednostka = jednostka.strip()
        skrot = skrot[:-1].strip()
        return jednostka, skrot

    return jednostka, None


def matchuj_jednostke(nazwa, wydzial=None):
    if nazwa is None:
        return

    nazwa = normalize_nazwa_jednostki(nazwa)
    skrot = nazwa

    if "(" in nazwa and ")" in nazwa:
        nazwa_bez_nawiasow, skrot = wytnij_skrot(nazwa)
        try:
            return Jednostka.objects.get(skrot=skrot)
        except Jednostka.DoesNotExist:
            pass

    try:
        return Jednostka.objects.get(Q(nazwa__iexact=nazwa) | Q(skrot__iexact=nazwa))
    except Jednostka.DoesNotExist:
        if nazwa.endswith("."):
            nazwa = nazwa[:-1].strip()

        try:
            return Jednostka.objects.get(
                Q(nazwa__istartswith=nazwa) | Q(skrot__istartswith=nazwa)
            )
        except Jednostka.MultipleObjectsReturned as e:
            if wydzial is None:
                raise e

        return Jednostka.objects.get(
            Q(nazwa__istartswith=nazwa) | Q(skrot__istartswith=nazwa),
            Q(wydzial__nazwa__iexact=wydzial),
        )

    except Jednostka.MultipleObjectsReturned as e:
        if wydzial is None:
            raise e

        return Jednostka.objects.get(
            Q(nazwa__iexact=nazwa) | Q(skrot__iexact=nazwa),
            Q(wydzial__nazwa__iexact=wydzial),
        )


def matchuj_autora(
    imiona: Union[str, None],
    nazwisko: Union[str, None],
    jednostka: Union[Jednostka, None] = None,
    bpp_id: Union[int, None] = None,
    pbn_uid_id: Union[str, None] = None,
    system_kadrowy_id: Union[int, None] = None,
    pbn_id: Union[int, None] = None,
    orcid: Union[str, None] = None,
    tytul_str: Union[Tytul, None] = None,
):
    if bpp_id is not None:
        try:
            return Autor.objects.get(pk=bpp_id)
        except Autor.DoesNotExist:
            pass

    if orcid:
        try:
            return Autor.objects.get(orcid__iexact=orcid.strip())
        except Autor.DoesNotExist:
            pass

    if pbn_uid_id is not None and pbn_uid_id.strip() != "":
        # Może być > 1 autor z takim pbn_uid_id
        _qset = Autor.objects.filter(pbn_uid_id=pbn_uid_id)
        if _qset.exists():
            return _qset.first()

    if system_kadrowy_id is not None:
        try:
            int(system_kadrowy_id)
        except (TypeError, ValueError):
            system_kadrowy_id = None

        if system_kadrowy_id is not None:
            try:
                return Autor.objects.get(system_kadrowy_id=system_kadrowy_id)
            except Autor.DoesNotExist:
                pass

    if pbn_id is not None:
        if isinstance(pbn_id, str):
            pbn_id = pbn_id.strip()

        try:
            pbn_id = int(pbn_id)
        except (TypeError, ValueError):
            pbn_id = None

        if pbn_id is not None:
            try:
                return Autor.objects.get(pbn_id=pbn_id)
            except Autor.DoesNotExist:
                pass

    if imiona is None:
        imiona = ""

    if nazwisko is None:
        nazwisko = ""

    queries = [
        Q(
            Q(nazwisko__iexact=nazwisko.strip())
            | Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
            imiona__iexact=imiona.strip(),
        ),
        Q(
            Q(nazwisko__iexact=nazwisko.strip())
            | Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
            imiona__iexact=imiona.strip().split(" ")[0],
        ),
    ]

    if tytul_str:
        for query in queries[: len(queries)]:
            queries.append(query & Q(tytul__skrot=tytul_str))

    for qry in queries:
        try:
            return Autor.objects.get(qry)
        except (Autor.DoesNotExist, Autor.MultipleObjectsReturned):
            pass

        try:
            return Autor.objects.get(qry & Q(aktualna_jednostka=jednostka))
        except (Autor.MultipleObjectsReturned, Autor.DoesNotExist):
            pass

    # Jesteśmy tutaj. Najwyraźniej poszukiwanie po aktualnej jednostce, imieniu, nazwisku,
    # tytule itp nie bardzo się powiodło. Spróbujmy innej strategii -- jednostka jest
    # określona, poszukajmy w jej autorach. Wszak nie musi być ta jednostka jednostką
    # aktualną...

    if jednostka:
        queries = [
            Q(
                Q(autor__nazwisko__iexact=nazwisko.strip())
                | Q(autor__poprzednie_nazwiska__icontains=nazwisko.strip()),
                autor__imiona__iexact=imiona.strip(),
            )
        ]
        if tytul_str:
            queries.append(queries[0] & Q(autor__tytul__skrot=tytul_str))

        for qry in queries:
            try:
                return jednostka.autor_jednostka_set.get(qry).autor
            except (
                Autor_Jednostka.MultipleObjectsReturned,
                Autor_Jednostka.DoesNotExist,
            ):
                pass

    # Jeżeli nie ma nadal jednego autora który spełnia te kryteria, spróbuj znaleźć konto
    # z ORCIDem i tytułem. W przypadku importów z PBNu często było tak, że autorzy byli zdublowani, ale
    # tylko jeden miał orcid i tytuł:

    try:
        return Autor.objects.get(
            Q(
                Q(nazwisko__iexact=nazwisko.strip())
                | Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
                imiona__iexact=imiona.strip(),
            ),
            orcid__isnull=False,
            tytul_id__isnull=False,
        )
    except (Autor.DoesNotExist, Autor.MultipleObjectsReturned):
        pass

    # .. albo tylko z tytułem:
    try:
        return Autor.objects.get(
            Q(
                Q(nazwisko__iexact=nazwisko.strip())
                | Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
                imiona__iexact=imiona.strip(),
            ),
            tytul_id__isnull=False,
        )
    except (Autor.DoesNotExist, Autor.MultipleObjectsReturned):
        pass

    return None


def matchuj_zrodlo(
    s: Union[str, None],
    issn: Union[str, None] = None,
    e_issn: Union[str, None] = None,
    alt_nazwa=None,
    disable_fuzzy=False,
    disable_skrot=False,
) -> Union[None, Zrodlo]:
    if s is None or str(s) == "":
        return

    if issn is not None:
        try:
            return Zrodlo.objects.get(issn=issn)
        except (Zrodlo.DoesNotExist, Zrodlo.MultipleObjectsReturned):
            pass

    if e_issn is not None:
        try:
            return Zrodlo.objects.get(e_issn=e_issn)
        except (Zrodlo.DoesNotExist, Zrodlo.MultipleObjectsReturned):
            pass

    for elem in s, alt_nazwa:
        if elem is None:
            continue

        elem = normalize_tytul_zrodla(elem)
        try:
            if disable_skrot is True:
                return Zrodlo.objects.get(nazwa__iexact=elem)
            return Zrodlo.objects.get(Q(nazwa__iexact=elem) | Q(skrot__iexact=elem))
        except Zrodlo.MultipleObjectsReturned:
            pass
        except Zrodlo.DoesNotExist:
            if not disable_fuzzy and elem.endswith("."):
                try:
                    return Zrodlo.objects.get(
                        Q(nazwa__istartswith=elem[:-1])
                        | Q(skrot__istartswith=elem[:-1])
                    )
                except Zrodlo.DoesNotExist:
                    pass
                except Zrodlo.MultipleObjectsReturned:
                    pass


def matchuj_dyscypline(kod, nazwa):
    if nazwa:
        for nazwa in [nazwa, nazwa.split("(", 2)[0]]:
            nazwa = normalize_nazwa_dyscypliny(nazwa)
            try:
                return Dyscyplina_Naukowa.objects.get(nazwa=nazwa)
            except Dyscyplina_Naukowa.DoesNotExist:
                pass
            except Dyscyplina_Naukowa.MultipleObjectsReturned:
                pass

    if kod:
        kod = normalize_kod_dyscypliny(kod)
        try:
            return Dyscyplina_Naukowa.objects.get(kod=kod)
        except Dyscyplina_Naukowa.DoesNotExist:
            pass
        except Dyscyplina_Naukowa.MultipleObjectsReturned:
            pass


def matchuj_wydawce(nazwa, pbn_uid_id=None, similarity=0.9):
    nazwa = normalize_nazwa_wydawcy(nazwa)
    try:
        return Wydawca.objects.get(nazwa=nazwa, alias_dla_id=None)
    except Wydawca.DoesNotExist:
        pass

    if pbn_uid_id is not None:
        try:
            return Wydawca.objects.get(pbn_uid_id=pbn_uid_id)
        except Wydawca.DoesNotExist:
            pass

    loose = (
        Wydawca.objects.annotate(similarity=TrigramSimilarity("nazwa", nazwa))
        .filter(similarity__gte=similarity)
        .order_by("-similarity")[:5]
    )
    if loose.count() > 0 and loose.count() < 2:
        return loose.first()


TITLE_LIMIT_SINGLE_WORD = 15
TITLE_LIMIT_MANY_WORDS = 25

MATCH_SIMILARITY_THRESHOLD = 0.95
MATCH_SIMILARITY_THRESHOLD_LOW = 0.90
MATCH_SIMILARITY_THRESHOLD_VERY_LOW = 0.80

# Znormalizowany tytuł w bazie danych -- wyrzucony ciąg znaków [online], podwójne
# spacje pozamieniane na pojedyncze, trim całości
normalized_db_title = Trim(
    Replace(
        Replace(Lower("tytul_oryginalny"), Value(" [online]"), Value("")),
        Value("  "),
        Value(" "),
    )
)

# Znormalizowany skrót nazwy źródła -- wyrzucone spacje i kropki, trim, zmniejszone
# znaki
normalized_db_zrodlo_skrot = Trim(
    Replace(
        Replace(
            Replace(Lower("skrot"), Value(" "), Value("")),
            Value("-"),
            Value(""),
        ),
        Value("."),
        Value(""),
    )
)


def normalize_zrodlo_skrot_for_db_lookup(s):
    return s.lower().replace(" ", "").strip().replace("-", "").replace(".", "")


def normalize_date(s):
    if s is None:
        return s

    if isinstance(s, str):
        s = s.strip()

        if not s:
            return

        return dateutil.parser.parse(s)

    return s


# Znormalizowany skrot zrodla do wyszukiwania -- wyrzucone wszystko procz kropek
normalized_db_zrodlo_nazwa = Trim(
    Replace(Lower("nazwa"), Value(" "), Value("")),
)


def normalize_zrodlo_nazwa_for_db_lookup(s):
    return s.lower().replace(" ", "").strip()


normalized_db_isbn = Trim(Replace(Lower("isbn"), Value("-"), Value("")))


def matchuj_publikacje(
    klass: [Wydawnictwo_Zwarte, Wydawnictwo_Ciagle, Rekord],
    title,
    year,
    doi=None,
    public_uri=None,
    isbn=None,
    zrodlo=None,
    DEBUG_MATCHOWANIE=False,
    isbn_matchuj_tylko_nadrzedne=True,
    doi_matchuj_tylko_nadrzedne=True,
):
    if doi is not None:
        doi = normalize_doi(doi)
        if doi:
            zapytanie = klass.objects.filter(doi__istartswith=doi, rok=year)

            if doi_matchuj_tylko_nadrzedne:
                if hasattr(klass, "wydawnictwo_nadrzedne_id"):
                    zapytanie = zapytanie.filter(wydawnictwo_nadrzedne_id=None)

            res = zapytanie.annotate(
                podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
            ).order_by("-podobienstwo")[:2]
            fail_if_seq_scan(res, DEBUG_MATCHOWANIE)
            if res.exists():
                if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD_VERY_LOW:
                    return res.first()

    title = normalize_tytul_publikacji(title)

    title_has_spaces = False

    if title is not None:
        title_has_spaces = title.find(" ") > 0

    if title is not None and (
        (not title_has_spaces and len(title) >= TITLE_LIMIT_SINGLE_WORD)
        or (title_has_spaces and len(title) >= TITLE_LIMIT_MANY_WORDS)
    ):
        if zrodlo is not None and hasattr(klass, "zrodlo"):
            try:
                return klass.objects.get(
                    tytul_oryginalny__istartswith=title, rok=year, zrodlo=zrodlo
                )
            except klass.DoesNotExist:
                pass
            except klass.MultipleObjectsReturned:
                print(
                    f"PPP ZZZ MultipleObjectsReturned dla title={title} rok={year} zrodlo={zrodlo}"
                )

    if (
        isbn is not None
        and isbn != ""
        and hasattr(klass, "isbn")
        and hasattr(klass, "e_isbn")
    ):
        ni = normalize_isbn(isbn)

        zapytanie = klass.objects.exclude(isbn=None, e_isbn=None).exclude(
            isbn="", e_isbn=""
        )

        if isbn_matchuj_tylko_nadrzedne:
            zapytanie = zapytanie.filter(wydawnictwo_nadrzedne_id=None)

            if klass == Rekord:
                zapytanie = zapytanie.filter(
                    pk__in=[
                        (ContentType.objects.get_for_model(Wydawnictwo_Zwarte).pk, x)
                        for x in Wydawnictwo_Zwarte.objects.wydawnictwa_nadrzedne_dla_innych()
                    ]
                )
            elif klass == Wydawnictwo_Zwarte:
                zapytanie = zapytanie.filter(
                    pk__in=Wydawnictwo_Zwarte.objects.wydawnictwa_nadrzedne_dla_innych()
                )
            else:
                raise NotImplementedError(
                    "Matchowanie po ISBN dla czegoś innego niż wydawnictwo zwarte nie opracowane"
                )

        #
        # Uwaga uwaga uwaga.
        #
        # Gdy matchujemy ISBN, to w BPP dochodzi do takiej nieciekawej sytuacji: wpisywany jest
        # ISBN zarówno dla rozdziałów jak i dla wydawnictw nadrzędnych.
        #
        # Zatem, na ten moment, aby usprawnić matchowanie ISBN, jeżeli ustawiona jest flaga
        # isbn_matchuj_tylko_nadrzedne, to system bedzie szukał tylko i wyłącznie wśród
        # rekordów będących wydawnictwami nadrzędnymi (czyli nie mającymi rekordów podrzędnych)
        #

        res = (
            zapytanie.filter(Q(isbn=ni) | Q(e_isbn=ni))
            .annotate(
                podobienstwo=TrigramSimilarity(
                    normalized_db_title,
                    title.lower(),
                )
            )
            .order_by("-podobienstwo")[:2]
        )
        fail_if_seq_scan(res, DEBUG_MATCHOWANIE)
        if res.exists():
            if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD_VERY_LOW:
                return res.first()

    public_uri = normalize_public_uri(public_uri)
    if public_uri:
        res = (
            klass.objects.filter(Q(www=public_uri) | Q(public_www=public_uri))
            .annotate(
                podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
            )
            .order_by("-podobienstwo")[:2]
        )
        fail_if_seq_scan(res, DEBUG_MATCHOWANIE)
        if res.exists():
            if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD:
                return res.first()

    if title is not None and (
        (not title_has_spaces and len(title) >= TITLE_LIMIT_SINGLE_WORD)
        or (title_has_spaces and len(title) >= TITLE_LIMIT_MANY_WORDS)
    ):
        res = (
            klass.objects.filter(tytul_oryginalny__istartswith=title, rok=year)
            .annotate(
                podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
            )
            .order_by("-podobienstwo")[:2]
        )

        fail_if_seq_scan(res, DEBUG_MATCHOWANIE)
        if res.exists():
            if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD:
                return res.first()

        # Ostatnia szansa, po podobieństwie, niski próg

        res = (
            klass.objects.filter(rok=year)
            .annotate(
                podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
            )
            .order_by("-podobienstwo")[:2]
        )

        fail_if_seq_scan(res, DEBUG_MATCHOWANIE)
        if res.exists():
            if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD_LOW:
                return res.first()


def normalize_kod_dyscypliny_pbn(kod):
    if kod is None:
        raise ValueError("kod = None")

    if kod.find(".") == -1:
        # Nie ma kropki, wiec juz znormalizowany
        return kod

    k1, k2 = (int(x) for x in kod.split(".", 2))
    return f"{k1}{k2:02}"


def matchuj_aktualna_dyscypline_pbn(kod, nazwa):
    kod = normalize_kod_dyscypliny_pbn(kod)

    from pbn_api.models import Discipline

    from django.utils import timezone

    d = timezone.now().date()
    parent_group_args = Q(parent_group__validityDateFrom__lte=d), Q(
        parent_group__validityDateTo=None
    ) | Q(parent_group__validityDateTo__gt=d)

    try:
        return Discipline.objects.get(*parent_group_args, code=kod)
    except Discipline.DoesNotExist:
        pass

    try:
        return Discipline.objects.get(*parent_group_args, name=nazwa)
    except Discipline.DoesNotExist:
        pass


def matchuj_nieaktualna_dyscypline_pbn(kod, nazwa, rok_min=2018, rok_max=2022):
    kod = normalize_kod_dyscypliny_pbn(kod)

    from pbn_api.models import Discipline

    nieaktualna_parent_group_args = Q(parent_group__validityDateFrom__year=rok_min), Q(
        parent_group__validityDateTo__year=rok_max
    )
    try:
        return Discipline.objects.get(*nieaktualna_parent_group_args, code=kod)
    except Discipline.DoesNotExist:
        pass

    try:
        return Discipline.objects.get(*nieaktualna_parent_group_args, name=nazwa)
    except Discipline.DoesNotExist:
        pass


def matchuj_uczelnie(nazwa):
    from pbn_api.models import Institution

    try:
        return Institution.objects.get(name=nazwa)
    except Institution.DoesNotExist:
        pass

    res = (
        Institution.objects.annotate(similarity=TrigramSimilarity("name", nazwa))
        .filter(similarity__gte=0.8)
        .order_by("-similarity")
    )

    if res.count() == 1:
        return res.first()

1	from typing import Union	4✔
2
3	import dateutil	4✔
4	from django.db.models import Q, Value	4✔
5	from django.db.models.functions import Lower, Replace, Trim	4✔
6
7	from .normalization import (	4✔
8	normalize_doi,
9	normalize_funkcja_autora,
10	normalize_grupa_pracownicza,
11	normalize_isbn,
12	normalize_kod_dyscypliny,
13	normalize_nazwa_dyscypliny,
14	normalize_nazwa_jednostki,
15	normalize_nazwa_wydawcy,
16	normalize_public_uri,
17	normalize_tytul_naukowy,
18	normalize_tytul_publikacji,
19	normalize_tytul_zrodla,
20	normalize_wymiar_etatu,
21	)
22
23	from django.contrib.contenttypes.models import ContentType	4✔
24	from django.contrib.postgres.search import TrigramSimilarity	4✔
25
26	from bpp.models import (	4✔
27	Autor,
28	Autor_Jednostka,
29	Dyscyplina_Naukowa,
30	Funkcja_Autora,
31	Grupa_Pracownicza,
32	Jednostka,
33	Rekord,
34	Tytul,
35	Wydawca,
36	Wydawnictwo_Ciagle,
37	Wydawnictwo_Zwarte,
38	Wydzial,
39	Wymiar_Etatu,
40	Zrodlo,
41	)
42	from bpp.util import fail_if_seq_scan	4✔
43
44
45	def matchuj_wydzial(nazwa: str \| None):	4✔
46	if nazwa is None:	×
47	return	×
48
49	try:	×
50	return Wydzial.objects.get(nazwa__iexact=nazwa.strip())	×
51	except Wydzial.DoesNotExist:	×
52	pass	×
53
54
55	def matchuj_tytul(tytul: str, create_if_not_exist=False) -> Tytul:	4✔
56	"""
57	Dostaje tytuł: pełną nazwę albo skrót
58	"""
59
60	try:	×
61	return Tytul.objects.get(nazwa__iexact=tytul)	×
62	except (Tytul.DoesNotExist, Tytul.MultipleObjectsReturned):	×
63	return Tytul.objects.get(skrot=normalize_tytul_naukowy(tytul))	×
64
65
66	def matchuj_funkcja_autora(funkcja_autora: str) -> Funkcja_Autora:	4✔
67	funkcja_autora = normalize_funkcja_autora(funkcja_autora)	×
68	return Funkcja_Autora.objects.get(	×
69	Q(nazwa__iexact=funkcja_autora) \| Q(skrot__iexact=funkcja_autora)
70	)
71
72
73	def matchuj_grupa_pracownicza(grupa_pracownicza: str) -> Grupa_Pracownicza:	4✔
74	grupa_pracownicza = normalize_grupa_pracownicza(grupa_pracownicza)	×
75	return Grupa_Pracownicza.objects.get(nazwa__iexact=grupa_pracownicza)	×
76
77
78	def matchuj_wymiar_etatu(wymiar_etatu: str) -> Wymiar_Etatu:	4✔
79	wymiar_etatu = normalize_wymiar_etatu(wymiar_etatu)	×
80	return Wymiar_Etatu.objects.get(nazwa__iexact=wymiar_etatu)	×
81
82
83	def wytnij_skrot(jednostka):	4✔
84	if jednostka.find("(") >= 0 and jednostka.find(")") >= 0:	×
85	jednostka, skrot = jednostka.split("(", 2)	×
86	jednostka = jednostka.strip()	×
87	skrot = skrot[:-1].strip()	×
88	return jednostka, skrot	×
89
90	return jednostka, None	×
91
92
93	def matchuj_jednostke(nazwa, wydzial=None):	4✔
94	if nazwa is None:	×
95	return	×
96
97	nazwa = normalize_nazwa_jednostki(nazwa)	×
98	skrot = nazwa	×
99
100	if "(" in nazwa and ")" in nazwa:	×
101	nazwa_bez_nawiasow, skrot = wytnij_skrot(nazwa)	×
102	try:	×
103	return Jednostka.objects.get(skrot=skrot)	×
104	except Jednostka.DoesNotExist:	×
105	pass	×
106
107	try:	×
108	return Jednostka.objects.get(Q(nazwa__iexact=nazwa) \| Q(skrot__iexact=nazwa))	×
109	except Jednostka.DoesNotExist:	×
110	if nazwa.endswith("."):	×
111	nazwa = nazwa[:-1].strip()	×
112
113	try:	×
114	return Jednostka.objects.get(	×
115	Q(nazwa__istartswith=nazwa) \| Q(skrot__istartswith=nazwa)
116	)
117	except Jednostka.MultipleObjectsReturned as e:	×
118	if wydzial is None:	×
119	raise e	×
120
121	return Jednostka.objects.get(	×
122	Q(nazwa__istartswith=nazwa) \| Q(skrot__istartswith=nazwa),
123	Q(wydzial__nazwa__iexact=wydzial),
124	)
125
126	except Jednostka.MultipleObjectsReturned as e:	×
127	if wydzial is None:	×
128	raise e	×
129
130	return Jednostka.objects.get(	×
131	Q(nazwa__iexact=nazwa) \| Q(skrot__iexact=nazwa),
132	Q(wydzial__nazwa__iexact=wydzial),
133	)
134
135
136	def matchuj_autora(	4✔
137	imiona: Union[str, None],
138	nazwisko: Union[str, None],
139	jednostka: Union[Jednostka, None] = None,
140	bpp_id: Union[int, None] = None,
141	pbn_uid_id: Union[str, None] = None,
142	system_kadrowy_id: Union[int, None] = None,
143	pbn_id: Union[int, None] = None,
144	orcid: Union[str, None] = None,
145	tytul_str: Union[Tytul, None] = None,
146	):
UNCOV 147	if bpp_id is not None:	1✔
148	try:	×
149	return Autor.objects.get(pk=bpp_id)	×
150	except Autor.DoesNotExist:	×
151	pass	×
152
UNCOV 153	if orcid:	1✔
154	try:	×
155	return Autor.objects.get(orcid__iexact=orcid.strip())	×
156	except Autor.DoesNotExist:	×
157	pass	×
158
UNCOV 159	if pbn_uid_id is not None and pbn_uid_id.strip() != "":	1✔
160	# Może być > 1 autor z takim pbn_uid_id
161	_qset = Autor.objects.filter(pbn_uid_id=pbn_uid_id)	×
162	if _qset.exists():	×
163	return _qset.first()	×
164
UNCOV 165	if system_kadrowy_id is not None:	1✔
166	try:	×
167	int(system_kadrowy_id)	×
168	except (TypeError, ValueError):	×
169	system_kadrowy_id = None	×
170
171	if system_kadrowy_id is not None:	×
172	try:	×
173	return Autor.objects.get(system_kadrowy_id=system_kadrowy_id)	×
174	except Autor.DoesNotExist:	×
175	pass	×
176
UNCOV 177	if pbn_id is not None:	1✔
178	if isinstance(pbn_id, str):	×
179	pbn_id = pbn_id.strip()	×
180
181	try:	×
182	pbn_id = int(pbn_id)	×
183	except (TypeError, ValueError):	×
184	pbn_id = None	×
185
186	if pbn_id is not None:	×
187	try:	×
188	return Autor.objects.get(pbn_id=pbn_id)	×
189	except Autor.DoesNotExist:	×
190	pass	×
191
UNCOV 192	if imiona is None:	1✔
193	imiona = ""	×
194
UNCOV 195	if nazwisko is None:	1✔
196	nazwisko = ""	×
197
UNCOV 198	queries = [	1✔
199	Q(
200	Q(nazwisko__iexact=nazwisko.strip())
201	\| Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
202	imiona__iexact=imiona.strip(),
203	),
204	Q(
205	Q(nazwisko__iexact=nazwisko.strip())
206	\| Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
207	imiona__iexact=imiona.strip().split(" ")[0],
208	),
209	]
210
UNCOV 211	if tytul_str:	1✔
212	for query in queries[: len(queries)]:	×
213	queries.append(query & Q(tytul__skrot=tytul_str))	×
214
UNCOV 215	for qry in queries:	1✔
UNCOV 216	try:	1✔
UNCOV 217	return Autor.objects.get(qry)	1✔
UNCOV 218	except (Autor.DoesNotExist, Autor.MultipleObjectsReturned):	1✔
UNCOV 219	pass	1✔
220
UNCOV 221	try:	1✔
UNCOV 222	return Autor.objects.get(qry & Q(aktualna_jednostka=jednostka))	1✔
UNCOV 223	except (Autor.MultipleObjectsReturned, Autor.DoesNotExist):	1✔
UNCOV 224	pass	1✔
225
226	# Jesteśmy tutaj. Najwyraźniej poszukiwanie po aktualnej jednostce, imieniu, nazwisku,
227	# tytule itp nie bardzo się powiodło. Spróbujmy innej strategii -- jednostka jest
228	# określona, poszukajmy w jej autorach. Wszak nie musi być ta jednostka jednostką
229	# aktualną...
230
UNCOV 231	if jednostka:	1✔
232	queries = [	×
233	Q(
234	Q(autor__nazwisko__iexact=nazwisko.strip())
235	\| Q(autor__poprzednie_nazwiska__icontains=nazwisko.strip()),
236	autor__imiona__iexact=imiona.strip(),
237	)
238	]
239	if tytul_str:	×
240	queries.append(queries[0] & Q(autor__tytul__skrot=tytul_str))	×
241
242	for qry in queries:	×
243	try:	×
244	return jednostka.autor_jednostka_set.get(qry).autor	×
245	except (	×
246	Autor_Jednostka.MultipleObjectsReturned,
247	Autor_Jednostka.DoesNotExist,
248	):
249	pass	×
250
251	# Jeżeli nie ma nadal jednego autora który spełnia te kryteria, spróbuj znaleźć konto
252	# z ORCIDem i tytułem. W przypadku importów z PBNu często było tak, że autorzy byli zdublowani, ale
253	# tylko jeden miał orcid i tytuł:
254
UNCOV 255	try:	1✔
UNCOV 256	return Autor.objects.get(	1✔
257	Q(
258	Q(nazwisko__iexact=nazwisko.strip())
259	\| Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
260	imiona__iexact=imiona.strip(),
261	),
262	orcid__isnull=False,
263	tytul_id__isnull=False,
264	)
UNCOV 265	except (Autor.DoesNotExist, Autor.MultipleObjectsReturned):	1✔
UNCOV 266	pass	1✔
267
268	# .. albo tylko z tytułem:
UNCOV 269	try:	1✔
UNCOV 270	return Autor.objects.get(	1✔
271	Q(
272	Q(nazwisko__iexact=nazwisko.strip())
273	\| Q(poprzednie_nazwiska__icontains=nazwisko.strip()),
274	imiona__iexact=imiona.strip(),
275	),
276	tytul_id__isnull=False,
277	)
UNCOV 278	except (Autor.DoesNotExist, Autor.MultipleObjectsReturned):	1✔
UNCOV 279	pass	1✔
280
UNCOV 281	return None	1✔
282
283
284	def matchuj_zrodlo(	4✔
285	s: Union[str, None],
286	issn: Union[str, None] = None,
287	e_issn: Union[str, None] = None,
288	alt_nazwa=None,
289	disable_fuzzy=False,
290	disable_skrot=False,
291	) -> Union[None, Zrodlo]:
292	if s is None or str(s) == "":	×
293	return	×
294
295	if issn is not None:	×
296	try:	×
297	return Zrodlo.objects.get(issn=issn)	×
298	except (Zrodlo.DoesNotExist, Zrodlo.MultipleObjectsReturned):	×
299	pass	×
300
301	if e_issn is not None:	×
302	try:	×
303	return Zrodlo.objects.get(e_issn=e_issn)	×
304	except (Zrodlo.DoesNotExist, Zrodlo.MultipleObjectsReturned):	×
305	pass	×
306
307	for elem in s, alt_nazwa:	×
308	if elem is None:	×
309	continue	×
310
311	elem = normalize_tytul_zrodla(elem)	×
312	try:	×
313	if disable_skrot is True:	×
314	return Zrodlo.objects.get(nazwa__iexact=elem)	×
315	return Zrodlo.objects.get(Q(nazwa__iexact=elem) \| Q(skrot__iexact=elem))	×
316	except Zrodlo.MultipleObjectsReturned:	×
317	pass	×
318	except Zrodlo.DoesNotExist:	×
319	if not disable_fuzzy and elem.endswith("."):	×
320	try:	×
321	return Zrodlo.objects.get(	×
322	Q(nazwa__istartswith=elem[:-1])
323	\| Q(skrot__istartswith=elem[:-1])
324	)
325	except Zrodlo.DoesNotExist:	×
326	pass	×
327	except Zrodlo.MultipleObjectsReturned:	×
328	pass	×
329
330
331	def matchuj_dyscypline(kod, nazwa):	4✔
332	if nazwa:	×
333	for nazwa in [nazwa, nazwa.split("(", 2)[0]]:	×
334	nazwa = normalize_nazwa_dyscypliny(nazwa)	×
335	try:	×
336	return Dyscyplina_Naukowa.objects.get(nazwa=nazwa)	×
337	except Dyscyplina_Naukowa.DoesNotExist:	×
338	pass	×
339	except Dyscyplina_Naukowa.MultipleObjectsReturned:	×
340	pass	×
341
342	if kod:	×
343	kod = normalize_kod_dyscypliny(kod)	×
344	try:	×
345	return Dyscyplina_Naukowa.objects.get(kod=kod)	×
346	except Dyscyplina_Naukowa.DoesNotExist:	×
347	pass	×
348	except Dyscyplina_Naukowa.MultipleObjectsReturned:	×
349	pass	×
350
351
352	def matchuj_wydawce(nazwa, pbn_uid_id=None, similarity=0.9):	4✔
UNCOV 353	nazwa = normalize_nazwa_wydawcy(nazwa)	1✔
UNCOV 354	try:	1✔
UNCOV 355	return Wydawca.objects.get(nazwa=nazwa, alias_dla_id=None)	1✔
UNCOV 356	except Wydawca.DoesNotExist:	1✔
UNCOV 357	pass	1✔
358
UNCOV 359	if pbn_uid_id is not None:	1✔
360	try:	×
361	return Wydawca.objects.get(pbn_uid_id=pbn_uid_id)	×
362	except Wydawca.DoesNotExist:	×
363	pass	×
364
UNCOV 365	loose = (	1✔
366	Wydawca.objects.annotate(similarity=TrigramSimilarity("nazwa", nazwa))
367	.filter(similarity__gte=similarity)
368	.order_by("-similarity")[:5]
369	)
UNCOV 370	if loose.count() > 0 and loose.count() < 2:	1✔
371	return loose.first()	×
372
373
374	TITLE_LIMIT_SINGLE_WORD = 15	4✔
375	TITLE_LIMIT_MANY_WORDS = 25	4✔
376
377	MATCH_SIMILARITY_THRESHOLD = 0.95	4✔
378	MATCH_SIMILARITY_THRESHOLD_LOW = 0.90	4✔
379	MATCH_SIMILARITY_THRESHOLD_VERY_LOW = 0.80	4✔
380
381	# Znormalizowany tytuł w bazie danych -- wyrzucony ciąg znaków [online], podwójne
382	# spacje pozamieniane na pojedyncze, trim całości
383	normalized_db_title = Trim(	4✔
384	Replace(
385	Replace(Lower("tytul_oryginalny"), Value(" [online]"), Value("")),
386	Value(" "),
387	Value(" "),
388	)
389	)
390
391	# Znormalizowany skrót nazwy źródła -- wyrzucone spacje i kropki, trim, zmniejszone
392	# znaki
393	normalized_db_zrodlo_skrot = Trim(	4✔
394	Replace(
395	Replace(
396	Replace(Lower("skrot"), Value(" "), Value("")),
397	Value("-"),
398	Value(""),
399	),
400	Value("."),
401	Value(""),
402	)
403	)
404
405
406	def normalize_zrodlo_skrot_for_db_lookup(s):	4✔
UNCOV 407	return s.lower().replace(" ", "").strip().replace("-", "").replace(".", "")	1✔
408
409
410	def normalize_date(s):	4✔
411	if s is None:	×
412	return s	×
413
414	if isinstance(s, str):	×
415	s = s.strip()	×
416
417	if not s:	×
418	return	×
419
420	return dateutil.parser.parse(s)	×
421
422	return s	×
423
424
425	# Znormalizowany skrot zrodla do wyszukiwania -- wyrzucone wszystko procz kropek
426	normalized_db_zrodlo_nazwa = Trim(	4✔
427	Replace(Lower("nazwa"), Value(" "), Value("")),
428	)
429
430
431	def normalize_zrodlo_nazwa_for_db_lookup(s):	4✔
UNCOV 432	return s.lower().replace(" ", "").strip()	1✔
433
434
435	normalized_db_isbn = Trim(Replace(Lower("isbn"), Value("-"), Value("")))	4✔
436
437
438	def matchuj_publikacje(	4✔
439	klass: [Wydawnictwo_Zwarte, Wydawnictwo_Ciagle, Rekord],
440	title,
441	year,
442	doi=None,
443	public_uri=None,
444	isbn=None,
445	zrodlo=None,
446	DEBUG_MATCHOWANIE=False,
447	isbn_matchuj_tylko_nadrzedne=True,
448	doi_matchuj_tylko_nadrzedne=True,
449	):
450	if doi is not None:	×
451	doi = normalize_doi(doi)	×
452	if doi:	×
453	zapytanie = klass.objects.filter(doi__istartswith=doi, rok=year)	×
454
455	if doi_matchuj_tylko_nadrzedne:	×
456	if hasattr(klass, "wydawnictwo_nadrzedne_id"):	×
457	zapytanie = zapytanie.filter(wydawnictwo_nadrzedne_id=None)	×
458
459	res = zapytanie.annotate(	×
460	podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
461	).order_by("-podobienstwo")[:2]
462	fail_if_seq_scan(res, DEBUG_MATCHOWANIE)	×
463	if res.exists():	×
464	if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD_VERY_LOW:	×
465	return res.first()	×
466
467	title = normalize_tytul_publikacji(title)	×
468
469	title_has_spaces = False	×
470
471	if title is not None:	×
472	title_has_spaces = title.find(" ") > 0	×
473
474	if title is not None and (	×
475	(not title_has_spaces and len(title) >= TITLE_LIMIT_SINGLE_WORD)
476	or (title_has_spaces and len(title) >= TITLE_LIMIT_MANY_WORDS)
477	):
478	if zrodlo is not None and hasattr(klass, "zrodlo"):	×
479	try:	×
480	return klass.objects.get(	×
481	tytul_oryginalny__istartswith=title, rok=year, zrodlo=zrodlo
482	)
483	except klass.DoesNotExist:	×
484	pass	×
485	except klass.MultipleObjectsReturned:	×
486	print(	×
487	f"PPP ZZZ MultipleObjectsReturned dla title={title} rok={year} zrodlo={zrodlo}"
488	)
489
490	if (	×
491	isbn is not None
492	and isbn != ""
493	and hasattr(klass, "isbn")
494	and hasattr(klass, "e_isbn")
495	):
496	ni = normalize_isbn(isbn)	×
497
498	zapytanie = klass.objects.exclude(isbn=None, e_isbn=None).exclude(	×
499	isbn="", e_isbn=""
500	)
501
502	if isbn_matchuj_tylko_nadrzedne:	×
503	zapytanie = zapytanie.filter(wydawnictwo_nadrzedne_id=None)	×
504
505	if klass == Rekord:	×
506	zapytanie = zapytanie.filter(	×
507	pk__in=[
508	(ContentType.objects.get_for_model(Wydawnictwo_Zwarte).pk, x)
509	for x in Wydawnictwo_Zwarte.objects.wydawnictwa_nadrzedne_dla_innych()
510	]
511	)
512	elif klass == Wydawnictwo_Zwarte:	×
513	zapytanie = zapytanie.filter(	×
514	pk__in=Wydawnictwo_Zwarte.objects.wydawnictwa_nadrzedne_dla_innych()
515	)
516	else:
517	raise NotImplementedError(	×
518	"Matchowanie po ISBN dla czegoś innego niż wydawnictwo zwarte nie opracowane"
519	)
520
521	#
522	# Uwaga uwaga uwaga.
523	#
524	# Gdy matchujemy ISBN, to w BPP dochodzi do takiej nieciekawej sytuacji: wpisywany jest
525	# ISBN zarówno dla rozdziałów jak i dla wydawnictw nadrzędnych.
526	#
527	# Zatem, na ten moment, aby usprawnić matchowanie ISBN, jeżeli ustawiona jest flaga
528	# isbn_matchuj_tylko_nadrzedne, to system bedzie szukał tylko i wyłącznie wśród
529	# rekordów będących wydawnictwami nadrzędnymi (czyli nie mającymi rekordów podrzędnych)
530	#
531
532	res = (	×
533	zapytanie.filter(Q(isbn=ni) \| Q(e_isbn=ni))
534	.annotate(
535	podobienstwo=TrigramSimilarity(
536	normalized_db_title,
537	title.lower(),
538	)
539	)
540	.order_by("-podobienstwo")[:2]
541	)
542	fail_if_seq_scan(res, DEBUG_MATCHOWANIE)	×
543	if res.exists():	×
544	if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD_VERY_LOW:	×
545	return res.first()	×
546
547	public_uri = normalize_public_uri(public_uri)	×
548	if public_uri:	×
549	res = (	×
550	klass.objects.filter(Q(www=public_uri) \| Q(public_www=public_uri))
551	.annotate(
552	podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
553	)
554	.order_by("-podobienstwo")[:2]
555	)
556	fail_if_seq_scan(res, DEBUG_MATCHOWANIE)	×
557	if res.exists():	×
558	if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD:	×
559	return res.first()	×
560
561	if title is not None and (	×
562	(not title_has_spaces and len(title) >= TITLE_LIMIT_SINGLE_WORD)
563	or (title_has_spaces and len(title) >= TITLE_LIMIT_MANY_WORDS)
564	):
565	res = (	×
566	klass.objects.filter(tytul_oryginalny__istartswith=title, rok=year)
567	.annotate(
568	podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
569	)
570	.order_by("-podobienstwo")[:2]
571	)
572
573	fail_if_seq_scan(res, DEBUG_MATCHOWANIE)	×
574	if res.exists():	×
575	if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD:	×
576	return res.first()	×
577
578	# Ostatnia szansa, po podobieństwie, niski próg
579
580	res = (	×
581	klass.objects.filter(rok=year)
582	.annotate(
583	podobienstwo=TrigramSimilarity(normalized_db_title, title.lower())
584	)
585	.order_by("-podobienstwo")[:2]
586	)
587
588	fail_if_seq_scan(res, DEBUG_MATCHOWANIE)	×
589	if res.exists():	×
590	if res.first().podobienstwo >= MATCH_SIMILARITY_THRESHOLD_LOW:	×
591	return res.first()	×
592
593
594	def normalize_kod_dyscypliny_pbn(kod):	4✔
595	if kod is None:	×
596	raise ValueError("kod = None")	×
597
598	if kod.find(".") == -1:	×
599	# Nie ma kropki, wiec juz znormalizowany
600	return kod	×
601
602	k1, k2 = (int(x) for x in kod.split(".", 2))	×
603	return f"{k1}{k2:02}"	×
604
605
606	def matchuj_aktualna_dyscypline_pbn(kod, nazwa):	4✔
607	kod = normalize_kod_dyscypliny_pbn(kod)	×
608
609	from pbn_api.models import Discipline	×
610
611	from django.utils import timezone	×
612
613	d = timezone.now().date()	×
614	parent_group_args = Q(parent_group__validityDateFrom__lte=d), Q(	×
615	parent_group__validityDateTo=None
616	) \| Q(parent_group__validityDateTo__gt=d)
617
618	try:	×
619	return Discipline.objects.get(*parent_group_args, code=kod)	×
620	except Discipline.DoesNotExist:	×
621	pass	×
622
623	try:	×
624	return Discipline.objects.get(*parent_group_args, name=nazwa)	×
625	except Discipline.DoesNotExist:	×
626	pass	×
627
628
629	def matchuj_nieaktualna_dyscypline_pbn(kod, nazwa, rok_min=2018, rok_max=2022):	4✔
630	kod = normalize_kod_dyscypliny_pbn(kod)	×
631
632	from pbn_api.models import Discipline	×
633
634	nieaktualna_parent_group_args = Q(parent_group__validityDateFrom__year=rok_min), Q(	×
635	parent_group__validityDateTo__year=rok_max
636	)
637	try:	×
638	return Discipline.objects.get(*nieaktualna_parent_group_args, code=kod)	×
639	except Discipline.DoesNotExist:	×
640	pass	×
641
642	try:	×
643	return Discipline.objects.get(*nieaktualna_parent_group_args, name=nazwa)	×
644	except Discipline.DoesNotExist:	×
645	pass	×
646
647
648	def matchuj_uczelnie(nazwa):	4✔
649	from pbn_api.models import Institution	×
650
651	try:	×
652	return Institution.objects.get(name=nazwa)	×
653	except Institution.DoesNotExist:	×
654	pass	×
655
656	res = (	×
657	Institution.objects.annotate(similarity=TrigramSimilarity("name", nazwa))
658	.filter(similarity__gte=0.8)
659	.order_by("-similarity")
660	)
661
662	if res.count() == 1:	×
663	return res.first()	×

iplweb / bpp / 58b9a630-8512-44e6-b730-daac96d1c4d7

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