"""Чистые функции маппинга формы ББ/ОФР (`statementDocument` v1.2) в черновики `financial_statement_*` (spec osv-result-mapping «Документ формы ББ/ОФР», «Статус документа формы»; design D-2, D-3, D-6, D-14).""" import datetime from collections.abc import Iterable, Sequence from dataclasses import dataclass, field from decimal import Decimal from typing import Any from src.contract.result import StatementColumn, StatementDocument, StatementValue from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementType __all__ = [ "ColumnDraft", "LineDraft", "StatementDraft", "ValueDraft", "build_statement", "edition_for", "quarter_of", "statement_period", "statement_status", "value_clean", "values_ok", ] FULL_CONFIDENCE = Decimal("100") EDITION_BOUNDARY = datetime.date(2024, 12, 31) _QUARTER_ENDS = { (3, 31): PeriodQuarter.Q1, (6, 30): PeriodQuarter.Q2, (9, 30): PeriodQuarter.Q3, (12, 31): PeriodQuarter.Q4, } _STATEMENT_TYPES = {DocumentType.BALANCE: StatementType.balance, DocumentType.INCOME: StatementType.income} @dataclass class ColumnDraft: """Колонка формы; `period_kind` (v5.19): `absolute` при дате колонки, иначе `none` (в `1.x` относительных нет).""" column_index: int label: str period_date: datetime.date | None is_current: bool | None period_kind: str = "none" label_confidence: Decimal | None = None @dataclass class ValueDraft: line_code: str column_index: int amount: Decimal | None string_value: str | None confidence: Decimal | None is_recognized: bool | None row_index: int = 0 # индекс строки в `doc.lines` — ключ строки при вставке (код может повторяться) source_text: str | None = None needs_review: bool = False @dataclass class LineDraft: line_code: str line_code_confidence: Decimal | None line_name: str line_name_confidence: Decimal | None row_index: int = 0 # индекс строки в `doc.lines` результата — ключ строки формы; повтор кода — свой индекс line_status: str = "ok" line_code_original_value: str | None = None source_ref: str | None = None line_code_text: str | None = None needs_review: bool = False @dataclass class StatementDraft: document_type: DocumentType statement_type: StatementType locator: str period_date: datetime.date | None period_year: int | None period_quarter: PeriodQuarter | None header: dict[str, Any] columns: list[ColumnDraft] lines: list[LineDraft] values: list[ValueDraft] warnings: list[str] = field(default_factory=list) def quarter_of(day: datetime.date) -> PeriodQuarter | None: return _QUARTER_ENDS.get((day.month, day.day)) def statement_period( report_date: datetime.date | None, column_dates: Sequence[datetime.date | None] ) -> datetime.date | None: """D-2: дата отчёта, иначе самая поздняя дата колонок.""" if report_date is not None: return report_date dates = [day for day in column_dates if day is not None] return max(dates) if dates else None def edition_for(period_date: datetime.date | None) -> str | None: """D-14: редакция формы для шаблона фронта.""" if period_date is None: return None return "66n" if period_date <= EDITION_BOUNDARY else "fsbu4" def _decimal(value: float | None) -> Decimal | None: return Decimal(str(value)) if value is not None else None def build_statement(doc: StatementDocument) -> StatementDraft: """Черновик формы `BALANCE`/`INCOME` (D-2): шапка, колонки, строки, значение на каждую ячейку. Повтор кода строки, `column_index` колонки и пары (строка, колонка) — берётся первое, о повторе — предупреждение.""" document_type = DocumentType(doc.document_type) header = doc.header warnings: list[str] = [] source_columns: list[StatementColumn] = [] seen_indexes: set[int] = set() for column in doc.columns: if column.column_index in seen_indexes: warnings.append(f"{doc.locator}: колонка {column.column_index} повторяется — оставлена первая") continue seen_indexes.add(column.column_index) source_columns.append(column) column_dates = [column.period_date for column in source_columns] period_date = statement_period(header.report_date, column_dates) period_year = period_date.year if period_date is not None else None period_quarter = quarter_of(period_date) if period_date is not None else None if period_date is None: warnings.append(f"{doc.locator}: период формы не определён (нет даты отчёта и дат колонок)") elif period_quarter is None: warnings.append(f"{doc.locator}: дата {period_date.isoformat()} не конец квартала — квартал не определён") latest = max((day for day in column_dates if day is not None), default=None) columns = [ ColumnDraft( column.column_index, column.label, column.period_date, None if latest is None else column.period_date == latest and column.column_index == min(c.column_index for c in source_columns if c.period_date == latest), period_kind="absolute" if column.period_date is not None else "none", ) for column in source_columns ] known_columns = seen_indexes lines: list[LineDraft] = [] values: list[ValueDraft] = [] seen: set[str] = set() for row_index, line in enumerate(doc.lines): if line.line_code in seen: warnings.append(f"{doc.locator}: код {line.line_code} повторяется — оставлена первая строка") continue seen.add(line.line_code) lines.append( LineDraft( line.line_code, _decimal(line.line_code_confidence), line.line_name, _decimal(line.line_name_confidence), row_index=row_index, line_code_original_value=line.line_code, ) ) by_column: dict[int, StatementValue] = {} for cell_value in line.values: if cell_value.column_index not in known_columns: warnings.append( f"{doc.locator}: строка {line.line_code}: значение колонки {cell_value.column_index}" " вне колонок формы пропущено" ) continue if cell_value.column_index in by_column: warnings.append( f"{doc.locator}: строка {line.line_code}: значение колонки {cell_value.column_index}" " повторяется — оставлено первое" ) continue by_column[cell_value.column_index] = cell_value for column in source_columns: value: StatementValue | None = by_column.get(column.column_index) values.append( ValueDraft( line.line_code, column.column_index, value.amount if value is not None else None, value.string_value if value is not None else None, _decimal(value.confidence) if value is not None else None, value.is_recognized if value is not None else None, row_index=row_index, source_text=value.string_value if value is not None else None, needs_review=value_needs_review(value), ) ) inn_confidence = _decimal(header.inn_confidence) header_source = { "organization_name": {"text": header.organization_name, "confidence": header.organization_name_confidence}, "inn": {"text": header.inn, "confidence": header.inn_confidence}, "report_date": {"text": header.report_date_text, "confidence": None}, "units": {"text": header.units, "confidence": None}, "okud": {"text": header.okud, "confidence": None}, "okpo": {"text": header.okpo, "confidence": None}, "okopf": {"text": header.okopf, "confidence": None}, "okfs": {"text": header.okfs, "confidence": None}, "okei": {"text": None, "confidence": None}, "address": {"text": header.address, "confidence": None}, } meta: dict[str, Any] = { "inn": header.inn, "inn_original_value": header.inn, "inn_confidence": inn_confidence, "inn_verified": inn_confidence == FULL_CONFIDENCE if header.inn is not None else None, "inn_modified": False, "organization_name": header.organization_name, "report_date": header.report_date, "report_date_original_value": header.report_date_text, "report_date_verified": header.report_date is not None, "report_date_modified": False, "units": header.units, "units_original_value": header.units, "units_verified": header.units is not None, "units_modified": False, "okud": header.okud, "okpo": header.okpo, "okopf": header.okopf, "okfs": header.okfs, "address": header.address, "header_source": header_source, "organization_name_original_value": header.organization_name, "organization_name_modified": False, "needs_review_fields": [], "form_variant": None, } return StatementDraft( document_type=document_type, statement_type=_STATEMENT_TYPES[document_type], locator=doc.locator, period_date=period_date, period_year=period_year, period_quarter=period_quarter, header=meta, columns=columns, lines=lines, values=values, warnings=warnings, ) def value_clean(confidence: Decimal | None, is_recognized: bool | None) -> bool: """Ячейка без сомнений: `is_recognized` не false, уверенность 100 или не задана (D-6, v5.19 `needs_review`).""" return is_recognized is not False and (confidence is None or confidence == FULL_CONFIDENCE) def value_needs_review(value: StatementValue | None) -> bool: """`needs_review` ячейки при приёме `1.x`: значение есть и распознано с сомнением (пустая ячейка — нет).""" if value is None: return False return not value_clean(_decimal(value.confidence), value.is_recognized) def values_ok(values: Iterable[tuple[bool, bool | None, Decimal | None]]) -> bool: """D-6: каждое значение правлено аналитиком или распознано без сомнений (`is_recognized` не false, уверенность 100/NULL).""" return all(modified or value_clean(confidence, recognized) for modified, recognized, confidence in values) def statement_status( # noqa: PLR0913 - условия D-6 перечислены явно *, job_done: bool, checks_passed: bool, counterparty: DocumentCheckResult, period_known: bool, values_ok: bool, has_lines: bool, warnings: Sequence[str], ) -> DocumentStatus: """D-6: `RECOGNIZED` только при всех условиях; иначе `NEEDS_CHECK`.""" recognized = ( job_done and checks_passed and counterparty != DocumentCheckResult.MISMATCH and period_known and values_ok and has_lines and not warnings ) return DocumentStatus.RECOGNIZED if recognized else DocumentStatus.NEEDS_CHECK