I add command /analytics and I improve code

It's version 0.2.3
This commit is contained in:
Niken
2025-10-05 23:16:19 +03:00
parent 3ef1327b67
commit cce8c7dc70
7 changed files with 156 additions and 86 deletions
+26 -6
View File
@@ -1,6 +1,7 @@
from collections import Counter
import re
from datetime import datetime
import statistics
import tempfile
import json
@@ -198,7 +199,7 @@ def calculate_schedule_success_rate(stats):
def create_statistics_text(stats):
"""Создает текстовый отчет статистики"""
"""Создает текстовый отчет статистики с расширенными метриками"""
if 'error' in stats:
return f"❌ Ошибка анализа логов: {stats['error']}"
@@ -212,15 +213,28 @@ def create_statistics_text(stats):
text += f"• Строк в логе: {stats['total_lines']:,}\n\n"
# Производительность
handling_times = stats.get("handling_times", []) # сохрани список в analyze_bot_logs
median_time = statistics.median(handling_times) if handling_times else 0
text += "⚡ ПРОИЗВОДИТЕЛЬНОСТЬ:\n"
text += f"• Среднее время: {stats['performance']['avg_handling_time']:.0f} мс\n"
text += f"• Медиана времени: {median_time:.0f} мс\n"
text += f"• Сообщений обработано: {stats['performance']['handling_count']}\n"
text += f"• Успешных операций: {stats['success_rate']}%\n\n"
# Статус работы
duration = stats['time_period'].get('duration_hours', 0)
errors_total = sum(stats['errors'].values())
errors_per_hour = round(errors_total / duration, 2) if duration else 0
restarts = stats['restarts']
mtbf = round(duration / restarts, 2) if restarts else duration
text += "🔄 СТАТУС РАБОТЫ:\n"
text += f"• Перезапусков: {stats['restarts']}\n"
text += f"• Uptime: {stats['uptime_percentage']}%\n\n"
text += f"• Перезапусков: {restarts}\n"
text += f"• Uptime: {stats['uptime_percentage']}%\n"
text += f"• MTBF (среднее время между перезапусками): {mtbf} ч\n"
text += f"• Ошибок в час: {errors_per_hour}\n\n"
# Расписание
text += "📅 РАСПИСАНИЕ:\n"
@@ -231,17 +245,23 @@ def create_statistics_text(stats):
# Ошибки
text += "🚨 ОШИБКИ:\n"
text += f"• Всего ошибок: {sum(stats['errors'].values())}\n"
text += f"• Всего ошибок: {errors_total}\n"
text += f"• Предупреждений: {sum(stats['warnings'].values())}\n"
text += f"• Сетевых: {stats['network_errors']}\n"
text += f"• Браузера: {stats['browser_errors']}\n\n"
text += f"• Браузера: {stats['browser_errors']}\n"
# Топ-3 ошибок
if stats['errors']:
text += "• Топ ошибок:\n"
for err, count in stats['errors'].most_common(3):
text += f" - {err} ({count})\n"
text += "\n"
# Пользователи
text += "👥 АКТИВНОСТЬ:\n"
text += f"• Команд: {sum(stats['user_commands'].values())}\n"
text += f"• Групп: {len(stats['groups'])}\n"
# Топ групп
if stats['groups']:
text += "• Топ групп:\n"
for group, count in stats['groups'].most_common(3):
+48 -9
View File
@@ -1,27 +1,66 @@
import asyncio
import os
async def get_macbook_battery_level():
"""
Асинхронная функция для получения уровня заряда батареи на macOS.
Использует утилиту pmset.
"""
# Запускаем команду pmset -g batt
process = await asyncio.create_subprocess_exec(
"pmset", "-g", "batt",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
raise RuntimeError(f"Ошибка при выполнении pmset: {stderr.decode().strip()}")
output = stdout.decode().strip()
# Пример строки: " - InternalBattery-0 (id=1234567) 85%; charging; ..."
for part in output.splitlines():
if "%" in part:
percent_str = part.split("\t")[-1].split(";")[0].strip()
return int(percent_str.replace("%", ""))
raise ValueError("Не удалось определить уровень заряда")
raise ValueError("Не удалось определить уровень заряда")
async def get_process_usage(pid=None):
"""
Возвращает CPU, MEM% и RSS в мегабайтах для указанного процесса.
По умолчанию берёт текущий процесс (os.getpid()).
"""
if pid is None:
pid = os.getpid()
process = await asyncio.create_subprocess_exec(
"ps", "-p", str(pid), "-o", "%cpu,%mem,rss,comm",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
raise RuntimeError(f"Ошибка ps: {stderr.decode().strip()}")
lines = stdout.decode().strip().splitlines()
if len(lines) < 2:
return None
cpu, mem_percent, rss_kb, comm = lines[1].split(None, 3)
return {
"pid": pid,
"command": comm,
"cpu_percent": float(cpu),
"mem_percent": float(mem_percent),
"rss_mb": int(rss_kb) / 1024 # переводим КБ → МБ
}
async def main():
battery = await get_macbook_battery_level()
usage = await get_process_usage()
print(f"🔋 Батарея: {battery}%")
print(f"🖥 CPU: {usage['cpu_percent']}% | MEM: {usage['mem_percent']}% | RSS: {usage['rss_mb']:.2f} MB")
if __name__ == "__main__":
asyncio.run(main())