arm-tabular-parser/api/src/parser/header.py

"""Шапка документа и схема группировок ОСВ (`05-domain-rules.md` §1.1, §1.5; spec «Шапка документа и схема…»).

Область шапки — строки выше заголовка таблицы. Организация, счёт, период, тип данных и валюта берутся по меткам
из `rules/osv_headers.yaml` (`header_markers`), квартал и признак нарастающего итога — по `rules/periods.yaml`,
тип группировки — по `rules/osv_groups.yaml`. Ни одна позиция ячейки заранее не известна; отсутствие поля —
пустая строка (для полей контракта-строк) или `None` и предупреждение на русском.
"""

import re
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any, Literal, get_args

from src.contract.common import DataType
from src.contract.result import OsvGroup
from src.parser.detect import TableHeader
from src.parser.grid import Sheet
from src.parser.rules import GroupType, MonthRange, Pattern, Rules, first_match, source_offset

__all__ = ["ACCOUNT_NOT_FOUND_WARNING", "DocHeader", "build_group_schema", "parse_doc_header"]

Quarter = Literal["Q1", "Q2", "Q3", "Q4"]
QUARTERS: tuple[Quarter, ...] = get_args(Quarter)
HeaderVariant = Literal[1, 2]
ACCOUNT_NOT_FOUND_WARNING = "код счёта (после «по счету») в шапке не найден"

# Код счёта начинается с цифры и идёт до пробела или запятой: «60», «58.03», «76.07.1»
_ACCOUNT_CODE = re.compile(r"\d[^\s,;]*")
# Метка отделяется от значения двоеточием («Валюта БУ, НУ: RUB»); перечисление субконто — запятыми
_LABEL_SEPARATOR = ":"
_LIST_SEPARATOR = ","
# Знаки, которыми код счёта может заканчиваться в тексте, но которые кодом не являются
_CODE_TRAILING = ".,;:"


@dataclass(frozen=True)
class DocHeader:
    """Поля шапки ОСВ (`osvDocument` контракта); `warnings` — что не найдено или не распознано."""

    organization_name: str
    account_code_raw: str
    period_text: str
    period_year: int | None
    period_quarter: Quarter | None
    is_cumulative: bool | None
    data_type: DataType | None
    currency: str | None
    header_variant: HeaderVariant
    warnings: tuple[str, ...]


def _group(group_type: GroupType, name: str | None = None, *, has_inn: bool = False) -> OsvGroup:
    """Группировка контракта; необязательные поля без значения опускаются (явный `null` схема не допускает)."""
    fields: dict[str, Any] = {"type": group_type}
    if name:
        fields["name"] = name
    if has_inn:
        fields["has_inn"] = True
    return OsvGroup(**fields)


@dataclass(frozen=True)
class _HeadCell:
    row: int
    col: int
    text: str


# --- Текст ---


def _tail(text: str, match: re.Match[str]) -> str:
    """Исходный текст после совпадения (позиции `match` — в нормализованном тексте)."""
    return text[source_offset(text, match.end()) :].strip()


def _find(patterns: Iterable[Pattern], text: str) -> re.Match[str] | None:
    for pattern in patterns:
        match = pattern.search(text)
        if match is not None:
            return match
    return None


def _value_after_colon(tail: str) -> str:
    """Значение метки: текст после двоеточия, а без двоеточия — весь хвост («Валюта БУ, НУ: RUB» -> «RUB»)."""
    _, separator, value = tail.partition(_LABEL_SEPARATOR)
    return (value if separator else tail).strip()


# --- Область шапки ---


def _head_cells(sheet: Sheet, header: TableHeader) -> list[_HeadCell]:
    """Текстовые ячейки выше заголовка таблицы по строкам сверху вниз, слева направо (числа и даты — не текст)."""
    return [
        _HeadCell(row, col, cell.text)
        for row in range(1, header.first_row)
        for col, cell in sheet.row_cells(row)
        if cell.kind == "text"
    ]


def _is_structural_marker(cell: _HeadCell, rules: Rules) -> bool:
    """Структурные метки шапки: лист, счёт, тип данных, параметры, субконто.

    Голые «за» и «валюта» меткой не считаются: это обычные слова, которые могут стоять в названии организации
    («ООО «За Родину»», «Валюта Трейд»). Период и валюта опознаются отдельно (`_find_period`, `_currency_cells`).
    """
    markers = rules.headers.header_markers
    patterns = [
        *rules.headers.sheet_markers,
        *markers["account"],
        *markers["data_type"],
        *markers["org_params"],
        *markers["subkonto"],
    ]
    return _find(patterns, cell.text) is not None


def _is_title(cell: _HeadCell, rules: Rules) -> bool:
    """Строка названия ведомости («Оборотно-сальдовая ведомость по счету … за …»): признак листа или метка счёта."""
    patterns = [*rules.headers.sheet_markers, *rules.headers.header_markers["account"]]
    return _find(patterns, cell.text) is not None


def _currency_cells(cells: list[_HeadCell], rules: Rules) -> list[_HeadCell]:
    """Строки-метки валюты: «Валюта БУ, НУ: RUB» (с двоеточием; «Валюта Трейд» — не метка)."""
    patterns = rules.headers.header_markers["currency"]
    return [cell for cell in cells if _LABEL_SEPARATOR in cell.text and _find(patterns, cell.text) is not None]


@dataclass(frozen=True)
class _PeriodHit:
    cell: _HeadCell
    text: str


def _cut_month_range(tail: str, rules: Rules) -> str | None:
    """Период — диапазон месяцев с годами («Октябрь 2024 г. - Сентябрь 2025 г.»): до года конца и «г.» после него.

    Первое «г.» диапазона (после года начала) период не заканчивает: «Октябрь 2024 г.» — лишь его начало.
    """
    if rules.periods.month_range(tail) is None:
        return None
    found = rules.periods.range.search(tail)
    if found is None:
        return None
    stop = source_offset(tail, found.end())
    rest = tail[stop:]
    mark = rules.periods.end.search(rest)
    if mark is not None and mark.start() == 0:
        stop += source_offset(rest, mark.end())
    return tail[:stop].strip()


def _cut_period(tail: str, rules: Rules) -> str:
    """Конец периода — «г.»/«год» включительно: самое короткое начало хвоста, которое словарь периодов узнаёт.

    Диапазон дат «01.01.2025 г. - 31.03.2025 г.» первым «г.» не обрезается: «01.01.2025 г.» — не период. Если ни одно
    начало не узнано, период — весь хвост до конца строки.
    """
    ranged = _cut_month_range(tail, rules)
    if ranged is not None:
        return ranged
    start = 0
    while start < len(tail):
        end = rules.periods.end.search(tail[start:])
        if end is None:
            break
        start += source_offset(tail[start:], end.end())
        candidate = tail[:start].strip()
        if rules.periods.match(candidate) is not None:
            return candidate
    return tail


def _period_candidates(cells: list[_HeadCell], rules: Rules) -> list[_PeriodHit]:
    """Ячейки с меткой периода («за …», «Период: …») и текст после метки; пустой текст — не кандидат.

    Какой из кандидатов — период, решает `_find_period`: «за» в названии организации не должно его перехватывать.
    """
    hits: list[_PeriodHit] = []
    for cell in cells:
        match = _find(rules.headers.header_markers["period"], cell.text)
        if match is None:
            continue
        tail = _tail(cell.text, match)
        if tail:
            hits.append(_PeriodHit(cell, _cut_period(tail, rules)))
    return hits


def _is_recognised(text: str, rules: Rules) -> bool:
    return rules.periods.match(text) is not None or rules.periods.year.search(text) is not None


def _standalone_period(cells: list[_HeadCell], rules: Rules) -> _PeriodHit | None:
    """Период отдельной ячейкой без метки: ячейка целиком — «9 месяцев 2025 г.» (словарь `periods.standalone`)."""
    cell = next((cell for cell in cells if _find(rules.periods.standalone, cell.text) is not None), None)
    return None if cell is None else _PeriodHit(cell, cell.text)


def _find_period(cells: list[_HeadCell], rules: Rules) -> _PeriodHit | None:
    """Строка периода: сначала с опознанным периодом по метке, затем период отдельной ячейкой, затем строка названия
    ведомости или метка с двоеточием.
    """
    hits = _period_candidates(cells, rules)
    recognised = next((hit for hit in hits if _is_recognised(hit.text, rules)), None)
    if recognised is not None:
        return recognised
    standalone = _standalone_period(cells, rules)
    if standalone is not None:
        return standalone
    return next((hit for hit in hits if _is_title(hit.cell, rules) or _LABEL_SEPARATOR in hit.cell.text), None)


def _standalone_data_type(cells: list[_HeadCell], rules: Rules) -> tuple[_HeadCell, DataType] | None:
    """Тип данных отдельной ячейкой без метки: ячейка целиком — «Бухгалтерский» (словарь `standalone_header`)."""
    for cell in cells:
        found = first_match(rules.data_types.standalone_header, cell.text)
        if found is not None:
            return cell, found
    return None


def _title_tail_organization(cells: list[_HeadCell], rules: Rules) -> str | None:
    """Организация в названии ведомости: «Оборотно-сальдовая ведомость НОВОРОСМЕТАЛЛ ООО» -> «НОВОРОСМЕТАЛЛ ООО».

    Хвост без метки счёта и периода («по счету 60 за …» — не организация); пустой хвост — `None`.
    """
    markers = rules.headers.header_markers
    for cell in cells:
        match = _find(rules.headers.sheet_markers, cell.text)
        if match is None:
            continue
        tail = _tail(cell.text, match).strip(" ,;:")
        if tail and _find([*markers["account"], *markers["period"]], tail) is None:
            return tail
    return None


def _organization(
    cells: list[_HeadCell], rules: Rules, period: _PeriodHit | None, data_type: _HeadCell | None
) -> str | None:
    """Первая текстовая ячейка, не являющаяся меткой шапки, строкой периода, меткой валюты или типом данных; иначе
    хвост названия ведомости.
    """
    taken = [
        cell for cell in (None if period is None else period.cell, data_type, *_currency_cells(cells, rules)) if cell
    ]
    found = next(
        (cell for cell in cells if cell not in taken and not _is_structural_marker(cell, rules)),
        None,
    )
    return _title_tail_organization(cells, rules) if found is None else found.text


def _account_code(cells: list[_HeadCell], rules: Rules) -> str | None:
    for cell in cells:
        match = _find(rules.headers.header_markers["account"], cell.text)
        if match is None:
            continue
        code = _ACCOUNT_CODE.match(_tail(cell.text, match))
        if code is not None:
            return code.group(0).rstrip(_CODE_TRAILING)
    return None


def _value_in_row(sheet: Sheet, cell: _HeadCell, patterns: list[Pattern]) -> str:
    """Значение метки: после двоеточия в той же ячейке, а если там пусто — в следующей ячейке строки."""
    match = _find(patterns, cell.text)
    value = _value_after_colon(_tail(cell.text, match)) if match is not None else ""
    if value:
        return value
    neighbours = [other.text for col, other in sheet.row_cells(cell.row) if col > cell.col]
    return neighbours[0] if neighbours else ""


def _labelled_value(sheet: Sheet, cells: list[_HeadCell], patterns: list[Pattern]) -> str | None:
    """Значение первой строки с меткой (`None` — метки нет)."""
    cell = next((cell for cell in cells if _find(patterns, cell.text) is not None), None)
    return None if cell is None else _value_in_row(sheet, cell, patterns)


def _subkonto_items(cells: list[_HeadCell], rules: Rules) -> list[str] | None:
    """Вариант 2: «Детализация по субсчетам, субконто: Контрагенты, Договоры» -> `["Контрагенты", "Договоры"]`."""
    for cell in cells:
        if _find(rules.headers.header_markers["subkonto"], cell.text) is None or _LABEL_SEPARATOR not in cell.text:
            continue
        _, _, listing = cell.text.partition(_LABEL_SEPARATOR)
        items = [item.strip() for item in listing.split(_LIST_SEPARATOR) if item.strip()]
        if items:
            return items
    return None


# --- Шапка документа ---


@dataclass(frozen=True)
class _Period:
    year: int | None
    quarter: Quarter | None
    is_cumulative: bool | None


_MONTHS_IN_QUARTER = 3


def _quarter_of_end(month: int) -> Quarter | None:
    """Квартал, которым заканчивается период, если конец диапазона — конец квартала (3, 6, 9, 12-й месяц)."""
    if month % _MONTHS_IN_QUARTER:
        return None
    return QUARTERS[month // _MONTHS_IN_QUARTER - 1]


def _month_range_period(span: MonthRange, text: str, warnings: list[str]) -> _Period:
    """Период-диапазон месяцев: с января по конец квартала — нарастающий, ровно квартал — квартальный; остальное
    (скользящие 12 месяцев, полгода с апреля) — квартал и признак не определены: год — год конца диапазона.

    Скользящий период нельзя выдать ни за квартальный, ни за нарастающий: ядро привяжет его к версии как не тот
    документ. `null` + предупреждение оставляют решение аналитику (OQ-131).
    """
    quarter = _quarter_of_end(span.end_month)
    same_year = span.start_year == span.end_year
    if quarter is not None and same_year and span.start_month == 1:
        return _Period(span.end_year, quarter, True)
    if quarter is not None and same_year and span.months == _MONTHS_IN_QUARTER:
        return _Period(span.end_year, quarter, False)
    if span.start_month != 1 or not same_year:
        warnings.append(
            f"период «{text}»: скользящий период {span.months} мес. "
            f"(с {span.start_month:02d}.{span.start_year} по {span.end_month:02d}.{span.end_year}), не с начала года: "
            "квартал и признак нарастающего итога не определены"
        )
    else:
        warnings.append(
            f"период «{text}»: конец диапазона не совпадает с концом квартала: "
            "квартал и признак нарастающего итога не определены"
        )
    return _Period(span.end_year, None, None)


def _parse_period(text: str, rules: Rules, warnings: list[str]) -> _Period:
    span = rules.periods.month_range(text)
    if span is not None:
        return _month_range_period(span, text, warnings)
    rule = rules.periods.match(text)
    year = rules.periods.year.search(text)
    if rule is None:
        warnings.append(f"период «{text}»: квартал и признак нарастающего итога не определены по словарю периодов")
    if year is None:
        warnings.append(f"период «{text}»: год не найден")
    # ASSUMPTION(OQ-02): is_cumulative=false для Q2–Q4 определяется по тексту «N квартал»; ядро может переопределить
    # по данным (opening Q3 == opening Q1 -> нарастающая)
    return _Period(
        year=None if year is None else int(year.group(0)),
        quarter=None if rule is None else rule.quarter,
        is_cumulative=None if rule is None else rule.is_cumulative,
    )


def _parse_data_type(sheet: Sheet, cells: list[_HeadCell], rules: Rules, warnings: list[str]) -> DataType | None:
    value = _labelled_value(sheet, cells, rules.headers.header_markers["data_type"])
    if value is None:
        standalone = _standalone_data_type(cells, rules)
        if standalone is not None:
            return standalone[1]
        warnings.append("тип выводимых данных (БУ/НУ/сумма) в шапке не найден")
        return None
    data_type = first_match(rules.data_types.header, value)
    if data_type is None:
        warnings.append(f"тип выводимых данных «{value}» не распознан")
    return data_type


def _parse_currency(sheet: Sheet, cells: list[_HeadCell], rules: Rules, warnings: list[str]) -> str | None:
    patterns = rules.headers.header_markers["currency"]
    candidates = _currency_cells(cells, rules)
    if not candidates:
        warnings.append("валюта в шапке не найдена")
        return None
    values = [_value_in_row(sheet, cell, patterns) for cell in candidates]
    # Метка валюты, значение которой опознано, важнее первой попавшейся
    code = next((found for found in (rules.data_types.currency_code(value) for value in values) if found), None)
    if code is None:
        warnings.append(f"валюта «{values[0]}» не распознана")
    return code


def parse_doc_header(sheet: Sheet, header: TableHeader, rules: Rules) -> DocHeader:
    """Шапка документа по области выше заголовка таблицы; что не найдено — `""`/`None` и предупреждение."""
    cells = _head_cells(sheet, header)
    warnings: list[str] = []

    period_hit = _find_period(cells, rules)
    standalone_type = _standalone_data_type(cells, rules)
    organization = _organization(cells, rules, period_hit, None if standalone_type is None else standalone_type[0])
    if organization is None:
        warnings.append("наименование организации в шапке не найдено")
    account = _account_code(cells, rules)
    if account is None:
        warnings.append(ACCOUNT_NOT_FOUND_WARNING)
    period_text = None if period_hit is None else period_hit.text
    if period_text is None:
        warnings.append("период (после «за» или «Период:») в шапке не найден")
        period = _Period(None, None, None)
    else:
        period = _parse_period(period_text, rules, warnings)

    data_type = _parse_data_type(sheet, cells, rules, warnings)
    currency = _parse_currency(sheet, cells, rules, warnings)
    variant: HeaderVariant = 1 if _subkonto_items(cells, rules) is None else 2
    return DocHeader(
        organization_name="" if organization is None else organization,
        account_code_raw=account or "",
        period_text=period_text or "",
        period_year=period.year,
        period_quarter=period.quarter,
        is_cumulative=period.is_cumulative,
        data_type=data_type,
        currency=currency,
        header_variant=variant,
        warnings=tuple(warnings),
    )


# --- Схема группировок ---


def _row_has_inn(sheet: Sheet, row: int, rules: Rules) -> bool:
    """В строке блока заголовка есть ячейка «ИНН» (отдельная или в тексте метки: «Контрагенты, ИНН»)."""
    patterns = rules.headers.service_columns["inn"]
    return any(_find(patterns, cell.text) is not None for _, cell in sheet.row_cells(row))


def _first_group_name(sheet: Sheet, header: TableHeader) -> str | None:
    if not header.group_label_rows:
        return None
    return sheet.text_at(header.group_label_rows[0], header.text_col) or None


def _variant1_groups(sheet: Sheet, header: TableHeader, rules: Rules) -> list[OsvGroup]:
    groups: list[OsvGroup] = []
    for index, row in enumerate(header.group_label_rows):
        label = sheet.text_at(row, header.text_col)
        # Первая группировка — всегда ACCOUNT, как бы она ни называлась («Счет», «Подразделение», «Субконто»)
        group_type: GroupType = "ACCOUNT" if index == 0 else rules.groups.classify(label)
        groups.append(_group(group_type, label, has_inn=_row_has_inn(sheet, row, rules)))
    return groups or [_group("ACCOUNT")]


def _variant2_groups(sheet: Sheet, header: TableHeader, rules: Rules, items: list[str]) -> list[OsvGroup]:
    groups = [_group("ACCOUNT", _first_group_name(sheet, header))]
    inn_pending = header.inn_col is not None
    for item in items:
        group_type = rules.groups.classify(item)
        has_inn = inn_pending and group_type == "COUNTERPARTY"
        inn_pending = inn_pending and not has_inn
        groups.append(_group(group_type, item, has_inn=has_inn))
    return groups


def build_group_schema(sheet: Sheet, header: TableHeader, rules: Rules) -> tuple[list[OsvGroup], HeaderVariant]:
    """Группировки в порядке вложенности (первая — ACCOUNT) и вариант шапки (1 — строки блока, 2 — «субконто: …»)."""
    items = _subkonto_items(_head_cells(sheet, header), rules)
    if items is None:
        return _variant1_groups(sheet, header, rules), 1
    return _variant2_groups(sheet, header, rules, items), 2