"""Сквозной сценарий Среза 4 на реальных данных группы «Русский бисквит» (tasks 5.1; spec consolidated-osv «Признаки на
фикстурах РБ», «Сопоставление из плана среза»; documents-api «Фикстуры, подтверждённые до консолидации»).

Golden-снимки парсера `rb/2024-Q1..Q3` и `schastl/2024-Q1..Q3` → боевой `map_file` → подтверждение через
`POST /documents/{id}/confirm` → создание консолидации 2024 Q3 (подхват подтверждённых) → блокировка → сводная ОСВ,
сводка и пробелы, ручное сопоставление «Пекарни Счастливочной» с автоприменением → xlsx-экспорт. Суммы свода сверяются с
суммой квартальных значений, прочитанных из golden-JSON независимо от доменного кода.
"""

import io
import json
import time
import uuid
from collections import defaultdict
from decimal import Decimal
from pathlib import Path
from typing import Any

import pytest
from httpx import AsyncClient
from loguru import logger
from openpyxl import load_workbook
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.domain.aliases import normalize_alias
from src.domain.osv_mapping import normalize_account_code
from src.models.access import AppUser
from src.models.consolidation import ConsolidationDocument
from src.models.directories import RegionalOffice
from src.models.enums import CounterpartyMatchKind, DocumentStatus
from src.models.files import Document
from src.models.osv import OsvEntry
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_legal_entity, make_membership, make_related_group
from tests.fixtures_helpers import map_golden_snapshot

pytestmark = [pytest.mark.integration, pytest.mark.fixtures]

HTTP_OK = 200
HTTP_CREATED = 201
GOLDEN = Path("golden") / "expected_osv_parse"
KA_RF = "ka_rf@armka.local"
ROSTOV = "Ростовский региональный филиал"  # филиал dev-пользователя ka_rf (в плане — СПб РФ: у ka_rf его нет)
RB_INN = "3528052718"  # АО «Русский бисквит» (ЕЦП-выписка 01.04.26)
SCHASTL_INN = "3528104476"  # ООО «Пекарня Счастливочная» (там же)
SCHASTL_FILE_NAME = "ПЕКАРНЯ СЧАСТЛИВОЧНАЯ"  # в файлах РБ контрагент записан «Юг-Финанс / ПЕКАРНЯ СЧАСТЛИВОЧНАЯ»
RB_NAME_IN_SCHASTL = "Русский бисквит"
TIME_BUDGET_SECONDS = 60.0
Q1, Q2, Q3 = "Q1", "Q2", "Q3"
# свод РБ 2024 Q3; 1379 = 1381 − 2: две строки находят контрагента по ИНН, извлечённому парсером из названия
# (tabular-parser #15, golden fixtures #8; issue #33)
ROWS_TOTAL, UNMAPPED_TOTAL = 1385, 1379
CHANGED_ROWS, PROPAGATED = 4, 17  # строк свода, получивших контрагента; записей osv_entry, кроме выбранной (18 всего)
SUMS = ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit", "closing_debit", "closing_credit")


def _is_schastl(name: str | None) -> bool:
    return SCHASTL_FILE_NAME.casefold() in normalize_alias(name or "")


def _golden_files(fixtures_dir: Path, folder: str) -> list[Path]:
    paths = [
        p for quarter in (1, 2, 3) for p in sorted((fixtures_dir / GOLDEN / folder / f"2024-Q{quarter}").glob("*.json"))
    ]
    assert paths, folder
    return paths


def _golden_rows(fixtures_dir: Path, folder: str, account_root: str) -> dict[str, list[dict[str, Any]]]:
    """Строки-контрагенты golden-снимков ОСВ счёта `account_root` по кварталам — для независимого расчёта."""
    result: dict[str, list[dict[str, Any]]] = {Q1: [], Q2: [], Q3: []}
    for path in _golden_files(fixtures_dir, folder):
        for sheet in json.loads(path.read_text("utf-8"))["documents"]:
            header = sheet["header"]
            if sheet["document_type"] != "OSV" or header["account_code_raw"].split(".")[0] != account_root:
                continue
            result[header["period_quarter"]].extend(r for r in sheet["rows"] if r["group_type"] == "COUNTERPARTY")
    return result


def _key(row: dict[str, Any]) -> tuple[str, str, str]:
    return (
        normalize_account_code(row["account_code_raw"]),
        (row["counterparty_name"] or "").strip(),
        (row["counterparty_inn"] or "").replace(" ", ""),
    )


def _dec(value: str | None) -> Decimal:
    return Decimal(value) if value is not None else Decimal(0)


async def _get(client: AsyncClient, url: str, **params: Any) -> Any:
    response = await client.get(url, params=params)
    assert response.status_code == HTTP_OK, response.text
    return response.json()


async def _all_rows(client: AsyncClient, version_url: str, **params: Any) -> list[dict[str, Any]]:
    page = await _get(client, f"{version_url}/osv", limit=2000, **params)
    assert page["total"] == len(page["items"]), "свод не помещается в одну страницу"
    items: list[dict[str, Any]] = page["items"]
    return items


async def test_consolidation_on_russian_biscuit_fixtures(  # noqa: PLR0912, PLR0914, PLR0915 - сквозной сценарий в одном месте
    auth_client: AuthClientFactory, seeded: AsyncSession, fixtures_dir: Path
) -> None:
    started = time.perf_counter()
    office = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
    user = (await seeded.execute(select(AppUser).where(AppUser.email == KA_RF))).scalar_one()
    group = await make_company_group(seeded, name="ГК Русский бисквит", office=office)
    gsk = await make_related_group(seeded, group=group, name="ГСК Русский бисквит")
    rb = await make_legal_entity(seeded, name="АО «Русский бисквит»", inn=RB_INN, group=group)
    schastl = await make_legal_entity(seeded, name="ООО «Пекарня Счастливочная»", inn=SCHASTL_INN, group=group)
    for entity in (rb, schastl):
        await make_membership(seeded, related_group=gsk, entity=entity)

    documents: list[Document] = []
    for folder, entity in (("rb", rb), ("schastl", schastl)):
        for path in _golden_files(fixtures_dir, folder):
            documents.extend(await map_golden_snapshot(seeded, entity=entity, path=path, uploaded_by=user.id))
    assert len(documents) == 11 + 11  # РБ: Q1 2, Q2 2, Q3 7; Счастливочная: Q1 3, Q2 3, Q3 5
    assert all(d.status is not DocumentStatus.NEEDS_CHECK for d in documents)  # подтверждаются без правок

    client = await auth_client(KA_RF)
    for document in documents:
        response = await client.post(f"/documents/{document.id}/confirm")
        assert response.status_code == HTTP_OK, (document.osv_account_code, document.period_quarter, response.text)

    created = await client.post(
        "/consolidations",
        json={"company_group_id": str(group.id), "related_group_id": str(gsk.id), "year": 2024, "quarter": "Q3"},
    )
    assert created.status_code == HTTP_CREATED, created.text
    card = created.json()
    assert card["status"] == "INITIATED"
    version_id = card["id"]
    version_url = f"/consolidations/versions/{version_id}"
    bound = (
        await seeded.execute(
            select(func.count())
            .select_from(ConsolidationDocument)
            .where(ConsolidationDocument.consolidation_version_id == uuid.UUID(version_id))
        )
    ).scalar_one()
    logger.info("привязано документов к версии: {}", bound)
    assert bound == len(documents)  # «Фикстуры, подтверждённые до консолидации»: подхвачены все подтверждённые ОСВ

    locked = await client.post(f"{version_url}/lock")
    assert locked.status_code == HTTP_CREATED, locked.text

    summary = await _get(client, f"{version_url}/osv/summary")
    rows = await _all_rows(client, version_url)
    logger.info(
        "сводная ОСВ РБ 2024 Q3: строк {}, сводка {}", len(rows), {k: v for k, v in summary.items() if k != "gaps"}
    )
    print(
        f"\nSVOD: rows={len(rows)} summary={ {k: v for k, v in summary.items() if k != 'gaps'} } gaps={summary['gaps']}"
    )  # noqa: T201
    # Числа сверены независимым подсчётом по сырым golden-JSON (склейка Q1–Q3, пробелы по наличию файлов).
    assert summary["rows_total"] == len(rows) == ROWS_TOTAL
    assert summary["unmapped_total"] == UNMAPPED_TOTAL

    # строки свода: 60 и 62 у обеих, 76 у Счастливочной
    present = {(row["legal_entity_id"], row["account_root"]) for row in rows}
    for account in ("60", "62"):
        assert (str(rb.id), account) in present
        assert (str(schastl.id), account) in present
    assert (str(schastl.id), "76") in present

    # пробелы (точный набор по файлам golden): у РБ за Q1–Q2 нет ОСВ 01, 58.03, 66, 67, 76, у Счастливочной — 01, 02
    gaps = {(g["legal_entity_id"], g["account_code"]): g["missing_quarters"] for g in summary["gaps"]}
    assert len(summary["gaps"]) == len(gaps)
    assert gaps == {
        **{(str(rb.id), account): [Q1, Q2] for account in ("01", "58.03", "66", "67", "76")},
        **{(str(schastl.id), account): [Q1, Q2] for account in ("01", "02")},
    }

    # чек-лист: ОСВ 58 РБ не подтверждена (есть только 58.03)
    checklist = await _get(client, f"{version_url}/checklist")
    column = next(
        i for i, rule in enumerate(checklist["rules"]) if (rule["document_type"], rule["account"]) == ("OSV", "58")
    )
    rb_row = next(r for r in checklist["rows"] if r["legal_entity_id"] == str(rb.id))
    assert rb_row["cells"][column]["status"] != "CONFIRMED"
    for account in ("60", "62"):  # ОСВ 60/62 РБ за Q1–Q3 подтверждены до консолидации — ячейка CONFIRMED, три документа
        cell = rb_row["cells"][
            next(i for i, r in enumerate(checklist["rules"]) if (r["document_type"], r["account"]) == ("OSV", account))
        ]
        assert (cell["status"], len(cell["document_ids"])) == ("CONFIRMED", 3), account  # noqa: PLR2004

    # несопоставленные строки РБ «ПЕКАРНЯ СЧАСТЛИВОЧНАЯ»
    unmapped = await _all_rows(client, version_url, unmapped="true", legal_entity_id=str(rb.id))
    target = [row for row in unmapped if _is_schastl(row["counterparty_name"])]
    assert target
    first = target[0]
    # записи склеенной строки идут по кварталам; сопоставляем запись последнего квартала: в строке свода показывается
    # способ сопоставления записи самого позднего квартала (D-3, решение №8), поэтому только она даёт `MANUAL`
    entry_id = first["entry_ids"][-1]
    unmapped_before = summary["unmapped_total"]
    assert (first["counterparty_id"], first["counterparty_type"]) == (None, "EXTERNAL")

    # кандидаты: ООО «Пекарня Счастливочная» есть в списке с признаком GSK. `recommended` на реальном имени «Юг-Финанс /
    # ПЕКАРНЯ СЧАСТЛИВОЧНАЯ» не ожидается: рекомендация — точное совпадение нормализованного имени/алиаса (сценарий
    # «Рекомендуемая Счастливочная» покрыт в test_api_counterparty_mapping на имени «Пекарня Счастливочная ООО»)
    candidates = await _get(client, f"{version_url}/counterparty-candidates", entry_id=entry_id)
    schastl_candidate = next(item for item in candidates if item["id"] == str(schastl.id))
    assert schastl_candidate["counterparty_type"] == "GSK"

    mapped = await client.post(
        f"{version_url}/osv/entries/{entry_id}/counterparty", json={"legal_entity_id": str(schastl.id)}
    )
    assert mapped.status_code == HTTP_OK, mapped.text
    body = mapped.json()
    assert body["row"]["counterparty_type"] == "GSK"
    assert (body["row"]["match_kind"], body["alias_created"]) == ("MANUAL", True)
    assert body["propagated"] == PROPAGATED

    after = await _get(client, f"{version_url}/osv/summary")
    rows_after = await _all_rows(client, version_url)
    # `propagated` считает записи `osv_entry` (в том числе в парах-пробелах 66/76, строк в своде не дающих), а
    # `unmapped_total` — строки свода: квартальные записи одного контрагента склеены (D-2). Поэтому `unmapped_total`
    # падает на число склеенных строк свода, получивших контрагента, а «1 + propagated» сверяется с записями в БД
    # (design #80).
    before_by_entries = {eid: row for row in rows for eid in row["entry_ids"]}
    changed = {
        tuple(row["entry_ids"]) for row in rows_after
        if row["counterparty_id"] is not None and before_by_entries[row["entry_ids"][0]]["counterparty_id"] is None
    }  # fmt: skip
    assert len(changed) == CHANGED_ROWS
    stored = (
        await seeded.execute(
            select(OsvEntry).where(OsvEntry.owner_legal_entity_id == rb.id, OsvEntry.counterparty_name.is_not(None))
            .execution_options(populate_existing=True)
        )
    ).scalars().all()  # fmt: skip
    named = [entry for entry in stored if _is_schastl(entry.counterparty_name)]
    assert len(named) == 1 + PROPAGATED == 1 + body["propagated"]
    assert {entry.counterparty_id for entry in named} == {schastl.id}
    assert {entry.counterparty_match_kind for entry in named} == {
        CounterpartyMatchKind.MANUAL,
        CounterpartyMatchKind.PROPAGATED,
    }
    assert {entry.account_root for entry in named} >= {"60", "62", "66", "76"}  # включая пары-пробелы
    assert after["unmapped_total"] == unmapped_before - len(changed)

    same_name = [r for r in rows_after if r["legal_entity_id"] == str(rb.id) and _is_schastl(r["counterparty_name"])]
    assert {r["account_root"] for r in same_name} >= {"60", "62"}  # 66 и 76 РБ — пробелы, строк в своде нет
    assert all(r["counterparty_id"] == str(schastl.id) and r["counterparty_type"] == "GSK" for r in same_name)

    # Счастливочная: контрагент «Русский бисквит» с ИНН — GSK по ИНН
    inn_rows = [
        r
        for r in rows_after
        if r["legal_entity_id"] == str(schastl.id)
        and r["counterparty_name"] == RB_NAME_IN_SCHASTL
        and r["counterparty_inn"] == RB_INN
    ]
    assert inn_rows
    assert all(r["counterparty_type"] == "GSK" and r["match_kind"] == "INN" for r in inn_rows)

    # сверка суммы периодов с golden: контрагент РБ по счёту 60, суммы Q1+Q2+Q3 считаются здесь, а не доменом
    golden = _golden_rows(fixtures_dir, "rb", "60")
    keys = {q: defaultdict(list) for q in golden}
    for quarter, quarter_rows in golden.items():
        for row in quarter_rows:
            keys[quarter][_key(row)].append(row)
    common = [
        k for k in keys[Q1] if k in keys[Q2] and k in keys[Q3] and any(_dec(r["turnover_debit"]) for r in keys[Q2][k])
    ]
    assert common
    checked = 0
    for key in common[:5]:
        account_code, name, inn = key
        found = [
            r for r in rows_after
            if r["legal_entity_id"] == str(rb.id) and r["account_code"] == account_code
            and (r["counterparty_name"] or "").strip() == name and (r["counterparty_inn"] or "").replace(" ", "") == inn
        ]  # fmt: skip
        assert len(found) == 1, key
        for field in SUMS:
            first_q, last_q = keys[Q1][key], keys[Q3][key]
            if field.startswith("opening"):
                expected = sum(_dec(r[field]) for r in first_q)
            elif field.startswith("closing"):
                expected = sum(_dec(r[field]) for r in last_q)
            else:
                expected = sum(_dec(r[field]) for q in (Q1, Q2, Q3) for r in keys[q][key])
            assert Decimal(found[0][field]) == expected, (key, field)
        assert found[0]["is_cumulative"] is False and len(found[0]["entry_ids"]) >= 3  # noqa: PLR2004
        checked += 1
    assert checked > 0

    # экспорт
    export = await client.get(f"{version_url}/osv/export")
    assert export.status_code == HTTP_OK, export.text
    book = load_workbook(io.BytesIO(export.content))
    main = list(book[book.sheetnames[0]].iter_rows(values_only=True))
    assert len(main) - 1 == after["rows_total"]

    elapsed = time.perf_counter() - started
    logger.info("сквозной сценарий РБ: {:.1f} с, строк свода {}", elapsed, len(rows_after))
    assert elapsed <= TIME_BUDGET_SECONDS