702ea1dff9
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
257 lines
11 KiB
Python
257 lines
11 KiB
Python
"""Конвейер листа ОСВ: лист → документ `OsvDocument`, предупреждения, признак «контрольные суммы сошлись».
|
||
|
||
Книгу (лимиты, чтение, дедлайн, вид листа, статус и карта ошибок) разбирает `book.BookParser`. Здесь: заголовок
|
||
таблицы → шапка и схема группировок → строки → каскад иерархии → классификация → контрольные суммы §1.4. Имя листа
|
||
только подставляется в `locator` и сообщения, поиск по нему не ведётся.
|
||
"""
|
||
|
||
import datetime
|
||
from collections.abc import Iterable, Sequence
|
||
from dataclasses import dataclass, replace
|
||
|
||
from src.contract import OsvDocument
|
||
from src.contract.result import OsvChecks, OsvGroup, OsvHeader, OsvRow
|
||
from src.parser.checks import run_checks
|
||
from src.parser.classify import SUM_FIELDS, ClassifiedRow, classify_rows, collect_body_rows
|
||
from src.parser.deadline import Deadline
|
||
from src.parser.detect import TableHeader, find_table_header
|
||
from src.parser.errors import InvalidFormatError
|
||
from src.parser.grid import Sheet, column_letter
|
||
from src.parser.header import ACCOUNT_NOT_FOUND_WARNING, DocHeader, build_group_levels, parse_doc_header
|
||
from src.parser.hierarchy import resolve_levels
|
||
from src.parser.inn_rows import InnGlue, glue_inn_rows
|
||
from src.parser.osv_warnings import (
|
||
OSV_HEADER,
|
||
OSV_INN_ROWS,
|
||
OSV_NO_DATA_ROWS,
|
||
OSV_ROWS,
|
||
OsvNote,
|
||
hierarchy_note,
|
||
notes_of,
|
||
)
|
||
from src.parser.rules import Rules
|
||
|
||
__all__ = ["OsvParser", "SheetResult", "data_rows", "sheet_line", "unique"]
|
||
|
||
LOCATOR_PREFIX = "sheet:"
|
||
INN_COLUMN_KEY = "inn"
|
||
NO_DATA_ROWS_WARNING = "нет строк данных"
|
||
|
||
|
||
def sheet_line(sheet: Sheet, text: str) -> str:
|
||
return f"sheet={sheet.name} {text}"
|
||
|
||
|
||
def unique(items: Iterable[str]) -> list[str]:
|
||
"""Без повторов, в порядке первого появления."""
|
||
return list(dict.fromkeys(items))
|
||
|
||
|
||
def _columns(header: TableHeader) -> dict[str, str]:
|
||
"""Диагностика `columns`: шесть колонок сумм и, если найдена, колонка ИНН (буквы — только здесь)."""
|
||
columns = header.columns.as_diagnostics()
|
||
if header.inn_col is not None:
|
||
columns[INN_COLUMN_KEY] = column_letter(header.inn_col)
|
||
return columns
|
||
|
||
|
||
def _from_year_start(start: datetime.date | None) -> bool | None:
|
||
return None if start is None else (start.month, start.day) == (1, 1)
|
||
|
||
|
||
def _header(doc: DocHeader) -> OsvHeader:
|
||
return OsvHeader(
|
||
organization_name=doc.organization_name,
|
||
organization_inn=None,
|
||
account_code_raw=doc.account_code_raw,
|
||
period_text=doc.period_text,
|
||
period_year=doc.period_year,
|
||
period_quarter=doc.period_quarter,
|
||
is_cumulative=doc.is_cumulative,
|
||
period_start=doc.period_start,
|
||
period_end=doc.period_end,
|
||
is_from_year_start=_from_year_start(doc.period_start),
|
||
data_type=doc.data_type,
|
||
currency=doc.currency,
|
||
header_variant=doc.header_variant,
|
||
)
|
||
|
||
|
||
def _row(item: ClassifiedRow) -> OsvRow:
|
||
sums = dict(zip(SUM_FIELDS, item.body.sums, strict=True))
|
||
return OsvRow(
|
||
row_number=item.body.row,
|
||
level=item.level,
|
||
parent_row_number=item.parent,
|
||
row_value=item.body.text,
|
||
group_type=item.group_type,
|
||
group_name=item.group_name,
|
||
account_code_raw=item.account_code_raw,
|
||
counterparty_name=item.counterparty_name,
|
||
counterparty_inn=item.counterparty_inn,
|
||
data_type=item.data_type,
|
||
opening_debit=sums["opening_debit"],
|
||
opening_credit=sums["opening_credit"],
|
||
turnover_debit=sums["turnover_debit"],
|
||
turnover_credit=sums["turnover_credit"],
|
||
closing_debit=sums["closing_debit"],
|
||
closing_credit=sums["closing_credit"],
|
||
extra=item.extra,
|
||
)
|
||
|
||
|
||
def _with_table_account(doc: DocHeader, rows: Sequence[ClassifiedRow]) -> DocHeader:
|
||
"""Код счёта шапки, которой нет («по счету …» в названии нет), — код первой строки-счёта таблицы.
|
||
|
||
«Счет: Код / Наименование»: код стоит в строке таблицы, а не в названии ведомости. Предупреждение «код счёта в
|
||
шапке не найден» заменяется пометкой, откуда код взят.
|
||
"""
|
||
if doc.account_code_raw:
|
||
return doc
|
||
accounts = (item.account_code_raw for item in rows if item.group_type == "ACCOUNT")
|
||
code = next((code for code in accounts if code), None)
|
||
if code is None:
|
||
return doc
|
||
warnings = [warning for warning in doc.warnings if warning != ACCOUNT_NOT_FOUND_WARNING]
|
||
warnings.append(f"код счёта «{code}» взят из первой строки-счёта таблицы: в шапке «по счету …» нет")
|
||
return replace(doc, account_code_raw=code, warnings=tuple(warnings))
|
||
|
||
|
||
def _is_ok(checks: OsvChecks) -> bool:
|
||
return (
|
||
bool(checks.total_row_matches_root)
|
||
and not checks.balance_identity_violations
|
||
and not (checks.parent_sum_violations)
|
||
)
|
||
|
||
|
||
def data_rows(document: OsvDocument) -> int:
|
||
return sum(1 for row in document.rows if row.group_type != "TOTAL")
|
||
|
||
|
||
def _unique_notes(notes: Iterable[OsvNote]) -> list[OsvNote]:
|
||
"""Без повторов по тексту, в порядке первого появления (как `unique` для строк вывода 1.x)."""
|
||
seen: dict[str, OsvNote] = {}
|
||
for note in notes:
|
||
seen.setdefault(note.message, note)
|
||
return list(seen.values())
|
||
|
||
|
||
def _invalid_format(sheet: Sheet, step: str, exc: InvalidFormatError) -> InvalidFormatError:
|
||
"""Сообщение `INVALID_FORMAT` с листом и шагом; префикс «лист «…»:» детектора не дублируется."""
|
||
reason = exc.message.removeprefix(f"лист «{sheet.name}»: ")
|
||
return InvalidFormatError(f"лист «{sheet.name}», шаг «{step}»: {reason}")
|
||
|
||
|
||
def _check_inn_owners(glue: InnGlue, classified: list[ClassifiedRow]) -> InnGlue:
|
||
"""Страховка (R-8, design №24): ИНН склеен со строкой, которая по схеме не контрагент (подразделение, договор,
|
||
OTHER) — ИНН пропадёт молча, поэтому предупреждение и `needs_check`."""
|
||
types = {item.body.row: item.group_type for item in classified}
|
||
extra = [
|
||
f"строка {glue.sources[owner]}: ИНН {inn} отнесён к строке {owner}, но она не контрагент — проверьте"
|
||
for owner, inn in glue.inns.items()
|
||
if owner in glue.sources and types.get(owner) != "COUNTERPARTY"
|
||
]
|
||
if not extra:
|
||
return glue
|
||
return replace(glue, warnings=[*glue.warnings, *extra], ok=False)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SheetResult:
|
||
"""Итог листа ОСВ: документ, заметки с кодами `OSV_*` (первая — иерархия) и признак «контрольные суммы сошлись»."""
|
||
|
||
document: OsvDocument
|
||
notes: tuple[OsvNote, ...]
|
||
ok: bool
|
||
|
||
|
||
class OsvParser:
|
||
"""Конвейер листа ОСВ по словарям `rules`; книгу разбирает `BookParser`."""
|
||
|
||
def __init__(self, rules: Rules) -> None:
|
||
self._rules = rules
|
||
|
||
def parse_sheet(self, sheet: Sheet, deadline: Deadline) -> SheetResult:
|
||
"""Документ листа ОСВ, его заметки (первая — иерархия) и признак «контрольные суммы сошлись»;
|
||
нет заголовка/колонок/иерархии — `InvalidFormatError` с листом и шагом.
|
||
"""
|
||
rules = self._rules
|
||
step = "заголовок таблицы"
|
||
try:
|
||
header = find_table_header(sheet, rules, deadline)
|
||
step = "шапка документа"
|
||
doc = parse_doc_header(sheet, header, rules)
|
||
schema = build_group_levels(sheet, header, rules)
|
||
deadline.check(f"лист «{sheet.name}»: строки таблицы")
|
||
step = "строки таблицы"
|
||
body, total_row, body_warnings = collect_body_rows(sheet, header, rules, deadline)
|
||
step = "иерархия строк"
|
||
levels = resolve_levels(body, total_row, deadline)
|
||
step = "строки ИНН"
|
||
glue = glue_inn_rows(
|
||
body,
|
||
levels,
|
||
total_row,
|
||
sheet,
|
||
inn_col=header.inn_col,
|
||
schema=schema.groups,
|
||
hint=schema.inn_hint,
|
||
rules=rules,
|
||
)
|
||
step = "классификация строк"
|
||
classified = classify_rows(
|
||
glue.rows,
|
||
glue.levels,
|
||
schema.groups,
|
||
header,
|
||
sheet,
|
||
doc.data_type,
|
||
total_row,
|
||
rules,
|
||
inns=glue.inns,
|
||
unpaired=glue.unpaired,
|
||
)
|
||
except InvalidFormatError as exc:
|
||
raise _invalid_format(sheet, step, exc) from exc
|
||
glue = _check_inn_owners(glue, classified)
|
||
doc = _with_table_account(doc, classified)
|
||
deadline.check(f"лист «{sheet.name}»: контрольные суммы")
|
||
checks, check_notes = run_checks(classified, rules.checks)
|
||
document = self._document(
|
||
sheet=sheet, header=header, doc=doc, schema=schema.groups, rows=classified, checks=checks
|
||
)
|
||
notes = [
|
||
hierarchy_note(levels.strategy),
|
||
*notes_of(OSV_HEADER, doc.warnings),
|
||
*notes_of(OSV_ROWS, body_warnings),
|
||
*notes_of(OSV_INN_ROWS, glue.warnings),
|
||
]
|
||
if not any(item.group_type != "TOTAL" for item in classified):
|
||
notes.append(OsvNote(OSV_NO_DATA_ROWS, NO_DATA_ROWS_WARNING))
|
||
notes.extend(check_notes)
|
||
# Строка ИНН без пары, похожая на ИНН строка или расхождение с колонкой (#35) — needs_check
|
||
return SheetResult(document, tuple(_unique_notes(notes)), _is_ok(checks) and glue.ok)
|
||
|
||
# --- внутреннее ---
|
||
|
||
@staticmethod
|
||
def _document( # noqa: PLR0913 - части документа собираются в parse_sheet и передаются явно
|
||
*,
|
||
sheet: Sheet,
|
||
header: TableHeader,
|
||
doc: DocHeader,
|
||
schema: list[OsvGroup],
|
||
rows: list[ClassifiedRow],
|
||
checks: OsvChecks,
|
||
) -> OsvDocument:
|
||
return OsvDocument(
|
||
document_type="OSV",
|
||
locator=f"{LOCATOR_PREFIX}{sheet.name}",
|
||
header=_header(doc),
|
||
group_schema=schema,
|
||
columns=_columns(header),
|
||
rows=[_row(item) for item in rows],
|
||
checks=checks,
|
||
)
|