"""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