Nero.py

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Flatten, Embedding, Flatten
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.models import load_model
import numpy as np
import os


import tensorflow as tf
print("Num GPUs Available: ", len(tf.config.list_physical_devices('GPU')))

import matplotlib.pyplot as plt
import pandas as pd
from bs4 import BeautifulSoup
import re
import nltk
from nltk.corpus import stopwords
from nltk.stem.porter import PorterStemmer
from nltk.tokenize.toktok import ToktokTokenizer
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score


import nltk
import ssl

try:
    _create_unverified_https_context = ssl._create_unverified_context
except AttributeError:
    pass
else:
    ssl._create_default_https_context = _create_unverified_https_context

nltk.download('stopwords')

stopword_list=nltk.corpus.stopwords.words('russian')

tokenizer=ToktokTokenizer()


from tensorflow.keras.preprocessing.text import Tokenizer

from datetime import datetime




def strip_html(text):
    soup = BeautifulSoup(text, "html.parser")
    return soup.get_text()

#удаляем квадратные скобки с помощью re
def remove_between_square_brackets(text):
    return re.sub('\[[^]]*\]', '', text)

#функция для очистки текста
def denoise_text(text):
    text = strip_html(text)
    text = remove_between_square_brackets(text)
    return text

#удаляем специальные символы
def remove_special_characters(text, remove_digits=True):
    pattern=r'[^а-яА-я0-9\s]'
    text=re.sub(pattern,'',text)
    return text

#стемминг
def simple_stemmer(text):
    ps=nltk.porter.PorterStemmer()
    text= ' '.join([ps.stem(word) for word in text.split()])
    return text

#удаляем стоп слова из отзывов
def remove_stopwords(text, is_lower_case=False):
    tokens = tokenizer.tokenize(text)
    tokens = [token.strip() for token in tokens]
    if is_lower_case:
        filtered_tokens = [token for token in tokens if token not in stopword_list]
    else:
        filtered_tokens = [token for token in tokens if token.lower() not in stopword_list]
    filtered_text = ' '.join(filtered_tokens)    
    return filtered_text


def get_max_length(kwargs):
    review_length = []
    for review in kwargs:
        review_length.append(len(review))
        
    return int(np.ceil(np.mean(review_length)))




pr = ''

data_set=pd.read_csv(f'{pr}geo-reviews-dataset-2023.csv')

test_data=pd.read_csv(f'{pr}srwniPars_df.csv')




def preparation_of_data_set():
    print('!!!!!!!!!!!!!!!Preparation_of_data_set')
    data_set['text']=data_set['text'].apply(denoise_text)

    data_set['text']=data_set['text'].apply(remove_special_characters)

    data_set['text']=data_set['text'].apply(simple_stemmer)

    data_set['text']=data_set['text'].apply(remove_stopwords)

    data_set['text']=data_set['text'].apply(lambda review: review.split(' '))

    #преобразуем оценку отзывов в отрицательную или положительную
    map_sentiment = {0:0,1:0,2:0,3:0,4:1,5:1}
    data_set['rating'] = data_set['rating'].map(map_sentiment)
    data_set.head()

    #преобразуем текст в числовые значения
    x_train, x_test, y_train, y_test = train_test_split(data_set['text'], data_set['rating'], test_size = 0.2)

    token = Tokenizer(lower=False)
    token.fit_on_texts(x_train)
    x_train = token.texts_to_sequences(x_train)
    x_test = token.texts_to_sequences(x_test)

    max_length = get_max_length(x_train)
    print('Maximum review length: ', max_length)
    x_train = pad_sequences(x_train, maxlen=max_length, padding='post', truncating='post')
    x_test = pad_sequences(x_test, maxlen=max_length, padding='post', truncating='post')

    total_words = len(token.word_index) + 1   # add 1 because of 0 padding

    print('Encoded X Train\n', x_train, '\n')
    print('Encoded X Test\n', x_test, '\n')
    print('Encoded Y Train\n', y_train, '\n')
    print('Encoded Y Test\n', y_test, '\n')
    
    return max_length, total_words, x_test, y_test, x_train, y_train




def preparation_of_test_data():
    print('!!!!!!!!!!!!!!!Preparation_of_test_data')
    test_data['reveiw']=test_data['reveiw'].apply(denoise_text)

    test_data['reveiw']=test_data['reveiw'].apply(remove_special_characters)

    test_data['reveiw']=test_data['reveiw'].apply(simple_stemmer)

    test_data['reveiw']=test_data['reveiw'].apply(remove_stopwords)

    test_data['reveiw']=test_data['reveiw'].apply(lambda review: review.split(' '))

    #преобразуем тестовую оценку отзывов в отрицательную или положительную
    map_sentiment = {'negative':0,'positive':1}
    test_data['sentiment']=test_data['sentiment'].map(map_sentiment)
    test_data.head()

    #преобразуем тестовый текст в числовые значения
    x_test2, y_test2 = test_data['reveiw'], test_data['sentiment']

    print('Encoded X2 Test\n', x_test2, '\n')
    print('Encoded Y2 Test\n', y_test2, '\n')

    token = Tokenizer(lower=False)
    token.fit_on_texts(x_test2)
    x_test2 = token.texts_to_sequences(x_test2)

    max_length = get_max_length(x_test2)
    print(max_length)
    x_test2 = pad_sequences(x_test2, maxlen=max_length, padding='post', truncating='post')

    total_words2 = len(token.word_index) + 1 

    print('Encoded X2 Test\n', x_test2, '\n')
    print('Encoded Y2 Test\n', y_test2, '\n')
    
    return x_test2, y_test2




def creating_neural_network(max_length, total_words):
    print('!!!!!!!!!!!!!!!Creating_neural_network')
    model = Sequential()
    model.add(Embedding(total_words, 50, input_length = max_length))
    model.add(Flatten())
    model.add(Dense(64))
    model.add(Dense(1, activation='sigmoid'))

    model.compile(optimizer='rmsprop', 
                loss='binary_crossentropy', 
                metrics=['accuracy'])
    
    return model




def neural_network_training(model, x_train, y_train):
    print('!!!!!!!!!!!!!!!Neural_Network_Training')
    history = model.fit(x_train, 
                        y_train, 
                        epochs=100,
                        batch_size=32,
                        validation_split=0.1)
    
    return history





def performance_chart(history):
    print('!!!!!!!!!!!!!!!Performance_chart')
    plt.plot(history.history['accuracy'], 
            label='Доля верных ответов на обучающем наборе')
    plt.plot(history.history['val_accuracy'], 
            label='Доля верных ответов на проверочном наборе')
    plt.xlabel('Эпоха обучения')
    plt.ylabel('Доля верных ответов')
    plt.legend()
    plt.show()




def checking_model_on_data_set(model, x_test, y_test):
    y_predit = model.predict(x_test)
    y_predit = y_predit.round()

    print(accuracy_score(y_test, y_predit))

    print('Predit\n', y_predit[20:40], '\n')
    print('Y_test\n', y_test[20:40], '\n')
    
    
    

def checking_model_on_test_data(model, x_test2, y_test2):
    y_predit2 = model.predict(x_test2)
    y_predit2 = y_predit2.round()

    print(accuracy_score(y_test2, y_predit2))

    print('Predit\n', y_predit2[20:40], '\n')
    print('Y_test2\n', y_test2[20:40], '\n')






                
def main():
    
    max_length, total_words, x_test, y_test, x_train, y_train = preparation_of_data_set()
                 
    x_test2, y_test2 = preparation_of_test_data()
    
    model = None 
    history = None
    
    while True:
        comand = input()

        if comand == "/Neural":
            model = creating_neural_network(max_length, total_words)
            history = neural_network_training(model, x_train, y_train)
            
        elif comand == "/Load":
            file = input('Path file: ')
            model = load_model(file)
            
        elif comand == "/Save":
            model.save(f'Model_{datetime.now().strftime("%d_%H_%M")}.h5')
            
        elif comand == "/Performance" and history:
            performance_chart(history)
            
        elif comand == "/Ch_data_set" and model:
            checking_model_on_data_set(model, x_test, y_test)
            
        elif comand == "/Ch_data_set" and model: 
            checking_model_on_test_data(model, x_test2, y_test2)
        
        elif comand == '/Stop': break
        
if __name__ == "__main__":
    main()