"""Шапка документа и схема группировок ОСВ (`05-domain-rules.md` §1.1, §1.5; spec «Шапка документа и схема…»). Область шапки — строки выше заголовка таблицы. Организация, счёт, период, тип данных и валюта берутся по меткам из `rules/osv_headers.yaml` (`header_markers`), квартал и признак нарастающего итога — по `rules/periods.yaml`, тип группировки — по `rules/osv_groups.yaml`. Ни одна позиция ячейки заранее не известна; отсутствие поля — пустая строка (для полей контракта-строк) или `None` и предупреждение на русском. """ import calendar import datetime 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", "GroupSchema", "InnAttribute", "InnHint", "build_group_levels", "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 = "," # Реквизит субконто отделяется от имени группировки точкой: «Контрагенты.ИНН» _ATTRIBUTE_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 period_start: datetime.date | None period_end: datetime.date | 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 start: datetime.date | None = None end: datetime.date | None = None _MONTHS_IN_QUARTER = 3 def _month_end(year: int, month: int) -> datetime.date: return datetime.date(year, month, calendar.monthrange(year, month)[1]) def _bounds( year: int | None, quarter: Quarter | None, is_cumulative: bool | None ) -> tuple[datetime.date | None, datetime.date | None]: """Границы квартального/годового периода: конец — конец квартала; начало — 1 января у нарастающего, иначе первый месяц квартала; без года или квартала — неизвестны.""" if year is None or quarter is None: return None, None end_month = (QUARTERS.index(quarter) + 1) * _MONTHS_IN_QUARTER start_month = 1 if is_cumulative else end_month - _MONTHS_IN_QUARTER + 1 return datetime.date(year, start_month, 1), _month_end(year, end_month) 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) start = datetime.date(span.start_year, span.start_month, 1) end = _month_end(span.end_year, 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, start, end) if quarter is not None and same_year and span.months == _MONTHS_IN_QUARTER: return _Period(span.end_year, quarter, False, start, end) 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, start, end) 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 -> нарастающая) year_value = 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 start, end = _bounds(year_value, quarter, is_cumulative) named = rules.periods.bounds_span(text) if named is not None: # Месяцы названы в тексте — границы берём из них, а не из квартала словаря («Январь - Май 2025 г.» — не год) start, end = datetime.date(named.start_year, named.start_month, 1), _month_end(named.end_year, named.end_month) return _Period(year_value, quarter, is_cumulative, start, end) 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, period_start=period.start, period_end=period.end, 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 @dataclass(frozen=True) class InnAttribute: """Реквизит ИНН группировки контрагентов («Контрагенты.ИНН», OQ-162): 1С выводит его строкой под строкой контрагента, на уровень глубже. Это не уровень аналитики: в `group_schema` контракта его нет.""" name: str @dataclass(frozen=True) class InnHint: """Подсказка шапки для строк ИНН (OQ-162, #35): элемент `<группировка>.ИНН` как в файле и направление по порядку элементов — после группировки-владельца ИНН стоит под наименованием (`below`), перед ней — над (`above`). С подсказкой контрольная сумма ИНН строки не проверяется (`inn_rows`).""" name: str direction: Literal["above", "below"] @dataclass(frozen=True) class GroupSchema: """Группировки контракта (`groups`) и позиции строк таблицы под счётом (`positions`). `positions[i]` — группировка строки на глубине i (0 — ACCOUNT) или реквизит ИНН, который глубину занимает, а группировкой не является. Без реквизитов `positions` совпадает с `groups`. """ groups: list[OsvGroup] positions: list[OsvGroup | InnAttribute] variant: HeaderVariant def _inn_owner(item: str, groups: list[OsvGroup], rules: Rules) -> int | None: """Индекс группировки контрагентов, чей реквизит ИНН — `item` («Контрагенты.ИНН»); иначе `None`.""" owner, separator, attribute = item.rpartition(_ATTRIBUTE_SEPARATOR) if not separator or _find(rules.headers.service_columns["inn"], attribute) is None: return None owner = owner.strip().casefold() for index, group in enumerate(groups): if group.type == "COUNTERPARTY" and (group.name or "").casefold() == owner: return index return None def _attach_inn_attribute(item: str, groups: list[OsvGroup], slots: list[int | InnAttribute], rules: Rules) -> bool: """`item` — реквизит ИНН одной из предыдущих группировок контрагентов: владелец получает `has_inn`, позиция строк — `InnAttribute`; `True`. Иначе ничего не меняется, `False`.""" owner = _inn_owner(item, groups, rules) if owner is None: return False groups[owner] = _group("COUNTERPARTY", groups[owner].name, has_inn=True) slots.append(InnAttribute(item)) return True def _positions(groups: list[OsvGroup], slots: list[int | InnAttribute]) -> list[OsvGroup | InnAttribute]: """Позиции строк: индекс группировки в `groups` или реквизит ИНН (OQ-162).""" return [slot if isinstance(slot, InnAttribute) else groups[slot] for slot in slots] def _variant1_groups(sheet: Sheet, header: TableHeader, rules: Rules) -> GroupSchema: groups: list[OsvGroup] = [] slots: list[int | InnAttribute] = [] for index, row in enumerate(header.group_label_rows): label = sheet.text_at(row, header.text_col) if index > 0 and _attach_inn_attribute(label, groups, slots, rules): continue # Первая группировка — всегда 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))) slots.append(len(groups) - 1) if not groups: groups, slots = [_group("ACCOUNT")], [0] return GroupSchema(groups, _positions(groups, slots), 1) def _variant2_groups(sheet: Sheet, header: TableHeader, rules: Rules, items: list[str]) -> GroupSchema: groups = [_group("ACCOUNT", _first_group_name(sheet, header))] slots: list[int | InnAttribute] = [0] inn_pending = header.inn_col is not None for item in items: if _attach_inn_attribute(item, groups, slots, rules): continue 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)) slots.append(len(groups) - 1) return GroupSchema(groups, _positions(groups, slots), 2) def build_group_levels(sheet: Sheet, header: TableHeader, rules: Rules) -> GroupSchema: """Группировки (первая — ACCOUNT), позиции строк и вариант шапки (1 — строки блока, 2 — «субконто: …»).""" items = _subkonto_items(_head_cells(sheet, header), rules) if items is None: return _variant1_groups(sheet, header, rules) return _variant2_groups(sheet, header, rules, items) def build_group_schema(sheet: Sheet, header: TableHeader, rules: Rules) -> tuple[list[OsvGroup], HeaderVariant]: """Группировки контракта в порядке вложенности (первая — ACCOUNT) и вариант шапки.""" schema = build_group_levels(sheet, header, rules) return schema.groups, schema.variant