scripts/dataset_auditor.py

#!/usr/bin/env python3
import json
import csv
import re
import argparse
from pathlib import Path
import unicodedata

KNOWN_RANGES = {
    "tacs": {
        "default": (4, 133),
        "variáveis e scope": (7, 9),
        "assincronia: promises": (119, 121),
    },
    "semio": {
        "default": (1, 60),
        "resumo sc": (17, 28),
        "semiótica resumos": (1, 30),
    }
}

def clean_text(text):
    text = (text or "").lower()
    text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode("ascii")
    return re.sub(r'\s+', ' ', re.sub(r'[^a-z0-9_]+', ' ', text)).strip()

def get_words(text):
    return set(clean_text(text).split())

def calculate_jaccard(text1, text2):
    w1 = get_words(text1)
    w2 = get_words(text2)
    if not w1 or not w2:
        return 0.0
    return len(w1.intersection(w2)) / len(w1.union(w2))

def parse_expected_sources(src_list):
    # Returns list of (file, page)
    parsed = []
    for src in src_list:
        # e.g., "Variáveis e Scope, pp. 6-8" or "Resumo SC, p. 18"
        parts = src.split(",")
        file_name = parts[0].strip() if parts else "Desconhecido"
        page = parts[1].strip() if len(parts) > 1 else "N/A"
        parsed.append((file_name, page))
    return parsed

def audit_dataset(dataset_path: Path, old_dataset_path: Path | None, subject: str, out_csv: Path) -> None:
    print(f"\n================ AUDITING DATASET: {dataset_path.name} ================")
    
    with open(dataset_path, "r", encoding="utf-8") as f:
        data = json.load(f)
        
    old_data = []
    if old_dataset_path and old_dataset_path.exists():
        with open(old_dataset_path, "r", encoding="utf-8") as f:
            old_data = json.load(f)
            
    # 1. Structural Integrity Check
    structural_ok = True
    reasons = []
    
    if len(data) != 50:
        structural_ok = False
        reasons.append(f"Número de itens incorreto: {len(data)} (esperado 50)")
        
    ids = [item.get("id") for item in data]
    if sorted(ids) != list(range(1, 51)):
        structural_ok = False
        reasons.append(f"IDs não são consecutivos de 1 a 50: {sorted(ids)[:5]}...")
        
    categories = [item.get("categoria") for item in data]
    from collections import Counter
    counts = Counter(categories)
    expected_counts = {
        "definicao": 10,
        "conceptual": 10,
        "procedimento": 10,
        "debug": 8,
        "aplicacao": 5,
        "fora_scope": 5,
        "ambigua": 2
    }
    for cat, expected in expected_counts.items():
        if counts[cat] != expected:
            structural_ok = False
            reasons.append(f"Categoria '{cat}' com contagem {counts[cat]} (esperado {expected})")
            
    # 2. Reference Validity Check
    reference_ok = True
    
    # 3. Internal Consistency Check (must_include in expected_context)
    consistency_ok = True
    
    # Generate CSV Rows
    csv_rows = []
    
    for item in data:
        item_id = item.get("id")
        cat = item.get("categoria")
        pergunta = item.get("pergunta")
        gt = item.get("contexto_esperado")
        must_inc = item.get("must_include", [])
        must_not = item.get("must_not_include", [])
        sources = item.get("expected_sources", [])
        notes = item.get("evaluation_notes", "")
        
        parsed_src = parse_expected_sources(sources)
        file_names = " | ".join(p[0] for p in parsed_src) if parsed_src else "N/A"
        pages = " | ".join(p[1] for p in parsed_src) if parsed_src else "N/A"
        
        # Validar páginas
        subject_ranges = KNOWN_RANGES.get(subject, {})
        for src in sources:
            matched_key = "default"
            for key in subject_ranges:
                if key != "default" and key in src.lower():
                    matched_key = key
                    break
            min_p, max_p = subject_ranges[matched_key]
            
            page_matches = re.findall(r'\b(?:p\.?|pp\.?|pag\.?|p[aá]gina|powerpoint|slide)\s*[:.]?\s*(\d+)', src, re.IGNORECASE)
            for p_str in page_matches:
                p_num = int(p_str)
                if p_num < min_p or p_num > max_p:
                    reference_ok = False
                    reasons.append(f"ID {item_id}: página {p_num} em '{src}' fora do intervalo ({min_p}-{max_p})")
                    
        # Validar consistência interna
        for must in must_inc:
            must_clean = clean_text(must)
            must_words = [w for w in must_clean.split() if len(w) >= 4]
            if must_words:
                matches = sum(1 for w in must_words if w in clean_text(gt))
                if (matches / len(must_words)) < 0.4:
                    consistency_ok = False
                    reasons.append(f"ID {item_id}: must_include '{must}' tem fraca consistência com o ground truth")
        
        csv_rows.append({
            "ID": item_id,
            "categoria": cat,
            "pergunta": pergunta,
            "resposta_de_referencia": gt,
            "ficheiro": file_names,
            "pagina": pages,
            "excerto_da_fonte": "[Preencher na auditoria manual]",
            "must_include": "; ".join(must_inc),
            "notas": notes,
            "manual_audit_status": "pending"
        })
        
    # Write CSV
    fields = ["ID", "categoria", "pergunta", "resposta_de_referencia", "ficheiro", "pagina", "excerto_da_fonte", "must_include", "notas", "manual_audit_status"]
    with open(out_csv, "w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        writer.writerows(csv_rows)
        
    print(f"Audit CSV written to: {out_csv.name}")
    
    # 4. Lexical Similarity Report against Previous Dataset
    similarities = []
    if old_data:
        for item in data:
            q_new = item.get("pergunta")
            id_new = item.get("id")
            for old_item in old_data:
                q_old = old_item.get("pergunta")
                id_old = old_item.get("id")
                jac = calculate_jaccard(q_new, q_old)
                similarities.append((id_new, id_old, jac, q_new, q_old))
                
        # Sort and get top 5 highest similarities
        similarities.sort(key=lambda x: x[2], reverse=True)
        print("\nTop 5 Pares com Maior Semelhança com Dataset Anterior:")
        for idx, (id_n, id_o, jac, q_n, q_o) in enumerate(similarities[:5]):
            print(f"  {idx+1}. Jaccard Similarity = {jac:.3f}")
            print(f"     Novo ID {id_n}: {q_n}")
            print(f"     Antigo ID {id_o}: {q_o}")
            
    # Audit summary
    print("\n---------------- AUDIT SUMMARY ----------------")
    print(f"1. Integridade Estrutural: {'✅ APROVADO' if structural_ok else '❌ FALHOU'}")
    print(f"2. Validade das Referências: {'✅ APROVADO' if reference_ok else '❌ FALHOU'}")
    print(f"3. Consistência Interna: {'✅ APROVADO' if consistency_ok else '⚠️ AVISOS EMITIDOS'}")
    print(f"4. Factualidade Científica: 🔎 PENDENTE DE AUDITORIA HUMANA (manual_audit_status=pending)")
    print("-----------------------------------------------")
    if reasons:
        print("Detalhes encontrados:")
        for r in reasons[:10]:
            print(f"  - {r}")
        if len(reasons) > 10:
            print(f"  ... e mais {len(reasons)-10} ocorrências.")

if __name__ == "__main__":
    audit_dataset(
        Path("dataset_tacs_mis4tel_v5_heldout_50.json"),
        Path("dataset_tacs_mis4tel_v4_50.json"),
        "tacs",
        Path("dataset_tacs_mis4tel_v5_heldout_audit.csv")
    )
    audit_dataset(
        Path("dataset_semio_mis4tel_v2_heldout_50.json"),
        Path("dataset_semio_mis4tel_v1.json"),
        "semio",
        Path("dataset_semio_mis4tel_v2_heldout_audit.csv")
    )