3c407b5b-8434-4f25-9e12-005b14d23a79

Committed 07 Sep 2025 02:53PM UTC coverage: 33.615% (-8.7%) from 42.328%

Build # 3c407b5b-8434-4f25-9e12-005b14d23a79

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Merge branch 'release/v202509.1221'

Run Details

1 of 45 new or added lines in 2 files covered. (2.22%)

3993 existing lines in 129 files now uncovered.

16522 of 49150 relevant lines covered (33.62%)

0.34 hits per line

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

26.93

src/bpp/util.py

import json
import multiprocessing
import operator
import os
import re
from datetime import datetime, timedelta
from functools import reduce
from math import ceil, floor
from pathlib import Path
from typing import Dict, List

import bleach
import lxml.html
import openpyxl.worksheet.worksheet
from django.apps import apps
from django.conf import settings
from django.db.models import Max, Min, Value
from openpyxl.utils import get_column_letter
from openpyxl.worksheet.filters import AutoFilter
from openpyxl.worksheet.table import Table, TableColumn, TableStyleInfo
from tqdm import tqdm
from unidecode import unidecode

from django.contrib.postgres.search import SearchQuery, SearchRank

from django.utils import timezone
from django.utils.html import strip_tags

non_url = re.compile(r"[^\w-]+")


def get_fixture(name):
    p = Path(__file__).parent / "fixtures" / ("%s.json" % name)
    ret = json.load(open(p, "rb"))
    ret = [x["fields"] for x in ret if x["model"] == ("bpp.%s" % name)]
    return {x["skrot"].lower().strip(): x for x in ret}


strip_nonalpha_regex = re.compile("\\W+")
strip_extra_spaces_regex = re.compile("\\s\\s+")


def strip_nonalphanumeric(s):
    """Usuń nie-alfanumeryczne znaki z ciągu znaków"""
    if s is None:
        return

    return strip_nonalpha_regex.sub(" ", s)


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

    return strip_extra_spaces_regex.sub("", s).strip()


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

    return [
        elem
        for elem in strip_extra_spaces(strip_nonalphanumeric(strip_tags(s))).split(" ")
        if elem
    ]


class FulltextSearchMixin:
    fts_field = "search"

    # włączaj typ wyszukoiwania web-search gdy podany jest znak minus
    # w tekscie zapytania:
    fts_enable_websearch_on_minus_or_quote = True

    def fulltext_empty(self):
        return self.none().annotate(**{self.fts_field + "__rank": Value(0)})

    def fulltext_annotate(self, search_query, normalization):
        return {
            self.fts_field
            + "__rank": SearchRank(
                self.fts_field, search_query, normalization=normalization
            )
        }

    def fulltext_filter(self, qstr, normalization=None):
        if qstr is None:
            return self.fulltext_empty()

        if isinstance(qstr, bytes):
            qstr = qstr.decode("utf-8")

        words = fulltext_tokenize(qstr)
        if not words:
            return self.fulltext_empty()

        if (
            qstr.find("-") >= 0 or qstr.find('"') >= 0
        ) and self.fts_enable_websearch_on_minus_or_quote:
            # Jezeli użytkownik podał cudzysłów lub minus w zapytaniu i dozwolone jest
            # przełączenie się na websearch (parametr ``self.fts_enable_websearch_on_minus``,
            # to skorzystaj z trybu zapytania ``websearch``:

            search_query = SearchQuery(
                qstr, search_type="websearch", config="bpp_nazwy_wlasne"
            )
        else:
            # Jeżeli nie ma minusów, cudzysłowów to możesz odpalić dodatkowo tryb wyszukiwania
            # 'phrase' żeby znaleźć wyrazy w kolejności, tak jak zostały podane

            q1 = reduce(
                operator.__and__,
                [
                    SearchQuery(
                        word + ":*", search_type="raw", config="bpp_nazwy_wlasne"
                    )
                    for word in words
                ],
            )

            q3 = SearchQuery(qstr, search_type="phrase", config="bpp_nazwy_wlasne")

            search_query = q1 | q3

        query = (
            self.filter(**{self.fts_field: search_query})
            .annotate(**self.fulltext_annotate(search_query, normalization))
            .order_by(f"-{self.fts_field}__rank")
        )
        return query


def strip_html(s):
    if not s:
        return s

    return lxml.html.fromstring(str(s)).text_content()


def slugify_function(s):
    s = unidecode(strip_html(s)).replace(" ", "-")
    while s.find("--") >= 0:
        s = s.replace("--", "-")
    return non_url.sub("", s)


def get_original_object(object_name, object_pk):
    from bpp.models import TABLE_TO_MODEL

    klass = TABLE_TO_MODEL.get(object_name)
    try:
        return klass.objects.get(pk=object_pk)
    except klass.DoesNotExist:
        return


def get_copy_from_db(instance):
    if not instance.pk:
        return None
    return instance.__class__._default_manager.get(pk=instance.pk)


def has_changed(instance, field_or_fields):
    try:
        original = get_copy_from_db(instance)
    except instance.__class__.DoesNotExist:
        return True
        # Jeżeli w bazie danych nie ma tego obiektu, no to bankowo
        # się zmienił...
        return True

    fields = field_or_fields
    if isinstance(field_or_fields, str):
        fields = [field_or_fields]

    for field in fields:
        if not getattr(instance, field) == getattr(original, field):
            return True

    return False


class Getter:
    """Klasa pomocnicza dla takich danych jak Typ_KBN czy
    Charakter_Formalny, umozliwia po zainicjowaniu pobieranie
    tych klas po atrybucie w taki sposob:

    >>> kbn = Getter(Typ_KBN)
    >>> kbn.PO == Typ_KBN.objects.get(skrot='PO')
    True
    """

    def __init__(self, klass, field="skrot"):
        self.field = field
        self.klass = klass

    def __getitem__(self, item):
        kw = {self.field: item}
        return self.klass.objects.get(**kw)

    __getattr__ = __getitem__


class NewGetter(Getter):
    """Zwraca KeyError zamiast DoesNotExist."""

    def __getitem__(self, item):
        kw = {self.field: item}
        try:
            return self.klass.objects.get(**kw)
        except self.klass.DoesNotExist as e:
            raise KeyError(e)

    __getattr__ = __getitem__


def zrob_cache(t):
    zle_znaki = [
        " ",
        ":",
        ";",
        "-",
        ",",
        "-",
        ".",
        "(",
        ")",
        "?",
        "!",
        "ę",
        "ą",
        "ł",
        "ń",
        "ó",
        "ź",
        "ż",
    ]
    for znak in zle_znaki:
        t = t.replace(znak, "")
    return t.lower()


def remove_old_objects(klass, file_field="file", field_name="created_on", days=7):
    since = datetime.now() - timedelta(days=days)

    kwargs = {}
    kwargs["%s__lt" % field_name] = since

    for rec in klass.objects.filter(**kwargs):
        try:
            path = getattr(rec, file_field).path
        except ValueError:
            path = None

        rec.delete()

        if path is not None:
            try:
                os.unlink(path)
            except OSError:
                pass


def rebuild_contenttypes():
    app_config = apps.get_app_config("bpp")
    from django.contrib.contenttypes.management import create_contenttypes

    create_contenttypes(app_config, verbosity=0)


#
# Progress bar
#


def pbar(
    query, count=None, label="Progres...", disable_progress_bar=False, callback=None
):
    if count is None:
        if hasattr(query, "count"):
            try:
                count = query.count()
            except TypeError:
                count = len(query)
        elif hasattr(query, "__len__"):
            count = len(query)

    if callback:
        # Callback provided but progress bar disabled - only use callback
        def callback_only_wrapper():
            for i, item in enumerate(query):
                callback.update(i + 1, count, label)
                yield item
            callback.clear()

        return callback_only_wrapper()

    return tqdm(
        query, total=count, desc=label, unit="items", disable=disable_progress_bar
    )


#
# Multiprocessing stuff
#


def partition(min, max, num_proc, fun=ceil):
    s = int(fun((max - min) / num_proc))
    cnt = min
    ret = []
    while cnt < max:
        ret.append((cnt, cnt + s))
        cnt += s
    return ret


def partition_ids(model, num_proc, attr="idt"):
    d = model.objects.aggregate(min=Min(attr), max=Max(attr))
    return partition(d["min"], d["max"], num_proc)


def partition_count(objects, num_proc):
    return partition(0, objects.count(), num_proc, fun=ceil)


def no_threads(multiplier=0.75):
    return max(int(floor(multiprocessing.cpu_count() * multiplier)), 1)


class safe_html_defaults:
    ALLOWED_TAGS = (
        "a",
        "abbr",
        "acronym",
        "b",
        "blockquote",
        "code",
        "em",
        "i",
        "li",
        "ol",
        "strong",
        "ul",
        "font",
        "div",
        "span",
        "br",
        "strike",
        "h2",
        "h3",
        "h4",
        "h5",
        "h6",
        "p",
        "table",
        "tr",
        "td",
        "th",
        "thead",
        "tbody",
        "dl",
        "dd",
        "u",
    )

    ALLOWED_ATTRIBUTES = {
        "*": ["class"],
        "a": ["href", "title", "rel"],
        "abbr": ["title"],
        "acronym": ["title"],
        "font": [
            "face",
            "size",
        ],
        "div": [
            "style",
        ],
        "span": [
            "style",
        ],
        "ul": [
            "style",
        ],
    }


def safe_html(html):
    html = html or ""

    ALLOWED_TAGS = getattr(settings, "ALLOWED_TAGS", safe_html_defaults.ALLOWED_TAGS)
    ALLOWED_ATTRIBUTES = getattr(
        settings, "ALLOWED_ATTRIBUTES", safe_html_defaults.ALLOWED_ATTRIBUTES
    )
    STRIP_TAGS = getattr(settings, "STRIP_TAGS", True)
    return bleach.clean(
        html,
        tags=ALLOWED_TAGS,
        attributes=ALLOWED_ATTRIBUTES,
        strip=STRIP_TAGS,
    )


def set_seq(s):
    if settings.DATABASES["default"]["ENGINE"].find("postgresql") >= 0:
        from django.db import connection

        cursor = connection.cursor()
        cursor.execute(f"SELECT setval('{s}_id_seq', (SELECT MAX(id) FROM {s}))")


def usun_nieuzywany_typ_charakter(klass, field, dry_run):
    from bpp.models import Rekord

    for elem in klass.objects.all():
        kw = {field: elem}
        if not Rekord.objects.filter(**kw).exists():
            print(f"Kasuje {elem}")
            if not dry_run:
                elem.delete()


isbn_regex = re.compile(
    r"^isbn\s*[0-9]*[-| ][0-9]*[-| ][0-9]*[-| ][0-9]*[-| ][0-9]*X?",
    flags=re.IGNORECASE,
)


def wytnij_isbn_z_uwag(uwagi):
    if uwagi is None:
        return

    if uwagi == "":
        return

    if uwagi.lower().find("isbn-10") >= 0 or uwagi.lower().find("isbn-13") >= 0:
        return None

    res = isbn_regex.search(uwagi)
    if res:
        res = res.group()
        isbn = res.replace("ISBN", "").replace("isbn", "").strip()
        reszta = uwagi.replace(res, "").strip()

        while (
            reszta.startswith(".") or reszta.startswith(";") or reszta.startswith(",")
        ):
            reszta = reszta[1:].strip()

        return isbn, reszta


def crispy_form_html(self, key):
    from crispy_forms_foundation.layout import HTML, Column, Row

    from django.utils.functional import lazy

    def _():
        return self.initial.get(key, None) or ""

    return Row(Column(HTML(lazy(_, str)())))


def formdefaults_html_before(form):
    return crispy_form_html(form, "formdefaults_pre_html")


def formdefaults_html_after(form):
    return crispy_form_html(form, "formdefaults_post_html")


def knapsack(W, wt, val, ids, zwracaj_liste_przedmiotow=True):
    """
    :param W: wielkosc plecaka -- maksymalna masa przedmiotów w plecaku (zbierany slot)
    :param wt: masy przedmiotów, które można włożyć do plecaka (sloty prac)
    :param val: ceny przedmiotów, które można włożyc do plecaka (punkty PKdAut prac)
    :param ids: ID prac, które można włożyć do plecaka (rekord.pk)
    :param zwracaj_liste_przedmiotow: gdy True (domyślnie) funkcja zwróci listę z identyfikatorami włożonych
    przedmiotów, gdy False zwrócona lista będzie pusta

    :returns: tuple(mp, lista), gdzie mp to maksymalna możliwa wartość włożonych przedmiotów, a lista to lista
    lub pusta lista gdy parametr `zwracaj_liste_przemiotów` był pusty
    """

    assert len(wt) == len(val) == len(ids), "Listy są różnej długości"

    sum_wt = sum(wt)
    if sum_wt <= W:
        # Jeżeli wszystkie przedmioty zmieszczą się w plecaku, to po co liczyć cokolwiek
        if zwracaj_liste_przedmiotow:
            return sum(val), ids
        return sum(val), []

    n = len(wt)

    K = [[0 for x in range(W + 1)] for x in range(n + 1)]

    for i in range(n + 1):
        for w in range(W + 1):
            if i == 0 or w == 0:
                K[i][w] = 0
            elif wt[i - 1] <= w:
                K[i][w] = max(val[i - 1] + K[i - 1][w - wt[i - 1]], K[i - 1][w])
            else:
                K[i][w] = K[i - 1][w]

    res = maks_punkty = K[n][W]
    lista = []

    if zwracaj_liste_przedmiotow:
        w = W
        for i in range(n, 0, -1):
            if res <= 0:
                break

            if res == K[i - 1][w]:
                continue
            else:
                lista.append(ids[i - 1])

                res = res - val[i - 1]
                w = w - wt[i - 1]

    return maks_punkty, lista


DEC2INT = 10000


def intsack(W, wt, val, ids):
    pkt, ids = knapsack(
        int(W * DEC2INT),
        [int(x * DEC2INT) for x in wt],
        [int(x * DEC2INT) for x in val],
        ids,
    )
    return pkt / DEC2INT, ids


def disable_multithreading_by_monkeypatching_pool(pool):
    def apply(fun, args=()):
        return fun(*args)

    pool.apply = apply

    def starmap(fun, lst):
        for elem in lst:
            fun(*elem)

    pool.starmap = starmap


def year_last_month():
    now = timezone.now().date()
    if now.month >= 2:
        return now.year
    return now.year - 1


#
# Seq Scan check
#


class PerformanceFailure(Exception):
    pass


def fail_if_seq_scan(qset, DEBUG):
    """
    Funkcja weryfikujaca, czy w wyjasnieniu zapytania (EXPLAIN) nie wystapi ciag znakow 'Seq Scan',
    jezeli tak to wyjatek PerformanceFailure z zapytaniem + wyjasnieniem
    """
    if DEBUG:
        explain = qset.explain()
        if explain.find("Seq Scan") >= 0:
            print("\r\n", explain)
            raise PerformanceFailure(str(qset.query), explain)


def rebuild_instances_of_models(modele, *args, **kw):
    from denorm import denorms
    from django.db import transaction

    with transaction.atomic():
        for model in modele:
            denorms.rebuild_instances_of(model, *args, **kw)


def worksheet_columns_autosize(
    ws: openpyxl.worksheet.worksheet.Worksheet,
    max_width: int = 55,
    column_widths: Dict[str, int] | None = None,
    dont_resize_those_columns: List[int] | None = None,
    right_margin=2,
    multiplier=1.1,
):
    if column_widths is None:
        column_widths = {}

    if dont_resize_those_columns is None:
        dont_resize_those_columns = []

    for ncol, col in enumerate(ws.columns):
        max_length = 0
        column = col[0].column_letter  # Get the column name

        # Nie ustawiaj szerokosci tym kolumnom, one będą jako auto-size
        if ncol in dont_resize_those_columns:
            continue

        if column in column_widths:
            adjusted_width = column_widths[column]
        else:
            for cell in col:
                if cell.value is None or not str(cell.value):
                    continue

                text = str(cell.value)

                if text.startswith("=HYPERLINK"):
                    try:
                        # Wyciągnij z hiperlinku jego faktyczny opis tekstowy na cele
                        # liczenia szerokości kolumny
                        text = text.split('"')[3]
                    except IndexError:
                        pass

                max_line_len = max(len(line) for line in text.split("\n"))
                max_length = max(max_length, max_line_len)

            adjusted_width = (max_length + right_margin) * multiplier
            if adjusted_width > max_width:
                adjusted_width = max_width

        ws.column_dimensions[column].width = adjusted_width


def worksheet_create_table(
    ws: openpyxl.worksheet.worksheet.Worksheet,
    title="Tabela",
    first_table_row=1,
    totals=False,
    table_columns=None,
):
    """
    Formatuje skoroszyt jako tabelę.

    :param first_table_row: pierwszy wiersz tabeli (licząc od nagłówka)

    :param table_columns: określa rodzaj kolumn w tabeli, jeżeli None to tytuły nagłówków zostaną pobrane
    z pierwszego wiersza w arkuszu.
    """
    max_column = ws.max_column
    max_column_letter = get_column_letter(max_column)
    max_row = ws.max_row

    style = TableStyleInfo(
        name="TableStyleMedium9",
        showFirstColumn=False,
        showLastColumn=False,
        showRowStripes=True,
        showColumnStripes=True,
    )

    if table_columns is None:
        table_columns = tuple(
            TableColumn(id=h, name=header.value)
            for h, header in enumerate(next(iter(ws.rows), None), start=1)
        )

    tab = Table(
        displayName=title,
        ref=f"A{first_table_row}:{max_column_letter}{max_row}",
        autoFilter=AutoFilter(
            ref=f"A{first_table_row}:{max_column_letter}{max_row - 1}"
        ),
        totalsRowShown=True if totals else False,
        totalsRowCount=1 if totals else False,
        tableStyleInfo=style,
        tableColumns=table_columns,
    )

    ws.add_table(tab)


def worksheet_create_urls(
    ws: openpyxl.worksheet.worksheet.Worksheet, default_link_name: str = "[link]"
):
    """Tworzy adresy URL w postaci klikalnego linku z domyslnym tekstem."""

    for column_cell in ws.iter_cols(1, ws.max_column):  # iterate column cell
        if hasattr(column_cell[0].value, "endswith") and column_cell[0].value.endswith(
            "_url"
        ):
            for data in column_cell[1:]:
                if data.value:
                    data.value = '=HYPERLINK("{}", "{}")'.format(
                        data.value, default_link_name
                    )


def dont_log_anonymous_crud_events(
    instance, object_json_repr, created, raw, using, update_fields, **kwargs
):
    """
    Za pomocą tej procedury  moduł django-easyaudit decyduje, czy zalogować dane
    zdarzenie, czy nie.

    Procedura ta sprawdza, czy w parametrach ``kwargs`` zawarty jest parametr
    ``request``, a jeżeli tak -- to czy ma on atrybut ``user`` czyli użytkownik. Jeśli tak,
    to zwracana jest wartość ``True``, aby dane zdarzenie mogło byc zalogowane.

    Jeżeli nie ma parametru ``user``, to takie zdarzenie logowane nie będzie.
    """
    if kwargs.get("request", None) and getattr(kwargs["request"], "user", None):
        return True

1	import json	1✔
2	import multiprocessing	1✔
3	import operator	1✔
4	import os	1✔
5	import re	1✔
6	from datetime import datetime, timedelta	1✔
7	from functools import reduce	1✔
8	from math import ceil, floor	1✔
9	from pathlib import Path	1✔
10	from typing import Dict, List	1✔
11
12	import bleach	1✔
13	import lxml.html	1✔
14	import openpyxl.worksheet.worksheet	1✔
15	from django.apps import apps	1✔
16	from django.conf import settings	1✔
17	from django.db.models import Max, Min, Value	1✔
18	from openpyxl.utils import get_column_letter	1✔
19	from openpyxl.worksheet.filters import AutoFilter	1✔
20	from openpyxl.worksheet.table import Table, TableColumn, TableStyleInfo	1✔
21	from tqdm import tqdm	1✔
22	from unidecode import unidecode	1✔
23
24	from django.contrib.postgres.search import SearchQuery, SearchRank	1✔
25
26	from django.utils import timezone	1✔
27	from django.utils.html import strip_tags	1✔
28
29	non_url = re.compile(r"[^\w-]+")	1✔
30
31
32	def get_fixture(name):	1✔
UNCOV 33	p = Path(__file__).parent / "fixtures" / ("%s.json" % name)	×
UNCOV 34	ret = json.load(open(p, "rb"))	×
UNCOV 35	ret = [x["fields"] for x in ret if x["model"] == ("bpp.%s" % name)]	×
UNCOV 36	return {x["skrot"].lower().strip(): x for x in ret}	×
37
38
39	strip_nonalpha_regex = re.compile("\\W+")	1✔
40	strip_extra_spaces_regex = re.compile("\\s\\s+")	1✔
41
42
43	def strip_nonalphanumeric(s):	1✔
44	"""Usuń nie-alfanumeryczne znaki z ciągu znaków"""
UNCOV 45	if s is None:	×
46	return	×
47
UNCOV 48	return strip_nonalpha_regex.sub(" ", s)	×
49
50
51	def strip_extra_spaces(s):	1✔
UNCOV 52	if s is None:	×
53	return	×
54
UNCOV 55	return strip_extra_spaces_regex.sub("", s).strip()	×
56
57
58	def fulltext_tokenize(s):	1✔
UNCOV 59	if s is None:	×
60	return	×
61
UNCOV 62	return [	×
63	elem
64	for elem in strip_extra_spaces(strip_nonalphanumeric(strip_tags(s))).split(" ")
65	if elem
66	]
67
68
69	class FulltextSearchMixin:	1✔
70	fts_field = "search"	1✔
71
72	# włączaj typ wyszukoiwania web-search gdy podany jest znak minus
73	# w tekscie zapytania:
74	fts_enable_websearch_on_minus_or_quote = True	1✔
75
76	def fulltext_empty(self):	1✔
77	return self.none().annotate(**{self.fts_field + "__rank": Value(0)})	×
78
79	def fulltext_annotate(self, search_query, normalization):	1✔
80	return {	×
81	self.fts_field
82	+ "__rank": SearchRank(
83	self.fts_field, search_query, normalization=normalization
84	)
85	}
86
87	def fulltext_filter(self, qstr, normalization=None):	1✔
UNCOV 88	if qstr is None:	×
89	return self.fulltext_empty()	×
90
UNCOV 91	if isinstance(qstr, bytes):	×
92	qstr = qstr.decode("utf-8")	×
93
UNCOV 94	words = fulltext_tokenize(qstr)	×
UNCOV 95	if not words:	×
96	return self.fulltext_empty()	×
97
UNCOV 98	if (	×
99	qstr.find("-") >= 0 or qstr.find('"') >= 0
100	) and self.fts_enable_websearch_on_minus_or_quote:
101	# Jezeli użytkownik podał cudzysłów lub minus w zapytaniu i dozwolone jest
102	# przełączenie się na websearch (parametr ``self.fts_enable_websearch_on_minus``,
103	# to skorzystaj z trybu zapytania ``websearch``:
104
105	search_query = SearchQuery(	×
106	qstr, search_type="websearch", config="bpp_nazwy_wlasne"
107	)
108	else:
109	# Jeżeli nie ma minusów, cudzysłowów to możesz odpalić dodatkowo tryb wyszukiwania
110	# 'phrase' żeby znaleźć wyrazy w kolejności, tak jak zostały podane
111
UNCOV 112	q1 = reduce(	×
113	operator.__and__,
114	[
115	SearchQuery(
116	word + ":*", search_type="raw", config="bpp_nazwy_wlasne"
117	)
118	for word in words
119	],
120	)
121
UNCOV 122	q3 = SearchQuery(qstr, search_type="phrase", config="bpp_nazwy_wlasne")	×
123
UNCOV 124	search_query = q1 \| q3	×
125
UNCOV 126	query = (	×
127	self.filter(**{self.fts_field: search_query})
128	.annotate(**self.fulltext_annotate(search_query, normalization))
129	.order_by(f"-{self.fts_field}__rank")
130	)
UNCOV 131	return query	×
132
133
134	def strip_html(s):	1✔
135	if not s:	1✔
136	return s	×
137
138	return lxml.html.fromstring(str(s)).text_content()	1✔
139
140
141	def slugify_function(s):	1✔
142	s = unidecode(strip_html(s)).replace(" ", "-")	1✔
143	while s.find("--") >= 0:	1✔
144	s = s.replace("--", "-")	1✔
145	return non_url.sub("", s)	1✔
146
147
148	def get_original_object(object_name, object_pk):	1✔
149	from bpp.models import TABLE_TO_MODEL	×
150
151	klass = TABLE_TO_MODEL.get(object_name)	×
152	try:	×
153	return klass.objects.get(pk=object_pk)	×
154	except klass.DoesNotExist:	×
155	return	×
156
157
158	def get_copy_from_db(instance):	1✔
159	if not instance.pk:	×
160	return None	×
161	return instance.__class__._default_manager.get(pk=instance.pk)	×
162
163
164	def has_changed(instance, field_or_fields):	1✔
165	try:	×
166	original = get_copy_from_db(instance)	×
167	except instance.__class__.DoesNotExist:	×
168	return True	×
169	# Jeżeli w bazie danych nie ma tego obiektu, no to bankowo
170	# się zmienił...
171	return True
172
173	fields = field_or_fields	×
174	if isinstance(field_or_fields, str):	×
175	fields = [field_or_fields]	×
176
177	for field in fields:	×
178	if not getattr(instance, field) == getattr(original, field):	×
179	return True	×
180
181	return False	×
182
183
184	class Getter:	1✔
185	"""Klasa pomocnicza dla takich danych jak Typ_KBN czy
186	Charakter_Formalny, umozliwia po zainicjowaniu pobieranie
187	tych klas po atrybucie w taki sposob:
188
189	>>> kbn = Getter(Typ_KBN)
190	>>> kbn.PO == Typ_KBN.objects.get(skrot='PO')
191	True
192	"""
193
194	def __init__(self, klass, field="skrot"):	1✔
195	self.field = field	×
196	self.klass = klass	×
197
198	def __getitem__(self, item):	1✔
199	kw = {self.field: item}	×
200	return self.klass.objects.get(**kw)	×
201
202	__getattr__ = __getitem__	1✔
203
204
205	class NewGetter(Getter):	1✔
206	"""Zwraca KeyError zamiast DoesNotExist."""
207
208	def __getitem__(self, item):	1✔
209	kw = {self.field: item}	×
210	try:	×
211	return self.klass.objects.get(**kw)	×
212	except self.klass.DoesNotExist as e:	×
213	raise KeyError(e)	×
214
215	__getattr__ = __getitem__	1✔
216
217
218	def zrob_cache(t):	1✔
219	zle_znaki = [	×
220	" ",
221	":",
222	";",
223	"-",
224	",",
225	"-",
226	".",
227	"(",
228	")",
229	"?",
230	"!",
231	"ę",
232	"ą",
233	"ł",
234	"ń",
235	"ó",
236	"ź",
237	"ż",
238	]
239	for znak in zle_znaki:	×
240	t = t.replace(znak, "")	×
241	return t.lower()	×
242
243
244	def remove_old_objects(klass, file_field="file", field_name="created_on", days=7):	1✔
245	since = datetime.now() - timedelta(days=days)	×
246
247	kwargs = {}	×
248	kwargs["%s__lt" % field_name] = since	×
249
250	for rec in klass.objects.filter(**kwargs):	×
251	try:	×
252	path = getattr(rec, file_field).path	×
253	except ValueError:	×
254	path = None	×
255
256	rec.delete()	×
257
258	if path is not None:	×
259	try:	×
260	os.unlink(path)	×
261	except OSError:	×
262	pass	×
263
264
265	def rebuild_contenttypes():	1✔
266	app_config = apps.get_app_config("bpp")	×
267	from django.contrib.contenttypes.management import create_contenttypes	×
268
269	create_contenttypes(app_config, verbosity=0)	×
270
271
272	#
273	# Progress bar
274	#
275
276
277	def pbar(	1✔
278	query, count=None, label="Progres...", disable_progress_bar=False, callback=None
279	):
280	if count is None:	×
281	if hasattr(query, "count"):	×
282	try:	×
283	count = query.count()	×
284	except TypeError:	×
285	count = len(query)	×
UNCOV 286	elif hasattr(query, "__len__"):	×
287	count = len(query)	×
288
UNCOV 289	if callback:	×
290	# Callback provided but progress bar disabled - only use callback
UNCOV 291	def callback_only_wrapper():	×
UNCOV 292	for i, item in enumerate(query):	×
UNCOV 293	callback.update(i + 1, count, label)	×
UNCOV 294	yield item	×
UNCOV 295	callback.clear()	×
296
297	return callback_only_wrapper()	×
298
299	return tqdm(	×
300	query, total=count, desc=label, unit="items", disable=disable_progress_bar
301	)
302
303
304	#
305	# Multiprocessing stuff
306	#
307
308
309	def partition(min, max, num_proc, fun=ceil):	1✔
UNCOV 310	s = int(fun((max - min) / num_proc))	×
311	cnt = min	×
UNCOV 312	ret = []	×
UNCOV 313	while cnt < max:	×
UNCOV 314	ret.append((cnt, cnt + s))	×
315	cnt += s	×
UNCOV 316	return ret	×
317
318
319	def partition_ids(model, num_proc, attr="idt"):	1✔
UNCOV 320	d = model.objects.aggregate(min=Min(attr), max=Max(attr))	×
UNCOV 321	return partition(d["min"], d["max"], num_proc)	×
322
323
324	def partition_count(objects, num_proc):	1✔
UNCOV 325	return partition(0, objects.count(), num_proc, fun=ceil)	×
326
327
328	def no_threads(multiplier=0.75):	1✔
UNCOV 329	return max(int(floor(multiprocessing.cpu_count() * multiplier)), 1)	×
330
331
332	class safe_html_defaults:	1✔
333	ALLOWED_TAGS = (	1✔
334	"a",
335	"abbr",
336	"acronym",
337	"b",
338	"blockquote",
339	"code",
340	"em",
341	"i",
342	"li",
343	"ol",
344	"strong",
345	"ul",
346	"font",
347	"div",
348	"span",
349	"br",
350	"strike",
351	"h2",
352	"h3",
353	"h4",
354	"h5",
355	"h6",
356	"p",
357	"table",
358	"tr",
359	"td",
360	"th",
361	"thead",
362	"tbody",
363	"dl",
364	"dd",
365	"u",
366	)
367
368	ALLOWED_ATTRIBUTES = {	1✔
369	"*": ["class"],
370	"a": ["href", "title", "rel"],
371	"abbr": ["title"],
372	"acronym": ["title"],
373	"font": [
374	"face",
375	"size",
376	],
377	"div": [
378	"style",
379	],
380	"span": [
381	"style",
382	],
383	"ul": [
384	"style",
385	],
386	}
387
388
389	def safe_html(html):	1✔
UNCOV 390	html = html or ""	×
391
392	ALLOWED_TAGS = getattr(settings, "ALLOWED_TAGS", safe_html_defaults.ALLOWED_TAGS)	×
393	ALLOWED_ATTRIBUTES = getattr(	×
394	settings, "ALLOWED_ATTRIBUTES", safe_html_defaults.ALLOWED_ATTRIBUTES
395	)
396	STRIP_TAGS = getattr(settings, "STRIP_TAGS", True)	×
UNCOV 397	return bleach.clean(	×
398	html,
399	tags=ALLOWED_TAGS,
400	attributes=ALLOWED_ATTRIBUTES,
401	strip=STRIP_TAGS,
402	)
403
404
405	def set_seq(s):	1✔
406	if settings.DATABASES["default"]["ENGINE"].find("postgresql") >= 0:	×
407	from django.db import connection	×
408
UNCOV 409	cursor = connection.cursor()	×
UNCOV 410	cursor.execute(f"SELECT setval('{s}_id_seq', (SELECT MAX(id) FROM {s}))")	×
411
412
413	def usun_nieuzywany_typ_charakter(klass, field, dry_run):	1✔
UNCOV 414	from bpp.models import Rekord	×
415
UNCOV 416	for elem in klass.objects.all():	×
417	kw = {field: elem}	×
418	if not Rekord.objects.filter(**kw).exists():	×
UNCOV 419	print(f"Kasuje {elem}")	×
420	if not dry_run:	×
421	elem.delete()	×
422
423
424	isbn_regex = re.compile(	1✔
425	r"^isbn\s[0-9][-\| ][0-9][-\| ][0-9][-\| ][0-9][-\| ][0-9]X?",
426	flags=re.IGNORECASE,
427	)
428
429
430	def wytnij_isbn_z_uwag(uwagi):	1✔
UNCOV 431	if uwagi is None:	×
432	return	×
433
UNCOV 434	if uwagi == "":	×
435	return	×
436
437	if uwagi.lower().find("isbn-10") >= 0 or uwagi.lower().find("isbn-13") >= 0:	×
UNCOV 438	return None	×
439
UNCOV 440	res = isbn_regex.search(uwagi)	×
441	if res:	×
UNCOV 442	res = res.group()	×
443	isbn = res.replace("ISBN", "").replace("isbn", "").strip()	×
UNCOV 444	reszta = uwagi.replace(res, "").strip()	×
445
446	while (	×
447	reszta.startswith(".") or reszta.startswith(";") or reszta.startswith(",")
448	):
UNCOV 449	reszta = reszta[1:].strip()	×
450
UNCOV 451	return isbn, reszta	×
452
453
454	def crispy_form_html(self, key):	1✔
UNCOV 455	from crispy_forms_foundation.layout import HTML, Column, Row	×
456
UNCOV 457	from django.utils.functional import lazy	×
458
UNCOV 459	def _():	×
UNCOV 460	return self.initial.get(key, None) or ""	×
461
UNCOV 462	return Row(Column(HTML(lazy(_, str)())))	×
463
464
465	def formdefaults_html_before(form):	1✔
UNCOV 466	return crispy_form_html(form, "formdefaults_pre_html")	×
467
468
469	def formdefaults_html_after(form):	1✔
UNCOV 470	return crispy_form_html(form, "formdefaults_post_html")	×
471
472
473	def knapsack(W, wt, val, ids, zwracaj_liste_przedmiotow=True):	1✔
474	"""
475	:param W: wielkosc plecaka -- maksymalna masa przedmiotów w plecaku (zbierany slot)
476	:param wt: masy przedmiotów, które można włożyć do plecaka (sloty prac)
477	:param val: ceny przedmiotów, które można włożyc do plecaka (punkty PKdAut prac)
478	:param ids: ID prac, które można włożyć do plecaka (rekord.pk)
479	:param zwracaj_liste_przedmiotow: gdy True (domyślnie) funkcja zwróci listę z identyfikatorami włożonych
480	przedmiotów, gdy False zwrócona lista będzie pusta
481
482	:returns: tuple(mp, lista), gdzie mp to maksymalna możliwa wartość włożonych przedmiotów, a lista to lista
483	lub pusta lista gdy parametr `zwracaj_liste_przemiotów` był pusty
484	"""
485
486	assert len(wt) == len(val) == len(ids), "Listy są różnej długości"	×
487
488	sum_wt = sum(wt)	×
489	if sum_wt <= W:	×
490	# Jeżeli wszystkie przedmioty zmieszczą się w plecaku, to po co liczyć cokolwiek
UNCOV 491	if zwracaj_liste_przedmiotow:	×
492	return sum(val), ids	×
UNCOV 493	return sum(val), []	×
494
495	n = len(wt)	×
496
497	K = [[0 for x in range(W + 1)] for x in range(n + 1)]	×
498
499	for i in range(n + 1):	×
500	for w in range(W + 1):	×
501	if i == 0 or w == 0:	×
UNCOV 502	K[i][w] = 0	×
503	elif wt[i - 1] <= w:	×
504	K[i][w] = max(val[i - 1] + K[i - 1][w - wt[i - 1]], K[i - 1][w])	×
505	else:
506	K[i][w] = K[i - 1][w]	×
507
508	res = maks_punkty = K[n][W]	×
509	lista = []	×
510
511	if zwracaj_liste_przedmiotow:	×
UNCOV 512	w = W	×
UNCOV 513	for i in range(n, 0, -1):	×
UNCOV 514	if res <= 0:	×
UNCOV 515	break	×
516
UNCOV 517	if res == K[i - 1][w]:	×
518	continue	×
519	else:
UNCOV 520	lista.append(ids[i - 1])	×
521
UNCOV 522	res = res - val[i - 1]	×
UNCOV 523	w = w - wt[i - 1]	×
524
UNCOV 525	return maks_punkty, lista	×
526
527
528	DEC2INT = 10000	1✔
529
530
531	def intsack(W, wt, val, ids):	1✔
UNCOV 532	pkt, ids = knapsack(	×
533	int(W * DEC2INT),
534	[int(x * DEC2INT) for x in wt],
535	[int(x * DEC2INT) for x in val],
536	ids,
537	)
UNCOV 538	return pkt / DEC2INT, ids	×
539
540
541	def disable_multithreading_by_monkeypatching_pool(pool):	1✔
542	def apply(fun, args=()):	×
543	return fun(*args)	×
544
UNCOV 545	pool.apply = apply	×
546
UNCOV 547	def starmap(fun, lst):	×
UNCOV 548	for elem in lst:	×
UNCOV 549	fun(*elem)	×
550
UNCOV 551	pool.starmap = starmap	×
552
553
554	def year_last_month():	1✔
UNCOV 555	now = timezone.now().date()	×
UNCOV 556	if now.month >= 2:	×
UNCOV 557	return now.year	×
UNCOV 558	return now.year - 1	×
559
560
561	#
562	# Seq Scan check
563	#
564
565
566	class PerformanceFailure(Exception):	1✔
567	pass	1✔
568
569
570	def fail_if_seq_scan(qset, DEBUG):	1✔
571	"""
572	Funkcja weryfikujaca, czy w wyjasnieniu zapytania (EXPLAIN) nie wystapi ciag znakow 'Seq Scan',
573	jezeli tak to wyjatek PerformanceFailure z zapytaniem + wyjasnieniem
574	"""
UNCOV 575	if DEBUG:	×
UNCOV 576	explain = qset.explain()	×
UNCOV 577	if explain.find("Seq Scan") >= 0:	×
UNCOV 578	print("\r\n", explain)	×
UNCOV 579	raise PerformanceFailure(str(qset.query), explain)	×
580
581
582	def rebuild_instances_of_models(modele, args, *kw):	1✔
UNCOV 583	from denorm import denorms	×
UNCOV 584	from django.db import transaction	×
585
586	with transaction.atomic():	×
UNCOV 587	for model in modele:	×
588	denorms.rebuild_instances_of(model, args, *kw)	×
589
590
591	def worksheet_columns_autosize(	1✔
592	ws: openpyxl.worksheet.worksheet.Worksheet,
593	max_width: int = 55,
594	column_widths: Dict[str, int] \| None = None,
595	dont_resize_those_columns: List[int] \| None = None,
596	right_margin=2,
597	multiplier=1.1,
598	):
599	if column_widths is None:	×
600	column_widths = {}	×
601
602	if dont_resize_those_columns is None:	×
603	dont_resize_those_columns = []	×
604
UNCOV 605	for ncol, col in enumerate(ws.columns):	×
606	max_length = 0	×
UNCOV 607	column = col[0].column_letter # Get the column name	×
608
609	# Nie ustawiaj szerokosci tym kolumnom, one będą jako auto-size
UNCOV 610	if ncol in dont_resize_those_columns:	×
UNCOV 611	continue	×
612
613	if column in column_widths:	×
614	adjusted_width = column_widths[column]	×
615	else:
616	for cell in col:	×
617	if cell.value is None or not str(cell.value):	×
UNCOV 618	continue	×
619
620	text = str(cell.value)	×
621
UNCOV 622	if text.startswith("=HYPERLINK"):	×
623	try:	×
624	# Wyciągnij z hiperlinku jego faktyczny opis tekstowy na cele
625	# liczenia szerokości kolumny
UNCOV 626	text = text.split('"')[3]	×
UNCOV 627	except IndexError:	×
UNCOV 628	pass	×
629
UNCOV 630	max_line_len = max(len(line) for line in text.split("\n"))	×
UNCOV 631	max_length = max(max_length, max_line_len)	×
632
UNCOV 633	adjusted_width = (max_length + right_margin) * multiplier	×
UNCOV 634	if adjusted_width > max_width:	×
UNCOV 635	adjusted_width = max_width	×
636
UNCOV 637	ws.column_dimensions[column].width = adjusted_width	×
638
639
640	def worksheet_create_table(	1✔
641	ws: openpyxl.worksheet.worksheet.Worksheet,
642	title="Tabela",
643	first_table_row=1,
644	totals=False,
645	table_columns=None,
646	):
647	"""
648	Formatuje skoroszyt jako tabelę.
649
650	:param first_table_row: pierwszy wiersz tabeli (licząc od nagłówka)
651
652	:param table_columns: określa rodzaj kolumn w tabeli, jeżeli None to tytuły nagłówków zostaną pobrane
653	z pierwszego wiersza w arkuszu.
654	"""
UNCOV 655	max_column = ws.max_column	×
UNCOV 656	max_column_letter = get_column_letter(max_column)	×
UNCOV 657	max_row = ws.max_row	×
658
659	style = TableStyleInfo(	×
660	name="TableStyleMedium9",
661	showFirstColumn=False,
662	showLastColumn=False,
663	showRowStripes=True,
664	showColumnStripes=True,
665	)
666
UNCOV 667	if table_columns is None:	×
UNCOV 668	table_columns = tuple(	×
669	TableColumn(id=h, name=header.value)
670	for h, header in enumerate(next(iter(ws.rows), None), start=1)
671	)
672
UNCOV 673	tab = Table(	×
674	displayName=title,
675	ref=f"A{first_table_row}:{max_column_letter}{max_row}",
676	autoFilter=AutoFilter(
677	ref=f"A{first_table_row}:{max_column_letter}{max_row - 1}"
678	),
679	totalsRowShown=True if totals else False,
680	totalsRowCount=1 if totals else False,
681	tableStyleInfo=style,
682	tableColumns=table_columns,
683	)
684
685	ws.add_table(tab)	×
686
687
688	def worksheet_create_urls(	1✔
689	ws: openpyxl.worksheet.worksheet.Worksheet, default_link_name: str = "[link]"
690	):
691	"""Tworzy adresy URL w postaci klikalnego linku z domyslnym tekstem."""
692
UNCOV 693	for column_cell in ws.iter_cols(1, ws.max_column): # iterate column cell	×
UNCOV 694	if hasattr(column_cell[0].value, "endswith") and column_cell[0].value.endswith(	×
695	"_url"
696	):
UNCOV 697	for data in column_cell[1:]:	×
UNCOV 698	if data.value:	×
UNCOV 699	data.value = '=HYPERLINK("{}", "{}")'.format(	×
700	data.value, default_link_name
701	)
702
703
704	def dont_log_anonymous_crud_events(	1✔
705	instance, object_json_repr, created, raw, using, update_fields, **kwargs
706	):
707	"""
708	Za pomocą tej procedury moduł django-easyaudit decyduje, czy zalogować dane
709	zdarzenie, czy nie.
710
711	Procedura ta sprawdza, czy w parametrach ``kwargs`` zawarty jest parametr
712	``request``, a jeżeli tak -- to czy ma on atrybut ``user`` czyli użytkownik. Jeśli tak,
713	to zwracana jest wartość ``True``, aby dane zdarzenie mogło byc zalogowane.
714
715	Jeżeli nie ma parametru ``user``, to takie zdarzenie logowane nie będzie.
716	"""
UNCOV 717	if kwargs.get("request", None) and getattr(kwargs["request"], "user", None):	×
UNCOV 718	return True	×

iplweb / bpp / 3c407b5b-8434-4f25-9e12-005b14d23a79

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