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)
-        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
+    # код — столбец с наибольшим числом известных кодов строк
+    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:
-            continue
-        parsed_rows.append(ParsedRow(row, cells, layout, kind))
+        return None
+    return ParsedRow(cells, layout, kind, ordinal)
 
-    return parsed_rows if layout is not None else None
+
+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:
+            layout = header_layout
+            continue
+        parsed = _parse_data_row(cells, layout, ordinal)
+        if parsed is not None:
+            parsed_rows.append(parsed)
+
+    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))