58b9a630-8512-44e6-b730-daac96d1c4d7

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

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

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Fix tests

Run Details

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

1342 existing lines in 64 files now uncovered.

19323 of 40686 relevant lines covered (47.49%)

1.51 hits per line

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

38.92

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):
    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")


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

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

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

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