310 lines
14 KiB
Python
310 lines
14 KiB
Python
"""Конвейер листа ОСВ: лист → документ `OsvDocument`, предупреждения, признак «контрольные суммы сошлись».
|
||
|
||
Книгу (лимиты, чтение, дедлайн, вид листа, статус и карта ошибок) разбирает `book.BookParser`. Здесь: заголовок
|
||
таблицы → шапка и схема группировок → строки → каскад иерархии → классификация → контрольные суммы §1.4 → статус
|
||
документа и его предупреждения с кодами `OSV_*` (контракт 2.0). Имя листа только подставляется в `locator` и
|
||
сообщения, поиск по нему не ведётся.
|
||
"""
|
||
|
||
import datetime
|
||
from collections.abc import Iterable, Sequence
|
||
from dataclasses import dataclass, replace
|
||
|
||
from src.contract import OsvDocument, ResultWarning
|
||
from src.contract.common import DataType
|
||
from src.contract.result import DocumentStatus, 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.code_blocks import apply_code_blocks
|
||
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 BodyRow
|
||
from src.parser.indicators import levels_with_indicators
|
||
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,
|
||
group_row_warnings,
|
||
hierarchy_note,
|
||
notes_of,
|
||
)
|
||
from src.parser.rules import Rules
|
||
|
||
__all__ = ["OsvParser", "SheetResult", "data_rows"]
|
||
|
||
LOCATOR_PREFIX = "sheet:"
|
||
INN_COLUMN_KEY = "inn"
|
||
NO_DATA_ROWS_WARNING = "нет строк данных"
|
||
DEFAULT_DATA_TYPE: DataType = "BU"
|
||
DEFAULT_DATA_TYPE_WARNING = "вид данных в шапке не указан — строки без метки приняты как БУ"
|
||
|
||
|
||
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 _without_foreign_bare_codes(body: list[BodyRow], header_account: str) -> list[BodyRow]:
|
||
"""Голый числовой код («441», «199») в таблице ОСВ с известным счётом шапки — код контрагента, а не счёт (#71).
|
||
|
||
Счётом остаётся голый код, совпадающий со счётом шапки («60» при «по счету 60.01»); коды с точкой и названием
|
||
(«60.01», «60, Расчеты») под фильтр не попадают. Без счёта шапки список не меняется.
|
||
"""
|
||
if not header_account:
|
||
return body
|
||
root = header_account.split(".", maxsplit=1)[0]
|
||
return [
|
||
replace(item, is_account=False, account_rank=None)
|
||
if item.is_account and item.text.isdigit() and item.text != root
|
||
else item
|
||
for item in body
|
||
]
|
||
|
||
|
||
def _by_account_warnings(rows: Sequence[ClassifiedRow]) -> list[str]:
|
||
"""По одному предупреждению на счёт, чьи строки признаны контрагентами по плану счетов (R-6), в порядке листа."""
|
||
codes = dict.fromkeys(item.account_code_raw or "" for item in rows if item.counterparty_by_account)
|
||
return [
|
||
f"вид субконто в шапке не назван: строки аналитики счёта {code} приняты за контрагентов по плану счетов — "
|
||
"проверьте"
|
||
for code in codes
|
||
]
|
||
|
||
|
||
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]:
|
||
"""Без повторов по паре (код, текст), в порядке первого появления (T6.1)."""
|
||
seen: dict[tuple[str, str], OsvNote] = {}
|
||
for note in notes:
|
||
seen.setdefault((note.code, 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, reader_notes: Sequence[OsvNote] = ()) -> SheetResult:
|
||
"""Документ листа ОСВ, его заметки (первая — иерархия) и признак «контрольные суммы сошлись»;
|
||
нет заголовка/колонок/иерархии — `InvalidFormatError` с листом и шагом. `reader_notes` — заметки читателя книги
|
||
(xlsb без outline): дописываются в конец предупреждений документа.
|
||
"""
|
||
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)
|
||
body = _without_foreign_bare_codes(body, doc.account_code_raw)
|
||
blocks = apply_code_blocks(body, total_row, rules, doc.account_code_raw)
|
||
body = blocks.rows
|
||
step = "иерархия строк"
|
||
levels = levels_with_indicators(body, total_row, deadline, header_account=bool(doc.account_code_raw))
|
||
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 = "классификация строк"
|
||
data_type = doc.data_type or DEFAULT_DATA_TYPE
|
||
# Вид данных шапки не указан: строки без метки приняты как БУ — предупреждение, если умолчание сработало
|
||
defaulted = doc.data_type is None and any(
|
||
not item.indicator and item.label_data_type is None for item in glue.rows
|
||
)
|
||
classified = classify_rows(
|
||
glue.rows,
|
||
glue.levels,
|
||
schema.groups,
|
||
header,
|
||
sheet,
|
||
data_type,
|
||
total_row,
|
||
rules,
|
||
inns=glue.inns,
|
||
unpaired=glue.unpaired,
|
||
header_account=doc.account_code_raw or None,
|
||
names=blocks.names,
|
||
)
|
||
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, secondary_totals=rules.headers.total_row_secondary)
|
||
notes = [
|
||
hierarchy_note(levels.strategy),
|
||
*notes_of(OSV_HEADER, doc.warnings),
|
||
*([OsvNote(OSV_HEADER, DEFAULT_DATA_TYPE_WARNING)] if defaulted else []),
|
||
*notes_of(OSV_ROWS, group_row_warnings(body_warnings)),
|
||
*notes_of(OSV_ROWS, _by_account_warnings(classified)),
|
||
*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)
|
||
notes.extend(reader_notes)
|
||
unique_notes = tuple(_unique_notes(notes))
|
||
# Строка ИНН без пары, похожая на ИНН строка или расхождение с колонкой (#35) — needs_check
|
||
ok = _is_ok(checks) and glue.ok
|
||
document = self._document(
|
||
sheet=sheet,
|
||
header=header,
|
||
doc=doc,
|
||
schema=schema.groups,
|
||
rows=classified,
|
||
checks=checks,
|
||
status="done" if ok else "needs_check",
|
||
notes=unique_notes,
|
||
)
|
||
return SheetResult(document, unique_notes, ok)
|
||
|
||
# --- внутреннее ---
|
||
|
||
@staticmethod
|
||
def _document( # noqa: PLR0913 - части документа собираются в parse_sheet и передаются явно
|
||
*,
|
||
sheet: Sheet,
|
||
header: TableHeader,
|
||
doc: DocHeader,
|
||
schema: list[OsvGroup],
|
||
rows: list[ClassifiedRow],
|
||
checks: OsvChecks,
|
||
status: DocumentStatus,
|
||
notes: Sequence[OsvNote],
|
||
) -> OsvDocument:
|
||
return OsvDocument(
|
||
document_type="OSV",
|
||
locator=f"{LOCATOR_PREFIX}{sheet.name}",
|
||
status=status,
|
||
warnings=[ResultWarning(code=note.code, severity="info", message=note.message) for note in notes],
|
||
header=_header(doc),
|
||
group_schema=schema,
|
||
columns=_columns(header),
|
||
rows=[_row(item) for item in rows],
|
||
checks=checks,
|
||
)
|