"""Переиспользуемая проверка контракта ответа на правку ячеек.

Каждая строка ответа PATCH /cell(s) должна совпасть (data и sort_order) со строкой
того же (row_type, depth, sort_order) из последующего GET /sheet.
Используется в AURORA-1844 и далее (AURORA-1850: пересчёт только затронутого поддерева).
"""


def row_key(row: dict) -> tuple:
    return (row["row_type"], row["depth"], row["sort_order"])


def assert_patch_rows_match_sheet(patch_rows: list[dict], sheet_rows: list[dict]) -> None:
    """Строки ответа на правку целиком совпадают со строками листа."""
    sheet_by_key = {row_key(r): r for r in sheet_rows}
    assert patch_rows, "ответ на правку пуст"
    keys = [row_key(r) for r in patch_rows]
    assert len(keys) == len(set(keys)), "в ответе дубли строк"
    for row in patch_rows:
        key = row_key(row)
        assert key in sheet_by_key, f"строки {key} нет в GET /sheet"
        assert row["data"] == sheet_by_key[key]["data"], f"data строки {key} расходится с GET"


# --- AURORA-1850: эталон (фильтр 0050 поверх полного v_form_view) и хелперы SQL ---

Row4 = tuple[str, int, int, str]  # (row_type, depth, sort_order, data::text)

_STRIP_ALL = [
    "plan", "seq_dfip", "seq_ssp", "approved", "contract", "booking",
    "contract_summary", "allocation", "collegial", "ckk",
    "sequestration", "reserve", "contract_detail",
    "q1", "q2", "q3", "q4", "totals",
    "fact_q1", "fact_q2", "fact_q3", "fact_q4",
]

_SAT_SHEETS = ("AHR_RENT", "AHR_UTILITY", "AHR_SECURITY")


def strip_keys(sections: list[str] | None) -> list[str]:
    """Ключи верхнего уровня исключённых разделов (как v_strip в 0050)."""
    if sections is None:
        return []
    return [k for k in _STRIP_ALL if k not in sections]


def sheet_kind(sheet: str) -> str:
    if sheet in _SAT_SHEETS:
        return "sat"
    if sheet == "AHR_LIMIT":
        return "limit"
    if sheet == "OTCH9F":
        return "otch9f"
    return "tree"


_REF_FROM = """
  SELECT v.row_type, v.depth, v.sort_order, (v.data - CAST(:strip AS text[]))::text
    FROM v3.v_form_view(:f, :s, CAST(:sec AS text[]), CAST(:d AS varchar)) v
   WHERE {where}
   ORDER BY v.sort_order"""

_REF_WHERE = {
    "sat": "v.row_type = 'INPUT' AND (v.data->>'id')::int = ANY(CAST(:ids AS int[]))",
    "limit": "(v.data->>'id')::int = ANY(v3._limit_template_path_set(CAST(:ids AS int[])))",
    "otch9f": """EXISTS (SELECT 1 FROM v3.fixed_asset_report far
                  WHERE far.id = ANY(CAST(:ids AS int[]))
                    AND far.expense_item_id = (v.data->>'expense_item_id')::int
                    AND far.equipment_name IS NOT DISTINCT FROM (v.data->>'equipment_name'))""",
    "tree": """(v.row_type IN ('ROOT','GROUP','ITEM','SUB_ITEM','SUB_SUB_ITEM')
                AND (v.data->'header'->>'expense_item_id')::int = ANY(v3._expense_item_path_set(
                      ARRAY(SELECT DISTINCT bl.expense_item_id FROM v3.budget_line bl
                             WHERE bl.id = ANY(CAST(:ids AS int[]))))))
               OR (v.row_type = 'INPUT' AND (v.data->>'line_id')::int = ANY(CAST(:ids AS int[])))""",
}


def _rows(db_exec, sql: str, **params) -> list:
    return [tuple(r) for r in db_exec(sql, **params)]


def reference_affected_rows(db_exec, form_id, sheet, ids, direction=None, sections=None, strip=True):
    sql = _REF_FROM.format(where=_REF_WHERE[sheet_kind(sheet)])
    return _rows(
        db_exec, sql, f=form_id, s=sheet, ids=list(ids), d=direction, sec=sections,
        strip=strip_keys(sections) if strip else [],
    )


def affected_rows(db_exec, form_id, sheet, ids, direction=None, sections=None):
    return _rows(
        db_exec,
        """SELECT r.row_type, r.depth, r.sort_order, r.data::text
             FROM v3._form_affected_rows(:f, :s, CAST(:ids AS int[]), CAST(:d AS varchar),
                                         CAST(:sec AS text[])) r
            ORDER BY r.sort_order""",
        f=form_id, s=sheet, ids=list(ids), d=direction, sec=sections,
    )


def focus_builder_rows(db_exec, form_type, form_id, sheet, ids, direction=None, sections=None):
    if form_type == "FORM_2":
        sql = """SELECT r.row_type, r.depth, r.sort_order, r.data::text
                   FROM v3.v_form2_view(:f, :s, CAST(:sec AS text[]), CAST(:ids AS int[])) r
                  ORDER BY r.sort_order"""
        params = dict(f=form_id, s=sheet, sec=sections, ids=list(ids))
    else:
        sql = """SELECT r.row_type, r.depth, r.sort_order, r.data::text
                   FROM v3.v_form1_view(:f, :s, CAST(:sec AS text[]), CAST(:d AS varchar),
                                        CAST(:ids AS int[])) r
                  ORDER BY r.sort_order"""
        params = dict(f=form_id, s=sheet, sec=sections, d=direction, ids=list(ids))
    return _rows(db_exec, sql, **params)


def assert_rows_equal(actual: list, expected: list, label: str) -> None:
    """Построчное сравнение; сообщение называет первую отличающуюся строку и длины."""
    if actual == expected:
        return
    n = min(len(actual), len(expected))
    first = next((i for i in range(n) if actual[i] != expected[i]), n)
    a = actual[first] if first < len(actual) else None
    e = expected[first] if first < len(expected) else None
    raise AssertionError(
        f"{label}: длины actual={len(actual)} expected={len(expected)}; "
        f"первое расхождение в позиции {first}:\n  actual   = {a}\n  expected = {e}"
    )