feat(statements): чистые функции маппинга формы — период, черновик, статус; counterparty_check общий (D-2, D-6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev 2026-10-07 11:17:27 +04:00
parent 4126a44e0e
commit 3a72702c3d
5 changed files with 412 additions and 21 deletions

View File

@ -14,7 +14,7 @@ from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from decimal import Decimal
from src.contract.result import OsvChecks, OsvDocument, OsvHeader, OsvRow
from src.contract.result import OsvChecks, OsvDocument, OsvRow
from src.domain.aliases import normalize_alias
from src.models.enums import (
CounterpartyMatchKind,
@ -206,25 +206,25 @@ def document_fields(doc: OsvDocument) -> DocumentDraft:
def counterparty_check(
header: OsvHeader, owner_inn: str, owner_names_normalized: frozenset[str]
inn: str | None, name: str | None, owner_inn: str, owner_names_normalized: frozenset[str]
) -> DocumentCheckResult:
"""D-8: организация в шапке ОСВ — владелец файла?
"""D-8, Срез 2c/3b: организация в шапке — владелец файла?
ИНН шапки есть → `MATCH`, если его цифры равны ИНН владельца, иначе `MISMATCH`. ИНН нет →
`normalize_alias(organization_name)` среди нормализованных имени и алиасов владельца → `MATCH`, иначе `MISMATCH`.
`normalize_alias(name)` среди нормализованных имени и алиасов владельца → `MATCH`, иначе `MISMATCH`.
Нет ни ИНН, ни имени (в том числе имя только из ОПФ и пунктуации) → `UNKNOWN`.
"""
header_inn = _NON_DIGITS_RE.sub("", header.organization_inn or "")
header_inn = _NON_DIGITS_RE.sub("", inn or "")
if header_inn:
return (
DocumentCheckResult.MATCH
if header_inn == _NON_DIGITS_RE.sub("", owner_inn)
else DocumentCheckResult.MISMATCH
)
name = normalize_alias(header.organization_name)
if not name:
normalized = normalize_alias(name or "")
if not normalized:
return DocumentCheckResult.UNKNOWN
return DocumentCheckResult.MATCH if name in owner_names_normalized else DocumentCheckResult.MISMATCH
return DocumentCheckResult.MATCH if normalized in owner_names_normalized else DocumentCheckResult.MISMATCH
def _checks_clean(checks: OsvChecks | None) -> bool:

View File

@ -0,0 +1,225 @@
"""Чистые функции маппинга формы ББ/ОФР (`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 StatementDocument, StatementValue
from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementType
__all__ = [
"LineDraft",
"StatementDraft",
"ValueDraft",
"build_statement",
"edition_for",
"quarter_of",
"statement_period",
"statement_status",
"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 ValueDraft:
line_code: str
column_index: int
amount: Decimal | None
string_value: str | None
confidence: Decimal | None
is_recognized: bool | None
@dataclass
class LineDraft:
line_code: str
line_code_confidence: Decimal | None
line_name: str
line_name_confidence: Decimal | None
@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[tuple[int, str, datetime.date | None, bool | None]]
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): шапка, колонки, строки (повтор кода — первая), значение
на каждую ячейку."""
document_type = DocumentType(doc.document_type)
header = doc.header
warnings: list[str] = []
column_dates = [column.period_date for column in doc.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 = [
(
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 doc.columns if c.period_date == latest),
)
for column in doc.columns
]
known_columns = {column.column_index for column in doc.columns}
lines: list[LineDraft] = []
values: list[ValueDraft] = []
seen: set[str] = set()
for line in 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),
)
)
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
by_column.setdefault(cell_value.column_index, cell_value)
for column in doc.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,
)
)
inn_confidence = _decimal(header.inn_confidence)
meta = {
"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,
}
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 values_ok(values: Iterable[tuple[bool, bool | None, Decimal | None]]) -> bool:
"""D-6: каждое значение правлено аналитиком или распознано без сомнений (`is_recognized` не false,
уверенность 100/NULL)."""
return all(
modified or (recognized is not False and (confidence is None or confidence == FULL_CONFIDENCE))
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

View File

@ -386,7 +386,9 @@ async def map_file( # noqa: PLR0913 - все входы маппинга фай
sheet_warnings.append(
f"sheet={sheet.locator}: период {sheet.header.period_text}: нарастающая по данным Q1"
)
check = counterparty_check(sheet.header, owner.inn, owner_names)
check = counterparty_check(
sheet.header.organization_inn, sheet.header.organization_name, owner.inn, owner_names
)
document = Document(
id=uuid.uuid4(),
stored_file_id=file.id,

View File

@ -14,6 +14,7 @@ import pytest
from scripts.seed_lib.loaders import load_chart_of_accounts
from src.contract.result import OsvChecks, OsvDocument, OsvHeader
from src.domain.aliases import normalize_alias
from src.domain.osv_mapping import (
DocumentDraft,
EntryDraft,
@ -244,21 +245,21 @@ def test_document_fields_on_golden_osv_66(fixtures_dir: Path) -> None:
@pytest.mark.parametrize(
("header", "expected"),
("inn", "name", "expected"),
[
({"organization_name": "АО «Русский бисквит»", "organization_inn": OWNER_INN}, DocumentCheckResult.MATCH),
({"organization_name": "", "organization_inn": " 3528 052718 "}, DocumentCheckResult.MATCH),
({"organization_name": "АО «Русский бисквит»", "organization_inn": "3528000000"}, DocumentCheckResult.MISMATCH),
({"organization_name": 'АО "Русский бисквит"'}, DocumentCheckResult.MATCH), # по нормализованному имени
({"organization_name": "ООО РБ"}, DocumentCheckResult.MATCH), # по алиасу
({"organization_name": "Пекарня Счастливочная"}, DocumentCheckResult.MISMATCH),
({"organization_name": "", "organization_inn": None}, DocumentCheckResult.UNKNOWN), # пустая шапка
({"organization_name": " ", "organization_inn": ""}, DocumentCheckResult.UNKNOWN),
({"organization_name": "ООО"}, DocumentCheckResult.UNKNOWN), # только ОПФ: нормализуется в пустое имя
(OWNER_INN, "АО «Русский бисквит»", DocumentCheckResult.MATCH),
(" 3528 052718 ", "", DocumentCheckResult.MATCH),
("3528000000", "АО «Русский бисквит»", DocumentCheckResult.MISMATCH),
(None, 'АО "Русский бисквит"', DocumentCheckResult.MATCH), # по нормализованному имени
(None, "ООО РБ", DocumentCheckResult.MATCH), # по алиасу
(None, "Пекарня Счастливочная", DocumentCheckResult.MISMATCH),
(None, "", DocumentCheckResult.UNKNOWN), # пустая шапка
("", " ", DocumentCheckResult.UNKNOWN),
(None, "ООО", DocumentCheckResult.UNKNOWN), # только ОПФ: нормализуется в пустое имя
],
)
def test_counterparty_check(header: dict[str, Any], expected: DocumentCheckResult) -> None:
assert counterparty_check(_header(**header), OWNER_INN, OWNER_NAMES) is expected
def test_counterparty_check(inn: str | None, name: str | None, expected: DocumentCheckResult) -> None:
assert counterparty_check(inn, name, OWNER_INN, OWNER_NAMES) is expected
# --- статус документа (§9.3, OQ-05) ---------------------------------------------------------------------------------
@ -319,3 +320,11 @@ def test_account_sort_key_survives_non_ascii_digits(code: str) -> None:
key = account_sort_key(code)
assert key == account_sort_key(code) # детерминирован
assert sorted(["A", code, "60"], key=account_sort_key)[0] == "60" # ASCII-числа — раньше нечисловых сегментов
def test_counterparty_check_plain_values() -> None:
names = frozenset({normalize_alias("ООО Ромашка")})
assert counterparty_check("7700000000", None, "7700000000", names) is DocumentCheckResult.MATCH
assert counterparty_check(None, "ООО «Ромашка»", "7700000001", names) is DocumentCheckResult.MATCH
assert counterparty_check("3121200351", "мусор", "7700000000", names) is DocumentCheckResult.MISMATCH
assert counterparty_check(None, None, "7700000000", names) is DocumentCheckResult.UNKNOWN

View File

@ -0,0 +1,155 @@
"""Чистые функции маппинга формы (spec osv-result-mapping «Документ формы ББ/ОФР»; design D-2, D-3, D-6, D-14)."""
import datetime
from decimal import Decimal
import pytest
from src.contract.result import StatementDocument
from src.domain.statement_mapping import (
build_statement,
edition_for,
quarter_of,
statement_period,
statement_status,
values_ok,
)
from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementType
d = datetime.date
@pytest.mark.parametrize(
("day", "quarter"),
[
(d(2025, 3, 31), PeriodQuarter.Q1),
(d(2025, 6, 30), PeriodQuarter.Q2),
(d(2025, 9, 30), PeriodQuarter.Q3),
(d(2024, 12, 31), PeriodQuarter.Q4),
(d(2025, 10, 31), None),
(d(2025, 6, 29), None),
],
)
def test_quarter_of(day: datetime.date, quarter: PeriodQuarter | None) -> None:
assert quarter_of(day) is quarter
def test_period_and_edition() -> None:
assert statement_period(d(2025, 9, 30), [d(2024, 12, 31)]) == d(2025, 9, 30)
assert statement_period(None, [d(2024, 12, 31), d(2025, 6, 30), None]) == d(2025, 6, 30)
assert statement_period(None, [None]) is None
assert edition_for(d(2024, 12, 31)) == "66n"
assert edition_for(d(2025, 3, 31)) == "fsbu4"
assert edition_for(None) is None
def _doc(**overrides: object) -> StatementDocument:
data: dict[str, object] = {
"document_type": "BALANCE",
"locator": "sheet:ББ",
"header": {
"form_name": "Бухгалтерский баланс",
"inn": "7700000000",
"inn_confidence": 100,
"report_date": "2025-09-30",
"report_date_text": "на 30 сентября 2025 г.",
"units": "тыс. руб.",
"organization_name": "ООО Пример",
"okud": "0710001",
},
"columns": [
{"column_index": 0, "label": "На 30.09.2025", "period_date": "2025-09-30"},
{"column_index": 1, "label": "На 31.12.2024", "period_date": "2024-12-31"},
],
"lines": [
{
"line_code": "1150",
"line_code_confidence": 100,
"line_name": "ОС",
"line_name_confidence": 100,
"values": [{"column_index": 0, "amount": "10", "confidence": 100, "is_recognized": True}],
},
{"line_code": "1150", "line_name": "дубль", "values": []},
{
"line_code": "1600",
"line_name": "БАЛАНС",
"values": [{"column_index": 5, "amount": "1", "confidence": 100, "is_recognized": True}],
},
],
}
data.update(overrides)
return StatementDocument.model_validate(data)
def test_build_statement_cells_and_header() -> None:
draft = build_statement(_doc())
expected_cells = 4 # 2 строки × 2 колонки
assert draft.document_type is DocumentType.BALANCE and draft.statement_type is StatementType.balance
assert (draft.period_year, draft.period_quarter, draft.period_date) == (2025, PeriodQuarter.Q3, d(2025, 9, 30))
assert draft.header["inn"] == "7700000000" and draft.header["inn_original_value"] == "7700000000"
assert draft.header["inn_verified"] is True
assert draft.header["report_date_original_value"] == "на 30 сентября 2025 г."
assert draft.header["units"] == draft.header["units_original_value"] == "тыс. руб."
assert draft.columns == [(0, "На 30.09.2025", d(2025, 9, 30), True), (1, "На 31.12.2024", d(2024, 12, 31), False)]
assert [line.line_code for line in draft.lines] == ["1150", "1600"] # дубль 1150 отброшен
cells = {(v.line_code, v.column_index): v for v in draft.values}
assert len(cells) == expected_cells
assert cells[("1150", 0)].amount == Decimal("10") and cells[("1150", 1)].amount is None
assert cells[("1150", 1)].is_recognized is None and cells[("1600", 0)].confidence is None
assert any("1150" in w and "повтор" in w for w in draft.warnings)
assert any("колонк" in w and "5" in w for w in draft.warnings)
def test_value_outside_columns_skipped() -> None:
"""Review Focus 3: значение с column_index вне колонок — пропущено с предупреждением, остальное смаплено."""
draft = build_statement(_doc())
assert all(v.column_index in (0, 1) for v in draft.values)
def test_period_from_columns_and_non_quarter() -> None:
no_date = _doc(header={"form_name": "ББ", "report_date": None})
draft = build_statement(no_date)
assert (draft.period_year, draft.period_quarter) == (2025, PeriodQuarter.Q3)
assert draft.header["report_date"] is None
october = _doc(header={"form_name": "ББ", "report_date": "2025-10-31"})
draft = build_statement(october)
assert (draft.period_year, draft.period_quarter) == (2025, None)
assert any("конец квартала" in w for w in draft.warnings)
def test_no_dates_at_all() -> None:
draft = build_statement(_doc(header={"form_name": "ББ"}, columns=[{"column_index": 0, "label": ""}]))
assert (draft.period_year, draft.period_quarter, draft.period_date) == (None, None, None)
assert draft.columns == [(0, "", None, None)]
def test_status_rules() -> None:
ok = dict(
job_done=True,
checks_passed=True,
counterparty=DocumentCheckResult.MATCH,
period_known=True,
values_ok=True,
has_lines=True,
warnings=[],
)
assert statement_status(**ok) is DocumentStatus.RECOGNIZED # type: ignore[arg-type]
for key, bad in [
("job_done", False),
("checks_passed", False),
("counterparty", DocumentCheckResult.MISMATCH),
("period_known", False),
("values_ok", False),
("has_lines", False),
("warnings", ["x"]),
]:
assert statement_status(**{**ok, key: bad}) is DocumentStatus.NEEDS_CHECK # type: ignore[arg-type]
assert statement_status(**{**ok, "counterparty": DocumentCheckResult.UNKNOWN}) is DocumentStatus.RECOGNIZED # type: ignore[arg-type]
def test_values_ok() -> None:
hundred = Decimal("100")
assert values_ok([(False, True, hundred), (False, None, None)])
assert not values_ok([(False, False, hundred)])
assert not values_ok([(False, True, Decimal("99"))])
assert values_ok([(True, False, Decimal("10"))]) # правленое аналитиком значение не тянет статус вниз