DFiP_Budget_planing/api/tests/svod2_oracle.py

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