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


# --- Поиск в svod2_lines (ускорение AURORA-1843): результат равен прежнему поиску по jsonb ---

# Прежний поиск дословно: значения header + contract_summary + contract_detail через jsonb_each,
# регистр svod2_casefold по каждому значению, числа через svod2_js_number_text.
REFERENCE_SEARCH_SQL = """
SELECT l.line_id
FROM v3.svod2_lines(:sh, 2031, NULL, NULL) l
WHERE EXISTS (
    SELECT 1
    FROM v3.budget_line bl
    JOIN v3.expense_item ei ON ei.id = bl.expense_item_id
    LEFT JOIN v3.contract_summary cs ON cs.line_id = bl.id
    LEFT JOIN v3.contract_detail cd ON cd.line_id = bl.id
    LEFT JOIN v3.ckk ck ON ck.line_id = bl.id
    CROSS JOIN LATERAL (
        SELECT
          to_jsonb(jsonb_populate_record(NULL::v3.contract_summary, '{}'::jsonb)) - 'line_id' AS cs0,
          to_jsonb(jsonb_populate_record(NULL::v3.contract_detail, '{}'::jsonb)) - 'line_id' AS cd0
    ) e
    CROSS JOIN LATERAL (
        SELECT
          jsonb_build_object(
            'section_code', ei.section_code, 'item_id', ei.item_id,
            'num_group', ei.num_group_id, 'name', COALESCE(bl.name, ei.name),
            'expense_item_id', ei.id, 'justification', bl.justification,
            'internal_order', bl.internal_order, 'vsp_id', bl.vsp_id,
            'project_id', bl.project_id, 'direction', bl.direction,
            'smeta_direction', CASE bl.direction
              WHEN 'Support' THEN 'Поддержка' WHEN 'Development' THEN 'Развитие' ELSE NULL END,
            'source_form', l.form_type_code, 'source_id', l.form_id,
            'source_org_unit_id', l.org_unit_id, 'org_name', l.org_name,
            'smeta_type', l.smeta_type) AS header_j,
          COALESCE(to_jsonb(cs) - 'line_id', e.cs0) AS cs_j,
          CASE WHEN :sh = 'RF'
            THEN COALESCE(to_jsonb(cd) - 'line_id', e.cd0) || jsonb_build_object(
              'expenses_q1', COALESCE(cd.expenses_q1, ck.expenses_q1),
              'expenses_q2', COALESCE(cd.expenses_q2, ck.expenses_q2),
              'expenses_q3', COALESCE(cd.expenses_q3, ck.expenses_q3),
              'expenses_q4', COALESCE(cd.expenses_q4, ck.expenses_q4))
            ELSE COALESCE(to_jsonb(cd) - 'line_id', e.cd0)
          END AS cd_j
    ) s
    CROSS JOIN LATERAL jsonb_each(s.header_j || s.cs_j || s.cd_j) kv
    WHERE bl.id = l.line_id
      AND strpos(
        v3.svod2_casefold(
          CASE jsonb_typeof(kv.value)
            WHEN 'string' THEN kv.value #>> '{}'
            WHEN 'number' THEN v3.svod2_js_number_text((kv.value #>> '{}')::float8)
            WHEN 'boolean' THEN kv.value #>> '{}'
            WHEN 'null' THEN ''
            ELSE kv.value::text
          END),
        :s) > 0
)
"""

# Данные для границ ускорения: соседние значения строки свода (name и justification идут подряд при склейке),
# U+0001 внутри значения, числа с экспонентой и отрицательные, ckk без contract_detail, дата.
SEARCH_EXTRA_DATA = [
    "UPDATE v3.budget_line SET name = 'ab', justification = 'cd' WHERE id = 910001",
    "UPDATE v3.budget_line SET internal_order = E'x\\u0001y' WHERE id = 910004",
    "INSERT INTO v3.contract_detail (line_id, counterparty, reference, ceiling_amount, expenses_q1, expenses_q2, "
    "exchange_rate, amount_foreign, contract_date, act) VALUES "
    "(910007, 'ЗАО Эй', 'Ref-А', 15000000000000000000000000, 12.5, 0, 0.0000001, -0.0000001234, '2031-03-04', 'Infinity')",
    "UPDATE v3.contract_detail SET exchange_rate = 'NaN', amount_foreign = '-Infinity' WHERE line_id = 910002",
    "INSERT INTO v3.ckk (line_id, ceiling_amount, expenses_q1, expenses_q2) VALUES (910009, 1, 555.5, 6)",
]

# Нормализованные (как приходят от v_svod2_*) и ненормализованные строки; по ветвям ускорения:
# только символы числа (числа переводятся в текст) и нет (не переводятся), есть/нет пар таблицы регистра,
# U+0001 в строке, стык значений, пустая строка.
SEARCH_EQUIV = [
    "", "аренд", "ё", "ёлк", "зао", "эй", "ref-а", "ооо", "д-", "zzzz", "а",
    "1", "0", "-", ".", "e", "+", "e+25", "1.5e+25", "1e-7", "-1.234e-7", "-1", "555.5", "777.25", "12.5", "1234.5",
    "infinity", "nan", "a", "nt", "inf", "2031-03-04", "03-04", "12",
    "abcd", "ab", "cd", "bc", "abc", "ab\u0001cd", "x\u0001y", "x", "y", "\u0001", "y\u0001", "\u0001x", "аренд\u0001",
    "АРЕНД", "Ё", "E", "Infinity", "ЭЙ", "Ab", "REF",
]


def test_search_columns_are_those_known_to_svod2_lines(svod2_sql):
    # svod2_lines перечисляет значения для поиска явно (по колонкам, без jsonb). Новая колонка в
    # contract_summary/contract_detail не попадёт в поиск, пока её не добавят в svod2_lines: тест просит это сделать.
    # Из contract_summary в поиск идут только колонки, которых нет в contract_detail: прежний
    # header || contract_summary || contract_detail затирал одноимённые.
    def columns(table: str) -> set[str]:
        return {r[0] for r in svod2_sql(
            "SELECT column_name FROM information_schema.columns "
            "WHERE table_schema = 'v3' AND table_name = :t", {"t": table})}

    assert columns("contract_summary") == {
        "line_id", "total_amount", "counterparty", "reference", "deadline", "comment",
        "future_payments_y1", "future_payments_y2", "other_ssp_amount", "centralized_flag"}
    assert columns("contract_detail") == {
        "line_id", "counterparty", "reference", "addenda", "subject", "currency", "ceiling_amount",
        "expenses_q1", "expenses_q2", "expenses_q3", "expenses_q4", "rf_schedule", "vat_rate",
        "exchange_rate", "amount_foreign", "deadline", "payment_scheme", "act", "comment", "contract_date"}
    # ключи header не пересекаются с колонками contract_*
    header_keys = {
        "section_code", "item_id", "num_group", "name", "expense_item_id", "justification",
        "internal_order", "vsp_id", "project_id", "direction", "smeta_direction", "source_form",
        "source_id", "source_org_unit_id", "org_name", "smeta_type"}
    assert not header_keys & (columns("contract_summary") | columns("contract_detail"))


@pytest.mark.parametrize("sheet", ["GO", "RF"])
@pytest.mark.parametrize("search", SEARCH_EQUIV, ids=[repr(s) for s in SEARCH_EQUIV])
def test_svod2_lines_search_equals_reference(svod2_sql, svod2_data, sheet, search):
    for statement in SEARCH_EXTRA_DATA:
        svod2_sql(statement)
    got = svod2_sql(
        "SELECT line_id FROM v3.svod2_lines(:sh, 2031, NULL, :s)", {"sh": sheet, "s": search})
    expected = svod2_sql(REFERENCE_SEARCH_SQL, {"sh": sheet, "s": search})
    assert sorted(r[0] for r in got) == sorted(r[0] for r in expected)


def test_svod2_lines_search_extra_data_is_searchable(svod2_sql, svod2_data):
    # Страховка от пустых сравнений: на этих данных поиск что-то находит и что-то отсекает.
    for statement in SEARCH_EXTRA_DATA:
        svod2_sql(statement)

    def found(sheet: str, search: str) -> int:
        return len(svod2_sql(
            "SELECT line_id FROM v3.svod2_lines(:sh, 2031, NULL, :s)", {"sh": sheet, "s": search}))

    assert found("GO", "ab") == 1 and found("GO", "cd") == 1
    assert found("GO", "abcd") == 0 and found("GO", "ab\u0001cd") == 0   # стык значений
    assert found("GO", "x\u0001y") == 1                                  # U+0001 внутри значения
    assert found("RF", "e+25") == 1 and found("RF", "-1.234e-7") == 1    # число как в JS
    assert found("RF", "555.5") == 1                                     # ckk без contract_detail
    assert found("RF", "zzzz") == 0