d739ad7a-8bce-4087-b0c8-83f9ea367ad1

Committed 18 Feb 2025 12:47AM UTC coverage: 48.187% (+0.7%) from 47.492%

Build # d739ad7a-8bce-4087-b0c8-83f9ea367ad1

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Merge branch 'release/v202502.1157'

Run Details

172 of 381 new or added lines in 31 files covered. (45.14%)

802 existing lines in 49 files now uncovered.

17072 of 35429 relevant lines covered (48.19%)

1.23 hits per line

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

42.41

src/bpp/admin/helpers.py

from urllib.parse import parse_qs
from urllib.parse import quote as urlquote

from django import forms
from django.db import models
from django.db.models import Q
from django.forms import BaseInlineFormSet
from django.forms.widgets import Textarea
from django.urls import reverse
from sentry_sdk import capture_exception

from import_common.normalization import normalize_isbn
from pbn_api.exceptions import (
    AccessDeniedException,
    AlreadyEnqueuedError,
    BrakZdefiniowanegoObiektuUczelniaWSystemieError,
    CharakterFormalnyNieobslugiwanyError,
    NeedsPBNAuthorisationException,
    PKZeroExportDisabled,
    PraceSerwisoweException,
    SameDataUploadedRecently,
)
from pbn_api.models import PBN_Export_Queue, SentData
from pbn_api.tasks import task_sprobuj_wyslac_do_pbn

from django.contrib import messages
from django.contrib.admin.utils import quote
from django.contrib.contenttypes.models import ContentType

from django.utils.html import format_html
from django.utils.safestring import mark_safe

from bpp import const
from bpp.const import PBN_MAX_ROK, PBN_MIN_ROK
from bpp.models import Status_Korekty, Uczelnia
from bpp.models.sloty.core import ISlot
from bpp.models.sloty.exceptions import CannotAdapt

CHARMAP_SINGLE_LINE = forms.TextInput(
    attrs={"class": "charmap", "style": "width: 500px"}
)


# Pomocnik dla klasy ModelZMetryczka


class ZapiszZAdnotacjaMixin:
    readonly_fields = ("ostatnio_zmieniony",)


class AdnotacjeZDatamiMixin:
    readonly_fields = ("utworzono", "ostatnio_zmieniony", "id")


class AdnotacjeZDatamiOrazPBNMixin:
    readonly_fields = (
        "utworzono",
        "ostatnio_zmieniony",
        "id",
        "pbn_id",
    )


ADNOTACJE_FIELDSET = (
    "Adnotacje",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": (ZapiszZAdnotacjaMixin.readonly_fields + ("adnotacje",)),
    },
)

ADNOTACJE_Z_DATAMI_FIELDSET = (
    "Adnotacje",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": AdnotacjeZDatamiMixin.readonly_fields + ("adnotacje",),
    },
)

ADNOTACJE_Z_DATAMI_ORAZ_PBN_FIELDSET = (
    "Adnotacje",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": AdnotacjeZDatamiOrazPBNMixin.readonly_fields + ("adnotacje",),
    },
)

OPENACCESS_FIELDSET = (
    "OpenAccess",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": (
            "openaccess_tryb_dostepu",
            "openaccess_licencja",
            "openaccess_wersja_tekstu",
            "openaccess_czas_publikacji",
            "openaccess_ilosc_miesiecy",
        ),
    },
)

DWA_TYTULY = (
    "tytul_oryginalny",
    "tytul",
)

MODEL_ZE_SZCZEGOLAMI = (
    "informacje",
    "szczegoly",
    "uwagi",
    "slowa_kluczowe",
    "slowa_kluczowe_eng",
    "strony",
    "tom",
)

MODEL_Z_ISSN = (
    "issn",
    "e_issn",
)

MODEL_Z_PBN_UID = ("pbn_uid",)

MODEL_Z_OPLATA_ZA_PUBLIKACJE = (
    "opl_pub_cost_free",
    "opl_pub_research_potential",
    "opl_pub_research_or_development_projects",
    "opl_pub_other",
    "opl_pub_amount",
)

MODEL_Z_OPLATA_ZA_PUBLIKACJE_FIELDSET = (
    "Opłata za publikację",
    {"classes": ("grp-collapse grp-closed",), "fields": MODEL_Z_OPLATA_ZA_PUBLIKACJE},
)

MODEL_Z_ISBN = (
    "isbn",
    "e_isbn",
)

MODEL_Z_WWW = (
    "www",
    "dostep_dnia",
    "public_www",
    "public_dostep_dnia",
)

MODEL_Z_PUBMEDID = ("pubmed_id", "pmc_id")

MODEL_Z_DOI = ("doi",)

MODEL_Z_LICZBA_CYTOWAN = ("liczba_cytowan",)

MODEL_Z_MIEJSCEM_PRZECHOWYWANIA = ("numer_odbitki",)

MODEL_Z_ROKIEM = ("rok",)

MODEL_TYPOWANY = (
    "jezyk",
    "jezyk_alt",
    "jezyk_orig",
    "typ_kbn",
)

MODEL_PUNKTOWANY_BAZA = (
    "punkty_kbn",
    "impact_factor",
    "index_copernicus",
    "punktacja_snip",
    "punktacja_wewnetrzna",
)

MODEL_PUNKTOWANY = MODEL_PUNKTOWANY_BAZA + ("weryfikacja_punktacji",)

MODEL_PUNKTOWANY_Z_KWARTYLAMI_BAZA = MODEL_PUNKTOWANY_BAZA + (
    "kwartyl_w_scopus",
    "kwartyl_w_wos",
)

MODEL_PUNKTOWANY_Z_KWARTYLAMI = MODEL_PUNKTOWANY_Z_KWARTYLAMI_BAZA + (
    "weryfikacja_punktacji",
)

MODEL_Z_INFORMACJA_Z = ("informacja_z",)

MODEL_Z_LICZBA_ZNAKOW_WYDAWNICZYCH = ("liczba_znakow_wydawniczych",)

MODEL_ZE_STATUSEM = ("status_korekty",)

MODEL_RECENZOWANY = ("recenzowana",)

MODEL_TYPOWANY_BEZ_CHARAKTERU_FIELDSET = (
    "Typ",
    {"classes": ("",), "fields": MODEL_TYPOWANY},
)

MODEL_TYPOWANY_FIELDSET = (
    "Typ",
    {"classes": ("",), "fields": ("charakter_formalny",) + MODEL_TYPOWANY},
)

MODEL_PUNKTOWANY_FIELDSET = (
    "Punktacja",
    {"classes": ("",), "fields": MODEL_PUNKTOWANY},
)

MODEL_PUNKTOWANY_WYDAWNICTWO_CIAGLE_FIELDSET = (
    "Punktacja",
    {
        "classes": ("",),
        "fields": MODEL_PUNKTOWANY_Z_KWARTYLAMI + ("uzupelnij_punktacje",),
    },
)

MODEL_OPCJONALNIE_NIE_EKSPORTOWANY_DO_API_FIELDSET = (
    "Eksport do API",
    {"classes": ("grp-collapse grp-closed",), "fields": ("nie_eksportuj_przez_api",)},
)

POZOSTALE_MODELE_FIELDSET = (
    "Pozostałe informacje",
    {
        "classes": ("",),
        "fields": MODEL_Z_INFORMACJA_Z + MODEL_ZE_STATUSEM + MODEL_RECENZOWANY,
    },
)

POZOSTALE_MODELE_WYDAWNICTWO_CIAGLE_FIELDSET = (
    "Pozostałe informacje",
    {
        "classes": ("",),
        "fields": MODEL_Z_LICZBA_ZNAKOW_WYDAWNICZYCH
        + MODEL_Z_INFORMACJA_Z
        + MODEL_ZE_STATUSEM
        + MODEL_RECENZOWANY,
    },
)

POZOSTALE_MODELE_WYDAWNICTWO_ZWARTE_FIELDSET = (
    "Pozostałe informacje",
    {
        "classes": ("",),
        "fields": MODEL_Z_LICZBA_ZNAKOW_WYDAWNICZYCH
        + MODEL_Z_INFORMACJA_Z
        + MODEL_ZE_STATUSEM
        + MODEL_RECENZOWANY,
    },
)

SERIA_WYDAWNICZA_FIELDSET = (
    "Seria wydawnicza",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": ("seria_wydawnicza", "numer_w_serii"),
    },
)

PRACA_WYBITNA_FIELDSET = (
    "Praca wybitna",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": ("praca_wybitna", "uzasadnienie_wybitnosci"),
    },
)

PRZED_PO_LISCIE_AUTOROW_FIELDSET = (
    "Dowolny tekst przed lub po liście autorów",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": ("tekst_przed_pierwszym_autorem", "tekst_po_ostatnim_autorze"),
    },
)

EKSTRA_INFORMACJE_WYDAWNICTWO_CIAGLE_FIELDSET = (
    "Ekstra informacje",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": MODEL_Z_PBN_UID
        + MODEL_Z_ISSN
        + MODEL_Z_WWW
        + MODEL_Z_PUBMEDID
        + MODEL_Z_DOI
        + MODEL_Z_LICZBA_CYTOWAN
        + MODEL_Z_MIEJSCEM_PRZECHOWYWANIA,
    },
)

EKSTRA_INFORMACJE_WYDAWNICTWO_ZWARTE_FIELDSET = (
    "Ekstra informacje",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": MODEL_Z_PBN_UID
        + MODEL_Z_ISSN
        + MODEL_Z_WWW
        + MODEL_Z_PUBMEDID
        + MODEL_Z_DOI
        + MODEL_Z_LICZBA_CYTOWAN
        + MODEL_Z_MIEJSCEM_PRZECHOWYWANIA,
    },
)

EKSTRA_INFORMACJE_DOKTORSKA_HABILITACYJNA_FIELDSET = (
    "Ekstra informacje",
    {
        "classes": ("grp-collapse grp-closed",),
        "fields": MODEL_Z_WWW
        + MODEL_Z_PUBMEDID
        + MODEL_Z_DOI
        + MODEL_Z_LICZBA_CYTOWAN
        + MODEL_Z_MIEJSCEM_PRZECHOWYWANIA,
    },
)


def js_openwin(url, handle, options):
    options = ",".join([f"{a}={b}" for a, b in list(options.items())])
    d = dict(url=url, handle=handle, options=options)
    return "window.open('%(url)s','\\%(handle)s','%(options)s')" % d


NIZSZE_TEXTFIELD_Z_MAPA_ZNAKOW = {
    models.TextField: {
        "widget": Textarea(attrs={"rows": 2, "cols": 90, "class": "charmap"})
    },
}


class DomyslnyStatusKorektyMixin:
    status_korekty = forms.ModelChoiceField(
        required=True,
        queryset=Status_Korekty.objects.all(),
        initial=lambda: Status_Korekty.objects.first(),
    )


class Wycinaj_W_z_InformacjiMixin:
    def clean_informacje(self):
        i = self.cleaned_data.get("informacje")
        if i:
            x = i.lower()
            n = 0
            if x.startswith("w:"):
                n = 2
            if x.startswith("w :"):
                n = 3
            if n:
                return i[n:].strip()
        return i


class LimitingFormset(BaseInlineFormSet):
    def get_queryset(self):
        if not hasattr(self, "_queryset_limited"):
            qs = super().get_queryset()
            self._queryset_limited = qs[:100]
        return self._queryset_limited


def link_do_obiektu(obj, friendly_name=None):
    opts = obj._meta
    obj_url = reverse(
        f"admin:{opts.app_label}_{opts.model_name}_change",
        args=(quote(obj.pk),),
        # current_app=self.admin_site.name,
    )
    # Add a link to the object's change form if the user can edit the obj.
    if friendly_name is None:
        friendly_name = mark_safe(obj)
    return format_html('<a href="{}">{}</a>', urlquote(obj_url), friendly_name)


def sprobuj_policzyc_sloty(request, obj):
    if obj.rok >= PBN_MIN_ROK and obj.rok <= PBN_MAX_ROK:
        try:
            ISlot(obj)
            messages.success(
                request,
                'Punkty dla dyscyplin dla "%s" będą mogły być obliczone.'
                % link_do_obiektu(obj),
            )
        except CannotAdapt as e:
            messages.error(
                request,
                'Nie można obliczyć punktów dla dyscyplin dla "%s": %s'
                % (link_do_obiektu(obj), e),
            )
    else:
        messages.warning(
            request,
            'Punkty dla dyscyplin dla "%s" nie będą liczone - rok poza zakresem (%i)'
            % (link_do_obiektu(obj), obj.rok),
        )


def sprawdz_czy_ustawiono_wysylke_tego_charakteru_formalnego(charakter_formalny):
    if charakter_formalny.rodzaj_pbn is None:
        raise CharakterFormalnyNieobslugiwanyError(
            "ten rekord nie może być wyeksportowany do PBN, gdyż ustawienia jego charakteru formalnego "
            "po stronie bazy BPP na to nie pozwalają"
        )


def sprawdz_czy_ustawiono_wysylke_tego_charakteru_formalnego_gui(request, obj):
    try:
        sprawdz_czy_ustawiono_wysylke_tego_charakteru_formalnego(obj.charakter_formalny)
    except CharakterFormalnyNieobslugiwanyError:
        messages.info(
            request,
            'Rekord "%s" nie będzie eksportowany do PBN zgodnie z ustawieniem dla charakteru formalnego.'
            % link_do_obiektu(obj),
        )
        return False
    return True


def sprawdz_wysylke_do_pbn_w_parametrach_uczelni(uczelnia):
    """
    :param uczelnia:
    :return: zwraca False jeżeli integracja wyłączona lub aktualizowanie na bieżąco wyłączone;
    zwraca obiekt uczelnia jeżeli jest OK,
    podnosi wyjątek jeżeli brak obiektu Uczelnia
    """
    if uczelnia is None:
        raise BrakZdefiniowanegoObiektuUczelniaWSystemieError()

    if not uczelnia.pbn_integracja or not uczelnia.pbn_aktualizuj_na_biezaco:
        return False

    return uczelnia


def sprawdz_wysylke_do_pbn_w_parametrach_uczelni_gui(request, obj):
    from bpp.models.uczelnia import Uczelnia

    uczelnia = Uczelnia.objects.get_for_request(request)
    try:
        res = sprawdz_wysylke_do_pbn_w_parametrach_uczelni(uczelnia)
    except BrakZdefiniowanegoObiektuUczelniaWSystemieError:
        messages.info(
            request,
            'Rekord "%s" nie zostanie wyeksportowany do PBN, ponieważ w systemie brakuje obiektu "Uczelnia", a'
            " co za tym idzie, jakchkolwiek ustawień" % link_do_obiektu(obj),
        )
        return

    return res


def sprobuj_utworzyc_zlecenie_eksportu_do_PBN_gui(request, obj):
    if not sprawdz_czy_ustawiono_wysylke_tego_charakteru_formalnego_gui(request, obj):
        return

    try:
        res = sprawdz_wysylke_do_pbn_w_parametrach_uczelni_gui(request, obj)
    except BrakZdefiniowanegoObiektuUczelniaWSystemieError:
        messages.error("Brak zdefiniowanego w systemie obiektu Uczelnia.")

    if res is False:
        messages.error("Wysyłka do PBN nie skonfigurowana w obiektu Uczelnia.")

    try:
        ret = PBN_Export_Queue.objects.sprobuj_utowrzyc_wpis(request.user, obj)
    except AlreadyEnqueuedError:
        messages.warning(
            request, f"Rekord {obj} jest już w kolejce do eksportu do PBN."
        )
        return

    link_do_kolejki = reverse("admin:pbn_api_pbn_export_queue_change", args=(ret.pk,))

    messages.info(
        request,
        f"Utworzono zlecenie wysyłki rekordu {obj} w tle do PBN. <a href={link_do_kolejki}>"
        f"Kliknij tutaj, aby śledzić stan.</a>",
    )

    task_sprobuj_wyslac_do_pbn.delay_on_commit(ret.pk)


class TextNotificator:
    def __init__(self, *args, **kw):
        self.output = []

    def info(self, msg, level="INFO"):
        self.output.append(f"{level} {msg}")

    def warning(self, msg):
        self.info(msg, level="WARNING")

    def error(self, msg):
        self.info(msg, level="WARNING")

    def success(self, msg):
        self.info(msg, level="WARNING")

    def as_text(self):
        return "\n".join(self.output)


class MessagesNotificator:
    def __init__(self, request):
        self.request = request

    def info(self, msg):
        messages.info(self.request, msg)

    def warning(self, msg):
        messages.warning(self.request, msg)

    def error(self, msg):
        messages.error(self.request, msg)

    def success(self, msg):
        messages.success(self.request, msg)


def sprobuj_wgrac_do_pbn_gui(request, obj, force_upload=False, pbn_client=None):
    if not sprawdz_czy_ustawiono_wysylke_tego_charakteru_formalnego_gui(request, obj):
        return

    uczelnia = sprawdz_wysylke_do_pbn_w_parametrach_uczelni_gui(request, obj)
    if uczelnia is None:
        return

    if uczelnia is False:
        messages.error(request, "Wysyłka do PBN nie skonfigurowana w obiektu Uczelnia.")
        return

    if pbn_client is None:
        pbn_client = uczelnia.pbn_client(request.user.pbn_token)

    sprobuj_wgrac_do_pbn(
        obj=obj,
        uczelnia=uczelnia,
        force_upload=force_upload,
        pbn_client=pbn_client,
        notificator=MessagesNotificator(request),
    )


def sprobuj_wgrac_do_pbn_celery(user, obj, force_upload=False, pbn_client=None):
    try:
        sprawdz_czy_ustawiono_wysylke_tego_charakteru_formalnego(obj.charakter_formalny)
    except CharakterFormalnyNieobslugiwanyError:
        return (
            None,
            [
                "Charakter formalny tego rekordu nie jest ustawiony jako wysyłany do PBN. Zmień konfigurację "
                "bazy BPP, Redagowanie -> Dane systemowe -> Charaktery formalne"
            ],
        )

    try:
        uczelnia = sprawdz_wysylke_do_pbn_w_parametrach_uczelni(
            Uczelnia.objects.get_default()
        )
    except BrakZdefiniowanegoObiektuUczelniaWSystemieError:
        return (None, ["W systemie brak obiektu Uczelnia."])

    if uczelnia is False:
        return (None, ["Wysyłka do PBN nie skonfigurowana w obiekcie Uczelnia"])

    if uczelnia is None:
        return (None, ["Brak obiektu uczelnia w systemie"])

    if user.pbn_token is None:
        return (
            None,
            [
                "Nie masz autoryzacji w PBN. Wykonaj autoryzację w PBN przez główną stronę serwisu. "
            ],
        )

    if pbn_client is None:
        pbn_client = uczelnia.pbn_client(user.pbn_token)

    try:
        pbn_client.get_institution_by_id(uczelnia.pbn_uid_id)
    except AccessDeniedException:
        return (
            None,
            [
                "Nie masz autoryzacji w PBN. Wykonaj autoryzację w PBN przez główną stronę serwisu. "
            ],
        )

    notificator = TextNotificator()

    sent_data = sprobuj_wgrac_do_pbn(
        obj=obj,
        uczelnia=uczelnia,
        force_upload=force_upload,
        pbn_client=pbn_client,
        notificator=notificator,
    )

    return (sent_data, notificator)


def sprobuj_wgrac_do_pbn(obj, pbn_client, uczelnia, notificator, force_upload=False):

    # Sprawdź, czy wydawnictwo nadrzędne ma odpowoednik PBN:
    if (
        hasattr(obj, "wydawnictwo_nadrzedne_id")
        and obj.wydawnictwo_nadrzedne_id is not None
    ):
        wn = obj.wydawnictwo_nadrzedne
        if wn.pbn_uid_id is None:
            notificator.info(
                "Wygląda na to, że wydawnictwo nadrzędne tego rekordu nie posiada odpowiednika "
                "w PBN, spróbuję go pobrać.",
            )
            udalo = False
            if wn.isbn:
                ni = normalize_isbn(wn.isbn)
                if ni:
                    from pbn_api.integrator import _pobierz_prace_po_elemencie

                    res = None
                    try:
                        res = _pobierz_prace_po_elemencie(pbn_client, "isbn", ni)
                    except PraceSerwisoweException:
                        notificator.warning(
                            "Pobieranie z PBN odpowiednika wydawnictwa nadrzędnego pracy po ISBN nie powiodło się "
                            "-- trwają prace serwisowe po stronie PBN. ",
                        )
                    except NeedsPBNAuthorisationException:
                        notificator.warning(
                            "Wyszukanie PBN UID wydawnictwa nadrzędnego po ISBN nieudane - "
                            "autoryzuj się najpierw w PBN. ",
                        )

                    if res:
                        notificator.info(
                            f"Udało się dopasować PBN UID wydawnictwa nadrzędnego po ISBN "
                            f"({', '.join([x.tytul_oryginalny for x in res])}). ",
                        )
                        udalo = True

            elif wn.doi:
                nd = normalize_isbn(wn.doi)
                if nd:
                    from pbn_api.integrator import _pobierz_prace_po_elemencie

                    res = None
                    try:
                        res = _pobierz_prace_po_elemencie(pbn_client, "doi", nd)
                    except PraceSerwisoweException:
                        notificator.warning(
                            "Pobieranie z PBN odpowiednika wydawnictwa nadrzędnego pracy po DOI nie powiodło się "
                            "-- trwają prace serwisowe po stronie PBN. ",
                        )
                    except NeedsPBNAuthorisationException:
                        notificator.warning(
                            "Wyszukanie PBN UID wydawnictwa nadrzędnego po DOI nieudane - "
                            "autoryzuj się najpierw w PBN. ",
                        )

                    if res:
                        notificator.info(
                            f"Udało się dopasować PBN UID wydawnictwa nadrzędnego po DOI. "
                            f"({', '.join([x.tytul_oryginalny for x in res])}). ",
                        )
                        udalo = True

            if not udalo:
                notificator.warning(
                    "Wygląda na to, że nie udało się dopasować rekordu nadrzędnego po ISBN/DOI do rekordu "
                    "po stronie PBN. Jeżeli jednak dokonano autoryzacji w PBN, to pewne rekordy z PBN "
                    "zostały teraz pobrane i możesz spróbować ustawić odpowiednik "
                    "PBN dla wydawnictwa nadrzędnego ręcznie. ",
                )

    try:
        pbn_client.sync_publication(
            obj,
            force_upload=force_upload,
            delete_statements_before_upload=uczelnia.pbn_api_kasuj_przed_wysylka,
            export_pk_zero=not uczelnia.pbn_api_nie_wysylaj_prac_bez_pk,
            always_affiliate_to_uid=(
                uczelnia.pbn_uid_id
                if uczelnia.pbn_api_afiliacja_zawsze_na_uczelnie
                else None
            ),
        )

    except SameDataUploadedRecently as e:
        link_do_wyslanych = reverse(
            "admin:pbn_api_sentdata_change",
            args=(SentData.objects.get_for_rec(obj).pk,),
        )

        notificator.info(
            f'Identyczne dane rekordu "{link_do_obiektu(obj)}" zostały wgrane do PBN w dniu {e}. '
            f"Nie aktualizuję w PBN API. Jeżeli chcesz wysłać ten rekord do PBN, musisz dokonać jakiejś zmiany "
            f"danych rekodu lub "
            f'usunąć informacje o <a target=_blank href="{link_do_wyslanych}">wcześniej wysłanych danych do PBN</a> '
            f"(Redagowanie -> PBN API -> Wysłane informacje). "
            f'<a target=_blank href="{obj.link_do_pbn()}">Kliknij tutaj, aby otworzyć w PBN</a>. ',
        )
        return

    except AccessDeniedException as e:
        notificator.warning(
            f'Nie można zsynchronizować obiektu "{link_do_obiektu(obj)}" z PBN pod adresem '
            f"API {e.url}. Brak dostępu -- najprawdopodobniej użytkownik nie posiada odpowiednich uprawnień "
            f"po stronie PBN/POLON. ",
        )
        return

    except PKZeroExportDisabled:
        notificator.warning(
            f"Eksport prac z PK=0 jest wyłączony w konfiguracji. Próba wysyłki do PBN rekordu "
            f'"{link_do_obiektu(obj)}" nie została podjęta z uwagi na konfigurację systemu. ',
        )
        return

    except NeedsPBNAuthorisationException as e:
        notificator.warning(
            f'Nie można zsynchronizować obiektu "{link_do_obiektu(obj)}" z PBN pod adresem '
            f"API {e.url}. Brak dostępu z powodu nieprawidłowego tokena -- wymagana autoryzacja w PBN. "
            f'<a target=_blank href="{reverse("pbn_api:authorize")}">Kliknij tutaj, aby autoryzować sesję</a>.',
        )
        return

    except Exception as e:
        try:
            link_do_wyslanych = reverse(
                "admin:pbn_api_sentdata_change",
                args=(SentData.objects.get_for_rec(obj).pk,),
            )
        except SentData.DoesNotExist:
            link_do_wyslanych = None

        extra = ""
        if link_do_wyslanych:
            extra = (
                '<a target=_blank href="%s">Kliknij, aby otworzyć widok wysłanych danych</a>.'
                % link_do_wyslanych
            )

        notificator.warning(
            'Nie można zsynchronizować obiektu "%s" z PBN. Kod błędu: %r. %s'
            % (link_do_obiektu(obj), e, extra),
        )

        # Zaloguj problem do Sentry, bo w sumie nie wiadomo, co to za problem na tym etapie...
        capture_exception(e)

        return

    sent_data = SentData.objects.get(
        content_type=ContentType.objects.get_for_model(obj), object_id=obj.pk
    )
    sent_data_link = link_do_obiektu(
        sent_data, "Kliknij tutaj, aby otworzyć widok wysłanych danych. "
    )

    publication_link = link_do_obiektu(
        sent_data.pbn_uid,
        "Kliknij tutaj, aby otworzyć zwrotnie otrzymane z PBN dane o rekordzie. ",
    )
    notificator.success(
        f"Dane w PBN dla rekordu {link_do_obiektu(obj)} zostały zaktualizowane. "
        f'<a target=_blank href="{obj.link_do_pbn()}">Kliknij tutaj, aby otworzyć w PBN</a>. '
        f"{sent_data_link}{publication_link}",
    )

    return sent_data


def get_rekord_id_from_GET_qs(request):
    flt = request.GET.get("_changelist_filters", "?")
    data = parse_qs(flt)  # noqa
    if "rekord__id__exact" in data:
        try:
            return int(data.get("rekord__id__exact")[0])
        except (ValueError, TypeError):
            pass


class OptionalPBNSaveMixin:
    def render_change_form(
        self, request, context, add=False, change=False, form_url="", obj=None
    ):
        from bpp.models import Uczelnia

        uczelnia = Uczelnia.objects.get_default()
        if uczelnia is not None:
            if uczelnia.pbn_integracja and uczelnia.pbn_aktualizuj_na_biezaco:
                context.update({"show_save_and_pbn": True})

        return super().render_change_form(request, context, add, change, form_url, obj)

    def response_post_save_change(self, request, obj):
        if "_continue_and_pbn" in request.POST:
            sprobuj_wgrac_do_pbn_gui(request, obj)

        elif "_continue_and_pbn_later" in request.POST:
            sprobuj_utworzyc_zlecenie_eksportu_do_PBN_gui(request, obj)

        else:
            # Otherwise, use default behavior
            return super().response_post_save_change(request, obj)

        # Przekieruj użytkownika na formularz zmian
        opts = self.model._meta
        route = f"admin:{opts.app_label}_{opts.model_name}_change"

        post_url = reverse(route, args=(obj.pk,))

        from django.http import HttpResponseRedirect

        return HttpResponseRedirect(post_url)


def sprawdz_duplikaty_www_doi(request, obj):
    from bpp.models.cache import Rekord

    IEXACT = "__iexact"
    for field, operator, label in [
        ("www", IEXACT, const.WWW_FIELD_LABEL),
        ("public_www", IEXACT, const.PUBLIC_WWW_FIELD_LABEL),
        ("doi", IEXACT, const.DOI_FIELD_LABEL),
        ("pbn_uid_id", "", const.PBN_UID_FIELD_LABEL),
    ]:
        if not hasattr(obj, field):
            continue

        v = getattr(obj, field)
        if v in [None, ""]:
            continue

        rekord_pk = (ContentType.objects.get_for_model(obj).pk, obj.pk)

        query = Q(**{field + operator: v})

        if field == "www":
            query |= Q(public_www__iexact=v)
        elif field == "public_www":
            query |= Q(www__iexact=v)

        if Rekord.objects.filter(query).exclude(pk=rekord_pk).exists():
            messages.warning(
                request, const.ZDUBLOWANE_POLE_KOMUNIKAT.format(label=label)
            )

iplweb / bpp / d739ad7a-8bce-4087-b0c8-83f9ea367ad1

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