Duygu Analizi

import pandas as pd
import torch
from torch.utils.data import DataLoader, TensorDataset
import os

# Veri kümesini okuma
def read_twitter_data(file_path):
    data = pd.read_csv(file_path)
    tweets = data['text'].tolist()
    labels = data['sentiment'].tolist()
    return tweets, labels

data_file = 'data/twitter_sentiment.csv'  # Veri kümesinin yolu
tweets, labels = read_twitter_data(data_file)

print(f'# of tweets: {len(tweets)}')
for tweet, label in zip(tweets[:3], labels[:3]):
    print('label:', label, 'tweet:', tweet[:60])

from collections import Counter
import re

# Belirteçleme ve kelime dağarcığı oluşturma
def tokenize(texts):
    tokens = [re.findall(r'\b\w+\b', text.lower()) for text in texts]
    return tokens

def build_vocab(tokens, min_freq=5):
    counter = Counter([token for line in tokens for token in line])
    vocab = {token: idx for idx, (token, freq) in enumerate(counter.items()) if freq >= min_freq}
    vocab['<pad>'] = len(vocab)
    return vocab

tokens = tokenize(tweets)
vocab = build_vocab(tokens)

print(f'Vocab size: {len(vocab)}')
 Dizi Uzunluğunu Sabitleme

# Dizi uzunluğunu sabitleme
def truncate_pad(tokens, max_len, padding_token='<pad>'):
    return [token if idx < max_len else padding_token for idx, token in enumerate(tokens)] + [padding_token] * (max_len - len(tokens))

max_len = 50
features = torch.tensor([[vocab.get(token, vocab['<pad>']) for token in truncate_pad(token, max_len)] for token in tokens])
labels = torch.tensor(labels)

print(features.shape)  # torch.Size([num_tweets, max_len])

# Veri yineleyici oluşturma
batch_size = 64
dataset = TensorDataset(features, labels)
data_loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

for X, y in data_loader:
    print('X:', X.shape, ', y:', y.shape)
    break

print('# batches:', len(data_loader))  # batch size kadar böldüğümüz veri kümesindeki toplam batch sayısı

def load_data_twitter(file_path, batch_size, max_len=50):
    tweets, labels = read_twitter_data(file_path)
    tokens = tokenize(tweets)
    vocab = build_vocab(tokens)
    features = torch.tensor([[vocab.get(token, vocab['<pad>']) for token in truncate_pad(token, max_len)] for token in tokens])
    labels = torch.tensor(labels)
    dataset = TensorDataset(features, labels)
    data_loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
    return data_loader, vocab

data_loader, vocab = load_data_twitter('data/twitter_sentiment.csv', batch_size=64)

import os
import tarfile
import requests
import torch
from torch import nn
from torch.utils.data import DataLoader, TensorDataset

# IMDb veri kümesini indirme ve çıkarma
def download_and_extract(url, download_folder):
    if not os.path.exists(download_folder):
        os.makedirs(download_folder)
    response = requests.get(url, stream=True)
    tar_path = os.path.join(download_folder, 'aclImdb_v1.tar.gz')
    with open(tar_path, 'wb') as file:
        file.write(response.raw.read())
    with tarfile.open(tar_path, 'r:gz') as tar:
        tar.extractall(path=download_folder)

url = 'http://ai.stanford.edu/~amaas/data/sentiment/aclImdb_v1.tar.gz'
data_dir = 'data/aclImdb'
download_and_extract(url, data_dir)

# IMDb veri kümesini okuma
def read_imdb(data_dir, is_train=True):
    data, labels = [], []
    for label in ('pos', 'neg'):
        folder_name = os.path.join(data_dir, 'train' if is_train else 'test', label)
        for file in os.listdir(folder_name):
            with open(os.path.join(folder_name, file), 'rb') as f:
                review = f.read().decode('utf-8').replace('\n', '')
                data.append(review)
                labels.append(1 if label == 'pos' else 0)
    return data, labels

train_data = read_imdb(data_dir, is_train=True)
test_data = read_imdb(data_dir, is_train=False)

print('# Training samples:', len(train_data[0]))
for review, label in zip(train_data[0][:3], train_data[1][:3]):
    print('label:', label, 'review:', review[:60])

class BiRNN(nn.Module):
    def __init__(self, vocab_size, embed_size, num_hiddens, num_layers):
        super(BiRNN, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.encoder = nn.LSTM(embed_size, num_hiddens, num_layers=num_layers, bidirectional=True)
        self.decoder = nn.Linear(4 * num_hiddens, 2)

    def forward(self, inputs):
        embeddings = self.embedding(inputs.T)
        self.encoder.flatten_parameters()
        outputs, _ = self.encoder(embeddings)
        encoding = torch.cat((outputs[0], outputs[-1]), dim=1)
        outs = self.decoder(encoding)
        return outs

embed_size, num_hiddens, num_layers, devices = 100, 100, 2, [torch.device('cuda' if torch.cuda.is_available() else 'cpu')]
net = BiRNN(len(vocab), embed_size, num_hiddens, num_layers)

def init_weights(m):
    if type(m) == nn.Linear:
        nn.init.xavier_uniform_(m.weight)
    elif type(m) == nn.LSTM:
        for param in m._flat_weights_names:
            if "weight" in param:
                nn.init.xavier_uniform_(m._parameters[param])

net.apply(init_weights)

# GloVe vektörlerini indirme ve yükleme
def load_glove_embeddings(vocab, embed_size):
    glove_path = 'data/glove.6B.100d.txt'
    if not os.path.exists(glove_path):
        url = 'http://nlp.stanford.edu/data/glove.6B.zip'
        download_and_extract(url, 'data')
    embeddings_index = {}
    with open(glove_path, 'r', encoding='utf-8') as f:
        for line in f:
            values = line.split()
            word = values[0]
            coefs = torch.tensor(list(map(float, values[1:])))
            embeddings_index[word] = coefs
    matrix_len = len(vocab)
    weights_matrix = torch.zeros((matrix_len, embed_size))
    for i, word in enumerate(vocab.idx_to_token):
        try:
            weights_matrix[i] = embeddings_index[word]
        except KeyError:
            weights_matrix[i] = torch.zeros(embed_size)
    return weights_matrix

weights_matrix = load_glove_embeddings(vocab, embed_size)
net.embedding.weight.data.copy_(weights_matrix)
net.embedding.weight.requires_grad = False

def train_model(net, train_iter, test_iter, loss, trainer, num_epochs, devices):
    for epoch in range(num_epochs):
        net.train()
        for X, y in train_iter:
            trainer.zero_grad()
            X, y = X.to(devices[0]), y.to(devices[0])
            y_hat = net(X)
            l = loss(y_hat, y)
            l.sum().backward()
            trainer.step()
        net.eval()
        correct, total = 0, 0
        with torch.no_grad():
            for X, y in test_iter:
                X, y = X.to(devices[0]), y.to(devices[0])
                y_hat = net(X)
                correct += (y_hat.argmax(axis=1) == y).sum().item()
                total += y.size(0)
        print(f'Epoch {epoch + 1}, accuracy {(correct / total):.4f}')

lr, num_epochs = 0.01, 5
trainer = torch.optim.Adam(net.parameters(), lr=lr)
loss = nn.CrossEntropyLoss()
train_iter = DataLoader(TensorDataset(train_features, torch.tensor(train_data[1])), batch_size=batch_size, shuffle=True)
test_iter = DataLoader(TensorDataset(test_features, torch.tensor(test_data[1])), batch_size=batch_size, shuffle=False)

train_model(net, train_iter, test_iter, loss, trainer, num_epochs, devices)

def predict_sentiment(net, vocab, sequence):
    sequence = torch.tensor([vocab[token] for token in sequence.split()], device=devices[0])
    label = torch.argmax(net(sequence.reshape(1, -1)), dim=1)
    return 'positive' if label.item() == 1 else 'negative'

print(predict_sentiment(net, vocab, 'this movie is so great'))
print(predict_sentiment(net, vocab, 'this movie is so bad'))