medtech-hackathon/etl/extractors/grid.py

"""Извлечение позиций прайса из «сетки» строк.

Алгоритм: найти строку заголовков → определить роли колонок (название, код,
единица, ценовые тарифы) → пройти строки, отделяя заголовки секций от позиций.
Один и тот же код работает для xlsx, xls, docx и таблиц, восстановленных из PDF.

Распознавание ролей колонок учитывает реальные ловушки прайсов:
- колонка «№» (последовательные 1, 2, 3…) — это индекс, а не цена;
- настоящая цена медуслуги — сотни и тысячи тенге, а не мелкое число;
- название услуги — самая «кириллическая» колонка, код — цифро-точечная.
"""

from __future__ import annotations

import statistics
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from contracts.models import RawRow  # noqa: E402
from etl.extractors.common import (  # noqa: E402
    _NAME_KEYS,
    _UNIT_KEYS,
    Grid,
    classify_price_tier,
    clean,
    find_header_row,
    is_real_name,
    looks_like_code,
    parse_price,
)

_TIER_ORDER = ["resident", "nonresident", "cis", "far", "insurance", "partner"]
_CODE_HEADER_KEYS = ("код", "тарификатор")
_INDEX_HEADER_KEYS = ("№", "п/п", "n п")
_MIN_REAL_PRICE = 100  # медуслуга дешевле сотни тенге не бывает — отсекает индексы и мелочь


def _col_values(sample: list[list[str]], c: int) -> list[str]:
    return [r[c] for r in sample if c < len(r) and r[c]]


def _col_prices(sample: list[list[str]], c: int) -> list[float]:
    return [p for r in sample if c < len(r) and (p := parse_price(r[c]))]


def _cyrillic_score(sample: list[list[str]], c: int) -> float:
    """Средняя доля кириллических букв в колонке — признак колонки названий."""
    total = 0
    for r in sample:
        if c < len(r):
            total += sum(1 for ch in r[c] if "а" <= ch.lower() <= "я" or ch.lower() == "ё")
    return total / max(1, len(sample))


def _is_index_column(values: list[float]) -> bool:
    """Колонка-индекс: целые, монотонные, шаг около 1, старт у единицы."""
    if len(values) < 5:
        return False
    integers = [v for v in values if abs(v - round(v)) < 1e-9]
    if len(integers) < 0.9 * len(values):
        return False
    ordered = sorted(round(v) for v in values)
    steps = [ordered[i + 1] - ordered[i] for i in range(len(ordered) - 1)]
    return ordered[0] <= 3 and statistics.mean(steps) < 1.5


def _assign_columns(grid: Grid, header_idx: int):
    """Определить роли колонок по заголовку и выборке строк под ним."""
    header = [clean(c) for c in grid[header_idx]]
    ncol = max((len(r) for r in grid), default=0)
    header += [""] * (ncol - len(header))
    sample = [[clean(c) for c in r] for r in grid[header_idx + 1 : header_idx + 60]]

    # --- ценовые колонки ---
    price_cols: dict[int, str | None] = {}
    for c in range(ncol):
        head = header[c].lower()
        if any(k in head for k in _INDEX_HEADER_KEYS) or any(k in head for k in _CODE_HEADER_KEYS):
            continue  # «№» и «Код» — не цены
        prices = _col_prices(sample, c)
        nonempty = _col_values(sample, c)
        ratio = len(prices) / len(nonempty) if nonempty else 0
        if not prices or _is_index_column(prices):
            continue
        tier = classify_price_tier(header[c])
        if tier:
            price_cols[c] = tier
        elif ratio >= 0.6 and statistics.median(prices) >= _MIN_REAL_PRICE:
            price_cols[c] = None  # числовая колонка без явного тарифа — назначим по позиции

    used = {t for t in price_cols.values() if t}
    fallback = [t for t in _TIER_ORDER if t not in used]
    for c in sorted(price_cols):
        if price_cols[c] is None:
            price_cols[c] = fallback.pop(0) if fallback else f"extra_{c}"

    # --- колонка названия: явный заголовок, иначе самая кириллическая ---
    # Колонки кода/индекса исключаем заранее: иначе ключ «услуг» из _NAME_KEYS ловит
    # заголовок «Код услуги» и колонка кодов ошибочно становится колонкой названий.
    def _is_code_or_index(c: int) -> bool:
        return any(k in header[c].lower() for k in _CODE_HEADER_KEYS + _INDEX_HEADER_KEYS)

    name_col = next(
        (
            c
            for c in range(ncol)
            if c not in price_cols
            and not _is_code_or_index(c)
            and any(k in header[c].lower() for k in _NAME_KEYS)
        ),
        None,
    )
    if name_col is None:
        candidates = [c for c in range(ncol) if c not in price_cols and not _is_code_or_index(c)]
        name_col = max(candidates, key=lambda c: _cyrillic_score(sample, c)) if candidates else 0

    # --- колонка кода: заголовок «Код»/«тарификатор» или код-подобные значения ---
    code_col = next(
        (
            c
            for c in range(ncol)
            if c != name_col
            and c not in price_cols
            and any(k in header[c].lower() for k in _CODE_HEADER_KEYS)
        ),
        None,
    )
    if code_col is None:
        for c in range(ncol):
            if c == name_col or c in price_cols:
                continue
            vals = _col_values(sample, c)
            if vals and sum(1 for v in vals if looks_like_code(v)) >= max(2, 0.5 * len(vals)):
                code_col = c
                break

    unit_col = next(
        (
            c
            for c in range(ncol)
            if c not in price_cols and any(k in header[c].lower() for k in _UNIT_KEYS)
        ),
        None,
    )
    return name_col, code_col, unit_col, price_cols


def extract_rows_from_grid(grid: Grid) -> list[RawRow]:
    """Превратить сетку в список позиций прайса."""
    grid = [[clean(c) for c in r] for r in grid if any(clean(c) for c in r)]
    if len(grid) < 2:
        return []

    header_idx = find_header_row(grid)
    name_col, code_col, unit_col, price_cols = _assign_columns(grid, header_idx)

    rows: list[RawRow] = []
    section: str | None = None
    for r in grid[header_idx + 1 :]:
        nonempty = [c for c in r if c]
        if not nonempty:
            continue

        prices: dict[str, float] = {}
        for c, tier in price_cols.items():
            if c < len(r):
                price = parse_price(r[c])
                if price:
                    prices[tier] = price

        # Заголовок секции: одна осмысленная ячейка и ни одной цены.
        if not prices and len(nonempty) == 1:
            section = nonempty[0]
            continue

        name = r[name_col] if name_col < len(r) else ""
        if not prices or not is_real_name(name):
            continue  # не позиция: нет цены либо в «названии» код/число, а не услуга

        rows.append(
            RawRow(
                service_name_raw=name,
                service_code_source=(
                    r[code_col]
                    if code_col is not None and code_col < len(r) and r[code_col]
                    else None
                ),
                prices=prices,
                unit=(
                    r[unit_col]
                    if unit_col is not None and unit_col < len(r) and r[unit_col]
                    else None
                ),
                section=section,
            )
        )
    return rows