"""Golden-прогон маппинга: все снимки парсера `../fixtures/golden/expected_osv_parse/**` в БД (spec osv-result-mapping
«Все golden-снимки парсера»; рулинг R-1; tasks 3.4).

Каждый снимок — файл ЮЛ в `PROCESSING` с текущим заданием `done` и `map_file` в одной откатываемой транзакции теста
(`seeded`: план счетов и справочники seed). Цикл маппинга и S3 здесь не нужны: чтение результата и выбор кандидата
проверяет `test_result_mapping.py`, а golden-прогон проверяет сам маппинг на реальных данных. Все снимки маппятся для
одного ЮЛ — файлы с одинаковым счётом и периодом по пути заменяют друг друга (автозамена), документы и строки
заменённых файлов сохраняются, поэтому инварианты проверяются для каждого файла.

Инварианты: строк `osv_entry` = строк результата без `TOTAL`; у каждой строки уровня > 0 есть родитель (строка того же
документа с `row_number = parent_row_number`); шесть сумм равны снимку; `account_id` = строка плана счетов seed для
нормализованного кода, а для кодов вне плана — NULL (известные коды вне плана 1С — `KNOWN_OUT_OF_PLAN`).

Число снимков проверяется снизу (`MIN_SNAPSHOTS`), а не на равенство (issue #33): новые снимки fixtures не ломают тест,
а проходят те же инварианты; пропажа снимков по-прежнему ловится.
"""

import json
import time
import uuid
from collections import Counter, defaultdict
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.contract.result import RecognitionResult
from src.domain.osv_mapping import normalize_account_code
from src.models.accounting import ChartOfAccount
from src.models.enums import DocumentType, FileStatus, RecognitionStatus
from src.models.files import Document, StoredFile
from src.models.osv import OsvEntry
from src.repositories.counterparties import CounterpartyIndex
from src.services.result_mapping import map_file
from tests.factories import make_legal_entity, make_stored_file
from tests.fixtures_helpers import golden_result

GOLDEN_DIR = Path("golden") / "expected_osv_parse"
# результатов ОСВ в fixtures на 07.10.2026: rb 42, schastl 30, osv-dkb 22 (пакет ДКБ), synthetic 19 — всего 113
MIN_SNAPSHOTS = 113
# коды строк-счетов, которых нет в плане счетов seed (выгрузка 1С, issue #23): account_id у них NULL, корень счёта
# известен. Новый код вне плана валит тест — сигнал о дыре в плане счетов (follow-up arm-core #28)
KNOWN_OUT_OF_PLAN = frozenset(
    {
        "60.АВ",  # synthetic/s09_subaccount_names
        "76.07.4",  # osv-dkb/npz
        "76.09.1",  # osv-dkb/nkz (76.09.1–76.09.4)
        "76.09.2",
        "76.09.3",
        "76.09.4",
        "76.77",  # osv-dkb/primanych, osv-dkb/tsimlyansky (76.77, 76.77.1, .2, .5)
        "76.77.1",
        "76.77.2",
        "76.77.5",
        "76.АФ",  # osv-dkb/primanych
        "76.Н.1",  # synthetic/s42_subaccounts_totals_reversed (tabular-parser#59)
        "60.111",  # synthetic/s46_counterparty_code_blocks (tabular-parser#71, AURORA-1714): 60.111, 60.112
        "60.112",
    }
)
TIME_BUDGET_SECONDS = 60.0  # tasks 3.4: маппинг всех снимков — не дольше минуты
SUM_FIELDS = ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit", "closing_debit", "closing_credit")
TOTAL = "TOTAL"

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


def _snapshots(fixtures_dir: Path) -> list[Path]:
    return sorted((fixtures_dir / GOLDEN_DIR).rglob("*.json"))


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


def _check_sheet(  # noqa: PLR0912 - все инварианты одного листа в одном месте
    name: str, sheet: dict[str, Any], entries: list[OsvEntry], chart: dict[str, uuid.UUID]
) -> tuple[list[str], set[str]]:
    """Нарушения инвариантов листа и нормализованные коды строк-счетов, которых нет в плане счетов."""
    problems: list[str] = []
    missing: set[str] = set()
    rows = [row for row in sheet["rows"] if row["group_type"] != TOTAL]
    if len(entries) != len(rows):
        problems.append(f"{name}: строк osv_entry {len(entries)}, в снимке без TOTAL {len(rows)}")
    by_row = {entry.row_number: entry for entry in entries}
    by_id = {entry.id: entry for entry in entries}
    for row in rows:
        entry = by_row.get(row["row_number"])
        if entry is None:
            problems.append(f"{name}: нет строки {row['row_number']}")
            continue
        where = f"{name} строка {row['row_number']}"
        if entry.level > 0:
            parent = by_id.get(entry.parent_entry_id) if entry.parent_entry_id is not None else None
            if parent is None or parent.row_number != row["parent_row_number"]:
                problems.append(f"{where}: уровень {entry.level} без родителя {row['parent_row_number']}")
        for field in SUM_FIELDS:
            if getattr(entry, field) != _decimal(row[field]):
                problems.append(f"{where}: {field} {getattr(entry, field)} != {row[field]}")
        if (entry.counterparty_id is None) != (entry.counterparty_match_kind is None):
            problems.append(f"{where}: counterparty_id {entry.counterparty_id}, kind {entry.counterparty_match_kind}")
        code = row["account_code_raw"]
        expected = chart.get(normalize_account_code(code)) if code else None
        if entry.account_id != expected:
            problems.append(f"{where}: account_id {entry.account_id}, ожидался {expected} для кода {code!r}")
        if row["group_type"] == "ACCOUNT" and code and expected is None:
            missing.add(normalize_account_code(code))
    return problems, missing


async def _form_problems(session: AsyncSession, file_id: uuid.UUID, raw: dict[str, Any], name: str) -> list[str]:
    """Формы смешанных снимков (s12 и др.): по документу на каждую форму ББ/ОФР снимка (повтор формы хранится, P3-4),
    `DISCLOSURE`/`OTHER` — по одному документу на тип."""
    types = Counter(
        document["document_type"] for document in raw["documents"] if document["document_type"] != DocumentType.OSV
    )
    stored = Counter(
        (
            await session.execute(
                select(Document.document_type).where(
                    Document.stored_file_id == file_id, Document.document_type != DocumentType.OSV
                )
            )
        ).scalars()
    )
    merged = (DocumentType.DISCLOSURE, DocumentType.OTHER)

    def expected(kind: str) -> int:
        return min(types[kind], 1) if kind in merged else types[kind]

    return [
        f"{name}: документов {kind} {stored.get(kind, 0)}, ожидалось {expected(kind)}"
        for kind in sorted({*types, *stored})
        if stored.get(kind, 0) != expected(kind)
    ]


async def test_all_golden_snapshots_mapped(seeded: AsyncSession, fixtures_dir: Path) -> None:  # noqa: PLR0914
    """«Все golden-снимки парсера» (не меньше 113) маппятся в БД, инварианты спеки на каждом документе, ≤ 60 с."""
    session = seeded
    snapshots = _snapshots(fixtures_dir)
    groups = Counter(path.relative_to(fixtures_dir / GOLDEN_DIR).parts[0] for path in snapshots)
    by_group = dict(sorted(groups.items()))
    assert len(snapshots) >= MIN_SNAPSHOTS, f"снимков {len(snapshots)} < {MIN_SNAPSHOTS}: {by_group}"
    owner = await make_legal_entity(session, name="АО «Русский бисквит»", inn="3528052718")
    chart = {
        code: account_id for code, account_id in await session.execute(select(ChartOfAccount.code, ChartOfAccount.id))
    }
    assert chart, "план счетов seed пуст — сверка account_id ничего не проверит"
    index = await CounterpartyIndex.load(session)

    mapped: list[tuple[Path, uuid.UUID, RecognitionResult]] = []
    mapping_seconds = 0.0
    for path in snapshots:
        file, job = await make_stored_file(
            session,
            entity=owner,
            status=FileStatus.PROCESSING,
            job_status=RecognitionStatus.done,
            original_name=path.name,
        )
        result = golden_result(path, job_id=job.id, file_id=file.id)
        started = time.perf_counter()
        await map_file(session, job=job, file=file, result=result, chart=chart, index=index, now=datetime.now(UTC))
        mapping_seconds += time.perf_counter() - started
        mapped.append((path, file.id, result))

    problems: list[str] = []
    missing_codes: set[str] = set()
    documents_total = entries_total = with_account = with_kind = 0
    for path, file_id, _ in mapped:
        name = str(path.relative_to(fixtures_dir / GOLDEN_DIR))
        raw = json.loads(path.read_text("utf-8"))
        documents = {
            document.locator: document
            for document in (
                await session.execute(
                    # формы ББ/ОФР смешанных снимков (s12) маппятся с Среза 3b — здесь сверяются только листы ОСВ
                    select(Document).where(
                        Document.stored_file_id == file_id, Document.document_type == DocumentType.OSV
                    )
                )
            ).scalars()
        }
        problems.extend(await _form_problems(session, file_id, raw, name))
        sheets = [sheet for sheet in raw["documents"] if sheet["document_type"] == "OSV"]
        if set(documents) != {sheet["locator"] for sheet in sheets}:
            problems.append(f"{name}: документы {sorted(documents)} не совпадают с листами снимка")
            continue
        entries_by_document: dict[uuid.UUID, list[OsvEntry]] = defaultdict(list)
        rows = await session.execute(
            select(OsvEntry).where(OsvEntry.document_id.in_([document.id for document in documents.values()]))
        )
        for entry in rows.scalars():
            entries_by_document[entry.document_id].append(entry)
        for sheet in sheets:
            entries = entries_by_document[documents[sheet["locator"]].id]
            sheet_problems, sheet_missing = _check_sheet(f"{name} {sheet['locator']}", sheet, entries, chart)
            problems.extend(sheet_problems)
            missing_codes |= sheet_missing
            documents_total += 1
            entries_total += len(entries)
            with_account += sum(1 for entry in entries if entry.account_id is not None)
            with_kind += sum(1 for entry in entries if entry.counterparty_match_kind is not None)
        file_status = (await session.execute(select(StoredFile.status).where(StoredFile.id == file_id))).scalar_one()
        if file_status not in {FileStatus.PROCESSED, FileStatus.SUPERSEDED}:
            problems.append(f"{name}: статус файла {file_status}")

    print(  # noqa: T201 - сводка прогона в отчёт задачи
        f"\nGOLDEN: снимков {len(mapped)}, документов {documents_total}, строк {entries_total}, "
        f"маппинг {mapping_seconds:.2f} с; кодов строк-счетов вне плана счетов seed: {len(missing_codes)} "
        f"{sorted(missing_codes)}"
    )
    assert problems == []
    # план счетов seed = выгрузка 1С (issue #23): вне плана — только известные коды (s09 и субсчета пакета ДКБ)
    unknown = sorted(missing_codes - KNOWN_OUT_OF_PLAN)
    assert not unknown, f"новые коды вне плана счетов: {unknown}"
    assert with_kind > 0  # есть сопоставленные строки: проверка «kind ⇔ counterparty_id» не вырождена
    assert with_account > 0  # строки со счётом плана счетов есть — сверка account_id не вырождена
    assert mapping_seconds <= TIME_BUDGET_SECONDS