import pandas as pd import numpy as np import re from typing import Dict, List, Tuple, Optional import warnings from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity warnings.filterwarnings('ignore') class RussianFeatureExtractor: """Исправленная версия экстрактора признаков с работающим composite_quality_score""" def __init__(self, use_heavy_models: bool = False): print("Инициализация исправленного экстрактора признаков...") self.use_heavy_models = use_heavy_models self.sbert_model = None # Инициализация моделей self._initialize_models() # Списки ключевых слов self.greeting_words = ['здравствуйте', 'привет', 'добрый', 'здравствуй', 'доброе', 'приветствую'] self.question_words = ['как', 'что', 'где', 'когда', 'почему', 'можно', 'сколько', 'какой', 'какая'] self.descriptive_words = ['вижу', 'изображен', 'находится', 'делает', 'одет', 'стоит', 'сидит'] self.connector_words = ['потому что', 'поэтому', 'так как', 'например', 'кроме того'] self.emotional_words = ['красиво', 'интересно', 'замечательно', 'прекрасно', 'нравится'] self.spatial_words = ['слева', 'справа', 'вверху', 'внизу', 'рядом', 'около'] print("✅ Инициализация завершена!") def _initialize_models(self): """Инициализация моделей""" if self.use_heavy_models: print("ℹ️ Тяжелые модели отключены для стабильности") print("ℹ️ Используем легкие методы (TF-IDF)") def clean_text(self, text: str) -> str: """Очистка текста""" if pd.isna(text): return "" text = str(text) text = re.sub(r'<[^>]+>', '', text) text = re.sub(r'[^\w\sа-яА-ЯёЁ.,!?;:()-]', '', text) text = re.sub(r'\s+', ' ', text).strip() return text def extract_basic_features(self, text: str) -> Dict[str, float]: """Базовые текстовые признаки""" text_clean = self.clean_text(text) if not text_clean: return { 'text_length': 0, 'word_count': 0, 'sentence_count': 0, 'avg_word_length': 0, 'lexical_diversity': 0, 'has_questions': 0, 'has_exclamations': 0 } # Базовые метрики words = re.findall(r'\b[а-яёa-z]+\b', text_clean.lower()) sentences = [s.strip() for s in re.split(r'[.!?]+', text_clean) if s.strip()] word_count = len(words) text_length = len(text_clean) sentence_count = len(sentences) features = { 'text_length': text_length, 'word_count': word_count, 'sentence_count': sentence_count, 'avg_word_length': sum(len(w) for w in words) / max(word_count, 1), 'lexical_diversity': len(set(words)) / max(word_count, 1), 'has_questions': int('?' in text_clean), 'has_exclamations': int('!' in text_clean), } return features def extract_semantic_features(self, question: str, answer: str) -> Dict[str, float]: """Семантические признаки""" question_clean = self.clean_text(question) answer_clean = self.clean_text(answer) features = { 'keyword_overlap': 0.0, 'response_relevance': 0.0 } if not answer_clean or not question_clean: return features try: # Упрощенный анализ ключевых слов question_words = set(re.findall(r'\b[а-яё]+\b', question_clean.lower())) answer_words = set(re.findall(r'\b[а-яё]+\b', answer_clean.lower())) if question_words: common_words = question_words.intersection(answer_words) features['keyword_overlap'] = len(common_words) / max(len(question_words), 1) features['response_relevance'] = min(1.0, len(answer_words) / max(len(question_words), 1)) except Exception as e: print(f"Ошибка семантических признаков: {e}") return features def extract_grammar_features(self, text: str) -> Dict[str, float]: """Грамматические признаки""" text_clean = self.clean_text(text) features = { 'grammar_quality': 0.5, # Базовая оценка 'has_punctuation': 0.0, 'sentence_completeness': 0.0 } if not text_clean: return features sentences = [s.strip() for s in re.split(r'[.!?]+', text_clean) if s.strip()] words = text_clean.split() if sentences: # Проверка пунктуации features['has_punctuation'] = 1.0 if any(mark in text_clean for mark in '.!?') else 0.0 # Полнота предложений complete_sentences = sum(1 for s in sentences if len(s.split()) >= 3) features['sentence_completeness'] = complete_sentences / max(len(sentences), 1) # Улучшенная эвристика грамматического качества grammar_score = 0.0 grammar_score += features['has_punctuation'] * 0.3 grammar_score += features['sentence_completeness'] * 0.4 # Дополнительные эвристики if len(words) > 5: avg_sentence_len = len(words) / len(sentences) if 5 <= avg_sentence_len <= 20: grammar_score += 0.2 elif avg_sentence_len > 20: grammar_score += 0.1 features['grammar_quality'] = min(1.0, grammar_score) return features def extract_style_features(self, text: str) -> Dict[str, float]: """Стилистические признаки""" text_clean = self.clean_text(text).lower() features = { 'has_greeting': 0.0, 'has_description': 0.0, 'has_connectors': 0.0, 'has_emotional_words': 0.0, 'style_score': 0.0 } if not text_clean: return features # Стилистические маркеры features.update({ 'has_greeting': float(any(greet in text_clean for greet in self.greeting_words)), 'has_description': float(any(desc in text_clean for desc in self.descriptive_words)), 'has_connectors': float(any(conn in text_clean for conn in self.connector_words)), 'has_emotional_words': float(any(emot in text_clean for emot in self.emotional_words)), }) # Оценка стиля style_indicators = sum([ features['has_greeting'], features['has_connectors'], features['has_emotional_words'] ]) features['style_score'] = min(1.0, style_indicators / 3) return features def extract_quality_features(self, text: str, question_type: int) -> Dict[str, float]: """Признаки качества ответа""" text_clean = self.clean_text(text) words = text_clean.split() word_count = len(words) features = { 'answer_length_sufficiency': min(1.0, word_count / 30), # Нормализованная длина 'content_richness': 0.0, 'engagement_level': 0.0 } if not text_clean: return features # Богатство контента (лексическое разнообразие + длина) lexical_diversity = len(set(words)) / max(word_count, 1) features['content_richness'] = min(1.0, (lexical_diversity + features['answer_length_sufficiency']) / 2) # Уровень вовлеченности engagement = 0.0 engagement += features['answer_length_sufficiency'] * 0.4 engagement += lexical_diversity * 0.3 engagement += (1.0 if '?' in text_clean else 0.0) * 0.3 features['engagement_level'] = engagement return features def extract_all_features(self, row: pd.Series) -> Dict[str, float]: """Извлечение всех признаков - ИСПРАВЛЕННАЯ ВЕРСИЯ""" try: # Безопасное извлечение данных question = row.get('Текст вопроса', row.get('Вопрос', '')) answer = row.get('Транскрибация ответа', row.get('Транскрипт', row.get('Ответ', ''))) question_type = row.get('№ вопроса', row.get('Тип вопроса', 1)) try: question_type = int(question_type) except: question_type = 1 features = {} # 1. Базовые признаки (надежные) basic_features = self.extract_basic_features(answer) features.update(basic_features) # 2. Семантические признаки semantic_features = self.extract_semantic_features(question, answer) features.update(semantic_features) # 3. Грамматические признаки grammar_features = self.extract_grammar_features(answer) features.update(grammar_features) # 4. Стилистические признаки style_features = self.extract_style_features(answer) features.update(style_features) # 5. Признаки качества quality_features = self.extract_quality_features(answer, question_type) features.update(quality_features) # 6. Тип вопроса features['question_type'] = float(question_type) # 7. ИСПРАВЛЕННЫЙ композитный показатель features['composite_quality_score'] = self._calculate_quality_score(features) return features except Exception as e: print(f"❌ Ошибка при извлечении признаков: {e}") # Возвращаем базовые признаки return self._get_fallback_features() def _calculate_quality_score(self, features: Dict[str, float]) -> float: """ИСПРАВЛЕННЫЙ расчет качества ответа""" # Веса для разных категорий weights = { # Семантика и релевантность (35%) 'keyword_overlap': 0.20, 'response_relevance': 0.15, # Грамматика и структура (25%) 'grammar_quality': 0.15, 'sentence_completeness': 0.10, # Стиль и вовлеченность (25%) 'style_score': 0.10, 'engagement_level': 0.15, # Содержание (15%) 'content_richness': 0.15 } total_score = 0.0 total_weight = 0.0 for feature, weight in weights.items(): if feature in features: value = features[feature] total_score += value * weight total_weight += weight # Нормализация на случай отсутствующих признаков if total_weight > 0: final_score = total_score / total_weight else: final_score = 0.5 # нейтральная оценка return min(1.0, max(0.0, final_score)) def _get_fallback_features(self) -> Dict[str, float]: """Базовые признаки при ошибке""" return { 'text_length': 0, 'word_count': 0, 'sentence_count': 0, 'avg_word_length': 0, 'lexical_diversity': 0, 'has_questions': 0, 'has_exclamations': 0, 'keyword_overlap': 0, 'response_relevance': 0, 'grammar_quality': 0.5, 'has_punctuation': 0, 'sentence_completeness': 0, 'has_greeting': 0, 'has_description': 0, 'has_connectors': 0, 'has_emotional_words': 0, 'style_score': 0, 'answer_length_sufficiency': 0, 'content_richness': 0, 'engagement_level': 0, 'question_type': 1, 'composite_quality_score': 0.5 } def extract_features_for_dataframe(self, df: pd.DataFrame, sample_size: int = None) -> pd.DataFrame: """Извлечение признаков для датафрейма""" if sample_size and sample_size < len(df): df = df.sample(sample_size, random_state=42) print(f"Взята выборка: {len(df)} строк") print(f"Извлечение признаков для {len(df)} строк...") features_list = [] successful = 0 for idx, row in df.iterrows(): if idx % 50 == 0 and idx > 0: print(f"Обработано {idx}/{len(df)} строк...") try: features = self.extract_all_features(row) features['original_index'] = idx features_list.append(features) successful += 1 except Exception as e: print(f"❌ Ошибка в строке {idx}: {e}") continue if features_list: features_df = pd.DataFrame(features_list) features_df.set_index('original_index', inplace=True) success_rate = successful / len(df) print(f"✅ Извлечение завершено! Успешно: {successful}/{len(df)} ({success_rate:.1%})") return features_df else: print("❌ Не удалось извлечь признаки") return pd.DataFrame() # Быстрая функция для тестирования def extract_quick_features(text: str) -> Dict[str, float]: extractor = RussianFeatureExtractor() return extractor.extract_basic_features(text) if __name__ == "__main__": # Тест исправленной версии extractor = RussianFeatureExtractor() test_data = { 'Текст вопроса': ['Расскажите о вашем городе'], 'Транскрибация ответа': ['Привет! Я живу в Москве. Это большой и красивый город с множеством парков и музеев.'], '№ вопроса': [1] } test_df = pd.DataFrame(test_data) features = extractor.extract_all_features(test_df.iloc[0]) print("🎯 ТЕСТ ИСПРАВЛЕННОЙ ВЕРСИИ:") print(f"Композитный показатель: {features['composite_quality_score']:.3f}") print(f"Грамматическое качество: {features['grammar_quality']:.3f}") print(f"Стилевой показатель: {features['style_score']:.3f}") print(f"Количество слов: {features['word_count']}")