"""Python-порт оракула экрана «до» свода 2 (perf/analysis/svod2_search_oracle.mjs). Поведение сверяется с векторами, снятыми в Node.js (tests/fixtures/*.json). """ from __future__ import annotations import math from decimal import Decimal FILTER_COLUMNS = ("smeta_type", "smeta_direction", "org_name") # ECMAScript WhiteSpace + LineTerminator (design § 3). _JS_WHITESPACE = ( "\u0009\u000a\u000b\u000c\u000d   " "           " "

   " ) def js_trim(s: str) -> str: return s.strip(_JS_WHITESPACE) def js_number_text(x: int | float) -> str: """ECMAScript Number::toString (radix 10).""" f = float(x) if math.isnan(f): return "NaN" if f == 0: return "0" if math.isinf(f): return "Infinity" if f > 0 else "-Infinity" sign = "-" if f < 0 else "" # Кратчайшая запись цифр: repr(float) -> Decimal -> (digits, exponent). t = Decimal(repr(abs(f))).as_tuple() digits = "".join(map(str, t.digits)).rstrip("0") # значение = 0.d1d2...dk * 10^n n = len(t.digits) + t.exponent k = len(digits) if not digits: return "0" # убрать ведущие нули (Decimal('0.001') даёт digits (1,), без лишних; страховка) stripped = digits.lstrip("0") n -= len(digits) - len(stripped) digits, k = stripped, len(stripped) if k <= n <= 21: body = digits + "0" * (n - k) elif 0 < n <= 21: body = digits[:n] + "." + digits[n:] elif -6 < n <= 0: body = "0." + "0" * (-n) + digits else: e = n - 1 es = ("+" if e >= 0 else "-") + str(abs(e)) body = (digits[0] + ("." + digits[1:] if k > 1 else "")) + "e" + es return sign + body def js_string(value: object) -> str: """String(value ?? '').""" if value is None: return "" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float)): return js_number_text(value) return str(value) def matches_search(row: dict, query: str) -> bool: data = row.get("data") or {} merged = { **(data.get("header") or {}), **(data.get("contract_summary") or {}), **(data.get("contract_detail") or {}), } return any(query in js_string(v).lower() for v in merged.values()) def _by_search(rows: list[dict], search: str) -> list[dict]: q = js_trim(search or "").lower() return [r for r in rows if matches_search(r, q)] if q else rows def _cell(row: dict, col: str): return ((row.get("data") or {}).get("header") or {}).get(col) def _equals(cell, filter_value: str) -> bool: if cell is None: return False return js_trim(js_string(cell).lower()) == js_trim(filter_value.lower()) def _by_filters(rows: list[dict], filters: dict[str, str], skip: str | None) -> list[dict]: active = [c for c in FILTER_COLUMNS if c != skip and filters.get(c)] return [r for r in rows if all(_equals(_cell(r, c), filters[c]) for c in active)] def screen_rows(rows, search: str = "", filters=None) -> list[dict]: return _by_filters(_by_search(rows, search), filters or {}, None) def screen_found(rows, search: str) -> tuple[int, int]: return len(_by_search(rows, search)), len(rows) def screen_facets(rows, search: str = "", filters=None) -> dict: filters = filters or {} searched = _by_search(rows, search) result: dict = {"total": len(rows), "search_total": len(searched)} for col in FILTER_COLUMNS: counts: dict = {} for r in _by_filters(searched, filters, col): v = _cell(r, col) if v is not None: counts[v] = counts.get(v, 0) + 1 result[col] = dict(sorted(counts.items())) return result _Q = ("adj_current", "adj_ssp", "adj_rf", "adj_reserve", "payment_date", "payment_amount", "payment_amount_ho", "payment_amount_rf", "payment_act", "booking_amount", "actual_m1", "actual_m2", "actual_m3") def _build_paths() -> tuple[str, ...]: p = ["line_id"] p += [f"header.{f}" for f in ("source_form", "source_id", "smeta_type", "smeta_direction", "org_name", "name", "item_id", "num_group", "justification")] p += [f"plan.plan_q{i}" for i in range(1, 5)] p += ["allocation.internal_order", "allocation.property_object"] p += [f"sequestration.DFIP.adj_q{i}" for i in range(1, 5)] p += [f"reserve.amount_q{i}" for i in range(1, 5)] p += [f"approved.approved_q{i}" for i in range(1, 5)] p += [f"contract_detail.{f}" for f in ("counterparty", "reference", "subject", "currency", "ceiling_amount")] p += [f"contract_detail.expenses_q{i}" for i in range(1, 5)] p += ["contract_detail.vat_rate", "contract_detail.deadline"] for q in range(1, 5): p += [f"q{q}.{f}" for f in _Q] if q == 4: p.append("q4.actual_spod") if q <= 3: p += [f"q{q}.transfer_to_q{k}" for k in range(q + 1, 5)] p.append(f"q{q}.transfer_to_economy") if q >= 2: p += [f"q{q}.{f}" for f in ("plan_revision_eco_change", "plan_revision_item_adj", "plan_revision_increase", "plan_revision_sequester")] return tuple(p) COMPACT_PATHS: tuple[str, ...] = _build_paths() def compact_projection(data: dict) -> dict: """jsonb_strip_nulls(jsonb_build_object(...)) по COMPACT_PATHS. Группы и подгруппы присутствуют всегда; листья None/отсутствующие опускаются. """ out: dict = {} for path in COMPACT_PATHS: parts = path.split(".") src = data for part in parts[:-1]: src = src.get(part) if isinstance(src, dict) else None dst = out for part in parts[:-1]: dst = dst.setdefault(part, {}) leaf = src.get(parts[-1]) if isinstance(src, dict) else None if len(parts) == 1: if leaf is not None: out[parts[0]] = leaf elif leaf is not None: dst[parts[-1]] = leaf return out