app/services/spreadsheet_reader.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from openpyxl import load_workbook

from app.services.normalization import clean_cell, normalize_key, to_float
from app.services.schema_detector import CANONICAL_COLUMNS, COLUMN_ALIASES, REQUIRED_MINIMUM, HeaderDetection


class ParsedSpreadsheet:
    def __init__(self, header: HeaderDetection, records: list[dict[str, Any]]):
        self.header = header
        self.records = records


def _map_header(value: Any) -> str | None:
    raw = clean_cell(value)
    if raw is None:
        return None
    if raw in CANONICAL_COLUMNS:
        return raw
    key = normalize_key(raw)
    if key in CANONICAL_COLUMNS:
        return key
    return COLUMN_ALIASES.get(key)


def _safe_cell(value: Any) -> Any:
    return clean_cell(value)


def _row_values(row: tuple[Any, ...]) -> list[Any]:
    return [_safe_cell(value) for value in row]


def _detect_header_from_rows(rows: list[list[Any]], max_scan_rows: int = 20) -> HeaderDetection:
    best: HeaderDetection | None = None
    best_score = -1

    scan_limit = min(len(rows), max_scan_rows)
    for row_idx0 in range(scan_limit):
        values = rows[row_idx0]
        raw_headers = [str(v).strip() if v is not None else '' for v in values]
        mapped: dict[str, str] = {}
        unknown: list[str] = []

        for raw in raw_headers:
            if not raw:
                continue
            canonical = _map_header(raw)
            if canonical:
                mapped[raw] = canonical
            else:
                # Text-heavy instruction rows in INPA templates should not dominate.
                if len(raw) <= 80:
                    unknown.append(raw)

        mapped_values = set(mapped.values())
        score = len(mapped_values & set(REQUIRED_MINIMUM)) * 3 + len(mapped_values & set(CANONICAL_COLUMNS))
        if score > best_score:
            missing = [col for col in REQUIRED_MINIMUM if col not in mapped_values]
            best = HeaderDetection(
                header_row=row_idx0 + 1,
                mapping=mapped,
                raw_headers=raw_headers,
                missing_minimum=missing,
                unknown_headers=unknown,
            )
            best_score = score

    if best is None or best_score < 3:
        raise ValueError('Não foi possível detectar a linha de cabeçalho da planilha.')
    return best


def _looks_like_template_example(row_idx: int, header_row: int, raw: dict[str, Any]) -> bool:
    if row_idx != header_row + 1:
        return False
    joined = ' '.join(str(v).strip().lower() for v in raw.values() if v is not None)
    example_markers = ['a. coletor', 'b. auxiliar', 'exemplo', 'encholirium horridum', 'bromeliaceae encholirium']
    return any(marker in joined for marker in example_markers)


def _open_workbook(path: Path):
    # read_only=True avoids a class of openpyxl errors in files with odd worksheet dimensions.
    # If a workbook cannot be streamed, fall back to normal mode.
    try:
        return load_workbook(path, data_only=True, read_only=True)
    except Exception:
        return load_workbook(path, data_only=True, read_only=False)


def _select_sheet(wb, sheet_name: str | None):
    if sheet_name:
        if sheet_name not in wb.sheetnames:
            available = ', '.join(wb.sheetnames)
            raise ValueError(f"A aba '{sheet_name}' não foi encontrada. Abas disponíveis: {available}")
        return wb[sheet_name]
    # Prefer the INPA/BRAHMS sheet when present, otherwise use the first sheet.
    for preferred in ('Espécimes', 'Especimes', 'Specimens', 'Sheet1'):
        if preferred in wb.sheetnames:
            return wb[preferred]
    return wb[wb.sheetnames[0]]


def parse_xlsx(path: Path, sheet_name: str | None = None) -> ParsedSpreadsheet:
    wb = _open_workbook(path)
    try:
        ws = _select_sheet(wb, sheet_name)

        rows: list[list[Any]] = []
        # Avoid ws.max_row/ws[1] because some browser-edited or legacy spreadsheets expose None-like dimensions.
        # IMPORTANT on Windows: read all rows before returning and close the workbook in finally,
        # otherwise TemporaryDirectory cleanup can fail with WinError 32 because openpyxl keeps
        # the uploaded .xlsx file handle open.
        for row in ws.iter_rows(values_only=True):
            values = _row_values(row)
            # Keep instruction rows; header detection needs them for row indices, but trim trailing empty cells.
            while values and values[-1] is None:
                values.pop()
            rows.append(values)
            if len(rows) >= 5000:
                break
    finally:
        close = getattr(wb, 'close', None)
        if callable(close):
            close()

    if not rows:
        raise ValueError('A planilha está vazia.')

    header = _detect_header_from_rows(rows)
    header_idx0 = header.header_row - 1
    header_values = header.raw_headers
    raw_to_canonical = header.mapping
    useful_header_count = max(1, len([h for h in header_values if h]))

    records: list[dict[str, Any]] = []
    for row_idx0 in range(header_idx0 + 1, len(rows)):
        row_idx = row_idx0 + 1
        values = rows[row_idx0]
        raw: dict[str, Any] = {}
        canonical: dict[str, Any] = {'_row_number': row_idx, '_raw': raw}
        empty_count = 0

        for pos, raw_header in enumerate(header_values):
            if not raw_header:
                continue
            value = values[pos] if pos < len(values) else None
            value = _safe_cell(value)
            raw[raw_header] = value
            if value is None:
                empty_count += 1
            canonical_name = raw_to_canonical.get(raw_header)
            if canonical_name:
                canonical[canonical_name] = value

        if empty_count >= useful_header_count - 1:
            continue
        if _looks_like_template_example(row_idx, header.header_row, raw):
            continue

        canonical['lat'] = to_float(canonical.get('lat'))
        canonical['long'] = to_float(canonical.get('long'))
        records.append(canonical)

    return ParsedSpreadsheet(header=header, records=records)

# TSIINO_SPREADSHEET_READER_DETECTOR_V34
def _detect_header_from_rows(rows, max_scan_rows: int = 20):
    try:
        from app.services.schema_detector import _tsiino_detect_header_from_values_v34
        return _tsiino_detect_header_from_values_v34(rows, max_scan_rows=max_scan_rows)
    except Exception:
        best = None
        best_score = -1
        for row_idx, row in enumerate(rows[:max_scan_rows], start=1):
            raw_headers = [clean_cell(v) or '' for v in row]
            mapped = {}
            unknown = []
            for raw in raw_headers:
                if not raw:
                    continue
                canonical = _map_header(raw)
                if canonical:
                    mapped.setdefault(raw, canonical)
                else:
                    unknown.append(raw)
            score = len(set(mapped.values()) & set(REQUIRED_MINIMUM)) + len(set(mapped.values()) & set(CANONICAL_COLUMNS))
            if score > best_score:
                missing = [col for col in REQUIRED_MINIMUM if col not in set(mapped.values())]
                best = HeaderDetection(row_idx, mapped, raw_headers, missing, unknown)
                best_score = score
        if best is None or best_score < 3:
            raise ValueError('Não foi possível detectar a linha de cabeçalho da planilha.')
        return best

# TSIINO_STANDARD_READER_DETECTOR_V36
def _detect_header_from_rows(rows, max_scan_rows: int = 25):
    try:
        from app.services.schema_detector import _tsiino_v36_detect_header_from_values
        return _tsiino_v36_detect_header_from_values(rows, max_scan_rows=max_scan_rows)
    except Exception as exc:
        # Propaga a falha para o fluxo de mapeamento, sem aceitar cabeçalhos descritivos como padrão.
        raise ValueError(str(exc))