image_resizer/workbook.py

"""Excel workbook parsing for product image URLs."""

from __future__ import annotations

import logging
from collections import Counter

import pandas as pd

from image_resizer.models import DuplicateRecord, ImageItem, WorkbookParseResult

logger = logging.getLogger(__name__)

_URL_KEYWORDS = ("url", "image", "link")
_TARGET_SHEETS = ("manual", "writebuffer")


def _normalize_item_code(raw) -> str | None:
  if pd.isna(raw):
    return None
  text = str(raw).strip()
  if not text:
    return None
  if text.replace(".", "", 1).isdigit():
    return text.split(".")[0]
  return text


def _looks_like_url(text: str) -> bool:
  lower = text.strip().lower()
  return lower.startswith("http://") or lower.startswith("https://")


def _url_columns(columns: list[str]) -> list[str]:
  return [c for c in columns if any(k in c.lower() for k in _URL_KEYWORDS)]


def _fallback_image_key(index: int) -> str:
  """Image, Image2, Image3, ... for rows without a product code."""
  return "Image" if index == 1 else f"Image{index}"


def _parse_sheet(sheet_name: str, df: pd.DataFrame) -> tuple[list[ImageItem], str | None]:
  """Extract image items from one sheet. Returns (items, error_message)."""
  df = df.copy()
  df.columns = [c.strip() for c in df.columns]

  item_col = next((c for c in df.columns if c.lower() == "itemcode"), None)
  if not item_col:
    return [], "Missing 'ItemCode' column"

  url_cols = _url_columns(list(df.columns))
  items: list[ImageItem] = []
  fallback_index = 0

  for df_index, row in df.iterrows():
    key = _normalize_item_code(row[item_col])
    item_code_url: str | None = None

    # URLs pasted into ItemCode (column A) are treated as the main image, not a SKU.
    if key and _looks_like_url(key):
      item_code_url = key
      key = None

    url_entries: list[tuple[str, str]] = []  # (url, column_label)
    if item_code_url:
      url_entries.append((item_code_url, item_col))

    for col in url_cols:
      if pd.notna(row[col]):
        url = str(row[col]).strip()
        if url:
          url_entries.append((url, col))

    if not url_entries:
      continue

    if not key:
      fallback_index += 1
      key = _fallback_image_key(fallback_index)

    excel_row = int(df_index) + 2
    for col_idx, (url, column_label) in enumerate(url_entries):
      name = f"{key}.MAIN" if col_idx == 0 else f"{key}.PT{col_idx:02d}"
      items.append({
        "url": url,
        "name": name,
        "sheet": sheet_name,
        "row": excel_row,
        "column": column_label,
      })

  return items, None


def _disambiguate_names(items: list[ImageItem]) -> list[ImageItem]:
  """Keep first use of each name; suffix later collisions with the sheet name."""
  seen: set[str] = set()
  result: list[ImageItem] = []

  for item in items:
    name = item["name"]
    if name in seen:
      sheet_tag = str(item.get("sheet", "")).replace(" ", "")
      renamed = dict(item)
      renamed["name"] = f"{name}_{sheet_tag}" if sheet_tag else f"{name}_dup"
      # Ensure the renamed name is also unique if multiple collisions occur.
      base = renamed["name"]
      suffix = 2
      while renamed["name"] in seen:
        renamed["name"] = f"{base}_{suffix}"
        suffix += 1
      result.append(renamed)  # type: ignore[arg-type]
      seen.add(renamed["name"])
    else:
      result.append(item)
      seen.add(name)

  return result


def detect_duplicates(
  items: list[ImageItem],
) -> tuple[list[ImageItem], list[DuplicateRecord]]:
  """Keep first occurrence of each URL; mark later rows as duplicates."""
  seen: dict[str, ImageItem] = {}
  unique: list[ImageItem] = []
  duplicates: list[DuplicateRecord] = []

  for item in items:
    url_key = item["url"].strip()
    if url_key in seen:
      original = seen[url_key]
      duplicates.append({
        "url": url_key,
        "sheet": item.get("sheet", ""),
        "row": item.get("row", 0),
        "column": item.get("column", ""),
        "name": item["name"],
        "original_sheet": original.get("sheet", ""),
        "original_row": original.get("row", 0),
        "original_column": original.get("column", ""),
        "original_name": original["name"],
      })
    else:
      seen[url_key] = item
      unique.append(item)

  return unique, duplicates


def format_duplicates_text(duplicates: list[DuplicateRecord]) -> str:
  """Format duplicate records for quick scanning in the UI."""
  if not duplicates:
    return "No duplicate URLs found."

  column_counts = Counter(dup["column"] for dup in duplicates)
  breakdown = " · ".join(
    f"{column}: {count}" for column, count in column_counts.most_common()
  )
  lines = [f"{len(duplicates)} duplicates found · {breakdown}", ""]
  lines.extend(
    f"R{dup['row']} {dup['column']} · {dup['name']} → "
    f"R{dup['original_row']} {dup['original_name']}"
    for dup in duplicates
  )
  return "\n".join(lines)


def _build_summary(
  total_links: int,
  duplicate_count: int,
  sheet_counts: dict[str, int],
  sheet_errors: list[str] | None = None,
) -> str:
  parts = [f"✅ {total_links} links"]
  for sheet, count in sheet_counts.items():
    parts.append(f"{sheet}: {count}")
  if duplicate_count:
    parts.append(f"{duplicate_count} duplicates found")
  else:
    parts.append("No duplicates")

  lines = [" · ".join(parts)]
  if sheet_errors:
    lines.extend(f"⚠️ {err}" for err in sheet_errors)
  return "\n".join(lines)


def read_uploaded_workbook(file) -> WorkbookParseResult:
  """Parse Manual and WriteBuffer sheets and extract image URL items."""
  if not file:
    return WorkbookParseResult([], "❌ No file uploaded", [], "")

  try:
    xls = pd.ExcelFile(file.name)
    sheets = [s for s in xls.sheet_names if s.lower() in _TARGET_SHEETS]
    if not sheets:
      return WorkbookParseResult([], "❌ No Manual or WriteBuffer sheet found", [], "")

    all_items: list[ImageItem] = []
    sheet_counts: dict[str, int] = {}
    sheet_errors: list[str] = []

    for sheet in sheets:
      df = pd.read_excel(file.name, sheet_name=sheet, engine="openpyxl")
      items, err = _parse_sheet(sheet, df)
      if err:
        sheet_errors.append(f"Sheet '{sheet}': {err}")
        sheet_counts[sheet] = 0
        continue
      sheet_counts[sheet] = len(items)
      all_items.extend(items)

    if not all_items:
      if sheet_errors:
        return WorkbookParseResult(
          [],
          "❌ " + " · ".join(sheet_errors),
          [],
          "",
        )
      return WorkbookParseResult([], "❌ No image URLs found in Manual or WriteBuffer", [], "")

    all_items = _disambiguate_names(all_items)
    _, duplicates = detect_duplicates(all_items)
    duplicates_text = format_duplicates_text(duplicates)
    summary = _build_summary(
      len(all_items),
      len(duplicates),
      sheet_counts,
      sheet_errors,
    )

    return WorkbookParseResult(all_items, summary, duplicates, duplicates_text)
  except Exception as e:
    logger.exception("Error reading workbook")
    return WorkbookParseResult([], f"❌ Error: {e}", [], "")


def workbook_upload_outputs(file) -> tuple[list[ImageItem], str, str]:
  """Adapter for Gradio workbook upload handler."""
  result = read_uploaded_workbook(file)
  return result.items, result.summary, result.duplicates_text