app.py

import ast
import datetime
import hashlib
import io
import json
import logging
import os
import re
import shutil
import tarfile
import tempfile
import zipfile
from typing import Any, Dict, Iterator, List, Optional, Tuple

import gradio as gr
import numpy as np
import pandas as pd
import requests
from bs4 import BeautifulSoup
from datasets import Dataset
from huggingface_hub import HfApi
from pypdf import PdfReader

try:
    from docx import Document as DocxDocument
    DOCX_SUPPORT = True
except ImportError:
    DOCX_SUPPORT = False

try:
    import tiktoken
    _TIKTOKEN_ENC = tiktoken.get_encoding("cl100k_base")
    TIKTOKEN_SUPPORT = True
except Exception:
    TIKTOKEN_SUPPORT = False
    _TIKTOKEN_ENC = None

try:
    import trafilatura
    TRAFILATURA_SUPPORT = True
except ImportError:
    TRAFILATURA_SUPPORT = False

try:
    import pyarrow as pa
    import pyarrow.parquet as pq
    PYARROW_SUPPORT = True
except ImportError:
    PYARROW_SUPPORT = False

# ─────────────────────────────────────────────
# CONSTANTS
# ─────────────────────────────────────────────

APP_TITLE = "Universal Dataset Builder Pro"
RANDOM_SEED = 42
CHUNK_SIZE = 800
CHUNK_OVERLAP = 100

CODE_LANGUAGE_MAP: Dict[str, str] = {
    ".py": "python", ".ipynb": "python",
    ".js": "javascript", ".jsx": "javascript",
    ".ts": "typescript", ".tsx": "typescript",
    ".c": "c", ".h": "c",
    ".cpp": "cpp", ".hpp": "cpp", ".cc": "cpp", ".cxx": "cpp",
    ".rs": "rust",
    ".go": "go",
    ".java": "java",
    ".rb": "ruby",
    ".php": "php",
    ".swift": "swift",
    ".kt": "kotlin",
    ".html": "html", ".htm": "html",
    ".css": "css",
    ".xml": "xml",
    ".yaml": "yaml", ".yml": "yaml",
    ".sql": "sql",
    ".sh": "bash", ".bash": "bash", ".zsh": "bash",
    ".json": "json", ".jsonl": "json",
    ".csv": "csv", ".tsv": "csv",
    ".md": "markdown", ".rst": "markdown",
    ".txt": "text",
    ".toml": "toml",
    ".dockerfile": "dockerfile",
    ".tf": "terraform",
    ".lua": "lua",
    ".r": "r", ".R": "r",
    ".scala": "scala",
    ".ex": "elixir", ".exs": "elixir",
    ".cs": "csharp",
    ".fs": "fsharp",
    ".hs": "haskell",
    ".ml": "ocaml",
}

IGNORE_DIR_PATTERNS = {
    ".git", "node_modules", "venv", ".venv", "env", "__pycache__",
    ".cache", "dist", "build", "target", "coverage", ".next", ".nuxt",
    "vendor", "out", "bin", "obj", ".tox", ".mypy_cache", ".pytest_cache",
    "htmlcov", "site-packages", "eggs", ".eggs",
}

SKIP_EXTENSIONS = {
    ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".svg", ".ico", ".webp",
    ".mp4", ".avi", ".mov", ".mp3", ".wav", ".ogg", ".flac",
    ".zip", ".tar", ".gz", ".tgz", ".xz", ".bz2", ".7z", ".rar",
    ".exe", ".dll", ".so", ".dylib", ".a", ".o", ".obj",
    ".pdf",  # handled separately
    ".docx",  # handled separately
    ".pyc", ".pyo", ".pyd",
    ".class", ".jar",
    ".whl", ".egg",
    ".lock",
    ".min.js", ".min.css",
}

PROCESS_EXTENSIONS = set(CODE_LANGUAGE_MAP.keys())

SPECIAL_FILENAMES = {
    "Dockerfile", "Makefile", "Kconfig", ".env.example",
    "CMakeLists.txt", "Vagrantfile", "Procfile", "Gemfile",
    "Rakefile", "Brewfile", "Justfile",
}

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)


# ─────────────────────────────────────────────
# UTILITY
# ─────────────────────────────────────────────

def normalize_whitespace(text: str) -> str:
    return re.sub(r"\s+", " ", str(text or "")).strip()

def ensure_dataframe(df: Optional[pd.DataFrame]) -> pd.DataFrame:
    return df if isinstance(df, pd.DataFrame) else pd.DataFrame()

def stable_hash(text: str) -> str:
    return hashlib.md5(text.encode("utf-8", errors="ignore")).hexdigest()

def count_tokens(text: str) -> int:
    if TIKTOKEN_SUPPORT and _TIKTOKEN_ENC:
        try:
            return len(_TIKTOKEN_ENC.encode(text))
        except Exception:
            pass
    words = text.split()
    return int(len(words) * 1.3)

def count_words(text: str) -> int:
    return len(text.split())

def should_ignore_dir(dirname: str) -> bool:
    return dirname in IGNORE_DIR_PATTERNS or dirname.startswith(".")

def should_skip_file(filename: str) -> bool:
    _, ext = os.path.splitext(filename)
    ext = ext.lower()
    if ext in SKIP_EXTENSIONS:
        return True
    # Skip minified files
    if filename.endswith(".min.js") or filename.endswith(".min.css"):
        return True
    return False

def get_language(filename: str) -> str:
    if filename in SPECIAL_FILENAMES:
        fname_lower = filename.lower()
        if fname_lower == "dockerfile":
            return "dockerfile"
        if fname_lower == "makefile":
            return "makefile"
        return "text"
    _, ext = os.path.splitext(filename)
    return CODE_LANGUAGE_MAP.get(ext.lower(), "text")

def get_file_size(file_path: str) -> int:
    try:
        return os.path.getsize(file_path)
    except Exception:
        return 0


# ─────────────────────────────────────────────
# BOILERPLATE CLEANING
# ─────────────────────────────────────────────

def clean_boilerplate(text: str) -> str:
    text = re.sub(r"<script[^>]*>.*?</script>", " ", text, flags=re.DOTALL)
    text = re.sub(r"<style[^>]*>.*?</style>", " ", text, flags=re.DOTALL)
    text = re.sub(r"<[^>]+>", " ", text)
    boilerplate = [
        r"©.*?all rights reserved",
        r"cookie.*?policy",
        r"terms.*?service",
        r"privacy.*?policy",
        r"subscribe.*?newsletter",
        r"follow us on",
    ]
    for pat in boilerplate:
        text = re.sub(pat, " ", text, flags=re.IGNORECASE)
    text = re.sub(r"\s+", " ", text).strip()
    return text


# ─────────────────────────────────────────────
# QUALITY SCORING V3
# ─────────────────────────────────────────────

_NEAR_DUP_CACHE: Dict[str, int] = {}

def compute_chunk_quality(text: str, language: str = "text") -> Dict[str, Any]:
    """Per-chunk quality metrics. Deterministic, no side-effects."""
    chars = len(text)
    words = count_words(text)
    tokens = count_tokens(text)

    # Noise: high ratio of non-alphanum characters
    non_alpha = len(re.findall(r"[^a-zA-Z0-9\s]", text))
    noise_score = round(min(1.0, non_alpha / max(chars, 1)), 4)

    # Content density: unique words / total words
    unique_words = len(set(text.lower().split()))
    content_density = round(unique_words / max(words, 1), 4)

    # Language confidence: heuristic
    if language in ("python", "cpp", "c", "rust", "go", "java", "javascript", "typescript"):
        # Code: check for typical code patterns
        code_signals = len(re.findall(r"(def |class |fn |func |import |return |if |for |while )", text))
        lang_confidence = min(1.0, code_signals / max(words / 10, 1))
    else:
        # Text: check for sentence-ending punctuation
        sentences = re.findall(r"[.!?]", text)
        lang_confidence = min(1.0, len(sentences) / max(words / 15, 1))
    lang_confidence = round(lang_confidence, 4)

    # Duplicate score (within-session hash)
    h = stable_hash(text)
    dup_count = _NEAR_DUP_CACHE.get(h, 0)
    _NEAR_DUP_CACHE[h] = dup_count + 1
    duplicate_score = round(min(1.0, dup_count * 0.5), 4)

    # Composite quality_score (0-100)
    q = 100.0
    if chars < 50:
        q -= 30
    elif chars < 150:
        q -= 15
    if noise_score > 0.5:
        q -= 20
    elif noise_score > 0.3:
        q -= 10
    if content_density < 0.3:
        q -= 10
    if lang_confidence < 0.1 and language not in ("text", "markdown", "csv"):
        q -= 5
    if duplicate_score > 0:
        q -= 25
    quality_score = max(0, min(100, int(q)))

    return {
        "quality_score": quality_score,
        "duplicate_score": duplicate_score,
        "noise_score": noise_score,
        "lang_confidence": lang_confidence,
        "content_density": content_density,
        "char_count": chars,
        "word_count": words,
        "token_count": tokens,
    }

def quality_grade(score: int) -> str:
    if score >= 95: return "A+"
    if score >= 90: return "A"
    if score >= 80: return "B"
    if score >= 70: return "C"
    return "D"


# ─────────────────────────────────────────────
# CODE UNDERSTANDING (AST-LEVEL)
# ─────────────────────────────────────────────

def extract_python_structure(code: str) -> Dict[str, Any]:
    """Extract imports, classes, functions from Python source."""
    result = {"imports": [], "classes": [], "functions": [], "methods": []}
    try:
        tree = ast.parse(code)
        for node in ast.walk(tree):
            if isinstance(node, (ast.Import, ast.ImportFrom)):
                if isinstance(node, ast.Import):
                    for alias in node.names:
                        result["imports"].append(alias.name)
                else:
                    module = node.module or ""
                    result["imports"].append(module)
            elif isinstance(node, ast.ClassDef):
                result["classes"].append(node.name)
                for item in node.body:
                    if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                        result["methods"].append(f"{node.name}.{item.name}")
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                result["functions"].append(node.name)
    except Exception:
        pass
    return result

def extract_generic_structure(code: str, language: str) -> Dict[str, Any]:
    """Regex-based structure extraction for non-Python languages."""
    result = {"imports": [], "classes": [], "functions": [], "methods": []}
    try:
        if language in ("javascript", "typescript"):
            result["imports"] = re.findall(r'(?:import|require)\s*\(?["\']([^"\']+)["\']', code)
            result["classes"] = re.findall(r'\bclass\s+(\w+)', code)
            result["functions"] = re.findall(r'(?:function\s+(\w+)|const\s+(\w+)\s*=\s*(?:async\s*)?\()', code)
            result["functions"] = [f[0] or f[1] for f in result["functions"] if f[0] or f[1]]
        elif language in ("c", "cpp"):
            result["includes"] = re.findall(r'#include\s*[<"]([^>"]+)[>"]', code)
            result["classes"] = re.findall(r'\bclass\s+(\w+)', code)
            result["functions"] = re.findall(r'(?:^|\n)\w[\w\s\*]+\s+(\w+)\s*\([^)]*\)\s*\{', code)
        elif language == "rust":
            result["imports"] = re.findall(r'\buse\s+([\w:]+)', code)
            result["functions"] = re.findall(r'\bfn\s+(\w+)', code)
            result["classes"] = re.findall(r'\b(?:struct|enum|impl)\s+(\w+)', code)
        elif language == "go":
            result["imports"] = re.findall(r'"([^"]+)"', re.findall(r'import\s*\(([^)]+)\)', code)[0] if re.search(r'import\s*\(', code) else "")
            result["functions"] = re.findall(r'\bfunc\s+(?:\(\w+\s+\*?\w+\)\s+)?(\w+)', code)
        elif language == "java":
            result["imports"] = re.findall(r'import\s+([\w.]+);', code)
            result["classes"] = re.findall(r'\bclass\s+(\w+)', code)
            result["functions"] = re.findall(r'(?:public|private|protected|static|\s)+[\w<>\[\]]+\s+(\w+)\s*\(', code)
    except Exception:
        pass
    return result

def extract_code_structure(code: str, language: str) -> Dict[str, Any]:
    if language == "python":
        return extract_python_structure(code)
    return extract_generic_structure(code, language)


# ─────────────────────────────────────────────
# ARCHIVE EXTRACTION (streaming)
# ─────────────────────────────────────────────

def extract_archive(archive_path: str, dest_dir: str) -> bool:
    """Extract .zip / .tar / .tar.gz / .tgz / .tar.xz / .gz archives."""
    try:
        if zipfile.is_zipfile(archive_path):
            with zipfile.ZipFile(archive_path, "r") as zf:
                zf.extractall(dest_dir)
            return True
        if tarfile.is_tarfile(archive_path):
            with tarfile.open(archive_path, "r:*") as tf:
                tf.extractall(dest_dir)
            return True
        # .gz single file
        if archive_path.endswith(".gz"):
            import gzip
            out_path = os.path.join(dest_dir, os.path.basename(archive_path[:-3]))
            with gzip.open(archive_path, "rb") as f_in:
                with open(out_path, "wb") as f_out:
                    shutil.copyfileobj(f_in, f_out)
            return True
    except Exception as e:
        logger.warning(f"Archive extraction failed for {archive_path}: {e}")
    return False


# ─────────────────────────────────────────────
# FILE STREAMING ITERATOR (linux-scale)
# ─────────────────────────────────────────────

def stream_files(root_dir: str, max_file_size_mb: float = 5.0) -> Iterator[Tuple[str, str, str]]:
    """
    Yield (abs_path, rel_path, language) for every processable file in root_dir.
    Streams — never builds full list in memory.
    Skips binary, ignored dirs, oversized files.
    """
    max_bytes = int(max_file_size_mb * 1024 * 1024)
    for dirpath, dirnames, filenames in os.walk(root_dir):
        # Prune ignored directories in-place (affects os.walk traversal)
        dirnames[:] = [d for d in dirnames if not should_ignore_dir(d)]
        for fname in filenames:
            if should_skip_file(fname):
                continue
            abs_path = os.path.join(dirpath, fname)
            # Size guard
            try:
                if os.path.getsize(abs_path) > max_bytes:
                    continue
            except OSError:
                continue
            _, ext = os.path.splitext(fname)
            ext_lower = ext.lower()
            if fname in SPECIAL_FILENAMES:
                lang = get_language(fname)
            else:
                lang = CODE_LANGUAGE_MAP.get(ext_lower, "")
            if not lang:
                # Try reading first bytes to detect text
                try:
                    with open(abs_path, "rb") as f:
                        sample = f.read(512)
                    if b"\x00" in sample:
                        continue  # binary
                    lang = "text"
                except Exception:
                    continue
            rel_path = os.path.relpath(abs_path, root_dir)
            yield abs_path, rel_path, lang or "text"


# ─────────────────────────────────────────────
# TEXT CHUNKING V2 (semantic, overlap)
# ─────────────────────────────────────────────

def chunk_text_v2(
    text: str,
    source_name: str = "",
    file_path: str = "",
    file_extension: str = "",
    file_size: int = 0,
    language: str = "text",
    chunk_size: int = CHUNK_SIZE,
    overlap: int = CHUNK_OVERLAP,
) -> Iterator[Dict[str, Any]]:
    """Streaming text chunker. Yields one chunk-dict at a time."""
    if not text or not text.strip():
        return
    text = normalize_whitespace(text)
    sentences = re.split(r"(?<=[.!?\n])\s+", text)
    current = ""
    chunk_idx = 0
    prev_tail = ""

    for sentence in sentences:
        if len(current) + len(sentence) < chunk_size:
            current = (current + " " + sentence).strip() if current else sentence
        else:
            if current:
                full_text = (prev_tail + " " + current).strip() if prev_tail else current
                metrics = compute_chunk_quality(full_text, language)
                if metrics["quality_score"] >= 20 and metrics["duplicate_score"] == 0:
                    cid = f"{stable_hash(source_name)}_{chunk_idx}"
                    yield {
                        "chunk_id": cid,
                        "source_name": source_name,
                        "source_kind": "text",
                        "file_path": file_path,
                        "file_extension": file_extension,
                        "file_size": file_size,
                        "language": language,
                        "text": full_text,
                        **metrics,
                    }
                    chunk_idx += 1
                    prev_tail = current[-overlap:] if len(current) >= overlap else current
            current = sentence

    if current:
        full_text = (prev_tail + " " + current).strip() if prev_tail else current
        metrics = compute_chunk_quality(full_text, language)
        if metrics["quality_score"] >= 20:
            cid = f"{stable_hash(source_name)}_{chunk_idx}"
            yield {
                "chunk_id": cid,
                "source_name": source_name,
                "source_kind": "text",
                "file_path": file_path,
                "file_extension": file_extension,
                "file_size": file_size,
                "language": language,
                "text": full_text,
                **metrics,
            }


# ─────────────────────────────────────────────
# CODE CHUNKING V2 (function/class aware)
# ─────────────────────────────────────────────

def chunk_code_v2(
    code: str,
    language: str,
    source_name: str = "",
    file_path: str = "",
    file_extension: str = "",
    file_size: int = 0,
    max_chunk: int = 1200,
) -> Iterator[Dict[str, Any]]:
    """Streaming code chunker. Yields chunk-dicts."""
    if not code or not code.strip():
        return

    structure = extract_code_structure(code, language)
    lines = code.split("\n")
    chunk_idx = 0

    def emit(block: str) -> Optional[Dict[str, Any]]:
        nonlocal chunk_idx
        block = block.strip()
        if not block:
            return None
        metrics = compute_chunk_quality(block, language)
        if metrics["quality_score"] < 10 or metrics["duplicate_score"] > 0:
            return None
        cid = f"{stable_hash(source_name)}_{chunk_idx}"
        chunk_idx += 1
        return {
            "chunk_id": cid,
            "source_name": source_name,
            "source_kind": "code",
            "file_path": file_path,
            "file_extension": file_extension,
            "file_size": file_size,
            "language": language,
            "content": block,
            "imports": json.dumps(structure.get("imports", [])),
            "classes": json.dumps(structure.get("classes", [])),
            "functions": json.dumps(structure.get("functions", [])),
            "methods": json.dumps(structure.get("methods", [])),
            **metrics,
        }

    if language == "python":
        # Split by top-level def/class
        pattern = re.compile(r"^(?:def |class |async def )", re.MULTILINE)
        splits = [m.start() for m in pattern.finditer(code)]
        if not splits:
            chunk = emit(code)
            if chunk:
                yield chunk
            return
        boundaries = splits + [len(code)]
        for i in range(len(splits)):
            block = code[boundaries[i]:boundaries[i + 1]]
            if len(block) > max_chunk:
                # Sub-chunk large blocks
                for sub in _split_by_lines(block, max_chunk):
                    chunk = emit(sub)
                    if chunk:
                        yield chunk
            else:
                chunk = emit(block)
                if chunk:
                    yield chunk
    elif language == "sql":
        for stmt in code.split(";"):
            chunk = emit(stmt)
            if chunk:
                yield chunk
    else:
        # Generic: split by line-count windows
        for block in _split_by_lines(code, max_chunk):
            chunk = emit(block)
            if chunk:
                yield chunk

def _split_by_lines(text: str, max_chars: int) -> Iterator[str]:
    lines = text.split("\n")
    current = []
    current_len = 0
    for line in lines:
        ll = len(line) + 1
        if current_len + ll > max_chars and current:
            yield "\n".join(current)
            current = [line]
            current_len = ll
        else:
            current.append(line)
            current_len += ll
    if current:
        yield "\n".join(current)


# ─────────────────────────────────────────────
# QA GENERATION V2
# ─────────────────────────────────────────────

def extract_key_phrases(text: str, max_phrases: int = 5) -> List[str]:
    words = re.findall(r"\b[a-zA-Z]{4,}\b", text)
    freq: Dict[str, int] = {}
    for w in words:
        freq[w.lower()] = freq.get(w.lower(), 0) + 1
    stopwords = {"this", "that", "with", "from", "have", "been", "they", "their",
                 "will", "would", "could", "should", "which", "what", "when", "where"}
    sorted_words = sorted(
        [(w, c) for w, c in freq.items() if w not in stopwords],
        key=lambda x: x[1], reverse=True
    )
    return [w for w, _ in sorted_words[:max_phrases]]

def generate_qa_v2(text: str, source_name: str = "") -> List[Dict[str, str]]:
    """Generate What/Why/How/When/Where QA pairs."""
    qa_pairs = []
    sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text) if len(s.strip()) >= 25]
    if len(sentences) < 2:
        return qa_pairs

    for sentence in sentences[:12]:
        keywords = extract_key_phrases(sentence, 3)
        if not keywords:
            continue

        # What
        qa_pairs.append({
            "question": f"What is {keywords[0]}?",
            "answer": sentence,
            "source_name": source_name,
            "qa_type": "what",
        })

        # How
        if any(kw in sentence.lower() for kw in ("how", "process", "method", "step", "using", "via", "through")):
            qa_pairs.append({
                "question": f"How does {keywords[0]} work?",
                "answer": sentence,
                "source_name": source_name,
                "qa_type": "how",
            })

        # Why
        if any(kw in sentence.lower() for kw in ("because", "reason", "therefore", "thus", "hence", "since")):
            qa_pairs.append({
                "question": f"Why is {keywords[0]} important?",
                "answer": sentence,
                "source_name": source_name,
                "qa_type": "why",
            })

        # When
        if any(kw in sentence.lower() for kw in ("when", "during", "after", "before", "once", "until")):
            qa_pairs.append({
                "question": f"When does {keywords[0]} occur?",
                "answer": sentence,
                "source_name": source_name,
                "qa_type": "when",
            })

        # Where
        if any(kw in sentence.lower() for kw in ("where", "location", "place", "region", "in ", "at ")):
            qa_pairs.append({
                "question": f"Where is {keywords[0]} found?",
                "answer": sentence,
                "source_name": source_name,
                "qa_type": "where",
            })

    return qa_pairs[:15]


# ─────────────────────────────────────────────
# INSTRUCTION GENERATION V2
# ─────────────────────────────────────────────

INSTRUCTION_TEMPLATES = [
    ("Summarize the following text in 2-3 sentences:", "summary"),
    ("Explain the main concept described below:", "explanation"),
    ("Extract the three most important key points from:", "key_points"),
    ("Analyze and describe the process or method in:", "analysis"),
    ("Write a concise technical description of:", "technical"),
    ("Identify and list the main entities mentioned in:", "entity_extraction"),
    ("Rewrite the following in simpler terms:", "simplification"),
    ("Generate a detailed description based on:", "description"),
]

def generate_instructions_v2(text: str, source_name: str = "") -> List[Dict[str, str]]:
    sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text) if len(s.strip()) >= 30]
    results = []
    for i, sentence in enumerate(sentences[:8]):
        template, itype = INSTRUCTION_TEMPLATES[i % len(INSTRUCTION_TEMPLATES)]
        results.append({
            "instruction": template,
            "input": sentence,
            "output": f"[Model should respond based on the provided input about: {extract_key_phrases(sentence, 2)}]",
            "source_name": source_name,
            "instruction_type": itype,
        })
    return results


# ─────────────────────────────────────────────
# FILE LOADERS (individual formats)
# ─────────────────────────────────────────────

def load_csv(file_path: str) -> pd.DataFrame:
    return pd.read_csv(file_path)

def load_excel(file_path: str) -> pd.DataFrame:
    return pd.read_excel(file_path)

def load_json_file(file_path: str) -> pd.DataFrame:
    try:
        df = pd.read_json(file_path)
        if isinstance(df, pd.DataFrame):
            return df
    except Exception:
        pass
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    if isinstance(data, list):
        return pd.json_normalize(data)
    return pd.json_normalize([data])

def load_jsonl_file(file_path: str) -> pd.DataFrame:
    records = []
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    try:
                        records.append(json.loads(line))
                    except json.JSONDecodeError:
                        continue
    except Exception:
        pass
    return pd.json_normalize(records) if records else pd.DataFrame({"text": ["Invalid JSONL"]})

def load_parquet_file(file_path: str) -> pd.DataFrame:
    try:
        return pd.read_parquet(file_path)
    except Exception:
        return pd.DataFrame({"text": ["Invalid Parquet file"]})

def load_docx_file(file_path: str) -> pd.DataFrame:
    if not DOCX_SUPPORT:
        return pd.DataFrame({"text": ["python-docx not installed"]})
    try:
        doc = DocxDocument(file_path)
        paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
        if paragraphs:
            return pd.DataFrame({"text": [clean_boilerplate("\n".join(paragraphs))]})
    except Exception:
        pass
    return pd.DataFrame({"text": ["Invalid DOCX file"]})

def load_pdf(file_path: str) -> pd.DataFrame:
    reader = PdfReader(file_path)
    pages = []
    for idx, page in enumerate(reader.pages, start=1):
        text = page.extract_text() or ""
        if text.strip():
            pages.append({"text": clean_boilerplate(text), "page_number": idx})
    return pd.DataFrame(pages) if pages else pd.DataFrame({"text": [""]})

def load_website(url: str) -> pd.DataFrame:
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
    if TRAFILATURA_SUPPORT:
        try:
            downloaded = trafilatura.fetch_url(url)
            text = trafilatura.extract(downloaded) or ""
            title = ""
        except Exception:
            text = ""
            title = ""
    else:
        resp = requests.get(url, timeout=20, headers=headers)
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, "html.parser")
        for tag in soup(["script", "style", "noscript", "nav", "footer"]):
            tag.decompose()
        title = soup.title.get_text(strip=True) if soup.title else ""
        text = soup.get_text(separator=" ", strip=True)
    return pd.DataFrame({"text": [clean_boilerplate(text)], "title": [title], "url": [url]})

def load_text_file_raw(file_path: str) -> str:
    with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
        return f.read()


# ─────────────────────────────────────────────
# ARCHIVE / REPO STREAMING PROCESSOR
# ─────────────────────────────────────────────

def stream_repo_chunks(
    root_dir: str,
    stats: Dict[str, Any],
    max_file_size_mb: float = 5.0,
) -> Iterator[Dict[str, Any]]:
    """
    Stream chunk dicts from a directory tree.
    Updates stats dict in-place.
    Never loads entire repo into RAM.
    """
    for abs_path, rel_path, lang in stream_files(root_dir, max_file_size_mb):
        _, ext = os.path.splitext(abs_path)
        fsize = get_file_size(abs_path)
        stats["files_processed"] = stats.get("files_processed", 0) + 1
        stats["languages"].add(lang)

        try:
            if ext.lower() == ".pdf":
                df = load_pdf(abs_path)
                for _, row in df.iterrows():
                    text = str(row.get("text", ""))
                    yield from chunk_text_v2(
                        text, source_name=rel_path, file_path=rel_path,
                        file_extension=ext, file_size=fsize, language="text"
                    )
            elif ext.lower() == ".docx":
                df = load_docx_file(abs_path)
                for _, row in df.iterrows():
                    yield from chunk_text_v2(
                        str(row.get("text", "")), source_name=rel_path,
                        file_path=rel_path, file_extension=ext,
                        file_size=fsize, language="text"
                    )
            else:
                code = load_text_file_raw(abs_path)
                if lang in CODE_LANGUAGE_MAP.values() and lang not in ("text", "markdown", "csv", "json"):
                    yield from chunk_code_v2(
                        code, lang, source_name=rel_path, file_path=rel_path,
                        file_extension=ext, file_size=fsize
                    )
                else:
                    yield from chunk_text_v2(
                        clean_boilerplate(code), source_name=rel_path,
                        file_path=rel_path, file_extension=ext,
                        file_size=fsize, language=lang
                    )
        except Exception as e:
            stats["files_skipped"] = stats.get("files_skipped", 0) + 1
            stats["errors"].append(f"{rel_path}: {str(e)[:60]}")
            continue


# ─────────────────────────────────────────────
# LOAD ANY FILE (single file handler)
# ─────────────────────────────────────────────

def load_any_file(file_obj) -> Tuple[pd.DataFrame, Dict[str, Any]]:
    file_path = file_obj.name
    name = os.path.basename(file_path)
    _, suffix = os.path.splitext(name)
    suffix = suffix.lower()
    fsize = get_file_size(file_path)

    diagnostics = {
        "filename": name, "extension": suffix or "none",
        "detected_type": "Unknown", "status": "processing", "message": ""
    }
    df = None
    error_msg = None

    try:
        if suffix == ".csv":
            df = load_csv(file_path); diagnostics["detected_type"] = "CSV"
        elif suffix in {".xlsx", ".xls"}:
            df = load_excel(file_path); diagnostics["detected_type"] = "Excel"
        elif suffix == ".json":
            df = load_json_file(file_path); diagnostics["detected_type"] = "JSON"
        elif suffix == ".jsonl":
            df = load_jsonl_file(file_path); diagnostics["detected_type"] = "JSONL"
        elif suffix == ".parquet":
            df = load_parquet_file(file_path); diagnostics["detected_type"] = "Parquet"
        elif suffix == ".pdf":
            df = load_pdf(file_path); diagnostics["detected_type"] = "PDF"
        elif suffix == ".docx":
            df = load_docx_file(file_path); diagnostics["detected_type"] = "DOCX"
        elif suffix in {".zip", ".tar", ".gz", ".tgz", ".xz", ".bz2"}:
            # Archive: extract and stream
            diagnostics["detected_type"] = "Archive"
            temp_dir = tempfile.mkdtemp()
            try:
                ok = extract_archive(file_path, temp_dir)
                if ok:
                    stats: Dict[str, Any] = {
                        "files_processed": 0, "files_skipped": 0,
                        "languages": set(), "errors": []
                    }
                    chunks = list(stream_repo_chunks(temp_dir, stats))
                    if chunks:
                        df = pd.DataFrame(chunks)
                        diagnostics["message"] = (
                            f"Archive: {stats['files_processed']} files, "
                            f"{len(chunks)} chunks"
                        )
                    else:
                        df = pd.DataFrame({"text": ["Archive produced no usable content"]})
                else:
                    error_msg = "Failed to extract archive"
            finally:
                shutil.rmtree(temp_dir, ignore_errors=True)
        elif suffix in CODE_LANGUAGE_MAP:
            lang = CODE_LANGUAGE_MAP[suffix]
            content = load_text_file_raw(file_path)
            if lang in ("text", "markdown", "csv", "json"):
                df = pd.DataFrame({
                    "text": [clean_boilerplate(content)],
                    "language": [lang],
                    "source_kind": ["file"],
                })
            else:
                chunks = list(chunk_code_v2(
                    content, lang, source_name=name,
                    file_path=file_path, file_extension=suffix, file_size=fsize
                ))
                df = pd.DataFrame(chunks) if chunks else pd.DataFrame({
                    "content": [content], "language": [lang]
                })
            diagnostics["detected_type"] = "Code"
        else:
            try:
                content = load_text_file_raw(file_path)
                if content.strip():
                    df = pd.DataFrame({"text": [clean_boilerplate(content)]})
                    diagnostics["detected_type"] = "Text (auto)"
                else:
                    error_msg = "Empty file"
            except Exception as e:
                error_msg = f"Cannot read: {str(e)[:60]}"

        if df is None and error_msg:
            df = pd.DataFrame({"error": [error_msg]})
            diagnostics["status"] = "warning"
            diagnostics["message"] = error_msg
        elif df is None:
            df = pd.DataFrame({"error": ["Failed to load"]})
            diagnostics["status"] = "error"
            diagnostics["message"] = "Load failed"
        else:
            if df.empty:
                df = pd.DataFrame({"warning": ["Empty dataset"]})
                diagnostics["status"] = "warning"
            else:
                diagnostics["status"] = "success"
                if not diagnostics["message"]:
                    diagnostics["message"] = f"Loaded {len(df)} rows"

    except Exception as e:
        error_msg = str(e)[:100]
        df = pd.DataFrame({"error": [error_msg]})
        diagnostics["status"] = "error"
        diagnostics["message"] = error_msg

    df = ensure_dataframe(df).copy()
    if "source_name" not in df.columns:
        df["source_name"] = name
    if "source_kind" not in df.columns:
        df["source_kind"] = "file"
    if "file_extension" not in df.columns:
        df["file_extension"] = suffix or "unknown"
    if "file_size" not in df.columns:
        df["file_size"] = fsize

    return df, diagnostics


# ─────────────────────────────────────────────
# PASTED CONTENT PARSER
# ─────────────────────────────────────────────

def parse_pasted_content(pasted_text: str, pasted_format: str) -> pd.DataFrame:
    text = (pasted_text or "").strip()
    if not text:
        return pd.DataFrame()
    if pasted_format == "Auto":
        try:
            parsed = json.loads(text)
            return pd.json_normalize(parsed if isinstance(parsed, list) else [parsed])
        except Exception:
            try:
                return pd.read_csv(io.StringIO(text))
            except Exception:
                return pd.DataFrame({"text": [clean_boilerplate(text)]})
    if pasted_format == "JSON":
        parsed = json.loads(text)
        return pd.json_normalize(parsed if isinstance(parsed, list) else [parsed])
    if pasted_format == "CSV":
        return pd.read_csv(io.StringIO(text))
    return pd.DataFrame({"text": [clean_boilerplate(text)]})


# ─────────────────────────────────────────────
# DATASET TYPE DETECTION
# ─────────────────────────────────────────────

def detect_dataset_type(df: pd.DataFrame) -> str:
    if df.empty:
        return "No dataset"
    cols = [str(c).lower() for c in df.columns]
    if "language" in cols and ("content" in cols or "code" in cols):
        return "Code Dataset"
    if "question" in cols and "answer" in cols:
        return "QA Dataset"
    if "instruction" in cols and ("output" in cols or "input" in cols):
        return "Instruction Dataset"
    if "prompt" in cols and "response" in cols:
        return "Chat Dataset"
    if "text" in cols and "label" in cols:
        return "Classification Dataset"
    if "chunk_id" in cols:
        return "RAG Dataset"
    if "text" in cols or "content" in cols:
        return "RAG Dataset"
    return "Text Corpus"


# ─────────────────────────────────────────────
# DEDUPLICATION
# ─────────────────────────────────────────────

def remove_duplicates_smart(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df
    non_source_cols = [c for c in df.columns if c != "source_name"]
    return df.drop_duplicates(subset=non_source_cols, keep="first").reset_index(drop=True)

def filter_dataset(df: pd.DataFrame) -> pd.DataFrame:
    df = ensure_dataframe(df).copy()
    if df.empty:
        return df
    df = remove_duplicates_smart(df)
    df = df.dropna(how="all")
    for col in df.columns:
        if df[col].dtype == object:
            df[col] = df[col].astype(str)
            df[col] = df[col].replace({"nan": np.nan, "None": np.nan, "": np.nan})
    text_cols = [c for c in df.columns if c in ("text", "content", "chunk", "instruction", "question", "prompt")]
    if text_cols:
        mask = df[text_cols].fillna("").astype(str).apply(
            lambda row: any(s.strip() for s in row), axis=1
        )
        df = df[mask]
    return df.reset_index(drop=True)


# ─────────────────────────────────────────────
# SOURCE DIVERSITY ENGINE
# ─────────────────────────────────────────────

def calculate_source_diversity(df: pd.DataFrame) -> Dict[str, Any]:
    if df.empty:
        return {"unique_files": 0, "unique_sources": 0, "unique_extensions": 0,
                "unique_languages": 0, "source_diversity_score": 0}
    unique_files = int(df["source_name"].nunique()) if "source_name" in df.columns else 1
    unique_sources = int(df["source_kind"].nunique()) if "source_kind" in df.columns else 1
    unique_extensions = int(df["file_extension"].nunique()) if "file_extension" in df.columns else 1
    unique_languages = int(df["language"].nunique()) if "language" in df.columns else 0
    actual = min(unique_files, 10) + min(unique_sources, 3) + min(unique_extensions, 10) + min(unique_languages, 5)
    diversity_score = min(100, int((actual / 28) * 100))
    return {
        "unique_files": unique_files, "unique_sources": unique_sources,
        "unique_extensions": unique_extensions, "unique_languages": unique_languages,
        "source_diversity_score": diversity_score,
    }


# ─────────────────────────────────────────────
# HF QUALITY SCORE V3
# ─────────────────────────────────────────────

def hf_dataset_score_v3(
    df: pd.DataFrame,
    chunk_count: int = 0,
    diversity_score: int = 0,
    total_tokens: int = 0,
) -> Tuple[int, str]:
    if df.empty:
        return 0, "D"
    score = 100.0

    # Dataset size (20 pts)
    n = len(df)
    if n < 100: score -= 18
    elif n < 500: score -= 10
    elif n < 1000: score -= 5

    # Chunk quality bonus (15 pts)
    if chunk_count > 0:
        cpr = chunk_count / max(n, 1)
        if 2 <= cpr <= 8: score += 8
        elif cpr > 8: score += 4

    # Source diversity (15 pts)
    score += min(15, diversity_score // 7)

    # Token coverage (10 pts)
    if total_tokens > 100000: score += 10
    elif total_tokens > 10000: score += 5

    # Missing values (10 pts)
    missing_pct = df.isna().sum().sum() / max(n * len(df.columns), 1) * 100
    if missing_pct > 10: score -= 8
    elif missing_pct > 5: score -= 4

    # Duplicates (10 pts)
    dup_pct = df.duplicated().sum() / max(n, 1) * 100
    if dup_pct > 10: score -= 8
    elif dup_pct > 5: score -= 4

    # Metadata completeness (10 pts)
    meta_cols = {"source_name", "source_kind", "file_extension", "language", "token_count"}
    present = sum(1 for c in meta_cols if c in df.columns)
    score += present * 2

    # Avg quality score of chunks (10 pts)
    if "quality_score" in df.columns:
        avg_q = df["quality_score"].mean()
        if avg_q >= 80: score += 10
        elif avg_q >= 60: score += 5

    score = max(0, min(100, int(score)))
    return score, quality_grade(score)

def hf_dataset_score_v2(df: pd.DataFrame, chunk_count: int = 0, diversity_score: int = 0) -> Tuple[int, str]:
    return hf_dataset_score_v3(df, chunk_count, diversity_score)

def hf_dataset_score(df: pd.DataFrame) -> Tuple[int, str]:
    return hf_dataset_score_v3(df)


# ─────────────────────────────────────────────
# READINESS ASSESSORS
# ─────────────────────────────────────────────

def assess_training_readiness(df: pd.DataFrame, dataset_type: str) -> str:
    if df.empty: return "Not Recommended"
    diversity_score = calculate_source_diversity(df).get("source_diversity_score", 0)
    total_tokens = int(df["token_count"].sum()) if "token_count" in df.columns else 0
    score, _ = hf_dataset_score_v3(df, chunk_count=len(df), diversity_score=diversity_score, total_tokens=total_tokens)
    if score >= 85: return "Ready"
    if score >= 70: return "Mostly Ready"
    if score >= 50: return "Needs Cleaning"
    return "Not Recommended"

def assess_rag_readiness(df: pd.DataFrame) -> str:
    if df.empty: return "Poor"
    text_col = next((c for c in df.columns if c in ("text", "content", "chunk")), None)
    if not text_col: return "Poor"
    avg_len = df[text_col].fillna("").astype(str).str.len().mean()
    if avg_len >= 500: return "Excellent"
    if avg_len >= 200: return "Good"
    if avg_len >= 50: return "Fair"
    return "Poor"


# ─────────────────────────────────────────────
# MULTI DATASET GENERATION
# ─────────────────────────────────────────────

def create_multi_datasets(df: pd.DataFrame) -> Dict[str, pd.DataFrame]:
    """Generate RAG, QA, Instruction, Code, Text datasets from base df."""
    _NEAR_DUP_CACHE.clear()
    result: Dict[str, pd.DataFrame] = {}

    # RAG
    rag_rows = []
    if "text" in df.columns:
        for _, row in df.iterrows():
            text = str(row.get("text", ""))
            source = str(row.get("source_name", "unknown"))
            fp = str(row.get("file_path", ""))
            ext = str(row.get("file_extension", ""))
            fsize = int(row.get("file_size", 0)) if pd.notna(row.get("file_size")) else 0
            lang = str(row.get("language", "text"))
            rag_rows.extend(list(chunk_text_v2(
                text, source_name=source, file_path=fp,
                file_extension=ext, file_size=fsize, language=lang
            )))
    result["rag"] = pd.DataFrame(rag_rows) if rag_rows else pd.DataFrame()

    # QA
    qa_rows = []
    if "text" in df.columns:
        for _, row in df.iterrows():
            qa_rows.extend(generate_qa_v2(str(row.get("text", "")), str(row.get("source_name", ""))))
    result["qa"] = pd.DataFrame(qa_rows) if qa_rows else pd.DataFrame()

    # Instruction
    inst_rows = []
    if "text" in df.columns:
        for _, row in df.iterrows():
            inst_rows.extend(generate_instructions_v2(str(row.get("text", "")), str(row.get("source_name", ""))))
    result["instruction"] = pd.DataFrame(inst_rows) if inst_rows else pd.DataFrame()

    # Code
    code_rows = []
    if "content" in df.columns and "language" in df.columns:
        for _, row in df.iterrows():
            lang = str(row.get("language", "text"))
            if lang in ("text", "markdown"):
                continue
            code = str(row.get("content", ""))
            source = str(row.get("source_name", "unknown"))
            fp = str(row.get("file_path", ""))
            ext = str(row.get("file_extension", ""))
            fsize = int(row.get("file_size", 0)) if pd.notna(row.get("file_size")) else 0
            code_rows.extend(list(chunk_code_v2(code, lang, source_name=source, file_path=fp, file_extension=ext, file_size=fsize)))
    elif "chunk_id" in df.columns and "content" in df.columns:
        code_rows = df.to_dict("records")
    result["code"] = pd.DataFrame(code_rows) if code_rows else pd.DataFrame()

    # Text corpus (full original)
    result["text"] = df.copy()

    return result


# ─────────────────────────────────────────────
# TRAIN / VAL / TEST SPLIT
# ─────────────────────────────────────────────

def split_dataset(df: pd.DataFrame, seed: int = RANDOM_SEED) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """Deterministic 80/10/10 split."""
    if df.empty:
        return df, pd.DataFrame(), pd.DataFrame()
    shuffled = df.sample(frac=1, random_state=seed).reset_index(drop=True)
    n = len(shuffled)
    t1, t2 = int(n * 0.8), int(n * 0.9)
    return shuffled.iloc[:t1], shuffled.iloc[t1:t2], shuffled.iloc[t2:]


# ─────────────────────────────────────────────
# DATASET CARD (README.md) GENERATION
# ─────────────────────────────────────────────

def generate_dataset_card(report: Dict[str, Any]) -> str:
    now = datetime.datetime.utcnow().strftime("%Y-%m-%d")
    langs = ", ".join(report.get("languages_detected", []) or ["mixed"])
    card = f"""---
language:
- en
tags:
- dataset
- universal-dataset-builder
- rag
- instruction-tuning
size_categories:
- {_size_category(report.get('rows', 0))}
---

# {APP_TITLE} — Generated Dataset

## Dataset Summary

Auto-generated by **Universal Dataset Builder Pro** on `{now}`.

This dataset was produced from `{report.get('unique_files', 'N/A')}` source files across
`{report.get('unique_sources', 'N/A')}` source types and is suitable for:

- Retrieval-Augmented Generation (RAG)
- Instruction Fine-Tuning
- QA Pair Training
- Code Understanding

## Dataset Statistics

| Metric | Value |
|--------|-------|
| Rows | {report.get('rows', 0):,} |
| Chunks | {report.get('chunk_count', 0):,} |
| Total Tokens | {report.get('total_tokens', 0):,} |
| Languages | {langs} |
| Quality Score | {report.get('hf_score', 0)}/100 ({report.get('dataset_grade', 'D')}) |
| Source Diversity | {report.get('source_diversity_score', 0)}/100 |
| Created | {now} |

## Quality Metrics

- **Training Readiness**: {report.get('training_readiness', 'N/A')}
- **RAG Readiness**: {report.get('rag_readiness', 'N/A')}
- **QA Readiness**: {report.get('qa_readiness', 'N/A')}
- **Instruction Readiness**: {report.get('instruction_readiness', 'N/A')}
- **Duplicates**: {report.get('duplicates', 0)}
- **Missing Values**: {report.get('missing_values', 0)}

## Dataset Variants

| Variant | Description |
|---------|-------------|
| `rag` | Text chunks with overlap, source metadata, quality scores |
| `qa` | Question-answer pairs (What/Why/How/When/Where) |
| `instruction` | Instruction-input-output triplets |
| `code` | Code chunks with language, imports, classes, functions |
| `text` | Full text corpus (cleaned, deduplicated) |

## Splits

Each variant is split 80% train / 10% validation / 10% test.

## Usage

```python
from datasets import load_dataset
ds = load_dataset("your-username/your-dataset-name")
```

## License

Depends on source material. Review before publishing.
"""
    return card

def _size_category(n: int) -> str:
    if n < 1000: return "1K<n<10K"
    if n < 10000: return "10K<n<100K"
    if n < 100000: return "100K<n<1M"
    return "1M<n<10M"


# ─────────────────────────────────────────────
# DATASET INFO JSON
# ─────────────────────────────────────────────

def generate_dataset_info(report: Dict[str, Any], multi_datasets: Dict[str, pd.DataFrame]) -> Dict[str, Any]:
    info: Dict[str, Any] = {
        "created_at": datetime.datetime.utcnow().isoformat(),
        "builder": APP_TITLE,
        "rows": report.get("rows", 0),
        "chunks": report.get("chunk_count", 0),
        "total_tokens": report.get("total_tokens", 0),
        "hf_score": report.get("hf_score", 0),
        "dataset_grade": report.get("dataset_grade", "D"),
        "source_diversity_score": report.get("source_diversity_score", 0),
        "languages": list(report.get("languages_detected", [])),
        "sources": {
            "unique_files": report.get("unique_files", 0),
            "unique_sources": report.get("unique_sources", 0),
            "unique_extensions": report.get("unique_extensions", 0),
        },
        "quality_stats": {},
        "token_stats": {},
        "splits": {"train": 0.8, "validation": 0.1, "test": 0.1},
        "variants": {},
    }
    for name, df in multi_datasets.items():
        if not df.empty:
            info["variants"][name] = {"rows": len(df), "columns": list(df.columns)}
            if "quality_score" in df.columns:
                qs = df["quality_score"]
                info["quality_stats"][name] = {
                    "mean": round(float(qs.mean()), 2),
                    "min": int(qs.min()), "max": int(qs.max()),
                }
            if "token_count" in df.columns:
                tc = df["token_count"]
                info["token_stats"][name] = {
                    "total": int(tc.sum()),
                    "mean": round(float(tc.mean()), 2),
                }
    return info


# ─────────────────────────────────────────────
# STREAMING EXPORT (chunked writes)
# ─────────────────────────────────────────────

def _write_csv_streaming(df: pd.DataFrame, path: str, chunksize: int = 5000) -> None:
    first = True
    for i in range(0, len(df), chunksize):
        chunk = df.iloc[i:i + chunksize]
        chunk.to_csv(path, mode="w" if first else "a", index=False, header=first)
        first = False

def _write_jsonl_streaming(df: pd.DataFrame, path: str, chunksize: int = 5000) -> None:
    with open(path, "w", encoding="utf-8") as f:
        for i in range(0, len(df), chunksize):
            chunk = df.iloc[i:i + chunksize]
            for record in chunk.to_dict("records"):
                f.write(json.dumps(record, ensure_ascii=False, default=str) + "\n")

def _write_parquet_streaming(df: pd.DataFrame, path: str, chunksize: int = 10000) -> None:
    if PYARROW_SUPPORT:
        try:
            schema = pa.Schema.from_pandas(df.iloc[:0])
            writer = pq.ParquetWriter(path, schema)
            for i in range(0, len(df), chunksize):
                chunk = df.iloc[i:i + chunksize]
                writer.write_table(pa.Table.from_pandas(chunk, schema=schema))
            writer.close()
            return
        except Exception:
            pass
    df.to_parquet(path, index=False, compression="snappy")

def save_exports_for_dataset(df: pd.DataFrame, name: str) -> Dict[str, Optional[str]]:
    """Save one dataset variant in all formats. Returns {format: path}."""
    if df.empty:
        return {}
    tmpdir = tempfile.mkdtemp(prefix=f"ds_{name}_")
    exports: Dict[str, Optional[str]] = {}

    # Stringify complex columns for CSV/Excel compatibility
    df_safe = df.copy()
    for col in df_safe.columns:
        if df_safe[col].dtype == object:
            pass  # already str
        # convert list/dict columns
    # Ensure all values are JSON-serialisable strings for mixed cols
    for col in df_safe.select_dtypes(include=["object"]).columns:
        df_safe[col] = df_safe[col].apply(
            lambda x: x if isinstance(x, (str, int, float, bool, type(None))) else str(x)
        )

    try:
        p = os.path.join(tmpdir, f"{name}.csv")
        _write_csv_streaming(df_safe, p)
        exports["csv"] = p
    except Exception as e:
        print(f"CSV error [{name}]: {e}")

    try:
        p = os.path.join(tmpdir, f"{name}.jsonl")
        _write_jsonl_streaming(df_safe, p)
        exports["jsonl"] = p
    except Exception as e:
        print(f"JSONL error [{name}]: {e}")

    try:
        p = os.path.join(tmpdir, f"{name}.parquet")
        _write_parquet_streaming(df_safe, p)
        exports["parquet"] = p
    except Exception as e:
        print(f"Parquet error [{name}]: {e}")

    try:
        p = os.path.join(tmpdir, f"{name}.csv.gz")
        df_safe.to_csv(p, index=False, compression="gzip")
        exports["csv_gz"] = p
    except Exception as e:
        print(f"CSV.GZ error [{name}]: {e}")

    try:
        p = os.path.join(tmpdir, f"{name}.jsonl.gz")
        import gzip as gzmod
        with gzmod.open(p, "wt", encoding="utf-8") as f:
            for rec in df_safe.to_dict("records"):
                f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")
        exports["jsonl_gz"] = p
    except Exception as e:
        print(f"JSONL.GZ error [{name}]: {e}")

    try:
        p = os.path.join(tmpdir, f"{name}.parquet.gz")
        df_safe.to_parquet(p, index=False, compression="gzip")
        exports["parquet_gz"] = p
    except Exception as e:
        print(f"Parquet.GZ error [{name}]: {e}")

    try:
        if len(df_safe) < 100000:
            p = os.path.join(tmpdir, f"{name}.xlsx")
            df_safe.to_excel(p, index=False)
            exports["excel"] = p
    except Exception as e:
        print(f"Excel error [{name}]: {e}")

    try:
        p = os.path.join(tmpdir, f"{name}.json")
        df_safe.to_json(p, orient="records", force_ascii=False, indent=2)
        exports["json"] = p
    except Exception as e:
        print(f"JSON error [{name}]: {e}")

    try:
        zip_p = os.path.join(tmpdir, f"{name}_bundle.zip")
        with zipfile.ZipFile(zip_p, "w", zipfile.ZIP_DEFLATED) as zf:
            for fmt, fp in exports.items():
                if fp and os.path.exists(fp):
                    zf.write(fp, arcname=os.path.basename(fp))
        exports["zip"] = zip_p
    except Exception as e:
        print(f"ZIP error [{name}]: {e}")

    return exports


def save_multi_exports(
    multi_datasets: Dict[str, pd.DataFrame],
    report: Optional[Dict[str, Any]] = None,
    dataset_info: Optional[Dict[str, Any]] = None,
) -> Dict[str, Dict[str, Optional[str]]]:
    """Save all dataset variants + master ZIP. Returns nested {name: {format: path}}."""
    exports: Dict[str, Dict[str, Optional[str]]] = {}

    for name, df in multi_datasets.items():
        if not df.empty:
            exports[name] = save_exports_for_dataset(df, name)
            # Generate train/val/test splits and save
            train_df, val_df, test_df = split_dataset(df)
            split_tmpdir = tempfile.mkdtemp(prefix=f"ds_{name}_splits_")
            try:
                for split_name, split_df in [("train", train_df), ("validation", val_df), ("test", test_df)]:
                    if not split_df.empty:
                        p = os.path.join(split_tmpdir, f"{name}_{split_name}.parquet")
                        _write_parquet_streaming(split_df, p)
                        exports[name][split_name] = p
            except Exception as e:
                print(f"Split error [{name}]: {e}")

    # Master ZIP
    try:
        master_tmpdir = tempfile.mkdtemp(prefix="ds_master_")
        master_zip_path = os.path.join(master_tmpdir, "all_datasets.zip")
        with zipfile.ZipFile(master_zip_path, "w", zipfile.ZIP_DEFLATED) as mz:
            for name, fmts in exports.items():
                for fmt, fp in fmts.items():
                    if fp and os.path.exists(fp):
                        mz.write(fp, arcname=f"{name}/{os.path.basename(fp)}")
            # README
            if report:
                readme = generate_dataset_card(report)
                mz.writestr("README.md", readme)
            # dataset_info.json
            if dataset_info:
                mz.writestr(
                    "dataset_info.json",
                    json.dumps(dataset_info, indent=2, default=str)
                )
        exports["master"] = {"zip": master_zip_path}
    except Exception as e:
        print(f"Master ZIP error: {e}")

    return exports


# ─────────────────────────────────────────────
# FULL PTF PIPELINE
# ─────────────────────────────────────────────

def process_dataset_ptf(
    files, website_urls: str, pasted_text: str, pasted_format: str
) -> Tuple[pd.DataFrame, Dict[str, pd.DataFrame], Dict[str, Any], str]:
    """
    Perceive → Transform → Filter pipeline.
    Returns (raw_df, multi_datasets, report, log_msg)
    """
    _NEAR_DUP_CACHE.clear()
    frames: List[pd.DataFrame] = []
    messages: List[str] = []
    all_diagnostics: List[Dict[str, Any]] = []
    job_stats: Dict[str, Any] = {
        "files_processed": 0, "files_skipped": 0,
        "languages": set(), "errors": [],
    }

    if files:
        for file_obj in files:
            try:
                df, diag = load_any_file(file_obj)
                all_diagnostics.append(diag)
                job_stats["files_processed"] += 1
                if "language" in df.columns:
                    job_stats["languages"].update(df["language"].dropna().unique().tolist())
                if diag["status"] in ("success", "warning"):
                    frames.append(df)
                    sym = "✓" if diag["status"] == "success" else "⚠"
                    messages.append(f"{sym} {diag['filename']} ({diag['detected_type']}) — {diag['message']}")
                else:
                    job_stats["files_skipped"] += 1
                    messages.append(f"✗ {diag['filename']} — {diag['message']}")
            except Exception as e:
                fname = os.path.basename(file_obj.name) if hasattr(file_obj, "name") else "unknown"
                messages.append(f"✗ {fname} — {str(e)[:60]}")
                job_stats["files_skipped"] += 1

    if website_urls and website_urls.strip():
        for url in website_urls.splitlines():
            url = url.strip()
            if not url:
                continue
            try:
                df = load_website(url)
                df["source_name"] = url
                df["source_kind"] = "website"
                df["file_extension"] = "web"
                df["file_size"] = 0
                frames.append(df)
                messages.append(f"✓ Website ({url[:60]}) — fetched")
                all_diagnostics.append({
                    "filename": url, "extension": "web",
                    "detected_type": "Website", "status": "success", "message": "fetched"
                })
                job_stats["files_processed"] += 1
            except Exception as e:
                messages.append(f"✗ Website ({url[:60]}) — {str(e)[:40]}")

    pasted_df = parse_pasted_content(pasted_text, pasted_format)
    if not pasted_df.empty:
        pasted_df["source_name"] = "pasted_input"
        pasted_df["source_kind"] = "pasted"
        pasted_df["file_extension"] = "text"
        pasted_df["file_size"] = 0
        frames.append(pasted_df)
        messages.append("✓ Pasted content — loaded")

    if not frames:
        return pd.DataFrame(), {}, {}, "\n".join(messages) or "No data loaded."

    merged_df = pd.concat(frames, ignore_index=True, sort=False)
    merged_df.columns = [normalize_whitespace(str(c)) for c in merged_df.columns]
    filtered_df = filter_dataset(merged_df)

    diversity_metrics = calculate_source_diversity(filtered_df)

    total_tokens = 0
    if "token_count" in filtered_df.columns:
        total_tokens = int(filtered_df["token_count"].sum())
    elif "text" in filtered_df.columns:
        total_tokens = int(filtered_df["text"].fillna("").astype(str).apply(count_tokens).sum())

    multi_datasets = create_multi_datasets(filtered_df)
    chunk_count = sum(len(v) for v in multi_datasets.values() if not v.empty)

    score, grade = hf_dataset_score_v3(
        filtered_df, chunk_count, diversity_metrics["source_diversity_score"], total_tokens
    )

    langs_detected = sorted(
        list(job_stats["languages"]) or
        (list(filtered_df["language"].dropna().unique()) if "language" in filtered_df.columns else [])
    )

    report: Dict[str, Any] = {
        "rows": int(len(filtered_df)),
        "columns": int(len(filtered_df.columns)),
        "chunk_count": chunk_count,
        "dataset_type": detect_dataset_type(filtered_df),
        "hf_score": int(score),
        "dataset_grade": grade,
        "training_readiness": assess_training_readiness(filtered_df, detect_dataset_type(filtered_df)),
        "rag_readiness": assess_rag_readiness(filtered_df),
        "qa_readiness": "Good" if len(multi_datasets.get("qa", pd.DataFrame())) > 10 else "Fair",
        "instruction_readiness": "Good" if len(multi_datasets.get("instruction", pd.DataFrame())) > 10 else "Fair",
        "duplicates": int(filtered_df.duplicated().sum()) if not filtered_df.empty else 0,
        "missing_values": int(filtered_df.isna().sum().sum()) if not filtered_df.empty else 0,
        "total_tokens": total_tokens,
        "languages_detected": langs_detected,
        "files_processed": job_stats["files_processed"],
        "files_skipped": job_stats["files_skipped"],
        **diversity_metrics,
    }

    diag_lines = "\n".join(
        f"  • {d['filename']}: {d['detected_type']} [{d['status'].upper()}]"
        for d in all_diagnostics
    )
    error_lines = ("\n⚠ Errors:\n" + "\n".join(f"  • {e}" for e in job_stats["errors"][:5])) if job_stats["errors"] else ""

    log_msg = (
        f"📋 Upload Diagnostics:\n{diag_lines}"
        f"\n\n📊 Detected Type: {report['dataset_type']}"
        f"\n✨ Quality Score: {score}/100 ({grade})"
        f"\n📈 Chunks: {chunk_count} | Tokens: {total_tokens:,}"
        f"\n🌍 Source Diversity: {diversity_metrics['source_diversity_score']}/100"
        f"\n🗂 Files: {job_stats['files_processed']} processed, {job_stats['files_skipped']} skipped"
        f"\n🌐 Languages: {', '.join(langs_detected[:10]) or 'N/A'}"
        + error_lines
        + (f"\n\nℹ️ Merged {len(frames)} sources into {len(multi_datasets)} dataset types" if len(frames) > 1 else "")
    )

    return filtered_df, multi_datasets, report, log_msg


# ─────────────────────────────────────────────
# PUSH TO HUGGING FACE
# ─────────────────────────────────────────────

def push_to_hugging_face(
    datasets_state_json: str, dataset_type: str, repo_name: str, private_repo: bool
) -> str:
    try:
        datasets_dict = json.loads(datasets_state_json) if datasets_state_json else {}
    except Exception:
        return "Error: Invalid datasets state."

    key = dataset_type.strip().lower()
    df_json = datasets_dict.get(key, "")
    if not df_json:
        return f"Error: No {dataset_type} dataset to upload."

    try:
        df = pd.read_json(io.StringIO(df_json), orient="split")
    except Exception as e:
        return f"Error parsing {dataset_type}: {str(e)[:120]}"

    if df.empty:
        return f"Error: {dataset_type} dataset is empty."

    token = os.getenv("HF_TOKEN")
    if not token:
        return "Error: HF_TOKEN not set in Space secrets."

    if not repo_name or "/" not in repo_name:
        return "Error: Use format 'username/dataset-name'"

    try:
        api = HfApi(token=token)
        api.create_repo(repo_id=repo_name, repo_type="dataset", private=private_repo, exist_ok=True)
        dataset = Dataset.from_pandas(df, preserve_index=False)
        dataset.push_to_hub(repo_name, private=private_repo, token=token)
        return f"✓ Uploaded {dataset_type} → https://huggingface.co/datasets/{repo_name}"
    except Exception as e:
        return f"Error: {str(e)[:200]}"


# ─────────────────────────────────────────────
# GRADIO UI
# ─────────────────────────────────────────────

with gr.Blocks(
    title=APP_TITLE,
    theme=gr.themes.Soft(),
    css="body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; } .container { max-width: 1200px; }",
) as demo:

    dataset_state = gr.State("")
    report_state = gr.State("")

    gr.Markdown(
        """
        # 🎨 Universal Dataset Builder Pro

        **Turn any knowledge source into a Hugging Face-grade AI dataset.**

        Upload files, paste URLs, or paste raw data. Automatically detect, chunk, score, and export.
        Supports: PDF · DOCX · CSV · JSON · JSONL · Parquet · Code · ZIP/TAR archives · Websites
        """
    )

    # ── TAB 1: UPLOAD SOURCES ──────────────────────────────────────────────
    with gr.Tab("📥 Upload Sources"):
        gr.Markdown("**Upload files, paste URLs, or paste raw content**")

        with gr.Row():
            with gr.Column():
                files_input = gr.File(
                    label="📎 Upload Files",
                    file_count="multiple",
                    file_types=[
                        ".csv", ".json", ".jsonl", ".parquet",
                        ".xlsx", ".xls",
                        ".pdf", ".docx", ".txt", ".md", ".rst",
                        ".py", ".ipynb", ".js", ".jsx", ".ts", ".tsx",
                        ".c", ".cpp", ".h", ".hpp", ".rs", ".go", ".java",
                        ".html", ".css", ".xml",
                        ".yaml", ".yml", ".sql", ".sh",
                        ".zip", ".tar", ".gz", ".tgz",
                    ]
                )

        with gr.Row():
            with gr.Column():
                urls_input = gr.Textbox(
                    label="🌐 Website URLs (one per line)",
                    lines=4,
                    placeholder="https://example.com\nhttps://docs.example.com/guide"
                )

        with gr.Row():
            with gr.Column():
                pasted_format = gr.Radio(
                    ["Auto", "Text", "CSV", "JSON"],
                    value="Auto",
                    label="📋 Pasted Content Format"
                )
            with gr.Column():
                pasted_input = gr.Textbox(
                    label="Paste Raw Data",
                    lines=6,
                    placeholder="Paste CSV, JSON, code, or plain text here..."
                )

        generate_btn = gr.Button("✨ Generate Dataset", variant="primary", size="lg")

        with gr.Row():
            with gr.Column():
                ingest_log = gr.Textbox(
                    label="📊 Processing Log",
                    lines=8,
                    interactive=False
                )

    # ── TAB 2: DATASET STUDIO ─────────────────────────────────────────────
    with gr.Tab("🎬 Dataset Studio"):
        gr.Markdown("**Multi-dataset generation, quality analysis & preview**")

        with gr.Row():
            with gr.Column(scale=2):
                gr.Markdown("### Dataset Type")
                dataset_selector = gr.Radio(
                    ["RAG", "QA", "Instruction", "Code", "Text"],
                    value="RAG",
                    label="View Dataset Type"
                )
            with gr.Column(scale=1):
                gr.Markdown("### Quality Metrics")
                dataset_type_display = gr.Textbox(label="Type", interactive=False)
                dataset_grade_display = gr.Textbox(label="Grade", interactive=False)
                hf_score_display = gr.Textbox(label="HF Score", interactive=False)

        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("### Dataset Readiness")
                training_readiness_display = gr.Textbox(label="Training", interactive=False)
                rag_readiness_display = gr.Textbox(label="RAG", interactive=False)
                qa_readiness_display = gr.Textbox(label="QA", interactive=False)
                instruction_readiness_display = gr.Textbox(label="Instruction", interactive=False)
            with gr.Column(scale=1):
                gr.Markdown("### Diversity & Coverage")
                rows_display = gr.Textbox(label="Rows", interactive=False)
                cols_display = gr.Textbox(label="Columns", interactive=False)
                chunks_display = gr.Textbox(label="Chunks", interactive=False)
                diversity_display = gr.Textbox(label="Diversity Score", interactive=False)

        with gr.Row():
            with gr.Column():
                gr.Markdown("### Job Statistics")
                files_processed_display = gr.Textbox(label="Files Processed", interactive=False)
                files_skipped_display = gr.Textbox(label="Files Skipped", interactive=False)
                tokens_display = gr.Textbox(label="Total Tokens", interactive=False)
                languages_display = gr.Textbox(label="Languages Detected", interactive=False)

        with gr.Row():
            with gr.Column():
                dataset_summary = gr.JSON(label="📊 Full Quality Report")

        with gr.Row():
            with gr.Column():
                dataset_preview = gr.Dataframe(label="👀 Dataset Preview (first 50 rows)", interactive=False)

    # ── TAB 3: EXPORT ─────────────────────────────────────────────────────
    with gr.Tab("💾 Export"):
        gr.Markdown(
            """
            **Production-grade HF-ready exports in every format**

            Each dataset variant is available in: CSV · JSONL · Parquet · CSV.GZ · JSONL.GZ · Parquet.GZ · Excel · JSON · ZIP Bundle
            Each variant also exports train/validation/test splits as Parquet.
            """
        )

        with gr.Row():
            with gr.Column():
                gr.Markdown("### 🧠 RAG Dataset")
                with gr.Row():
                    rag_csv = gr.File(label="CSV")
                    rag_jsonl = gr.File(label="JSONL")
                    rag_parquet = gr.File(label="Parquet")
                with gr.Row():
                    rag_csv_gz = gr.File(label="CSV.GZ")
                    rag_jsonl_gz = gr.File(label="JSONL.GZ")
                    rag_parquet_gz = gr.File(label="Parquet.GZ")
                with gr.Row():
                    rag_excel = gr.File(label="Excel")
                    rag_json = gr.File(label="JSON")
                    rag_zip = gr.File(label="ZIP Bundle")
                with gr.Row():
                    rag_train = gr.File(label="train.parquet")
                    rag_val = gr.File(label="validation.parquet")
                    rag_test = gr.File(label="test.parquet")

            with gr.Column():
                gr.Markdown("### ❓ QA Dataset")
                with gr.Row():
                    qa_csv = gr.File(label="CSV")
                    qa_jsonl = gr.File(label="JSONL")
                    qa_parquet = gr.File(label="Parquet")
                with gr.Row():
                    qa_csv_gz = gr.File(label="CSV.GZ")
                    qa_jsonl_gz = gr.File(label="JSONL.GZ")
                    qa_parquet_gz = gr.File(label="Parquet.GZ")
                with gr.Row():
                    qa_excel = gr.File(label="Excel")
                    qa_json = gr.File(label="JSON")
                    qa_zip = gr.File(label="ZIP Bundle")
                with gr.Row():
                    qa_train = gr.File(label="train.parquet")
                    qa_val = gr.File(label="validation.parquet")
                    qa_test = gr.File(label="test.parquet")

        with gr.Row():
            with gr.Column():
                gr.Markdown("### 📝 Instruction Dataset")
                with gr.Row():
                    inst_csv = gr.File(label="CSV")
                    inst_jsonl = gr.File(label="JSONL")
                    inst_parquet = gr.File(label="Parquet")
                with gr.Row():
                    inst_csv_gz = gr.File(label="CSV.GZ")
                    inst_jsonl_gz = gr.File(label="JSONL.GZ")
                    inst_parquet_gz = gr.File(label="Parquet.GZ")
                with gr.Row():
                    inst_excel = gr.File(label="Excel")
                    inst_json = gr.File(label="JSON")
                    inst_zip = gr.File(label="ZIP Bundle")
                with gr.Row():
                    inst_train = gr.File(label="train.parquet")
                    inst_val = gr.File(label="validation.parquet")
                    inst_test = gr.File(label="test.parquet")

            with gr.Column():
                gr.Markdown("### 💻 Code Dataset")
                with gr.Row():
                    code_csv = gr.File(label="CSV")
                    code_jsonl = gr.File(label="JSONL")
                    code_parquet = gr.File(label="Parquet")
                with gr.Row():
                    code_csv_gz = gr.File(label="CSV.GZ")
                    code_jsonl_gz = gr.File(label="JSONL.GZ")
                    code_parquet_gz = gr.File(label="Parquet.GZ")
                with gr.Row():
                    code_excel = gr.File(label="Excel")
                    code_json = gr.File(label="JSON")
                    code_zip = gr.File(label="ZIP Bundle")
                with gr.Row():
                    code_train = gr.File(label="train.parquet")
                    code_val = gr.File(label="validation.parquet")
                    code_test = gr.File(label="test.parquet")

        gr.Markdown("---\n### 🎁 Master Export Bundle")
        gr.Markdown("All datasets · all formats · README.md · dataset_info.json")
        with gr.Row():
            master_zip = gr.File(label="all_datasets.zip")

        gr.Markdown("---\n### 🚀 Push to Hugging Face Hub")
        gr.Markdown("Set `HF_TOKEN` in Space secrets. Format: `username/dataset-name`")
        with gr.Row():
            with gr.Column():
                hf_repo = gr.Textbox(label="Dataset Repository", placeholder="username/my-dataset")
            with gr.Column():
                hf_private = gr.Checkbox(label="Private Dataset", value=False)
        hf_push_btn = gr.Button("🚀 Push to Hub", variant="primary")
        hf_status = gr.Textbox(label="Status", interactive=False, lines=3)

    # ── EVENT HANDLERS ────────────────────────────────────────────────────

    def handle_generate(files, urls, pasted_text, pasted_format):
        df, multi_datasets, report, log = process_dataset_ptf(files, urls, pasted_text, pasted_format)

        _empty_file = [None] * 12  # per-variant: 9 formats + 3 splits
        _empty_base = [
            "", "", log,
            "", "", "",
            "", "", "", "",
            "", "", "", "",
            "", "", "", "",
            report, pd.DataFrame(),
        ]
        if df.empty:
            return _empty_base + _empty_file * 4 + [None]

        di = generate_dataset_info(report, multi_datasets)
        exports = save_multi_exports(multi_datasets, report, di)

        def _e(name: str) -> List:
            ex = exports.get(name, {})
            return [
                ex.get("csv"), ex.get("jsonl"), ex.get("parquet"),
                ex.get("csv_gz"), ex.get("jsonl_gz"), ex.get("parquet_gz"),
                ex.get("excel"), ex.get("json"), ex.get("zip"),
                ex.get("train"), ex.get("validation"), ex.get("test"),
            ]

        datasets_json = {
            k: multi_datasets.get(k, pd.DataFrame()).head(50).to_json(orient="split", force_ascii=False)
            for k in ("rag", "qa", "instruction", "code", "text")
        }
        datasets_state_json = json.dumps(datasets_json)

        base = [
            datasets_state_json, json.dumps(report, default=str), log,
            report.get("dataset_type", ""),
            f"{report.get('dataset_grade','D')} ({report.get('hf_score',0)}/100)",
            f"{report.get('hf_score',0)}/100",
            report.get("training_readiness", ""),
            report.get("rag_readiness", ""),
            report.get("qa_readiness", ""),
            report.get("instruction_readiness", ""),
            f"{report.get('rows',0)} rows",
            f"{report.get('columns',0)} cols",
            f"{report.get('chunk_count',0)} chunks",
            f"{report.get('source_diversity_score',0)}/100",
            f"{report.get('files_processed',0)} files",
            f"{report.get('files_skipped',0)} skipped",
            f"{report.get('total_tokens',0):,} tokens",
            ", ".join(report.get("languages_detected", [])[:10]) or "N/A",
            report,
            multi_datasets.get("text", pd.DataFrame()).head(50),
        ]

        return base + _e("rag") + _e("qa") + _e("instruction") + _e("code") + [exports.get("master", {}).get("zip")]

    def handle_hf_push(datasets_state_json, dataset_type, repo_name, private):
        return push_to_hugging_face(datasets_state_json, dataset_type, repo_name, private)

    def switch_dataset_view(dataset_type, datasets_state_json):
        try:
            if not datasets_state_json:
                return pd.DataFrame()
            d = json.loads(datasets_state_json)
            key = str(dataset_type).strip().lower()
            df_json = d.get(key, "{}")
            if df_json and df_json != "{}":
                return pd.read_json(io.StringIO(df_json), orient="split")
        except Exception:
            pass
        return pd.DataFrame()

    # ── WIRE OUTPUTS ──────────────────────────────────────────────────────
    _all_outputs = [
        dataset_state, report_state, ingest_log,
        dataset_type_display, dataset_grade_display, hf_score_display,
        training_readiness_display, rag_readiness_display, qa_readiness_display, instruction_readiness_display,
        rows_display, cols_display, chunks_display, diversity_display,
        files_processed_display, files_skipped_display, tokens_display, languages_display,
        dataset_summary, dataset_preview,
        # RAG (12)
        rag_csv, rag_jsonl, rag_parquet,
        rag_csv_gz, rag_jsonl_gz, rag_parquet_gz,
        rag_excel, rag_json, rag_zip,
        rag_train, rag_val, rag_test,
        # QA (12)
        qa_csv, qa_jsonl, qa_parquet,
        qa_csv_gz, qa_jsonl_gz, qa_parquet_gz,
        qa_excel, qa_json, qa_zip,
        qa_train, qa_val, qa_test,
        # Instruction (12)
        inst_csv, inst_jsonl, inst_parquet,
        inst_csv_gz, inst_jsonl_gz, inst_parquet_gz,
        inst_excel, inst_json, inst_zip,
        inst_train, inst_val, inst_test,
        # Code (12)
        code_csv, code_jsonl, code_parquet,
        code_csv_gz, code_jsonl_gz, code_parquet_gz,
        code_excel, code_json, code_zip,
        code_train, code_val, code_test,
        # Master ZIP (1)
        master_zip,
    ]

    generate_btn.click(
        fn=handle_generate,
        inputs=[files_input, urls_input, pasted_input, pasted_format],
        outputs=_all_outputs,
    )

    dataset_selector.change(
        fn=switch_dataset_view,
        inputs=[dataset_selector, dataset_state],
        outputs=dataset_preview,
    )

    hf_push_btn.click(
        fn=handle_hf_push,
        inputs=[dataset_state, dataset_selector, hf_repo, hf_private],
        outputs=hf_status,
    )


if __name__ == "__main__":
    demo.launch()