export/csv_export.py

"""EuropaLex CSV Export — creates a zipped folder containing CSV + media files."""

import csv
import re
import shutil
from pathlib import Path
from typing import Any

# ISO 639-1 language abbreviation mapping
_LANGUAGE_ABBREVS: dict[str, str] = {
    "Bulgarian": "BG",
    "Croatian": "HR",
    "Czech": "CS",
    "Danish": "DA",
    "Dutch": "NL",
    "Estonian": "ET",
    "Finnish": "FI",
    "French": "FR",
    "German": "DE",
    "Greek": "EL",
    "Hungarian": "HU",
    "Irish": "GA",
    "Italian": "IT",
    "Latvian": "LV",
    "Lithuanian": "LT",
    "Maltese": "MT",
    "Polish": "PL",
    "Portuguese": "PT",
    "Romanian": "RO",
    "Slovak": "SK",
    "Slovenian": "SL",
    "Spanish": "ES",
    "Swedish": "SV",
}

# Project root for resolving relative paths
_PROJECT_ROOT = Path(__file__).resolve().parent.parent


def _sanitize_folder_name(scenario: str) -> str:
    """Convert scenario text to a filesystem-safe folder name slug.

    Rules: lowercase, remove special characters (keep alphanumeric, spaces, underscores),
    replace spaces with underscores, collapse multiple underscores, strip leading/trailing underscores.

    Args:
        scenario: Free-form scenario/topic string from the user.

    Returns:
        Sanitized slug suitable for use as a directory name.
    """
    slug = scenario.strip().lower()
    slug = re.sub(r'[^a-z0-9\s_]', '', slug)   # remove special chars
    slug = re.sub(r'\s+', '_', slug)             # spaces → underscores
    slug = re.sub(r'_+', '_', slug)              # collapse multiple underscores
    return slug.strip('_')


def _get_language_abbrev(language: str) -> str:
    """Return the ISO 639-1 abbreviation for a language name.

    Args:
        language: Language name (e.g., 'Latvian', 'Spanish').

    Returns:
        Two-letter ISO 639-1 code.

    Raises:
        ValueError: If the language is not in the mapping.
    """
    if language not in _LANGUAGE_ABBREVS:
        raise ValueError(
            f"Unknown language '{language}'. "
            f"Supported: {', '.join(sorted(_LANGUAGE_ABBREVS.keys()))}"
        )
    return _LANGUAGE_ABBREVS[language]


def export_csv_zip(
    cards: list[dict[str, Any]],
    scenario: str,
    cefr_level: str,
    target_language: str,
) -> str:
    """Export cards as a zipped folder containing CSV + media files.

    Creates a folder under {models_dir}/output/export/ with the following flat structure:
        {folder_name}/
            cards.csv                          (CSV with 7 columns, one row per card)
            {scenario_slug}_{CEFR}_{LANG}_0.wav   (copied from TTS output)
            {scenario_slug}_{CEFR}_{LANG}_0.png   (copied from image generation)
        {folder_name}.zip                       (zipped archive of the above)

    CSV columns: scenario, cefr_level, target_language, english_text, translated_text,
                 audio_filename, image_filename

    Media filenames are bare filenames in the export folder (e.g., 'ordering_coffee_A2_LV_0.wav').
    Missing media files are silently skipped — CSV entries remain empty strings.

    Args:
        cards: List of card dicts with keys: 'text', 'translation',
               'audio_path' (str or None), 'image_path' (str or None).
        scenario: Free-form scenario/topic string.
        cefr_level: CEFR level string (e.g., 'A2', 'B1').
        target_language: Target language name (e.g., 'Latvian').

    Returns:
        Absolute path to the generated .zip file.

    Raises:
        ValueError: If target_language is not in the supported mapping.
        RuntimeError: If zip creation fails.
    """
    # Resolve output directory from project root
    export_base = _PROJECT_ROOT / ".local" / "models" / "output" / "export"
    export_base.mkdir(parents=True, exist_ok=True)

    # Build folder name
    scenario_slug = _sanitize_folder_name(scenario)
    lang_abbrev = _get_language_abbrev(target_language)
    folder_name = f"{scenario_slug}_{cefr_level}_{lang_abbrev}"
    export_dir = export_base / folder_name
    export_dir.mkdir(parents=True, exist_ok=True)

    # Copy media files and build CSV rows (flat folder — no subfolders)
    csv_path = export_dir / "cards.csv"
    with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.writer(csvfile, quoting=csv.QUOTE_ALL)
        # Header row
        writer.writerow([
            'scenario', 'cefr_level', 'target_language',
            'english_text', 'translated_text',
            'audio_filename', 'image_filename'
        ])
        for i, card in enumerate(cards):
            audio_path = card.get('audio_path')
            image_path = card.get('image_path')

            # Build the common prefix: {scenario_slug}_{CEFR}_{LANG_ABBREV}
            base_name = f"{scenario_slug}_{cefr_level}_{lang_abbrev}"

            # Copy audio file if it exists — flat naming
            audio_filename = ''
            if audio_path and Path(audio_path).exists():
                audio_dst = export_dir / f"{base_name}_{i}.wav"
                shutil.copy2(audio_path, audio_dst)
                audio_filename = f"{base_name}_{i}.wav"

            # Copy image file if it exists — flat naming
            image_filename = ''
            if image_path and Path(image_path).exists():
                image_dst = export_dir / f"{base_name}_{i}.png"
                shutil.copy2(image_path, image_dst)
                image_filename = f"{base_name}_{i}.png"

            writer.writerow([
                scenario,
                cefr_level,
                target_language,
                card.get('text', ''),
                card.get('translation', ''),
                audio_filename,
                image_filename,
            ])

    # Create zip archive
    zip_path = shutil.make_archive(
        str(export_base / folder_name),
        'zip',
        export_dir,
    )

    return str(zip_path)