7b14276b1a
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
197 lines
8.3 KiB
Python
197 lines
8.3 KiB
Python
"""Сумма и значение ячейки (шаг 4): `parse_amount` — единственное место, где разбор зависит от источника."""
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from decimal import ROUND_HALF_UP, Decimal
|
||
|
||
from arm_statements.codes import code_parts
|
||
from arm_statements.loader import Rules
|
||
from arm_statements.model import Cell, SignForm, Source, StatementValue
|
||
from arm_statements.text import replace_lookalikes
|
||
|
||
MAX_INTEGER_DIGITS = 16
|
||
_CENT = Decimal("0.01")
|
||
_DIGITS = frozenset("0123456789")
|
||
_CORE_START = _DIGITS | frozenset("(-−+")
|
||
_CORE_END = _DIGITS | frozenset(")}]")
|
||
_DASHES = frozenset("-–—−")
|
||
_MINUSES = frozenset("-−")
|
||
_BODY = re.compile(r"[\d\s,.]*")
|
||
_DIGIT_SPACE_RUN = re.compile(r"\d+(?:\s+\d+)+")
|
||
_SPACE_GROUPS = re.compile(r"\d{1,3}(?:\s\d{3})+")
|
||
_COMMA_THOUSANDS = re.compile(r"[1-9]\d{0,2}(?:,\d{3})+")
|
||
_DOT_THOUSANDS = re.compile(r"[1-9]\d*(?:\.\d{3})+")
|
||
_SEPARATORS = ",."
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class AmountParse:
|
||
amount: Decimal | None
|
||
string_value: str | None
|
||
is_recognized: bool
|
||
sign_form: SignForm | None
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ValueBuild:
|
||
value: StatementValue
|
||
out_of_range: bool
|
||
|
||
|
||
def _quantize(amount: Decimal) -> Decimal:
|
||
return amount.quantize(_CENT, rounding=ROUND_HALF_UP)
|
||
|
||
|
||
def _too_long(amount: Decimal) -> bool:
|
||
return len(str(abs(int(amount)))) > MAX_INTEGER_DIGITS
|
||
|
||
|
||
def _is_blank(text: str) -> bool:
|
||
return all(char.isspace() or char in _DASHES for char in text)
|
||
|
||
|
||
def _strip_edge_bars(text: str) -> tuple[str, bool]:
|
||
"""Крайние `|` (след границы таблицы) — до замены двойников; второй элемент — было ли удаление."""
|
||
stripped = text.strip()
|
||
while stripped.startswith("|"):
|
||
stripped = stripped[1:].strip()
|
||
while stripped.endswith("|"):
|
||
stripped = stripped[:-1].strip()
|
||
return stripped, stripped != text.strip()
|
||
|
||
|
||
def _core_bounds(text: str) -> tuple[int, int] | None:
|
||
starts = [index for index, char in enumerate(text) if char in _CORE_START]
|
||
ends = [index for index, char in enumerate(text) if char in _CORE_END]
|
||
if not starts or not ends or starts[0] > ends[-1]:
|
||
return None
|
||
return starts[0], ends[-1]
|
||
|
||
|
||
def _guess(digits: str, text: str, sign_form: SignForm, *, negative: bool) -> AmountParse:
|
||
"""«Лучшая догадка»: цифры подряд, не длиннее 16, иначе значение не число."""
|
||
if not digits or len(digits) > MAX_INTEGER_DIGITS:
|
||
return AmountParse(None, text, False, sign_form)
|
||
amount = Decimal(digits)
|
||
return AmountParse(-amount if negative else amount, None, False, sign_form)
|
||
|
||
|
||
def _split_core(core: str) -> tuple[str, SignForm, bool, bool] | None:
|
||
"""Ядро → (тело без знака и скобок, форма знака, отрицательное ли, исправлена ли скобка) или None."""
|
||
fixed = False
|
||
sign_form = SignForm.UNSIGNED
|
||
negative = False
|
||
if core[0] == "(":
|
||
if len(core) < 2 or core[-1] not in ")}]": # noqa: PLR2004 — скобки в ядре минимум пара
|
||
return None
|
||
fixed = core[-1] != ")"
|
||
core = core[1:-1].strip()
|
||
sign_form, negative = SignForm.PARENTHESES, True
|
||
if core and core[0] in "-−+":
|
||
core = core[1:].strip()
|
||
elif core[0] in "-−+":
|
||
if core[0] in _MINUSES:
|
||
sign_form, negative = SignForm.MINUS, True
|
||
core = core[1:].strip()
|
||
if not core or not _BODY.fullmatch(core) or not core[0].isdigit():
|
||
return None
|
||
return core, sign_form, negative, fixed
|
||
|
||
|
||
def _number(body: str, source: Source, text: str, sign_form: SignForm, *, negative: bool) -> AmountParse:
|
||
"""Стадия 6: разделители. Возвращает догадку с `is_recognized = false` или точный разбор."""
|
||
runs = _DIGIT_SPACE_RUN.findall(body)
|
||
spaces_ok = all(_SPACE_GROUPS.fullmatch(run) for run in runs)
|
||
compact = "".join(body.split())
|
||
digits = compact.replace(",", "").replace(".", "")
|
||
if not spaces_ok:
|
||
return _guess(digits, text, sign_form, negative=negative)
|
||
if source is Source.OCR and _COMMA_THOUSANDS.fullmatch(compact):
|
||
compact = compact.replace(",", "")
|
||
elif source is Source.OCR and _DOT_THOUSANDS.fullmatch(compact):
|
||
compact = compact.replace(".", "")
|
||
separators = [char for char in compact if char in _SEPARATORS]
|
||
if not separators:
|
||
integer, fraction = compact, ""
|
||
elif len(separators) > 1: # смешанные или повторяющиеся разделители
|
||
return _guess(digits, text, sign_form, negative=negative)
|
||
else:
|
||
integer, fraction = compact.replace(",", ".").split(".")
|
||
if not integer:
|
||
return AmountParse(None, text, False, sign_form)
|
||
if source is Source.OCR and len(fraction) >= 3: # noqa: PLR2004 — три знака: не копейки
|
||
return AmountParse(None, text, False, sign_form)
|
||
if len(integer) > MAX_INTEGER_DIGITS:
|
||
return AmountParse(None, text, False, sign_form)
|
||
amount = _quantize(Decimal(f"{integer}.{fraction}" if fraction else integer))
|
||
return AmountParse(-amount if negative else amount, None, True, sign_form)
|
||
|
||
|
||
def parse_amount(text: str | None, source: Source, rules: Rules) -> AmountParse:
|
||
"""Стадии 1–6 шага «значение ячейки»; от источника зависят только десятичный разделитель и тысячи."""
|
||
if text is None or _is_blank(text):
|
||
return AmountParse(None, None, True, None)
|
||
work, fixed = _strip_edge_bars(text)
|
||
replaced = replace_lookalikes(work, rules.lookalikes)
|
||
fixed = fixed or replaced != work
|
||
if not any(char in _DIGITS for char in replaced): # стадия 5
|
||
return AmountParse(None, text, False, None)
|
||
bounds = _core_bounds(replaced)
|
||
if bounds is None:
|
||
return AmountParse(None, text, False, None)
|
||
start, end = bounds
|
||
core = replaced[start : end + 1]
|
||
trash = bool((replaced[:start] + replaced[end + 1 :]).strip())
|
||
split = _split_core(core)
|
||
if split is None:
|
||
return AmountParse(None, text, False, None)
|
||
body, sign_form, negative, bracket_fixed = split
|
||
parsed = _number(body, source, text, sign_form, negative=negative)
|
||
clean = parsed.is_recognized and not (fixed or trash or bracket_fixed)
|
||
return AmountParse(parsed.amount, parsed.string_value, clean, parsed.sign_form)
|
||
|
||
|
||
def sign_form_of(text: str | None, rules: Rules) -> SignForm | None:
|
||
return parse_amount(text, Source.OCR, rules).sign_form
|
||
|
||
|
||
def _plain(cell: Cell, column_index: int, parsed: AmountParse, *, recognized: bool | None = None) -> StatementValue:
|
||
return StatementValue(
|
||
column_index=column_index,
|
||
amount=parsed.amount,
|
||
string_value=parsed.string_value,
|
||
text=cell.text,
|
||
confidence=cell.confidence,
|
||
is_recognized=parsed.is_recognized if recognized is None else recognized,
|
||
)
|
||
|
||
|
||
def build_value(
|
||
cell: Cell | None,
|
||
column_index: int,
|
||
source: Source,
|
||
rules: Rules,
|
||
*,
|
||
split_part: bool = False,
|
||
) -> ValueBuild:
|
||
if cell is None:
|
||
return ValueBuild(StatementValue(column_index, None, None, None, 100.0, True), out_of_range=False)
|
||
text = cell.text
|
||
string_text = text if text.strip() else None
|
||
value = cell.value
|
||
if isinstance(value, Decimal):
|
||
amount = _quantize(value)
|
||
if _too_long(amount):
|
||
return ValueBuild(_plain(cell, column_index, AmountParse(None, text, False, None)), out_of_range=True)
|
||
return ValueBuild(
|
||
_plain(cell, column_index, AmountParse(amount, None, True, None), recognized=not split_part),
|
||
out_of_range=False,
|
||
)
|
||
if isinstance(value, date | bool):
|
||
return ValueBuild(_plain(cell, column_index, AmountParse(None, string_text, False, None)), out_of_range=False)
|
||
if len(code_parts(text)) > 1 and not split_part:
|
||
return ValueBuild(_plain(cell, column_index, AmountParse(None, text, False, None)), out_of_range=False)
|
||
parsed = parse_amount(text, source, rules)
|
||
return ValueBuild(_plain(cell, column_index, parsed, recognized=False if split_part else None), out_of_range=False)
|