ba4676b0af
✅ Регистрация и верификация жильцов ✅ Рейтинг активности (уровни) ✅ Детект шпионов (Score 0-100) ✅ Анти-мат система (3 предупреждения → бан) ✅ Админ-панель с рассылкой ✅ Учёт квартир (несколько жильцов) ✅ Телефоны экстренных служб и мастеров ✅ Еженедельный экспорт JSON ✅ Прокси (socks5) для обхода РКН Бекап: backups/versions/v1.0_mvp_2026-02-26/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""
|
||
Topics Analyzer — анализ тем чата, облако тегов
|
||
Классификация сообщений по темам и категориям
|
||
"""
|
||
import logging
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, List, Tuple, Optional
|
||
from collections import defaultdict, Counter
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class TopicsAnalyzer:
|
||
"""Анализатор тем чата"""
|
||
|
||
def __init__(self, session: AsyncSession):
|
||
self.session = session
|
||
self.topics = self._load_topics()
|
||
|
||
def _load_topics(self) -> Dict:
|
||
"""Загрузить темы и ключевые слова"""
|
||
return {
|
||
'лифты': {
|
||
'keywords': ['лифт', 'лифты', 'лифтовая', 'кабина', 'шахта лифта', 'лифт застрял', 'лифт не работает'],
|
||
'emoji': '🛗'
|
||
},
|
||
'парковка': {
|
||
'keywords': ['парковк', 'машина', 'автомобиль', 'место парковк', 'припаркован', 'шлагбаум'],
|
||
'emoji': '🚗'
|
||
},
|
||
'уборка': {
|
||
'keywords': ['уборк', 'грязн', 'мусор', 'подъезд', 'лестниц', 'пол', 'пыль'],
|
||
'emoji': '🧹'
|
||
},
|
||
'соседи': {
|
||
'keywords': ['сосед', 'соседи', 'шум', 'громко', 'музыка', 'ремонт', 'сверлит'],
|
||
'emoji': '👥'
|
||
},
|
||
'вода': {
|
||
'keywords': ['вод', 'кран', 'труб', 'протечк', 'засор', 'канализаци', 'стояк'],
|
||
'emoji': '💧'
|
||
},
|
||
'электричество': {
|
||
'keywords': ['свет', 'электричеств', 'розетк', 'проводк', 'автомат', 'лампочк', 'мигает'],
|
||
'emoji': '⚡'
|
||
},
|
||
'отопление': {
|
||
'keywords': ['отоплен', 'батаре', 'радиатор', 'тепло', 'холод', 'температур'],
|
||
'emoji': '🔥'
|
||
},
|
||
'домофон': {
|
||
'keywords': ['домофон', 'ключ', 'карта', 'доступ', 'вход', 'дверь'],
|
||
'emoji': '🔑'
|
||
},
|
||
'мусор': {
|
||
'keywords': ['мусор', 'контейнер', 'баки', 'вывоз', 'раздельный', 'переработк'],
|
||
'emoji': '🗑️'
|
||
},
|
||
'дети': {
|
||
'keywords': ['ребенок', 'дети', 'детск', 'коляск', 'площадк', 'песочниц'],
|
||
'emoji': '👶'
|
||
},
|
||
'животные': {
|
||
'keywords': ['собак', 'кошк', 'животн', 'питомец', 'лают', 'мяукают'],
|
||
'emoji': '🐕'
|
||
},
|
||
'объявления': {
|
||
'keywords': ['объявлен', 'продам', 'куплю', 'отдам', 'ищу', 'предлагаю'],
|
||
'emoji': '📢'
|
||
},
|
||
'жалобы': {
|
||
'keywords': ['жалоб', 'жалуюсь', 'недовольн', 'возмущен', 'требую', 'безобраз'],
|
||
'emoji': '😠'
|
||
},
|
||
'благодарности': {
|
||
'keywords': ['спасиб', 'благодар', 'благодарю', 'молодец', 'хорошо', 'отличн'],
|
||
'emoji': '🙏'
|
||
},
|
||
'вопросы': {
|
||
'keywords': ['вопрос', 'подскажит', 'помогит', 'кто знает', 'как сделать'],
|
||
'emoji': '❓'
|
||
},
|
||
'предложения': {
|
||
'keywords': ['предлагаю', 'предложен', 'давайте', 'может быть', 'идея'],
|
||
'emoji': '💡'
|
||
},
|
||
'собрание': {
|
||
'keywords': ['собран', 'встреч', 'совещан', 'голосован', 'опрос', 'обсужден'],
|
||
'emoji': '📅'
|
||
},
|
||
'ук': {
|
||
'keywords': ['ук', 'управляющ', 'компан', 'жкх', 'паспортист', 'диспетчер'],
|
||
'emoji': '🏢'
|
||
},
|
||
}
|
||
|
||
async def analyze_message(self, text: str) -> Optional[str]:
|
||
"""
|
||
Классифицировать сообщение по теме
|
||
Returns: название темы или None
|
||
"""
|
||
if not text:
|
||
return None
|
||
|
||
text_lower = text.lower()
|
||
topic_scores = defaultdict(int)
|
||
|
||
for topic_name, topic_data in self.topics.items():
|
||
for keyword in topic_data['keywords']:
|
||
if keyword in text_lower:
|
||
topic_scores[topic_name] += 1
|
||
|
||
if not topic_scores:
|
||
return None
|
||
|
||
# Возвращаем тему с наибольшим score
|
||
best_topic = max(topic_scores, key=topic_scores.get)
|
||
return best_topic
|
||
|
||
async def get_topics_stats(self, days: int = 7) -> Dict:
|
||
"""
|
||
Получить статистику тем за период
|
||
Returns: dict с темами и количеством сообщений
|
||
"""
|
||
from database.models import Message
|
||
|
||
time_window = datetime.utcnow() - timedelta(days=days)
|
||
|
||
stmt = (
|
||
select(Message.text)
|
||
.where(Message.timestamp >= time_window)
|
||
.where(Message.text.isnot(None))
|
||
)
|
||
result = await self.session.execute(stmt)
|
||
messages = [row[0] for row in result.all()]
|
||
|
||
topic_counts = defaultdict(int)
|
||
|
||
for text in messages:
|
||
topic = await self.analyze_message(text)
|
||
if topic:
|
||
topic_counts[topic] += 1
|
||
|
||
# Считаем общее количество классифицированных сообщений
|
||
total = sum(topic_counts.values())
|
||
unclassified = len(messages) - total
|
||
|
||
# Формируем результат
|
||
result = {
|
||
'total_messages': len(messages),
|
||
'classified': total,
|
||
'unclassified': unclassified,
|
||
'topics': {}
|
||
}
|
||
|
||
for topic_name, count in sorted(topic_counts.items(), key=lambda x: x[1], reverse=True):
|
||
percent = (count / total * 100) if total > 0 else 0
|
||
emoji = self.topics[topic_name]['emoji']
|
||
result['topics'][topic_name] = {
|
||
'count': count,
|
||
'percent': round(percent, 1),
|
||
'emoji': emoji
|
||
}
|
||
|
||
return result
|
||
|
||
async def get_topics_cloud(self, days: int = 7) -> str:
|
||
"""
|
||
Получить облако тегов в текстовом виде
|
||
Returns: форматированный текст
|
||
"""
|
||
stats = await self.get_topics_stats(days)
|
||
|
||
if not stats['topics']:
|
||
return 'ℹ️ Нет данных за указанный период'
|
||
|
||
text = f'☁️ <b>Облако тем чата</b> (за {days} дн.)\n\n'
|
||
text += f'Всего сообщений: {stats["total_messages"]}\n'
|
||
text += f'Распознано: {stats["classified"]} ({stats["classified"]/stats["total_messages"]*100:.0f}%)\n\n'
|
||
|
||
# Формируем облако
|
||
sorted_topics = sorted(
|
||
stats['topics'].items(),
|
||
key=lambda x: x[1]['percent'],
|
||
reverse=True
|
||
)
|
||
|
||
for topic_name, data in sorted_topics:
|
||
bar_length = int(data['percent'] / 5) # 1 символ = 5%
|
||
bar = '█' * bar_length + '░' * (20 - bar_length)
|
||
|
||
text += f'{data["emoji"]} <b>{topic_name.capitalize()}</b>: {bar} {data["percent"]}% ({data["count"]})\n'
|
||
|
||
return text
|
||
|
||
async def get_top_topics(self, days: int = 7, limit: int = 5) -> List[Tuple[str, Dict]]:
|
||
"""
|
||
Получить топ тем за период
|
||
Returns: список (название_темы, данные)
|
||
"""
|
||
stats = await self.get_topics_stats(days)
|
||
|
||
sorted_topics = sorted(
|
||
stats['topics'].items(),
|
||
key=lambda x: x[1]['percent'],
|
||
reverse=True
|
||
)
|
||
|
||
return sorted_topics[:limit]
|
||
|
||
async def get_topic_trend(self, topic: str, days: int = 14) -> Dict:
|
||
"""
|
||
Получить тренд темы по дням
|
||
Returns: dict с датами и количеством сообщений
|
||
"""
|
||
from database.models import Message
|
||
|
||
time_window = datetime.utcnow() - timedelta(days=days)
|
||
|
||
stmt = (
|
||
select(Message.text, func.date(Message.timestamp).label('msg_date'))
|
||
.where(Message.timestamp >= time_window)
|
||
.where(Message.text.isnot(None))
|
||
)
|
||
result = await self.session.execute(stmt)
|
||
messages = [(row[0], row[1]) for row in result.all()]
|
||
|
||
daily_counts = defaultdict(int)
|
||
|
||
for text, date in messages:
|
||
topic = await self.analyze_message(text)
|
||
if topic == topic:
|
||
daily_counts[date] += 1
|
||
|
||
return {
|
||
'topic': topic,
|
||
'days': days,
|
||
'daily_counts': dict(sorted(daily_counts.items())),
|
||
'total': sum(daily_counts.values())
|
||
}
|
||
|
||
async def get_weekly_report(self) -> str:
|
||
"""
|
||
Сформировать еженедельный отчёт по темам
|
||
Returns: форматированный текст отчёта
|
||
"""
|
||
stats = await self.get_topics_stats(7)
|
||
prev_stats = await self.get_topics_stats(14)
|
||
|
||
text = '📊 <b>Еженедельный отчёт по темам чата</b>\n\n'
|
||
text += f'Период: последние 7 дней\n\n'
|
||
|
||
# Топ тем
|
||
text += '🔥 <b>Топ тем недели:</b>\n\n'
|
||
top_topics = await self.get_top_topics(7, 5)
|
||
|
||
for i, (topic_name, data) in enumerate(top_topics, 1):
|
||
text += f'{i}. {data["emoji"]} {topic_name.capitalize()}: {data["percent"]}% ({data["count"]} сообщ.)\n'
|
||
|
||
# Сравнение с прошлой неделей
|
||
text += '\n📈 <b>Динамика:</b>\n\n'
|
||
|
||
if prev_stats['topics']:
|
||
for topic_name, data in stats['topics'].items():
|
||
prev_data = prev_stats['topics'].get(topic_name, {'count': 0})
|
||
diff = data['count'] - prev_data['count']
|
||
|
||
if diff > 0:
|
||
text += f'↗️ {topic_name.capitalize()}: +{diff} ({data["emoji"]})\n'
|
||
elif diff < 0:
|
||
text += f'↘️ {topic_name.capitalize()}: {diff} ({data["emoji"]})\n'
|
||
|
||
text += '\n💡 Используйте /topics для детальной статистики'
|
||
|
||
return text
|