"""SQL-нормализаторы свода 2 (миграция 0053): регистр, обрезка, запись числа как в JS, компактная проекция."""
import json
from pathlib import Path

import pytest

from tests.integration.conftest import legacy_full
from tests.svod2_oracle import compact_projection

NUM = json.loads(
    (Path(__file__).resolve().parent.parent / "fixtures" / "js_number_text_vectors.json").read_text(
        encoding="utf-8"
    )
)


def test_js_number_text_matches_node_vectors(svod2_sql):
    for v in NUM:
        got = svod2_sql(
            "SELECT v3.svod2_js_number_text((CAST(:x AS jsonb) #>> '{}')::float8) AS t",
            {"x": v["input"]},
        )[0][0]
        assert got == v["js"], v


def test_js_number_text_ignores_session_extra_float_digits(svod2_sql):
    svod2_sql("SET LOCAL extra_float_digits = 0", None)
    got = svod2_sql("SELECT v3.svod2_js_number_text(0.1::float8 + 0.2::float8) AS t", None)[0][0]
    assert got == "0.30000000000000004"


@pytest.mark.parametrize(
    "src",
    ["АБВГДЕЁЖЗИЙКЛМНОПРСТУФХЦЧШЩЪЫЬЭЮЯ", "ЄІЇЎЂЏҐ", "ABCxyz", "ÀÉÖÞ", "×", "аРЕНДА 1"],
)
def test_casefold_matches_js_lower(svod2_sql, src):
    assert svod2_sql("SELECT v3.svod2_casefold(:s) AS t", {"s": src})[0][0] == src.lower()


def test_casefold_independent_of_collation(svod2_sql):
    got = svod2_sql("""SELECT v3.svod2_casefold('ЁЛКА' COLLATE "C") AS t""", None)[0][0]
    assert got == "ёлка"


def test_js_trim_ecmascript_whitespace(svod2_sql):
    assert svod2_sql("SELECT v3.svod2_js_trim(:s) AS t", {"s": " 　 x  \t"})[0][0] == "x"
    assert svod2_sql("SELECT v3.svod2_js_trim(:s) AS t", {"s": "​ x"})[0][0] == "​ x"


def test_compact_matches_python_projection(svod2_sql, svod2_data, client, admin_tokens, auth_headers):
    full = legacy_full(client, "RF", auth_headers(admin_tokens))[0]["data"]
    got = svod2_sql("SELECT v3.svod2_compact(CAST(:d AS jsonb)) AS c", {"d": json.dumps(full)})[0][0]
    assert got == compact_projection(full)


def test_single_signature_without_overloads(svod2_sql):
    rows = svod2_sql(
        "SELECT proname, pronargs FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace "
        "WHERE nspname = 'v3' AND proname IN ('v_svod2_go', 'v_svod2_rf', 'v_svod2_go_export')",
        None,
    )
    assert sorted((r.proname, r.pronargs) for r in rows) == [
        ("v_svod2_go", 11), ("v_svod2_go_export", 4), ("v_svod2_rf", 11)]


@pytest.mark.parametrize("fn", ["v_svod2_go", "v_svod2_rf"])
def test_page_total_count_and_order(svod2_sql, svod2_data, admin_tokens, fn):
    # admin_tokens: запрос к API нужен, чтобы откат сессии клиента сработал (loop_ref задаётся при обращении к get_db)
    full = svod2_sql(f"SELECT source_line_id FROM v3.{fn}(2031)", None)
    page = svod2_sql(
        f"SELECT source_line_id, total_count FROM v3.{fn}(2031, NULL, 0, 2, 'org_name', 'asc', "
        "NULL, NULL, NULL, NULL, false)", None)
    assert [r.source_line_id for r in page] == [r.source_line_id for r in full][:2]
    assert {r.total_count for r in page} == {len(full)}


def test_facets_totals(svod2_sql, svod2_data, admin_tokens):
    rows = {(r.kind, r.value): r.n for r in svod2_sql(
        "SELECT * FROM v3.v_svod2_facets('GO', 2031, NULL, NULL, NULL, NULL, NULL)", None)}
    assert rows[("total", None)] == rows[("search_total", None)] == len(
        svod2_sql("SELECT 1 FROM v3.v_svod2_go(2031)", None))


SEARCHES = ["", "   ", "аренда", "АРЕНДА", "ё", "1234.5", "1234.50", " кв ", "1", "true", "777.25", "нет такого"]


@pytest.mark.parametrize("sheet,fn", [("GO", "v_svod2_go"), ("RF", "v_svod2_rf")])
@pytest.mark.parametrize("search", SEARCHES)
def test_search_matches_screen_oracle(svod2_sql, svod2_data, client, admin_tokens, auth_headers,
                                      sheet, fn, search):
    from tests.svod2_oracle import screen_rows

    rows = legacy_full(client, sheet, auth_headers(admin_tokens))
    expected = [r["data"]["line_id"] for r in screen_rows(rows, search)]
    got = svod2_sql(
        f"SELECT source_line_id, total_count FROM v3.{fn}(2031, NULL, NULL, NULL, 'org_name', 'asc', :s)",
        {"s": search},
    )
    assert sorted(r.source_line_id for r in got) == sorted(expected)
    if got:
        assert {r.total_count for r in got} == {len(expected)}


@pytest.mark.parametrize("sheet", ["GO", "RF"])
@pytest.mark.parametrize("search,smeta_type", [("", None), ("1", "кв "), ("аренда", "АХР"), ("", "Операц")])
def test_facets_match_screen_oracle(svod2_sql, svod2_data, client, admin_tokens, auth_headers,
                                    sheet, search, smeta_type):
    from tests.svod2_oracle import screen_facets

    rows = legacy_full(client, sheet, auth_headers(admin_tokens))
    filters = {"smeta_type": smeta_type} if smeta_type else {}
    expected = screen_facets(rows, search, filters)
    got = svod2_sql(
        "SELECT kind, value, n FROM v3.v_svod2_facets(:sh, 2031, NULL, :s, :t, NULL, NULL)",
        {"sh": sheet, "s": search or None, "t": smeta_type},
    )
    assert next(r.n for r in got if r.kind == "total") == expected["total"]
    assert next(r.n for r in got if r.kind == "search_total") == expected["search_total"]
    for col in ("smeta_type", "smeta_direction", "org_name"):
        assert {r.value: r.n for r in got if r.kind == col} == expected[col]


def test_facets_search_subplan_is_not_correlated(svod2_sql):
    import re

    plan = "\n".join(r[0] for r in svod2_sql(
        "EXPLAIN (VERBOSE, COSTS OFF) "
        "SELECT * FROM v3.v_svod2_facets('GO', 2031, NULL, 'аренда', NULL, NULL, NULL)", None))
    # IN по svod2_lines должен быть hashed SubPlan (считается один раз); коррелированный
    # считался бы на каждую строку. Не-hashed SubPlan в Filter внутри svod2_lines (EXISTS поиска)
    # законен, поэтому смотрим только на выражения в Output.
    outputs = [l for l in plan.splitlines() if "Output:" in l and "SubPlan" in l]
    assert any("hashed SubPlan" in l for l in outputs), plan
    assert not any(re.search(r"(?<!hashed )SubPlan \d+", l) for l in outputs), plan