kdp_generator/csv_reader.py

import csv
import logging
import unicodedata
from pathlib import Path
from typing import Optional

from models import WordClue

logger = logging.getLogger(__name__)

MIN_WORD_LENGTH = 3
REQUIRED_COLUMNS = {"word", "clue"}
OPTIONAL_COLUMNS = {"difficulty", "category"}


class CSVValidationError(Exception):
    """Raised when the CSV file fails validation."""


def _strip_accents(text: str) -> str:
    """Replace accented characters with their ASCII equivalents."""
    nfkd = unicodedata.normalize("NFKD", text)
    return "".join(c for c in nfkd if not unicodedata.combining(c))


def _normalize_word(raw: str) -> Optional[str]:
    """Normalize a word to uppercase A-Z only. Returns None if invalid."""
    word = _strip_accents(raw.strip()).upper()
    word = word.replace(" ", "").replace("-", "")
    if not word.isalpha():
        return None
    return word


def read_csv(
    filepath: str | Path,
    min_words: int = 1,
    max_word_length: Optional[int] = None,
) -> list[WordClue]:
    """Read and validate a CSV file of words and clues.

    Args:
        filepath: Path to the CSV file.
        min_words: Minimum number of valid words required.
        max_word_length: If set, words longer than this trigger a warning.

    Returns:
        A deduplicated, validated list of WordClue objects.

    Raises:
        CSVValidationError: If the file is missing required columns or
            contains fewer than min_words valid entries.
        FileNotFoundError: If the file does not exist.
    """
    path = Path(filepath)
    if not path.exists():
        raise FileNotFoundError(f"CSV file not found: {path}")

    with open(path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)

        if reader.fieldnames is None:
            raise CSVValidationError("CSV file is empty or has no header row.")

        headers = {h.strip().lower() for h in reader.fieldnames}
        missing = REQUIRED_COLUMNS - headers
        if missing:
            raise CSVValidationError(
                f"CSV is missing required columns: {', '.join(sorted(missing))}. "
                f"Found: {', '.join(sorted(headers))}"
            )

        has_difficulty = "difficulty" in headers
        has_category = "category" in headers

        words_seen: set[str] = set()
        results: list[WordClue] = []
        skipped = 0

        for line_num, row in enumerate(reader, start=2):
            raw_word = row.get("word", "").strip()
            clue = row.get("clue", "").strip()

            if not raw_word or not clue:
                logger.warning("Line %d: empty word or clue, skipping.", line_num)
                skipped += 1
                continue

            word = _normalize_word(raw_word)
            if word is None:
                logger.warning(
                    "Line %d: word %r contains non-alpha characters, skipping.",
                    line_num,
                    raw_word,
                )
                skipped += 1
                continue

            if len(word) < MIN_WORD_LENGTH:
                logger.warning(
                    "Line %d: word %r is too short (min %d letters), skipping.",
                    line_num,
                    word,
                    MIN_WORD_LENGTH,
                )
                skipped += 1
                continue

            if max_word_length and len(word) > max_word_length:
                logger.warning(
                    "Line %d: word %r has %d letters, exceeds grid size of %d.",
                    line_num,
                    word,
                    len(word),
                    max_word_length,
                )

            if word in words_seen:
                logger.warning(
                    "Line %d: duplicate word %r, skipping.", line_num, word
                )
                skipped += 1
                continue

            words_seen.add(word)

            difficulty = row.get("difficulty", "").strip().lower() or None if has_difficulty else None
            category = row.get("category", "").strip() or None if has_category else None

            results.append(WordClue(
                word=word,
                clue=clue,
                difficulty=difficulty,
                category=category,
            ))

    if len(results) < min_words:
        raise CSVValidationError(
            f"CSV contains only {len(results)} valid word(s), "
            f"but at least {min_words} are required."
        )

    if skipped:
        logger.info("Skipped %d invalid row(s). Loaded %d word(s).", skipped, len(results))
    else:
        logger.info("Loaded %d word(s) from %s.", len(results), path.name)

    return results


def group_by_category(words: list[WordClue]) -> dict[str, list[WordClue]]:
    """Group words by their category. Uncategorized words go under 'General'."""
    groups: dict[str, list[WordClue]] = {}
    for wc in words:
        key = wc.category or "General"
        groups.setdefault(key, []).append(wc)
    return groups


def group_by_difficulty(words: list[WordClue]) -> dict[str, list[WordClue]]:
    """Group words by difficulty. Words without difficulty go under 'medium'."""
    groups: dict[str, list[WordClue]] = {}
    for wc in words:
        key = wc.difficulty or "medium"
        groups.setdefault(key, []).append(wc)
    return groups