src/bigru_t/data/streaming_datasets.py

"""streaming_datasets_v13_9.py — Streaming dataset loader for v13.9.1 training.

Carrega os 9 datasets especificados pelo usuário via streaming (IterableDataset),
um por vez, exaustivamente dentro do limite max_samples_per_dataset.

Datasets:
  1. CEIA-POSITIVO/ultrachat_br_clustred_balanced_v1
  2. Madras1/corpus-ptbr-v2
  3. rhaymison/multmodal_175k_portuguese
  4. TucanoBR/GigaVerbo
  5. nvidia/OpenMathReasoning
  6. MathLLMs/MathVision
  7. nvidia/OpenMathInstruct-2
  8. dominguesm/restore-punctuation-ptbr-dataset
  9. carolina-c4ai/corpus-carolina

Memory-efficient: streaming=True, no full materialization.
Returns ProcessedSample with raw_text only (tokenizer handles encoding).
"""
from __future__ import annotations

import logging
import os
import time
import traceback
from dataclasses import dataclass, field
from typing import Any, Dict, Iterator, List, Optional, Tuple

import torch

logger = logging.getLogger(__name__)

# Datasets V13.9.1 — exatamente os 9 especificados pelo usuário
# V6.5-final: adicionado dominguesm/Canarim-Instruct-PTBR-Dataset
# V6.5-7ds: adicionados adalbertojunior/punctuation-ptbr,
#           iara-project/news-articles-ptbr-dataset, manoela/noticias_ptbr
DEFAULT_DATASETS = [
    "CEIA-POSITIVO/ultrachat_br_clustred_balanced_v1",
    "Madras1/corpus-ptbr-v2",
    "rhaymison/multmodal_175k_portuguese",
    "TucanoBR/GigaVerbo",
    "nvidia/OpenMathReasoning",
    "MathLLMs/MathVision",
    "nvidia/OpenMathInstruct-2",
    "dominguesm/restore-punctuation-ptbr-dataset",
    "carolina-c4ai/corpus-carolina",
    "dominguesm/Canarim-Instruct-PTBR-Dataset",
    "adalbertojunior/punctuation-ptbr",
    "iara-project/news-articles-ptbr-dataset",
    "manoela/noticias_ptbr",
]

# Dataset format descriptors — Lista priorizada de campos de texto
DATASET_FORMATS: Dict[str, Dict[str, Any]] = {
    "CEIA-POSITIVO/ultrachat_br_clustred_balanced_v1": {
        "type": "chat",
        "text_fields": ["conversa", "text", "content", "conversation", "messages"],
        "label_fields": [],
        "split": "train",
        "config": None,
    },
    "Madras1/corpus-ptbr-v2": {
        "type": "text",
        "text_fields": ["text", "content", "document", "body"],
        "label_fields": [],
        "split": "train",
        "config": None,
    },
    "rhaymison/multmodal_175k_portuguese": {
        "type": "multimodal",
        "text_fields": ["description", "text", "question", "prompt", "instruction", "input"],
        "image_fields": ["image", "image_url"],
        "label_fields": ["answer", "response", "output"],
        "split": "train",
        "config": None,
    },
    "TucanoBR/GigaVerbo": {
        "type": "text",
        "text_fields": ["text", "content", "document", "body"],
        "label_fields": [],
        "split": "train",
        "config": None,
    },
    "nvidia/OpenMathReasoning": {
        "type": "math",
        "text_fields": ["problem", "question", "input", "prompt"],
        "label_fields": ["solution", "answer", "output", "response"],
        # BUG FIX: OpenMathReasoning has splits ['cot', 'tir', 'genselect', 'additional_problems']
        "split": "cot",  # Use 'cot' (chain-of-thought) as primary split
        "config": None,
    },
    "MathLLMs/MathVision": {
        "type": "math_multimodal",
        "text_fields": ["question", "problem", "text", "query"],
        "image_fields": ["image", "image_url"],
        "label_fields": ["answer", "solution", "response"],
        # BUG FIX: MathVision has only 'test' and 'testmini' splits (no train)
        "split": "test",
        "config": None,
    },
    "nvidia/OpenMathInstruct-2": {
        "type": "math",
        "text_fields": ["problem", "question", "input", "prompt"],
        # V3 FIX: campos reais do dataset são generated_solution e expected_answer
        "label_fields": ["generated_solution", "expected_answer", "solution", "answer", "output", "response"],
        "split": "train",
        "config": None,
    },
    "dominguesm/restore-punctuation-ptbr-dataset": {
        "type": "punctuation",
        "text_fields": ["text", "original", "unpunctuated", "input"],
        "label_fields": ["punctuated", "restored", "target", "output"],
        "split": "train",
        "config": None,
    },
    "carolina-c4ai/corpus-carolina": {
        "type": "text",
        "text_fields": ["text", "content", "document", "body", "xml"],
        "label_fields": [],
        "split": "train",
        "config": None,
        # V13.9.2-carolina: dataset original usa script Python (não suportado em
        # datasets 5.0+). Workaround: stream_xml_gz direto via lxml.iterparse,
        # sem usar o script Python. Implementado em _stream_carolina_direct().
    },
    # ── V13.9.2-finetune-v2: novos datasets PT-BR ───────────────────────────
    # Adicionados para fine-tuning contínuo do modelo v13.9.2 (preserva
    # arquitetura e parâmetros). Nenhuma mudança estrutural — apenas novos
    # descritores de formato para que stream_dataset() reconheça os datasets.
    #
    # V13.9.2-finetune-v3: padronização para o formato unificado
    # "### Instruction:\n...\n\n### Response:\n..." conforme requisitado pelo
    # usuário. Apenas os datasets com estrutura instruction/response (orion,
    # cnmoro) usam o template; strak2005/bratao continuam como texto puro
    # (corpus sem estrutura de instrução). Nenhuma mudança nos 9 datasets
    # base — retrocompatibilidade total.
    "orion-research/translations-en_US-pt_BR": {
        "type": "translation",
        # string = EN (instruction), string_translation = PT-BR (response).
        # V13.9.2-finetune-v3: separa instruction/response para o template
        # unificado em vez de concatenar como texto único.
        "text_fields": ["string"],
        "label_fields": ["string_translation"],
        "split": "train",
        "config": None,
        "format_template": "instruction_response",
        "instruction_prefix": "Translate the following text to Portuguese:",
    },
    "cnmoro/Instruct-PTBR-10M": {
        "type": "instruct",
        # NB: campos em MAIÚSCULAS no schema do dataset
        # V13.9.2-finetune-v3: INSTRUCTION -> instruction, RESPONSE -> response
        # via template unificado.
        "text_fields": ["INSTRUCTION"],
        "label_fields": ["RESPONSE"],
        "split": "train",
        "config": None,
        "format_template": "instruction_response",
        # V13.9.2-finetune-v2: dataset é um único parquet de 10 GB — streaming
        # via datasets.load_dataset() é extremamente lento para materializar
        # a primeira amostra. Fallback via HF datasets-server rows API (HTTP),
        # que retorna batches de 100 rows rapidamente. Implementado em
        # _stream_via_rows_api().
        "loader": "rows_api",
    },
    "strak2005/corpus-ptbr-v1": {
        "type": "text",
        "text_fields": ["text", "content"],
        "label_fields": [],
        "split": "train",
        "config": None,
        # Fallback: se strak2005 cair, tenta bratao (mirror idêntico)
        "fallback": "bratao/corpus-ptbr-v1",
        # V13.9.2-finetune-v2: dataset tem 18 shards parquet de ~1GB cada.
        # Streaming via datasets.load_dataset() não materializa primeira
        # amostra em tempo aceitável. Fallback: baixa 1 shard localmente e
        # itera com pyarrow. Implementado em _stream_via_parquet_first_shard().
        "loader": "parquet_first_shard",
    },
    "bratao/corpus-ptbr-v1": {
        "type": "text",
        "text_fields": ["text", "content"],
        "label_fields": [],
        "split": "train",
        "config": None,
        "loader": "parquet_first_shard",
    },
    # V6.7 — BrunoN-Dev/corpus-ptbr-v1 (user requirement: FASE2 com punição
    # ativa para 'BrunoN-Dev/corpus-ptbr-v1'). Dataset usa 18 parquet shards
    # grandes (~1GB total), então usamos parquet_first_shard loader (igual ao
    # bratao/corpus-ptbr-v1).
    "BrunoN-Dev/corpus-ptbr-v1": {
        "type": "text",
        "text_fields": ["text", "content"],
        "label_fields": [],
        "split": "train",
        "config": None,
        "loader": "parquet_first_shard",
    },
    # ── V6.5: novos datasets para esgotar ────────────────────────────────
    # User requirement: "esgotar 'dominguesm/restore-punctuation-ptbr-dataset'
    # e 'carolina-c4ai/corpus-carolina' e 'nvidia/OpenMathInstruct-2' e
    # 'nvidia/OpenMathReasoning' e 'CEIA-POSITIVO/ultrachat_br_clustred_balanced_v1'
    # e 'Dexavator/English-PTBR'"
    #
    # Dexavator/English-PTBR: dataset de tradução EN->PT-BR.
    # Estrutura típica: {"english": "...", "portuguese": "..."} ou
    # {"en": "...", "pt": "..."}.
    "Dexavator/English-PTBR": {
        "type": "translation",
        "text_fields": ["english", "en", "text", "source"],
        "label_fields": ["portuguese", "pt", "target", "translation"],
        "split": "train",
        "config": None,
        "format_template": "instruction_response",
        "instruction_prefix": "Translate the following text to Portuguese:",
    },
    # ── V6.5-final: Canarim-Instruct-PTBR-Dataset ────────────────────────
    # User requirement: "esgotar 'dominguesm/Canarim-Instruct-PTBR-Dataset'"
    # Canarim-Instruct: dataset PT-BR de instruções (~430k samples).
    # Estrutura típica: {"instruction": "...", "input": "...", "output": "..."}
    # ou {"text": "...", "conversation": [{"role": "user", "content": "..."}, ...]}
    "dominguesm/Canarim-Instruct-PTBR-Dataset": {
        "type": "instruct",
        "text_fields": ["instruction", "input", "text", "prompt", "question"],
        "label_fields": ["output", "response", "answer"],
        "split": "train",
        "config": None,
        "format_template": "instruction_response",
        "instruction_prefix": "Instrução:",
    },
    # ── V6.5-7ds: novos 3 datasets PT-BR ─────────────────────────────────
    # User requirement (latest): "streaming até esgotar nesta sequência
    # 'dominguesm/restore-punctuation-ptbr-dataset' e
    # 'carolina-c4ai/corpus-carolina' e
    # 'CEIA-POSITIVO/ultrachat_br_clustred_balanced_v1' e
    # 'dominguesm/Canarim-Instruct-PTBR-Dataset' e
    # 'adalbertojunior/punctuation-ptbr' e
    # 'iara-project/news-articles-ptbr-dataset' e
    # 'manoela/noticias_ptbr'"
    #
    # adalbertojunior/punctuation-ptbr:
    #   Dataset de restauração de pontuação PT-BR. Schema:
    #     {id: str, tokens: list[str], pos_tags, chunk_tags, ner_tags}
    #   O campo "tokens" é uma lista de palavras (sem pontuação). Precisamos
    #   fazer join com espaços para obter texto bruto. Como o dataset usa
    #   script Python (datasets 5.0+ não suporta), usamos rows_api loader.
    #   Splits: train/validation/test com config "punctuation-ptbr".
    "adalbertojunior/punctuation-ptbr": {
        "type": "punctuation_tokens",
        "text_fields": ["tokens"],  # list[str] — join com espaços
        "label_fields": [],
        "split": "train",
        "config": "punctuation-ptbr",  # necessário para rows_api
        "loader": "rows_api",
        "join_tokens": True,  # flag especial para _normalize_sample
    },
    # iara-project/news-articles-ptbr-dataset:
    #   Dataset de notícias PT-BR da Folha de S.Paulo (via IARA project).
    #   Schema: {title, text, date, category, category_natural_language, link}
    #   Streaming normal via parquet — primeira amostra materializa rápido.
    #   "text" é o corpo da notícia, "title" é o título. Concatenamos ambos
    #   via format_template="title_text_concat" para enriquecer contexto.
    #   O campo "category" é usado como label (categoria da notícia).
    "iara-project/news-articles-ptbr-dataset": {
        "type": "news",
        # text_fields em ordem de prioridade: "text" (corpo) primeiro
        "text_fields": ["text", "title", "content", "body"],
        "label_fields": ["category", "category_natural_language", "title"],
        "split": "train",
        "config": None,
        # V6.5-7ds: title_text_concat usa title como label (prefixo)
        # e text como corpo. Aqui label_fields pega "category" primeiro
        # para dar contexto temático.
        "format_template": "title_text_concat",
    },
    # manoela/noticias_ptbr:
    #   Mirror do iara-project/news-articles-ptbr-dataset (mesmo parquet,
    #   mesmo schema). Streaming normal via parquet.
    "manoela/noticias_ptbr": {
        "type": "news",
        "text_fields": ["text", "title", "content", "body"],
        "label_fields": ["category", "category_natural_language", "title"],
        "split": "train",
        "config": None,
        "format_template": "title_text_concat",
    },
}


# Lista de arquivos XML.gz do Carolina que tentaremos baixar (ordem: menores
# taxonomias primeiro para minimizar banda e RAM). Selecionamos 2 arquivos
# por taxonomia para garantir diversidade sem explodir a memória.
CAROLINA_XML_FILES = [
    # taxonomia "datasets_and_other_corpora" (DAT = pt-BR notícias)
    "corpus/datasets_and_other_corpora/pt-BR/DATa.xml.gz",
    "corpus/datasets_and_other_corpora/pt-BR/DATb.xml.gz",
    # taxonomia "wik" (wikis)
    "corpus/wik/WIKa.xml.gz",
    "corpus/wik/WIKb.xml.gz",
    # taxonomia "uni" (university_domains)
    "corpus/uni/UNIa.xml.gz",
    "corpus/uni/UNIb.xml.gz",
    # taxonomia "pub" (public_domain_works)
    "corpus/pub/PUBa.xml.gz",
    "corpus/pub/PUBb.xml.gz",
]


def _stream_carolina_direct(
    max_samples: int,
    hf_token: Optional[str] = None,
    timeout_per_file: int = 60,
) -> Iterator[Dict[str, Any]]:
    """Workaround V13.9.2-carolina: carrega Carolina sem usar o script Python.

    Baixa arquivos XML.gz diretamente do repo e faz parse streaming com
    lxml.etree.iterparse (compatível com datasets 5.0+). Cada <TEI> vira um
    documento de texto (concatenação dos <p> dentro de <body>).

    Args:
        max_samples: número máximo de amostras a retornar
        hf_token: token HF opcional (dataset é público, mas token evita rate limit)
        timeout_per_file: tempo máximo (segundos) por arquivo XML

    Yields:
        dict com chave "text" (compatível com _normalize_sample)
    """
    from huggingface_hub import hf_hub_download
    import gzip
    from lxml import etree

    TEI_NS = "{http://www.tei-c.org/ns/1.0}"
    BODY_P_TAG = f".//{TEI_NS}body/{TEI_NS}p"

    count = 0
    for xml_path in CAROLINA_XML_FILES:
        if count >= max_samples:
            return
        try:
            local_path = hf_hub_download(
                repo_id="carolina-c4ai/corpus-carolina",
                filename=xml_path,
                repo_type="dataset",
                token=hf_token,
            )
        except Exception as e:
            logger.warning(f"Carolina: não foi possível baixar {xml_path}: {e}")
            continue

        try:
            with gzip.open(local_path, "rb") as gz:
                # iterparse streaming: não carrega árvore inteira na memória
                for _, tei in etree.iterparse(
                    gz, huge_tree=True, encoding="utf-8", tag=f"{TEI_NS}TEI"
                ):
                    if count >= max_samples:
                        tei.clear()
                        return
                    # Extrai texto dos <p> dentro de <body>
                    parts = []
                    for p in tei.findall(BODY_P_TAG):
                        if p.text:
                            parts.append(p.text)
                    text = " ".join(parts).strip()
                    if text and len(text) >= 10:
                        yield {"text": text, "meta": ""}
                    tei.clear()  # libera memória da árvore TEI
                    count += 1
        except Exception as e:
            logger.warning(f"Carolina: erro processando {xml_path}: {e}")
            continue

    logger.info(f"Carolina: streamou {count} amostras no total")


# ═══════════════════════════════════════════════════════════════════════════
# V13.9.2-finetune-v2: Loaders alternativos para datasets parquet grandes
# onde datasets.load_dataset(streaming=True) é inviável (primeira amostra
# demora minutos para materializar).
# ═══════════════════════════════════════════════════════════════════════════

def _stream_via_rows_api(
    dataset_name: str,
    max_samples: int,
    hf_token: Optional[str] = None,
    batch_size: int = 100,
    max_offset: int = 100_000,
    config: Optional[str] = None,
    split: str = "train",
) -> Iterator[Dict[str, Any]]:
    """Carrega amostras via HF datasets-server /rows API (HTTP).

    Usado para datasets cujo parquet é muito grande para streaming eficiente
    (ex: cnmoro/Instruct-PTBR-10M com 10 GB em arquivo único) OU datasets
    que usam script Python como loader (não suportado em datasets 5.0+,
    ex: adalbertojunior/punctuation-ptbr). A rows API retorna batches de
    até 100 rows via HTTP, sem precisar baixar o parquet.

    Args:
        dataset_name: repo_id do dataset (ex: "cnmoro/Instruct-PTBR-10M")
        max_samples: nº máximo de amostras a retornar
        hf_token: token HF (opcional — dataset público)
        batch_size: nº de rows por requisição (máx 100)
        max_offset: offset máximo a tentar (sai do loop se excedido)
        config: nome da config do dataset (default: "default"). NECESSÁRIO
                para datasets cuja config não é "default" (ex:
                adalbertojunior/punctuation-ptbr usa "punctuation-ptbr").
        split: nome do split (default: "train").

    Yields:
        dict com campos do dataset (compatível com _normalize_sample)
    """
    import json as _json
    import urllib.request
    import urllib.parse

    base_url = "https://datasets-server.huggingface.co/rows"
    config_name = config or "default"
    count = 0
    offset = 0

    while count < max_samples and offset < max_offset:
        params = urllib.parse.urlencode({
            "dataset": dataset_name,
            "config": config_name,
            "split": split,
            "offset": offset,
            "length": min(batch_size, max_samples - count),
        })
        url = f"{base_url}?{params}"
        try:
            req = urllib.request.Request(url)
            if hf_token:
                req.add_header("Authorization", f"Bearer {hf_token}")
            with urllib.request.urlopen(req, timeout=30) as r:
                data = _json.loads(r.read())
        except Exception as e:
            logger.warning(f"rows_api {dataset_name} config={config_name} offset={offset}: {e}")
            break

        rows = data.get("rows", [])
        if not rows:
            break

        for row in rows:
            if count >= max_samples:
                return
            row_data = row.get("row", {})
            # Remove campos internos como __index_level_0__
            clean = {k: v for k, v in row_data.items()
                     if not k.startswith("__")}
            yield clean
            count += 1

        offset += len(rows)
        if len(rows) < batch_size:
            break  # chegou ao fim do dataset

    logger.info(f"rows_api {dataset_name} config={config_name}: streamou {count} amostras")


def _stream_via_parquet_first_shard(
    dataset_name: str,
    max_samples: int,
    hf_token: Optional[str] = None,
) -> Iterator[Dict[str, Any]]:
    """Baixa primeiro shard parquet do dataset e itera localmente com pyarrow.

    Usado para datasets com múltiplos shards parquet grandes (ex:
    strak2005/corpus-ptbr-v1 com 18 shards de ~1GB). Streaming via
    datasets.load_dataset() não materializa primeira amostra em tempo
    aceitável. Esta função baixa o primeiro shard (~26s para 1GB) e itera
    localmente — rápido após download.

    Args:
        dataset_name: repo_id do dataset
        max_samples: nº máximo de amostras
        hf_token: token HF opcional

    Yields:
        dict com campos do dataset
    """
    from huggingface_hub import HfApi, hf_hub_download
    import pyarrow.parquet as pq

    api = HfApi(token=hf_token)

    # Lista arquivos parquet no repo
    try:
        info = api.dataset_info(dataset_name)
    except Exception as e:
        logger.warning(f"parquet_shard {dataset_name}: dataset_info failed: {e}")
        return

    parquet_files = sorted([
        s.rfilename for s in info.siblings
        if s.rfilename.endswith(".parquet")
    ])
    if not parquet_files:
        logger.warning(f"parquet_shard {dataset_name}: nenhum .parquet encontrado")
        return

    # Baixa apenas o primeiro shard
    first_shard = parquet_files[0]
    logger.info(f"parquet_shard {dataset_name}: baixando {first_shard}...")
    try:
        local_path = hf_hub_download(
            repo_id=dataset_name,
            filename=first_shard,
            repo_type="dataset",
            token=hf_token,
        )
    except Exception as e:
        logger.warning(f"parquet_shard {dataset_name}: download failed: {e}")
        return

    # Itera com pyarrow.iter_batches — método eficiente que lê batches pequenos
    # do parquet sem materializar o row group inteiro na memória.
    # BUG FIX V13.9.2-finetune-v2: read_row_group() materializa TODAS as ~1M
    # rows de uma vez (lento + ~1GB RAM). iter_batches(batch_size=10) lê só
    # 10 rows por vez (rápido + ~10KB RAM).
    try:
        pf = pq.ParquetFile(local_path)
        count = 0
        # batch_read_size: nº de rows por batch (pequeno = menos RAM)
        for batch in pf.iter_batches(batch_size=100):
            if count >= max_samples:
                break
            cols = batch.schema.names
            # to_pylist() é eficiente para batches pequenos
            pydict = batch.to_pydict()
            n_rows = len(pydict[cols[0]]) if cols else 0
            for i in range(n_rows):
                if count >= max_samples:
                    break
                row = {col: pydict[col][i] for col in cols}
                yield row
                count += 1
    except Exception as e:
        logger.warning(f"parquet_shard {dataset_name}: pyarrow iter failed: {e}")
        return

    logger.info(f"parquet_shard {dataset_name}: streamou {count} amostras do shard {first_shard}")


@dataclass
class ProcessedSample:
    """Uma amostra processada: texto bruto + metadados."""
    raw_text: str
    dataset_name: str
    sample_idx: int
    text_field: str = ""
    label_text: str = ""


def _extract_field(sample: Dict[str, Any], candidate_fields: List[str]) -> Optional[str]:
    """Extrai texto da primeira field disponível no sample.

    Suporta:
    - strings diretas
    - listas de strings (join com espaço — útil para datasets tokenizados
      como adalbertojunior/punctuation-ptbr onde "tokens": ["hoje","a",...])
    - listas de dicts (chat messages — concatena role: content)
    - dicts aninhados (pega text/content/body)
    """
    for field_name in candidate_fields:
        if field_name in sample:
            val = sample[field_name]
            if val is None:
                continue
            if isinstance(val, str):
                if val.strip():
                    return val
            elif isinstance(val, list):
                # V6.5-7ds: se for lista de strings, join com espaços
                # (ex: adalbertojunior/punctuation-ptbr tokens field)
                string_parts = []
                chat_parts = []
                all_strings = True
                for item in val:
                    if isinstance(item, str):
                        if item.strip():
                            string_parts.append(item)
                    elif isinstance(item, (int, float)):
                        # tokens podem vir como ints (token IDs) — skip
                        all_strings = False
                        break
                    elif isinstance(item, dict):
                        all_strings = False
                        role = item.get("role", "")
                        content = item.get("content", "")
                        if isinstance(content, str) and content.strip():
                            chat_parts.append(f"{role}: {content}")
                    else:
                        all_strings = False
                if all_strings and string_parts:
                    joined = " ".join(string_parts).strip()
                    if joined:
                        return joined
                if chat_parts:
                    return "\n".join(chat_parts)
            elif isinstance(val, dict):
                # Pode ser nested
                text = val.get("text") or val.get("content") or val.get("body")
                if isinstance(text, str) and text.strip():
                    return text
    return None


def _normalize_sample(sample: Dict[str, Any], dataset_name: str, idx: int) -> Optional[ProcessedSample]:
    """Normaliza uma amostra bruta em ProcessedSample."""
    fmt = DATASET_FORMATS.get(dataset_name, {})
    text_fields = fmt.get("text_fields", ["text"])
    label_fields = fmt.get("label_fields", [])

    text = _extract_field(sample, text_fields)
    if not text or len(text.strip()) < 10:  # skip very short
        return None

    label = ""
    if label_fields:
        label = _extract_field(sample, label_fields) or ""

    # Se tem label, aplica template apropriado
    template = fmt.get("format_template")
    full_text = text

    # V13.9.2-finetune-v3: para o template instruction_response, exige ambos
    # os lados (instruction + response). Samples sem response são descartados
    # para evitar treinar em texto parcial.
    if template == "instruction_response" and not label:
        return None

    if label:
        # V13.9.2-finetune-v3: format_template unificado "### Instruction:/### Response:"
        # quando o dataset declarar explicitamente. Caso contrário, mantém o
        # comportamento legado (concatenação simples text\nlabel) para os 9
        # datasets base — retrocompatibilidade total.
        if template == "instruction_response":
            instr_prefix = fmt.get("instruction_prefix", "")
            if instr_prefix:
                full_text = (
                    f"### Instruction:\n{instr_prefix}\n{text}\n\n"
                    f"### Response:\n{label}"
                )
            else:
                full_text = (
                    f"### Instruction:\n{text}\n\n"
                    f"### Response:\n{label}"
                )
        elif template == "title_text_concat":
            # V6.5-7ds: para datasets de notícias (iara-project, manoela),
            # o campo "text" (corpo da notícia) é o conteúdo principal e
            # "category" (categoria) é o contexto temático. Formatamos como
            # "TEXTO: <corpo>\nCATEGORIA: <categoria>".
            full_text = f"TEXTO: {text}\nCATEGORIA: {label}"
        else:
            # Comportamento legado (9 datasets base): concatenação simples
            full_text = f"{text}\n{label}"

    return ProcessedSample(
        raw_text=full_text,
        dataset_name=dataset_name,
        sample_idx=idx,
        text_field=text_fields[0] if text_fields else "",
        label_text=label,
    )


def _load_dataset_streaming(
    dataset_name: str,
    split: str = "train",
    config: Optional[str] = None,
    hf_token: Optional[str] = None,
):
    """Carrega dataset em modo streaming.

    BUG FIX V13.9.1: Removido trust_remote_code (deprecated).
    Se o dataset não carregar com split especificado, tenta 'train' como fallback.
    BUG FIX V13.9.2: Adicionado fallback para datasets com script (carolina-c4ai).
    """
    from datasets import load_dataset

    # Lista de splits para tentar, em ordem
    splits_to_try = [split]
    if split != "train":
        splits_to_try.append("train")
    # Adiciona outros splits comuns
    for s in ["test", "validation", "cot", "tir"]:
        if s not in splits_to_try:
            splits_to_try.append(s)

    fmt = DATASET_FORMATS.get(dataset_name, {})
    fallback_name = fmt.get("fallback")
    datasets_to_try = [dataset_name]
    if fallback_name:
        datasets_to_try.append(fallback_name)

    for try_ds_name in datasets_to_try:
        for try_split in splits_to_try:
            try:
                if config:
                    ds = load_dataset(
                        try_ds_name, config, split=try_split, streaming=True,
                        token=hf_token,
                    )
                else:
                    ds = load_dataset(
                        try_ds_name, split=try_split, streaming=True,
                        token=hf_token,
                    )
                logger.info(f"Loaded {try_ds_name} split={try_split}")
                return ds
            except Exception as e:
                err_str = str(e).lower()
                # If "Bad split" error, try next split
                if "bad split" in err_str or "available splits" in err_str:
                    logger.info(f"  Split {try_split} not available for {try_ds_name}, trying next")
                    continue
                # If "scripts no longer supported", try next dataset (fallback)
                if "scripts are no longer supported" in err_str and fallback_name:
                    logger.info(f"  {try_ds_name} uses script — trying fallback {fallback_name}")
                    break
                # For other errors, log and try next split
                logger.warning(f"Failed to load {try_ds_name} split={try_split}: {str(e)[:120]}")
                continue

    logger.warning(f"All split attempts failed for {dataset_name} (will be skipped)")
    return None


def stream_dataset(
    dataset_name: str,
    max_samples: int = 500,
    hf_token: Optional[str] = None,
) -> Iterator[ProcessedSample]:
    """Faz streaming de um dataset, retornando até max_samples amostras.

    Args:
        dataset_name: nome do dataset no HuggingFace
        max_samples: número máximo de amostras a retornar
        hf_token: token HF opcional

    Yields:
        ProcessedSample
    """
    fmt = DATASET_FORMATS.get(dataset_name)
    if fmt is None:
        logger.error(f"Unknown dataset: {dataset_name}")
        return

    # ── V13.9.2-carolina: bypass do script Python ─────────────────────────
    # O dataset carolina-c4ai/corpus-carolina usa um script Python como loader,
    # que não é suportado por datasets 5.0+. Streamamos os XML.gz diretamente.
    if dataset_name == "carolina-c4ai/corpus-carolina":
        logger.info(f"Carolina: usando carregador direto (bypass do script Python)")
        count = 0
        try:
            for raw_sample in _stream_carolina_direct(
                max_samples=max_samples, hf_token=hf_token
            ):
                processed = _normalize_sample(raw_sample, dataset_name, count)
                if processed is not None:
                    yield processed
                    count += 1
        except Exception as e:
            logger.warning(f"Carolina direct stream error at sample {count}: {e}")
            return
        logger.info(f"Streamed {count} samples from {dataset_name} (direct XML)")
        return

    # ── V13.9.2-finetune-v2: loaders alternativos para parquets grandes ───
    # Alguns datasets (cnmoro, strak2005/bratao) têm parquets tão grandes que
    # datasets.load_dataset(streaming=True) demora minutos para materializar
    # a primeira amostra. Usamos loaders alternativos especificados no campo
    # "loader" do formato.
    loader = fmt.get("loader")
    if loader == "rows_api":
        logger.info(f"{dataset_name}: usando rows_api loader (datasets-server HTTP)")
        count = 0
        try:
            for raw_sample in _stream_via_rows_api(
                dataset_name, max_samples=max_samples, hf_token=hf_token,
                config=fmt.get("config"),
                split=fmt.get("split", "train"),
            ):
                processed = _normalize_sample(raw_sample, dataset_name, count)
                if processed is not None:
                    yield processed
                    count += 1
        except Exception as e:
            logger.warning(f"rows_api stream error at sample {count}: {e}")
            return
        logger.info(f"Streamed {count} samples from {dataset_name} (rows_api)")
        return

    if loader == "parquet_first_shard":
        logger.info(f"{dataset_name}: usando parquet_first_shard loader")
        count = 0
        try:
            for raw_sample in _stream_via_parquet_first_shard(
                dataset_name, max_samples=max_samples, hf_token=hf_token
            ):
                processed = _normalize_sample(raw_sample, dataset_name, count)
                if processed is not None:
                    yield processed
                    count += 1
        except Exception as e:
            logger.warning(f"parquet_shard stream error at sample {count}: {e}")
            # Tenta fallback (ex: strak2005 -> bratao)
            fallback_name = fmt.get("fallback")
            if fallback_name and fallback_name != dataset_name:
                logger.info(f"Tentando fallback: {fallback_name}")
                fallback_fmt = DATASET_FORMATS.get(fallback_name, {})
                # Recursão respeitando o loader do fallback
                yield from stream_dataset(
                    fallback_name, max_samples=max_samples, hf_token=hf_token
                )
            return
        logger.info(f"Streamed {count} samples from {dataset_name} (parquet_first_shard)")
        return

    ds = _load_dataset_streaming(
        dataset_name,
        split=fmt.get("split", "train"),
        config=fmt.get("config"),
        hf_token=hf_token,
    )
    if ds is None:
        return

    count = 0
    try:
        for idx, sample in enumerate(ds):
            if count >= max_samples:
                break
            processed = _normalize_sample(sample, dataset_name, idx)
            if processed is not None:
                yield processed
                count += 1
    except Exception as e:
        logger.warning(f"Error streaming {dataset_name} at sample {idx}: {e}")
        return

    logger.info(f"Streamed {count} samples from {dataset_name}")


def stream_all_datasets(
    datasets: Optional[List[str]] = None,
    max_samples_per_dataset: int = 500,
    hf_token: Optional[str] = None,
) -> Iterator[ProcessedSample]:
    """Faz streaming sequencial de múltiplos datasets.

    Args:
        datasets: lista de nomes (default: DEFAULT_DATASETS)
        max_samples_per_dataset: limite por dataset
        hf_token: token HF opcional

    Yields:
        ProcessedSample
    """
    if datasets is None:
        datasets = DEFAULT_DATASETS

    for ds_name in datasets:
        logger.info(f"Starting dataset: {ds_name}")
        yield from stream_dataset(
            ds_name,
            max_samples=max_samples_per_dataset,
            hf_token=hf_token,
        )


def collate_samples_to_tensors(
    samples: List[ProcessedSample],
    tokenizer,
    max_seq_len: int = 64,
    pad_token_id: int = 0,
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
    """Converte lista de ProcessedSample em tensors para treino.

    BUG FIX V13.9.1: Tokenizer.encode() retorna 'Encoding' object, não list.
    Use .ids attribute to get token IDs.

    Returns:
        input_ids: (B, T)
        attention_mask: (B, T)
        labels: (B, T) — same as input_ids (LM training)
    """
    batch_input_ids = []
    batch_attn_mask = []

    for sample in samples:
        # Tokenize
        encoding = tokenizer.encode(sample.raw_text, add_special_tokens=True)

        # BUG FIX: Handle both `tokenizers.Encoding` and `transformers` tokenizer outputs
        if hasattr(encoding, 'ids'):
            tokens = encoding.ids  # tokenizers library
        elif isinstance(encoding, list):
            tokens = encoding  # already a list of ints
        elif hasattr(encoding, 'input_ids'):
            tokens = encoding.input_ids  # transformers BatchEncoding
        else:
            tokens = list(encoding)  # fallback

        # Truncate to max_seq_len
        if len(tokens) > max_seq_len:
            tokens = tokens[:max_seq_len]
        # Pad
        attn_mask = [1] * len(tokens) + [0] * (max_seq_len - len(tokens))
        tokens = list(tokens) + [pad_token_id] * (max_seq_len - len(tokens))
        batch_input_ids.append(tokens)
        batch_attn_mask.append(attn_mask)

    input_ids = torch.tensor(batch_input_ids, dtype=torch.long)
    attention_mask = torch.tensor(batch_attn_mask, dtype=torch.long)
    labels = input_ids.clone()

    return input_ids, attention_mask, labels


# ═══════════════════════════════════════════════════════════════════════════
# Self-test (NÃO enviar para HuggingFace)
# ═══════════════════════════════════════════════════════════════════════════
if __name__ == "__main__":
    print("=== Streaming Datasets V13.9.1 Self-Test ===\n")

    # Test 1: Test field extraction
    print("Test 1: Field extraction")
    sample_chat = {
        "messages": [
            {"role": "user", "content": "Olá"},
            {"role": "assistant", "content": "Como posso ajudar?"},
        ]
    }
    text = _extract_field(sample_chat, ["messages", "text"])
    print(f"  Chat text: {text!r}")
    assert text and "Olá" in text, "Field extraction failed for chat"

    sample_text = {"text": "Hello world"}
    text = _extract_field(sample_text, ["text", "content"])
    assert text == "Hello world"
    print(f"  Text field: {text!r}")
    print(f"  OK\n")

    # Test 2: Test list of datasets
    print("Test 2: Dataset list")
    print(f"  Configured datasets ({len(DEFAULT_DATASETS)}):")
    for i, ds in enumerate(DEFAULT_DATASETS, 1):
        print(f"    {i}. {ds}")
    assert len(DEFAULT_DATASETS) == 13
    print(f"  OK\n")

    print("=== ALL STREAMING TESTS PASSED ===")