7d423bde-8caf-4c8b-8bf5-12ad3f831949

Committed 22 Aug 2025 01:06PM UTC coverage: 45.781% (+0.06%) from 45.719%

Build # 7d423bde-8caf-4c8b-8bf5-12ad3f831949

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Merge branch 'release/v202508.1201'

Run Details

11 of 18 new or added lines in 3 files covered. (61.11%)

1245 existing lines in 104 files now uncovered.

17481 of 38184 relevant lines covered (45.78%)

1.18 hits per line

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

34.17

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

iplweb / bpp / 7d423bde-8caf-4c8b-8bf5-12ad3f831949

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