backend/tests/fixtures/client_examples.py

"""Parse client golden examples from PDC data/Examples .xlsx."""

from __future__ import annotations

import re
from pathlib import Path

import pandas as pd

EXAMPLES_PATH = Path(__file__).resolve().parents[3] / "PDC data" / "Examples .xlsx"

AI_FIELD_MAP = {
    "Warp Count": "warp_count",
    "Weft Count": "weft_count",
    "Reed Count": "reed_count",
    "Ends Per Dent": "ends_per_dent",
    "Onloom Epi": "on_loom_epi",
    "Onloom Ppi": "on_loom_ppi",
    "Greige Epi": "greige_epi",
    "Greige Ppi": "greige_ppi",
    "Finish Epi": "finish_epi",
    "Finish Ppi": "finish_ppi",
}


def _parse_analysis(text: str) -> dict:
    """Parse analysis lines including twill weaves and multi-line blends."""
    out: dict = {}
    cleaned = text.replace("Analysis:", "").strip()
    lines = [ln.strip() for ln in cleaned.split("\n") if ln.strip()]
    main = lines[0] if lines else cleaned

    for ln in lines[1:]:
        if "%" in ln:
            out["blend"] = ln.strip()

    m = re.search(
        r"([\d.]+)'?s?\*([\d.]+)'?s?-(\d+)\*(\d+)-(.+?)\s+GSM:\s*([\d.]+)",
        main,
        re.IGNORECASE,
    )
    if m:
        out["warp_count"] = float(m.group(1))
        out["weft_count"] = float(m.group(2))
        out["finish_epi"] = float(m.group(3))
        out["finish_ppi"] = float(m.group(4))
        out["weave"] = m.group(5).strip().upper()
        out["target_gsm"] = float(m.group(6))

    if "blend" not in out and "cotton" in text.lower():
        out["blend"] = "100% COTTON"
    return out


def _parse_key_value_block(df: pd.DataFrame, key_col: int, val_col: int, start_row: int) -> dict:
    ai: dict = {}
    for j in range(start_row, min(start_row + 15, len(df))):
        key = df.iloc[j, key_col]
        if pd.isna(key):
            break
        key_s = str(key).strip()
        if key_s.lower().startswith("master"):
            break
        val = df.iloc[j, val_col]
        if pd.notna(val):
            try:
                ai[key_s] = float(val)
            except (TypeError, ValueError):
                ai[key_s] = val
    return ai


def _parse_ai_block(df: pd.DataFrame) -> tuple[dict, str]:
    ai: dict = {}
    note = ""
    for i in range(len(df)):
        for col in range(len(df.columns)):
            cell = df.iloc[i, col]
            if not isinstance(cell, str) or "AI Suggestion" not in cell:
                continue
            parts = cell.split(":", 1)
            if len(parts) > 1 and parts[1].strip():
                note = parts[1].strip()
            val_col = col + 2
            key_col = col + 1
            if val_col >= len(df.columns):
                continue
            ai = _parse_key_value_block(df, key_col, val_col, i)
            return ai, note

    for i in range(len(df) - 1, -1, -1):
        for col in range(len(df.columns) - 1):
            key = df.iloc[i, col]
            if pd.isna(key) or str(key).strip() != "Warp Count":
                continue
            val = df.iloc[i, col + 1]
            try:
                float(val)
            except (TypeError, ValueError):
                continue
            ai = _parse_key_value_block(df, col, col + 1, i)
            if ai:
                return ai, note
    return ai, note


def _find_analysis_row(df: pd.DataFrame) -> str:
    for i in range(len(df)):
        for col in range(len(df.columns)):
            cell = df.iloc[i, col]
            if isinstance(cell, str) and "Analysis:" in cell:
                return cell.replace("Analysis:", "").strip()
    return ""


def _detect_archive_cols(df: pd.DataFrame) -> dict:
    header = [str(c).strip().lower() if pd.notna(c) else "" for c in df.iloc[0]]

    def idx(*names: str) -> int | None:
        for i, h in enumerate(header):
            if any(n in h for n in names):
                return i
        return None

    return {
        "warp": idx("warp code"),
        "weft": idx("weft code"),
        "reed": idx("reed count"),
        "epd": idx("ends per dent"),
        "onloom_epi": idx("on loom epi"),
        "onloom_ppi": idx("on loom ppi"),
        "greige_epi": idx("greige epi"),
        "greige_ppi": idx("greige ppi"),
        "finish_epi": idx("finish epi"),
        "finish_ppi": idx("finish ppi"),
    }


def _parse_gsm_cases(df: pd.DataFrame) -> list[dict]:
    cases: list[dict] = []
    for i in range(len(df)):
        for col in range(len(df.columns)):
            if str(df.iloc[i, col]).strip() != "Case 1":
                continue
            for j in range(i, min(i + 6, len(df))):
                label = str(df.iloc[j, col]).strip()
                if not label.startswith("Case"):
                    continue
                try:
                    cases.append({
                        "case": label,
                        "warp_count": float(df.iloc[j, col + 1]),
                        "weft_count": float(df.iloc[j, col + 2]),
                        "finish_epi": float(df.iloc[j, col + 3]),
                        "finish_ppi": float(df.iloc[j, col + 4]),
                        "gsm": float(df.iloc[j, col + 5]),
                    })
                except (TypeError, ValueError, IndexError):
                    break
            return cases
    return cases


def load_client_examples() -> list[dict]:
    if not EXAMPLES_PATH.exists():
        return []

    xl = pd.ExcelFile(EXAMPLES_PATH)
    examples: list[dict] = []
    for sheet in xl.sheet_names:
        df = pd.read_excel(EXAMPLES_PATH, sheet_name=sheet, header=None)
        analysis_text = _find_analysis_row(df)
        inputs = _parse_analysis(analysis_text)
        ai, note = _parse_ai_block(df)
        if not inputs or not ai:
            continue

        cols = _detect_archive_cols(df)
        archive_matches: list[dict] = []
        for r in range(1, 20):
            if r >= len(df) or pd.isna(df.iloc[r, 0]):
                break
            master = str(df.iloc[r, 0]).strip()
            if not master or master.lower() in {"nan", "sr no"}:
                break
            try:
                archive_matches.append({
                    "master_article": master,
                    "score": float(r),
                    "construction": {
                        "warp_count": _safe_float_col(df.iloc[r, cols["warp"]]) if cols["warp"] is not None else None,
                        "weft_count": _safe_float_col(df.iloc[r, cols["weft"]]) if cols["weft"] is not None else None,
                        "reed_count": float(df.iloc[r, cols["reed"]]),
                        "ends_per_dent": float(df.iloc[r, cols["epd"]]),
                        "on_loom_epi": float(df.iloc[r, cols["onloom_epi"]]),
                        "on_loom_ppi": float(df.iloc[r, cols["onloom_ppi"]]),
                        "greige_epi": float(df.iloc[r, cols["greige_epi"]]),
                        "greige_ppi": float(df.iloc[r, cols["greige_ppi"]]),
                        "finish_epi": float(df.iloc[r, cols["finish_epi"]]),
                        "finish_ppi": float(df.iloc[r, cols["finish_ppi"]]),
                    },
                })
            except (TypeError, ValueError, IndexError):
                break

        expected = {field: ai.get(label) for label, field in AI_FIELD_MAP.items()}
        examples.append({
            "sheet": sheet,
            "inputs": inputs,
            "gsm_cases": _parse_gsm_cases(df),
            "archive_matches": archive_matches,
            "expected": expected,
            "note": note,
        })
    return examples


def _safe_float_col(value) -> float | None:
    if pd.isna(value):
        return None
    text = str(value).strip()
    if "/" in text:
        try:
            return float(text.split("/")[-1])
        except ValueError:
            return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None