ba6f9e1f-4683-40a1-aae1-40dd0fcb64e3

Committed 25 Aug 2025 06:57PM UTC coverage: 43.284% (+0.6%) from 42.715%

Build # ba6f9e1f-4683-40a1-aae1-40dd0fcb64e3

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Merge branch 'release/v202508.1208'

Run Details

77 of 961 new or added lines in 27 files covered. (8.01%)

731 existing lines in 54 files now uncovered.

17273 of 39906 relevant lines covered (43.28%)

0.78 hits per line

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

35.74

src/bpp/util.py

import json
import multiprocessing
import operator
import os
import re
import sys
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):
    if sys.stdout.isatty() and not disable_progress_bar:
        if count is None:
            if hasattr(query, "count"):
                try:
                    count = query.count()
                except TypeError:
                    count = len(query)
            elif hasattr(query, "__len__"):
                count = len(query)

        return tqdm(query, total=count, desc=label, unit="items")
    else:
        # You're being piped or redirected
        return query


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

iplweb / bpp / ba6f9e1f-4683-40a1-aae1-40dd0fcb64e3

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