diff --git a/api/src/core/PDFWorker/utils.py b/api/src/core/PDFWorker/utils.py index c39bbe9..10c621f 100644 --- a/api/src/core/PDFWorker/utils.py +++ b/api/src/core/PDFWorker/utils.py @@ -57,7 +57,7 @@ class ExtractResult(TypedDict): @dataclass(frozen=True) class ColumnLayout: - """Индексы значимых столбцов таблицы OCR, определённые по строке заголовка.""" + """Индексы значимых столбцов таблицы OCR: из строки заголовка или выведенные по строкам данных.""" label_idx: Optional[int] code_idx: int value_idxs: tuple[int, ...] @@ -65,15 +65,17 @@ class ColumnLayout: @dataclass class ParsedRow: - row: dict cells: dict[int, dict] layout: ColumnLayout kind: FslRowKind + ordinal: int # column_index в БД: 0 - наименование, 1 - код, значения начинаются с 2 _FIRST_VALUE_COLUMN = 2 _YEAR_RE = re.compile(r"\d{4}") +_BLANK_CHARS_RE = re.compile(r"[\s\-–—]") +_HEADER_CODE = 'Код' _TABLE_TYPE_MAP: dict[str, StatementType] = { @@ -198,33 +200,114 @@ def _keep_digits_only(raw: str) -> str: return res.strip() -def _parse_table_rows(table_rows: list[dict]) -> Optional[list[ParsedRow]]: - """Разбирает строки таблицы по раскладке из заголовка. +def _is_blank(value: str) -> bool: + # прочерк в отчётности означает ноль, считаем его пустым значением + return not _BLANK_CHARS_RE.sub('', value) - Раскладка берётся из строки с ячейкой «Код»; каждый повторный заголовок - перезаписывает индексы столбцов. В результат попадает только первый заголовок, - строки до первого заголовка игнорируются. Если заголовка нет, возвращает None. + +def _infer_layout(rows_cells: list[dict[int, dict]]) -> Optional[ColumnLayout]: + """Выводит раскладку столбцов по строкам данных, когда заголовка нет. + + Столбец оценивается по всем строкам сразу, поэтому пустые ячейки в отдельных + строках (например, в начале таблицы) не влияют на результат. """ - layout: Optional[ColumnLayout] = None - parsed_rows: list[ParsedRow] = [] + columns: dict[int, list[str]] = {} + for cells in rows_cells: + for idx, cell in cells.items(): + columns.setdefault(idx, []).append(str(cell['value'] or '').strip()) - for row in table_rows: - cells = _cells_by_index(row) + # код — столбец с наибольшим числом известных кодов строк + code_hits = {idx: sum(1 for v in values if v in codes_mapping) for idx, values in columns.items()} + code_idx = min(code_hits, key=lambda idx: (-code_hits[idx], idx), default=None) + if code_idx is None or code_hits[code_idx] == 0: + return None + + # наименование — столбец левее кода с наибольшим числом текстовых ячеек + text_hits = { + idx: sum(1 for v in values if any(c.isalpha() for c in v)) + for idx, values in columns.items() if idx < code_idx + } + label_idx = min(text_hits, key=lambda idx: (-text_hits[idx], -idx), default=None) + if label_idx is not None and text_hits[label_idx] == 0: + label_idx = None + + # значения — столбцы правее кода, где суммы преобладают над мусором (полоса скана), + # либо полностью пустые (период без данных) + value_idxs = [] + for idx in sorted(columns): + if idx <= code_idx: + continue + filled = [v for v in columns[idx] if not _is_blank(v)] + amounts = sum(1 for v in filled if _parse_amount(v)[0] is not None) + if amounts >= len(filled) - amounts: + value_idxs.append(idx) + + return ColumnLayout(label_idx=label_idx, code_idx=code_idx, value_idxs=tuple(value_idxs)) + + +def _parse_data_row(cells: dict[int, dict], layout: ColumnLayout, ordinal: int) -> Optional[ParsedRow]: + kind = _classify_row(cells, layout) + if kind is None or kind == FslRowKind.HEADER: + return None + return ParsedRow(cells, layout, kind, ordinal) + + +def _synthetic_header_cells(layout: ColumnLayout) -> dict[int, dict]: + """Ячейки восстановленного заголовка: код и наименование, тексты периодов пустые.""" + cells = {layout.code_idx: {'value': _HEADER_CODE, 'confidence': _FULL_CONFIDENCE}} + if layout.label_idx is not None: + cells[layout.label_idx] = {'value': codes_mapping[_HEADER_CODE], 'confidence': _FULL_CONFIDENCE} + return cells + + +def _parse_table_rows(table_rows: list[dict]) -> Optional[list[ParsedRow]]: + """Разбирает строки таблицы; первым элементом результата всегда идёт заголовок. + + - Раскладка берётся из строки с ячейкой «Код»; каждый повторный заголовок + перезаписывает индексы столбцов. + - Строки до первого заголовка разбираются по раскладке, выведенной из них самих + (индексы заголовка могут не совпадать с ними). Заголовок пишется в начало таблицы. + - Если заголовка нет, раскладка выводится по всем строкам, а заголовок создаётся + с кодом и наименованием и пустыми текстами периодов. + Если не удалось определить ни заголовок, ни столбец кода, возвращает None. + """ + rows_cells = [(row['row_index'], _cells_by_index(row)) for row in table_rows] + first_header = next( + (pos for pos, (_, cells) in enumerate(rows_cells) if _build_layout(cells) is not None), + len(rows_cells), + ) + + before_header = rows_cells[:first_header] + inferred = _infer_layout([cells for _, cells in before_header]) if before_header else None + leading_rows = [] + if inferred is not None: + for ordinal, cells in before_header: + parsed = _parse_data_row(cells, inferred, ordinal) + if parsed is not None: + leading_rows.append(parsed) + + if first_header == len(rows_cells): + if not leading_rows: + return None + header_ordinal = min(p.ordinal for p in leading_rows) - 1 + header = ParsedRow(_synthetic_header_cells(inferred), inferred, FslRowKind.HEADER, header_ordinal) + return [header, *leading_rows] + + header_row_index, header_cells = rows_cells[first_header] + layout = _build_layout(header_cells) + header_ordinal = min([header_row_index, *(p.ordinal - 1 for p in leading_rows)]) + parsed_rows = [ParsedRow(header_cells, layout, FslRowKind.HEADER, header_ordinal), *leading_rows] + + for ordinal, cells in rows_cells[first_header + 1:]: header_layout = _build_layout(cells) if header_layout is not None: - if layout is None: - parsed_rows.append(ParsedRow(row, cells, header_layout, FslRowKind.HEADER)) layout = header_layout continue - if layout is None: - continue + parsed = _parse_data_row(cells, layout, ordinal) + if parsed is not None: + parsed_rows.append(parsed) - kind = _classify_row(cells, layout) - if kind is None or kind == FslRowKind.HEADER: - continue - parsed_rows.append(ParsedRow(row, cells, layout, kind)) - - return parsed_rows if layout is not None else None + return parsed_rows def _resolve_table_type(parsed_rows: list[ParsedRow]) -> StatementType: @@ -296,7 +379,7 @@ def _process_file( parsed_rows = _parse_table_rows(table["rows"]) if parsed_rows is None: logger.warning( - "В таблице не найден заголовок со столбцом «Код», таблица пропущена. " + "В таблице нет заголовка и по строкам не удалось определить столбец кода, таблица пропущена. " "file_uid={file_uid}, table_number={table_number}, table_label={table_label}", file_uid=file.get("file_uid"), table_number=table_number, @@ -312,7 +395,7 @@ def _process_file( metas.append(asdict(meta)) for parsed in parsed_rows: - row, cells, layout, kind = parsed.row, parsed.cells, parsed.layout, parsed.kind + cells, layout, kind = parsed.cells, parsed.layout, parsed.kind code_cell = cells[layout.code_idx] is_line_code_valid = True line_code = _clear_value(code_cell["value"]) @@ -334,7 +417,7 @@ def _process_file( line_name=line_name, line_name_recognized=line_name_recognized, row_kind=kind, - ordinal=row["row_index"], + ordinal=parsed.ordinal, ) rows.append(asdict(db_row))