04 - Sentiment Analysis per Italiano: feel-it e Custom

# Problemi comuni con i tokenizzatori per l'italiano

# Enclitici: pronomi attaccati al verbo
examples = [
    "Dimmelo",      # dimmi + lo
    "Portarmelo",   # portare + mi + lo
    "Fallo",        # fai + lo
    "Dateglielo",   # date + glie + lo
]

# Truncation sbagliata con tokenizzatori non italiani
from transformers import BertTokenizer
tokenizer_en = BertTokenizer.from_pretrained('bert-base-uncased')
tokenizer_it = BertTokenizer.from_pretrained('dbmdz/bert-base-italian-cased')

word = "Dimmelo"
print(f"EN tokenizer: {tokenizer_en.tokenize(word)}")
# ['dim', '##mel', '##o'] - non coglie la struttura

print(f"IT tokenizer: {tokenizer_it.tokenize(word)}")
# ['Dim', '##me', '##lo'] - migliore ma non perfetto

# La soluzione ottimale e la lemmatizzazione prima del tokenize

# Installa il modello italiano: python -m spacy download it_core_news_lg
import spacy

nlp = spacy.load("it_core_news_lg")

def preprocess_italian(text: str,
                        remove_stopwords: bool = True,
                        lemmatize: bool = True) -> str:
    """Preprocessing completo per testi italiani."""
    doc = nlp(text)

    tokens = []
    for token in doc:
        # Salta punteggiatura, spazi, numeri (se non rilevanti)
        if token.is_punct or token.is_space:
            continue

        # Normalizza a minuscolo
        word = token.text.lower()

        # Rimuovi stopwords italiane
        if remove_stopwords and token.is_stop:
            continue

        # Lemmatizza
        if lemmatize:
            word = token.lemma_.lower()

        tokens.append(word)

    return ' '.join(tokens)

# Test
texts = [
    "I prodotti sono stati consegnati rapidamente e tutto funzionava perfettamente",
    "Ho comprato questo telefono tre mesi fa e sono rimasto deluso dalla batteria",
    "PRODOTTO FANTASTICO! Lo consiglio assolutamente a tutti voi amici!!!"
]

for text in texts:
    processed = preprocess_italian(text)
    print(f"Original:  {text}")
    print(f"Processed: {processed}")
    print()

import re
import unicodedata

def normalize_italian_text(text: str) -> str:
    """
    Normalizzazione per testi italiani informali (social media, recensioni).
    """
    # 1. Normalizza unicode (accenti)
    text = unicodedata.normalize('NFC', text)

    # 2. Sostituisci accenti con apostrofo (comune online)
    accent_map = {
        "a'": "a",   # può' -> può
        "è": "è",  # mantenuto per 'è (voce del verbo essere)
        "i'": "i",
        "o'": "o",
        "u'": "u"
    }
    # Non sostituiamo indiscriminatamente per evitare ambiguità

    # 3. Espandi abbreviazioni comuni
    abbreviations = {
        r'\bcmq\b': 'comunque',
        r'\bnn\b': 'non',
        r'\bxke\b': 'perchè',
        r'\bxche\b': 'perchè',
        r'\bx\b': 'per',
        r'\bke\b': 'che',
        r'\bkm\b': 'come',
        r'\bqs\b': 'questo',
        r'\btv\b': 'televisione',
        r'\bgg\b': 'giorni',
        r'\bprof\b': 'professore',
    }
    for abbr, expanded in abbreviations.items():
        text = re.sub(abbr, expanded, text, flags=re.IGNORECASE)

    # 4. Rimuovi caratteri ripetuti eccessivi (bellissimoooo -> bellissimo)
    text = re.sub(r'(.)\1{2,}', r'\1\1', text)  # max 2 ripetizioni

    # 5. Rimuovi emoji (opzionale - può essere informativo per il sentiment)
    # text = re.sub(r'[^\w\s.,!?;:\'\"()-]', ' ', text)

    # 6. Normalizza spazi multipli
    text = re.sub(r'\s+', ' ', text).strip()

    return text

# Test
informal_texts = [
    "cmq il prodotto è fantasticooo!!!",
    "nn mi e piaciuto x niente, sto cercando di restituirlo xke nn funziona",
    "Amici... COMPRATE QUESTOOO!!! è il TOP del TOP!!!",
]

for text in informal_texts:
    normalized = normalize_italian_text(text)
    print(f"Originale:   {text}")
    print(f"Normalizzato: {normalized}")
    print()

from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch

# feel-it per sentiment (positivo/negativo)
sentiment_model = pipeline(
    "text-classification",
    model="MilaNLProc/feel-it-italian-sentiment",
    tokenizer="MilaNLProc/feel-it-italian-sentiment"
)

# feel-it per emozioni (gioia, tristezza, rabbia, paura, disgusto, sorpresa)
emotion_model = pipeline(
    "text-classification",
    model="MilaNLProc/feel-it-italian-emotion",
    tokenizer="MilaNLProc/feel-it-italian-emotion"
)

# Test su testi italiani
texts = [
    "Sono molto felice del mio acquisto, qualità eccellente!",
    "Ho perso tutto il mio lavoro, sono devastato.",
    "Questa e la situazione più ridicola che abbia mai visto.",
    "Non credevo che potesse funzionare cosi bene, sono stupito!",
]

print("=== SENTIMENT ===")
for text in texts:
    result = sentiment_model(text)[0]
    print(f"  [{result['label']}: {result['score']:.3f}] {text[:60]}")

print("\n=== EMOTION ===")
for text in texts:
    result = emotion_model(text)[0]
    print(f"  [{result['label']}: {result['score']:.3f}] {text[:60]}")

from transformers import AutoTokenizer, AutoModel
import torch

# AlBERTo - BERT uncased per Twitter italiano
alberto_name = "m-polignano-uniba/bert_uncased_L-12_H-768_A-12_Italian_alb3rt0"
tokenizer = AutoTokenizer.from_pretrained(alberto_name)
model = AutoModel.from_pretrained(alberto_name)

# Test di tokenizzazione su testo colloquiale
informal_texts = [
    "PRODOTTO TOP! ma la spedizione ha fatto schifo cmq",
    "mizzica quanto e bello sto telefono!! ci ho messo 2gg ma ne valeva la pena",
    "ok mi avete rotto... non lo compro più #delusione",
]

for text in informal_texts:
    tokens = tokenizer.tokenize(text)
    print(f"Testo: {text[:50]}")
    print(f"Tokens ({len(tokens)}): {tokens[:10]}...")
    print()

# Estrazione embeddings
def get_sentence_embedding(text, model, tokenizer, pooling='cls'):
    inputs = tokenizer(text, return_tensors='pt',
                      truncation=True, max_length=128, padding=True)
    with torch.no_grad():
        outputs = model(**inputs)
    if pooling == 'cls':
        return outputs.last_hidden_state[:, 0, :]  # [CLS] token
    elif pooling == 'mean':
        mask = inputs['attention_mask'].unsqueeze(-1)
        return (outputs.last_hidden_state * mask).sum(1) / mask.sum(1)

emb = get_sentence_embedding(informal_texts[0], model, tokenizer)
print(f"Embedding shape: {emb.shape}")  # (1, 768)

from transformers import BertTokenizer, BertForSequenceClassification
from transformers import TrainingArguments, Trainer
from datasets import Dataset
import torch

# Modello base per l'italiano
MODEL = "dbmdz/bert-base-italian-cased"
tokenizer = BertTokenizer.from_pretrained(MODEL)

# Crea un classificatore di sentiment per l'italiano
model = BertForSequenceClassification.from_pretrained(
    MODEL,
    num_labels=2,
    id2label={0: "NEGATIVO", 1: "POSITIVO"},
    label2id={"NEGATIVO": 0, "POSITIVO": 1}
)

# Dataset di esempio in italiano
train_data = {
    "text": [
        "Il prodotto e arrivato in perfette condizioni, molto soddisfatto",
        "qualità pessima, si e rotto dopo due giorni",
        "Eccellente rapporto qualità/prezzo, lo consiglio",
        "Imballaggio scarso, prodotto danneggiato alla consegna",
        "Supera le aspettative, ottimo acquisto",
        "Servizio clienti inesistente, rimborso impossibile",
        "Materiali di qualità, costruzione solida",
        "Non corrisponde alla descrizione, immagine ingannevole",
    ],
    "label": [1, 0, 1, 0, 1, 0, 1, 0]
}

def tokenize_fn(examples):
    return tokenizer(examples["text"], truncation=True,
                    padding="max_length", max_length=128)

dataset = Dataset.from_dict(train_data)
tokenized = dataset.map(tokenize_fn, batched=True)

# Training veloce (pochi dati = pochissime epoche)
args = TrainingArguments(
    output_dir="./models/bert-italian-sentiment",
    num_train_epochs=5,
    per_device_train_batch_size=8,
    learning_rate=3e-5,
    warmup_ratio=0.1,
    weight_decay=0.01,
    save_steps=100,
    logging_steps=10,
    report_to="none"
)

trainer = Trainer(
    model=model,
    args=args,
    train_dataset=tokenized,
)
trainer.train()

from datasets import load_dataset

# SENTIPOLC 2016 - dataset italiano per polarity detection su Twitter
# Disponibile su: http://www.di.unito.it/~tutreeb/sentipolc-evalita16/
# Etichette: OBJ (oggettivo), POS (positivo), NEG (negativo), MIX

# Dataset disponibile su HuggingFace
try:
    dataset = load_dataset("gsarti/itacola")
    print("ITA-CoLA dataset:", dataset)
except Exception:
    print("Dataset non disponibile direttamente, usa URL manuale")

# Costruzione dataset personalizzato da CSV
import pandas as pd
from datasets import Dataset

# Esempio: caricare recensioni italiane da CSV
# Formato atteso: colonne 'text' e 'label'
def load_italian_dataset(csv_path):
    df = pd.read_csv(csv_path)

    # Validazione
    assert 'text' in df.columns, "Manca colonna 'text'"
    assert 'label' in df.columns, "Manca colonna 'label'"

    # Rimuovi righe con testo vuoto
    df = df.dropna(subset=['text', 'label'])
    df = df[df['text'].str.strip() != '']

    # Normalizza etichette
    label_map = {
        'positivo': 1, 'pos': 1, '1': 1, 1: 1,
        'negativo': 0, 'neg': 0, '0': 0, 0: 0
    }
    df['label'] = df['label'].map(label_map)
    df = df.dropna(subset=['label'])
    df['label'] = df['label'].astype(int)

    return Dataset.from_pandas(df[['text', 'label']])

import re
import spacy
from transformers import pipeline as hf_pipeline
from typing import Optional
import unicodedata

class ItalianSentimentPipeline:
    """
    Pipeline completa per il sentiment analysis in italiano.
    Combina preprocessing specifico e feel-it per il sentiment.
    """

    def __init__(self,
                 sentiment_model: str = "MilaNLProc/feel-it-italian-sentiment",
                 emotion_model: Optional[str] = "MilaNLProc/feel-it-italian-emotion",
                 use_spacy: bool = True,
                 confidence_threshold: float = 0.6):
        # Carica modelli sentiment ed emotion
        self.sentiment = hf_pipeline(
            "text-classification",
            model=sentiment_model,
            truncation=True,
            max_length=128
        )
        self.emotion = hf_pipeline(
            "text-classification",
            model=emotion_model,
            truncation=True,
            max_length=128
        ) if emotion_model else None

        # spaCy per preprocessing avanzato
        if use_spacy:
            try:
                self.nlp = spacy.load("it_core_news_sm")
            except OSError:
                print("Modello spaCy 'it_core_news_sm' non trovato.")
                print("Installa con: python -m spacy download it_core_news_sm")
                self.nlp = None
        else:
            self.nlp = None

        self.threshold = confidence_threshold

    def preprocess(self, text: str) -> str:
        """Preprocessing specifico per italiano."""
        if not text or not text.strip():
            return ""

        # Normalizza unicode
        text = unicodedata.normalize('NFC', text)

        # Abbreviazioni comuni italiane
        abbr_map = {
            r'\bcmq\b': 'comunque',
            r'\bnn\b': 'non',
            r'\bxke\b': 'perchè',
            r'\bx\b': 'per',
        }
        for pattern, replacement in abbr_map.items():
            text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)

        # Riduzione caratteri ripetuti
        text = re.sub(r'(.)\1{2,}', r'\1\1', text)

        # Normalizza spazi
        text = re.sub(r'\s+', ' ', text).strip()

        return text

    def analyze(self, text: str) -> dict:
        """Analisi completa: sentiment + emozione + preprocessing."""
        if not text or not text.strip():
            return {"error": "Testo vuoto"}

        preprocessed = self.preprocess(text)

        # Sentiment
        sent_result = self.sentiment(preprocessed)[0]
        sentiment_label = sent_result['label']
        sentiment_score = sent_result['score']

        result = {
            "text_originale": text,
            "text_preprocessato": preprocessed,
            "sentiment": sentiment_label,
            "sentiment_score": round(sentiment_score, 4),
            "confident": sentiment_score >= self.threshold
        }

        # Emozione (se disponibile)
        if self.emotion:
            em_result = self.emotion(preprocessed)[0]
            result["emotion"] = em_result['label']
            result["emotion_score"] = round(em_result['score'], 4)

        return result

    def analyze_batch(self, texts: list) -> list:
        return [self.analyze(t) for t in texts]

# Utilizzo
pipeline = ItalianSentimentPipeline()

test_texts = [
    "Il prodotto e arrivato in perfette condizioni, sono molto soddisfatto dell'acquisto!",
    "Pessima esperienza. Il pacco era danneggiato e il servizio clienti non risponde.",
    "Mah, diciamo che si poteva fare meglio. Non e ne buono ne cattivo.",
    "INCREDIBILE! Non avrei mai pensato che fosse cosi bello!!! Sto piangendo di gioia",
    "Nn ci credo... mi ha di nuovo fregato sto negozio di schifo",
]

for text in test_texts:
    result = pipeline.analyze(text)
    print(f"Testo: {text[:60]}...")
    print(f"Sentiment: {result['sentiment']} ({result['sentiment_score']:.3f})")
    if 'emotion' in result:
        print(f"Emozione:  {result['emotion']} ({result['emotion_score']:.3f})")
    print(f"Affidabile: {result['confident']}")
    print()

from transformers import (
    AutoTokenizer, AutoModelForSequenceClassification,
    TrainingArguments, Trainer
)
from datasets import Dataset
import evaluate
import numpy as np

# Strategia 1: Fine-tuning di feel-it su dati dominio
def finetune_for_domain(
    base_model: str,
    train_texts: list,
    train_labels: list,
    val_texts: list,
    val_labels: list,
    output_dir: str,
    num_epochs: int = 3
):
    tokenizer = AutoTokenizer.from_pretrained(base_model)
    model = AutoModelForSequenceClassification.from_pretrained(
        base_model,
        num_labels=2,
        ignore_mismatched_sizes=True   # per modelli già fine-tuned
    )

    def tokenize(examples):
        return tokenizer(examples["text"], truncation=True,
                        padding="max_length", max_length=128)

    train_ds = Dataset.from_dict({"text": train_texts, "label": train_labels})
    val_ds = Dataset.from_dict({"text": val_texts, "label": val_labels})

    train_tok = train_ds.map(tokenize, batched=True)
    val_tok = val_ds.map(tokenize, batched=True)

    accuracy = evaluate.load("accuracy")
    def compute_metrics(eval_pred):
        logits, labels = eval_pred
        preds = np.argmax(logits, axis=-1)
        return accuracy.compute(predictions=preds, references=labels)

    args = TrainingArguments(
        output_dir=output_dir,
        num_train_epochs=num_epochs,
        per_device_train_batch_size=16,
        per_device_eval_batch_size=32,
        learning_rate=2e-5,
        warmup_ratio=0.1,
        weight_decay=0.01,
        evaluation_strategy="epoch",
        save_strategy="epoch",
        load_best_model_at_end=True,
        metric_for_best_model="accuracy",
        report_to="none"
    )

    trainer = Trainer(
        model=model,
        args=args,
        train_dataset=train_tok,
        eval_dataset=val_tok,
        compute_metrics=compute_metrics
    )

    trainer.train()
    trainer.save_model(output_dir)
    tokenizer.save_pretrained(output_dir)
    return trainer

# Strategia 2: Confronto modelli italiani
def compare_italian_models(texts, true_labels):
    """Confronto automatico di diversi modelli BERT italiani."""
    models = {
        "feel-it": "MilaNLProc/feel-it-italian-sentiment",
        "AlBERTo-fine": "m-polignano-uniba/bert_uncased_L-12_H-768_A-12_Italian_alb3rt0",
        "mBERT": "bert-base-multilingual-cased"
    }

    results = {}
    for name, model_id in models.items():
        try:
            clf = hf_pipeline("text-classification", model=model_id,
                              truncation=True, max_length=128)
            preds = clf(texts)
            # ... calcola metriche
            print(f"{name}: modello caricato correttamente")
        except Exception as e:
            print(f"{name}: errore - {e}")

    return results

from sklearn.metrics import classification_report, confusion_matrix
import numpy as np

def benchmark_italian_sentiment(model_pipeline, test_data):
    """
    Benchmark completo per modelli di sentiment italiano.
    test_data: lista di tuple (testo, label)
    """
    texts = [d[0] for d in test_data]
    true_labels = [d[1] for d in test_data]

    predictions = model_pipeline(texts)
    pred_labels = []

    for pred in predictions:
        label = pred['label'].upper()
        if label in ['POSITIVE', 'POSITIVO', 'POS']:
            pred_labels.append(1)
        else:
            pred_labels.append(0)

    print("=== REPORT CLASSIFICAZIONE ===")
    print(classification_report(
        true_labels, pred_labels,
        target_names=['NEGATIVO', 'POSITIVO'],
        digits=4
    ))

    # Analisi per categoria di testo
    categories = {
        'formale': [i for i, t in enumerate(texts) if len(t.split()) > 20],
        'informale': [i for i, t in enumerate(texts) if len(t.split()) <= 20],
    }

    for cat_name, indices in categories.items():
        if indices:
            cat_true = [true_labels[i] for i in indices]
            cat_pred = [pred_labels[i] for i in indices]
            report = classification_report(cat_true, cat_pred, output_dict=True)
            acc = report['accuracy']
            print(f"\nCategoria '{cat_name}' ({len(indices)} esempi): accuracy={acc:.4f}")

    return pred_labels

from transformers import (
    AutoTokenizer, AutoModelForSequenceClassification,
    TrainingArguments, Trainer
)
from datasets import Dataset
import numpy as np
import evaluate

# 1. Dataset personalizzato (es. recensioni e-commerce italiane)
custom_data = {
    "text": [
        "Prodotto eccellente, consegna rapidissima. Super consigliato!",
        "qualità pessima, si e rotto dopo una settimana. Deluso.",
        "Ok, nella media. Ne potevo fare a meno.",
        "Fantastico! Esattamente come descritto, sono molto soddisfatto.",
        "Spedizione veloce ma il prodotto non corrisponde alla descrizione.",
        "Materiale scadente, non vale il prezzo. Non ricompro.",
        "Ottimo rapporto qualità/prezzo, lo consiglio a tutti.",
        "Funziona perfettamente, esattamente quello che cercavo.",
    ],
    "label": [1, 0, 0, 1, 0, 0, 1, 1]  # 0=negativo, 1=positivo
}

# 2. Carica tokenizer feel-it
model_name = "MilaNLProc/feel-it-italian-sentiment"
tokenizer = AutoTokenizer.from_pretrained(model_name)

def tokenize(examples):
    return tokenizer(
        examples["text"],
        padding="max_length",
        truncation=True,
        max_length=128
    )

dataset = Dataset.from_dict(custom_data)
dataset = dataset.train_test_split(test_size=0.2, seed=42)
tokenized = dataset.map(tokenize, batched=True)

# 3. Carica modello con nuova classification head
model = AutoModelForSequenceClassification.from_pretrained(
    model_name,
    num_labels=2,
    ignore_mismatched_sizes=True,  # la head originale ha etichette diverse
    id2label={0: "NEGATIVO", 1: "POSITIVO"},
    label2id={"NEGATIVO": 0, "POSITIVO": 1}
)

# 4. Metriche di valutazione
accuracy_metric = evaluate.load("accuracy")
f1_metric = evaluate.load("f1")

def compute_metrics(eval_pred):
    logits, labels = eval_pred
    preds = np.argmax(logits, axis=-1)
    return {
        "accuracy": accuracy_metric.compute(predictions=preds, references=labels)["accuracy"],
        "f1": f1_metric.compute(predictions=preds, references=labels)["f1"]
    }

# 5. Training arguments calibrati per dataset piccoli
training_args = TrainingArguments(
    output_dir="./feel-it-finetuned",
    num_train_epochs=5,         # più epoche per dataset piccoli
    per_device_train_batch_size=8,
    per_device_eval_batch_size=16,
    learning_rate=2e-5,
    warmup_ratio=0.2,           # warmup più lungo per stabilità
    weight_decay=0.01,
    evaluation_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
    metric_for_best_model="f1",
    fp16=False,                 # disabilita per GPU piccole o CPU
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized["train"],
    eval_dataset=tokenized["test"],
    compute_metrics=compute_metrics,
)

trainer.train()
results = trainer.evaluate()
print(f"Accuracy: {results['eval_accuracy']:.4f}")
print(f"F1: {results['eval_f1']:.4f}")

# 6. Salva e publica su HuggingFace Hub (opzionale)
trainer.save_model("./feel-it-custom-ecommerce")
tokenizer.save_pretrained("./feel-it-custom-ecommerce")