18634744198

Committed 19 Oct 2025 07:00PM UTC coverage: 31.618% (-29.9%) from 61.514%

Build # 18634744198

Build Type

push

github

Committed by

mpasternak

Commit Message

Merge branch 'release/v202510.1270'

Run Details

657 of 9430 branches covered (6.97%)

Branch coverage included in aggregate %.

229 of 523 new or added lines in 42 files covered. (43.79%)

11303 existing lines in 316 files now uncovered.

14765 of 39346 relevant lines covered (37.53%)

0.38 hits per line

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

33.26

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

import bleach
import lxml.html
import openpyxl.worksheet.worksheet
from django.apps import apps
from django.conf import settings
from django.contrib.postgres.search import SearchQuery, SearchRank
from django.db.models import Max, Min, Value
from django.utils import timezone
from django.utils.html import strip_tags
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

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


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


strip_nonalpha_regex = re.compile(r"\W+")
strip_extra_spaces_regex = re.compile(r"\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) from 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[f"{field_name}__lt"] = since

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

        rec.delete()

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


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

    create_contenttypes(app_config, verbosity=0)


#
# Progress bar
#


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

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

        return callback_only_wrapper()

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


#
# Multiprocessing stuff
#


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


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


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


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


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

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


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

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


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

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


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

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


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


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

    if uwagi == "":
        return

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

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

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

        return isbn, reszta


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

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

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


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


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


def _build_knapsack_table(n, W, wt, val):
    """Build the dynamic programming table for knapsack problem."""
    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]

    return K


def _reconstruct_knapsack_items(K, n, W, wt, val, ids):
    """Reconstruct which items were selected in the optimal solution."""
    res = K[n][W]
    lista = []
    w = W

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

        if res != K[i - 1][w]:
            lista.append(ids[i - 1])
            res = res - val[i - 1]
            w = w - wt[i - 1]

    return lista


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 = _build_knapsack_table(n, W, wt, val)

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

    if zwracaj_liste_przedmiotow:
        lista = _reconstruct_knapsack_items(K, n, W, wt, val, ids)

    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 _extract_hyperlink_text(text):
    """Extract display text from hyperlink formula."""
    if text.startswith("=HYPERLINK"):
        try:
            # Wyciągnij z hiperlinku jego faktyczny opis tekstowy na cele
            # liczenia szerokości kolumny
            return text.split('"')[3]
        except IndexError:
            pass
    return text


def _calculate_column_width(col, right_margin, multiplier, max_width):
    """Calculate optimal width for a column based on its content."""
    max_length = 0

    for cell in col:
        if cell.value is None or not str(cell.value):
            continue

        text = str(cell.value)
        text = _extract_hyperlink_text(text)

        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

    return adjusted_width


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):
        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:
            adjusted_width = _calculate_column_width(
                col, right_margin, multiplier, 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 = f'=HYPERLINK("{data.value}", "{default_link_name}")'


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

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

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

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

iplweb / bpp / 18634744198

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