FastTextSocialNetworkModel - master - co1dsun/ContentAnalysisSocialMedia

FastTextSocialNetworkModel
143 строки6 KB
co1dsun
create FastTextSocialNetworkModel
17 ноя 2024, 12:55
17 ноя 2024, 12:555757ed4
from dostoevsky.tokenization import RegexTokenizer
from dostoevsky.models import FastTextSocialNetworkModel
from natasha import (
    Segmenter,
    MorphVocab,
    NewsEmbedding,
    NewsMorphTagger,
    NewsSyntaxParser,
    NewsNERTagger,
    PER,
    NamesExtractor,
    Doc
)
from livejournal_parser import parse_livejournal_api
from vkontakte_parser import parse_vkontakte
from inference import inference
from pymystem3 import Mystem

# Загрузка списка негативных ключевых слов из файла
with open('negative_keywords.txt', 'r', encoding='utf-8') as file:
    negative_keywords = [line.strip().lower() for line in file]

# Инициализация лемматизатора
mystem = Mystem()

# Инициализация модели анализа тональности Dostoevsky
tokenizer = RegexTokenizer()
model = FastTextSocialNetworkModel(tokenizer=tokenizer)

# Инициализация инструментов Natasha для предобработки текста
segmenter = Segmenter()
morph_vocab = MorphVocab()
emb = NewsEmbedding()
morph_tagger = NewsMorphTagger(emb)
syntax_parser = NewsSyntaxParser(emb)
ner_tagger = NewsNERTagger(emb)
names_extractor = NamesExtractor(morph_vocab)

# Функция для анализа тональности текста с помощью Dostoevsky
def analyze_sentiment(text):
    results = model.predict([text], k=2)[0]
    sentiment = max(results, key=results.get)
    sentiment_value = results[sentiment]
    return sentiment, sentiment_value

# Функция для предобработки текста и извлечения сущностей с помощью Natasha
def preprocess_text(text):
    doc = Doc(text)
    doc.segment(segmenter)
    doc.tag_morph(morph_tagger)
    doc.parse_syntax(syntax_parser)
    doc.tag_ner(ner_tagger)
    for span in doc.spans:
        span.normalize(morph_vocab)

    tokens = []
    for token in doc.tokens:
        token.lemmatize(morph_vocab)
        tokens.append(token.lemma)

    persons = []
    for span in doc.spans:
        if span.type == PER:
            persons.append(span.text)

    return " ".join(tokens), ", ".join(persons)

# Функция для проверки наличия негативных ключевых слов в тексте
def contains_negative_keywords(text):
    lemmas = mystem.lemmatize(text.lower())
    
    for lemma in lemmas:
        if lemma.strip() in negative_keywords:
            return True
    return False

def analyze_posts_with_offset(user_id, platform, offset, progress_callback=None):
    if platform == 'livejournal':
        posts = parse_livejournal_api(user_id, offset=offset)
    elif platform == 'vkontakte':
        posts = parse_vkontakte(user_id, offset=offset)
    else:
        raise ValueError(f"Неподдерживаемая платформа: {platform}. Поддерживаются только 'livejournal' и 'vkontakte'.")

    if not posts:
        return ["Не удалось найти посты для указанного пользователя на выбранной платформе."], 0

    # Анализ тональности и предобработка текстов постов
    negative_posts = []
    negative_political_posts = []
    total_posts = len(posts)

    for i, post in enumerate(posts, start=1):
        text = post["text"]
        link = post["link"]
        sentiment, sentiment_value = analyze_sentiment(text)
        
        # Проверка наличия негативных ключевых слов
        has_negative_keywords = contains_negative_keywords(text)
        
        # Инференс для определения политической тематики
        predicted_labels = inference(text)
        is_political = "политика" in [label.lower() for label in predicted_labels]
        
        if sentiment == "negative" or has_negative_keywords:
            processed_text, persons = preprocess_text(text)
            post_info = f"Ссылка: {link}, Значение негативности: {sentiment_value:.2f}"
            if has_negative_keywords:
                post_info += " [Негативные ключевые слова]"
            if is_political:
                post_info += " [Политический пост]"
            if persons:
                post_info += f", Персоны: {persons}"
            if is_political:
                negative_political_posts.append(post_info)
            else:
                negative_posts.append(post_info)
        
        # Обновляем прогресс
        if progress_callback:
            progress = int(i / total_posts * 100)
            progress_callback(progress, i)

    # Формирование результатов анализа
    response_parts = []
    response_parts.append(f"Результаты анализа постов для пользователя {user_id} на платформе {platform}:\n\n")
    response_parts.append(f"Всего обработано постов: {total_posts}\n\n")
    
    if negative_posts:
        response_parts.append("Негативные посты:\n")
        for post_info in negative_posts:
            response_parts.append(post_info + "\n")
    else:
        response_parts.append("Негативные посты не найдены.\n\n")
    
    if negative_political_posts:
        response_parts.append("Негативные посты на политическую тематику:\n")
        for post_info in negative_political_posts:
            response_parts.append(post_info + "\n")
    else:
        response_parts.append("Негативные посты на политическую тематику не найдены.")

    return response_parts, total_posts