306 lines
12 KiB
Python
306 lines
12 KiB
Python
"""
|
||
Секретная страница статистики в стиле Spotify Wrapped / Яндекс.Музыка.
|
||
Красивые анимации, диаграммы и визуализации.
|
||
"""
|
||
import json
|
||
from datetime import timedelta, datetime
|
||
from collections import defaultdict
|
||
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
|
||
from django.db.models import Count, Q, Min, Max, Avg, Sum
|
||
from django.db.models.functions import TruncDate, TruncMonth, ExtractWeekDay, ExtractHour
|
||
from django.utils import timezone
|
||
from django.views.generic import TemplateView
|
||
|
||
from ..models import ObjItem, Source, Satellite, Geo, Parameter
|
||
|
||
|
||
class AdminOnlyMixin(UserPassesTestMixin):
|
||
"""Mixin to restrict access to admin role only."""
|
||
|
||
def test_func(self):
|
||
return (
|
||
self.request.user.is_authenticated and
|
||
hasattr(self.request.user, 'customuser') and
|
||
self.request.user.customuser.role == 'admin'
|
||
)
|
||
|
||
def handle_no_permission(self):
|
||
from django.contrib import messages
|
||
from django.shortcuts import redirect
|
||
messages.error(self.request, 'Доступ запрещён. Требуется роль администратора.')
|
||
return redirect('mainapp:home')
|
||
|
||
|
||
class SecretStatsView(LoginRequiredMixin, AdminOnlyMixin, TemplateView):
|
||
"""Секретная страница статистики - итоги года в стиле Spotify Wrapped."""
|
||
|
||
template_name = 'mainapp/secret_stats.html'
|
||
|
||
def get_year_range(self):
|
||
"""Получает диапазон дат для текущего года."""
|
||
now = timezone.now()
|
||
year = self.request.GET.get('year', now.year)
|
||
try:
|
||
year = int(year)
|
||
except (ValueError, TypeError):
|
||
year = now.year
|
||
|
||
date_from = datetime(year, 1, 1).date()
|
||
date_to = datetime(year, 12, 31).date()
|
||
|
||
return date_from, date_to, year
|
||
|
||
def get_base_queryset(self, date_from, date_to):
|
||
"""Возвращает базовый queryset ObjItem с фильтрами по дате ГЛ."""
|
||
qs = ObjItem.objects.filter(
|
||
geo_obj__isnull=False,
|
||
geo_obj__timestamp__isnull=False,
|
||
geo_obj__timestamp__date__gte=date_from,
|
||
geo_obj__timestamp__date__lte=date_to
|
||
)
|
||
return qs
|
||
|
||
def get_main_stats(self, date_from, date_to):
|
||
"""Основная статистика: точки и объекты."""
|
||
base_qs = self.get_base_queryset(date_from, date_to)
|
||
|
||
total_points = base_qs.count()
|
||
total_sources = base_qs.filter(source__isnull=False).values('source').distinct().count()
|
||
|
||
return {
|
||
'total_points': total_points,
|
||
'total_sources': total_sources,
|
||
}
|
||
|
||
def get_new_emissions(self, date_from, date_to):
|
||
"""
|
||
Новые излучения - объекты, у которых имя появилось впервые в выбранном периоде.
|
||
"""
|
||
# Получаем все имена объектов, которые появились ДО выбранного периода
|
||
existing_names = set(
|
||
ObjItem.objects.filter(
|
||
geo_obj__isnull=False,
|
||
geo_obj__timestamp__isnull=False,
|
||
geo_obj__timestamp__date__lt=date_from,
|
||
name__isnull=False
|
||
).exclude(name='').values_list('name', flat=True).distinct()
|
||
)
|
||
|
||
# Базовый queryset для выбранного периода
|
||
period_qs = self.get_base_queryset(date_from, date_to).filter(
|
||
name__isnull=False
|
||
).exclude(name='')
|
||
|
||
# Получаем уникальные имена в выбранном периоде
|
||
period_names = set(period_qs.values_list('name', flat=True).distinct())
|
||
|
||
# Новые имена = имена в периоде, которых не было раньше
|
||
new_names = period_names - existing_names
|
||
|
||
if not new_names:
|
||
return {'count': 0, 'objects': [], 'sources_count': 0}
|
||
|
||
# Получаем данные о новых объектах
|
||
objitems_data = period_qs.filter(
|
||
name__in=new_names
|
||
).select_related(
|
||
'source__info', 'source__ownership'
|
||
).values(
|
||
'name',
|
||
'source__info__name',
|
||
'source__ownership__name'
|
||
).distinct()
|
||
|
||
seen_names = set()
|
||
new_objects = []
|
||
|
||
for item in objitems_data:
|
||
name = item['name']
|
||
if name not in seen_names:
|
||
seen_names.add(name)
|
||
new_objects.append({
|
||
'name': name,
|
||
'info': item['source__info__name'] or '-',
|
||
'ownership': item['source__ownership__name'] or '-',
|
||
})
|
||
|
||
new_objects.sort(key=lambda x: x['name'])
|
||
|
||
# Количество источников для новых излучений
|
||
new_sources_count = period_qs.filter(
|
||
name__in=new_names, source__isnull=False
|
||
).values('source').distinct().count()
|
||
|
||
return {
|
||
'count': len(new_names),
|
||
'objects': new_objects[:20], # Топ-20 для отображения
|
||
'sources_count': new_sources_count
|
||
}
|
||
|
||
def get_satellite_stats(self, date_from, date_to):
|
||
"""Статистика по спутникам."""
|
||
base_qs = self.get_base_queryset(date_from, date_to)
|
||
|
||
stats = base_qs.filter(
|
||
parameter_obj__id_satellite__isnull=False
|
||
).values(
|
||
'parameter_obj__id_satellite__id',
|
||
'parameter_obj__id_satellite__name'
|
||
).annotate(
|
||
points_count=Count('id'),
|
||
sources_count=Count('source', distinct=True),
|
||
unique_names=Count('name', distinct=True)
|
||
).order_by('-points_count')
|
||
|
||
return list(stats)
|
||
|
||
def get_monthly_stats(self, date_from, date_to):
|
||
"""Статистика по месяцам."""
|
||
base_qs = self.get_base_queryset(date_from, date_to)
|
||
|
||
monthly = base_qs.annotate(
|
||
month=TruncMonth('geo_obj__timestamp')
|
||
).values('month').annotate(
|
||
points=Count('id'),
|
||
sources=Count('source', distinct=True)
|
||
).order_by('month')
|
||
|
||
return list(monthly)
|
||
|
||
def get_weekday_stats(self, date_from, date_to):
|
||
"""Статистика по дням недели."""
|
||
base_qs = self.get_base_queryset(date_from, date_to)
|
||
|
||
weekday = base_qs.annotate(
|
||
weekday=ExtractWeekDay('geo_obj__timestamp')
|
||
).values('weekday').annotate(
|
||
points=Count('id')
|
||
).order_by('weekday')
|
||
|
||
return list(weekday)
|
||
|
||
def get_hourly_stats(self, date_from, date_to):
|
||
"""Статистика по часам."""
|
||
base_qs = self.get_base_queryset(date_from, date_to)
|
||
|
||
hourly = base_qs.annotate(
|
||
hour=ExtractHour('geo_obj__timestamp')
|
||
).values('hour').annotate(
|
||
points=Count('id')
|
||
).order_by('hour')
|
||
|
||
return list(hourly)
|
||
|
||
def get_top_objects(self, date_from, date_to):
|
||
"""Топ объектов по количеству точек."""
|
||
base_qs = self.get_base_queryset(date_from, date_to)
|
||
|
||
top = base_qs.filter(
|
||
name__isnull=False
|
||
).exclude(name='').values('name').annotate(
|
||
points=Count('id')
|
||
).order_by('-points')[:10]
|
||
|
||
return list(top)
|
||
|
||
def get_busiest_day(self, date_from, date_to):
|
||
"""Самый активный день."""
|
||
base_qs = self.get_base_queryset(date_from, date_to)
|
||
|
||
daily = base_qs.annotate(
|
||
date=TruncDate('geo_obj__timestamp')
|
||
).values('date').annotate(
|
||
points=Count('id')
|
||
).order_by('-points').first()
|
||
|
||
return daily
|
||
|
||
def get_context_data(self, **kwargs):
|
||
context = super().get_context_data(**kwargs)
|
||
|
||
date_from, date_to, year = self.get_year_range()
|
||
|
||
# Основная статистика
|
||
main_stats = self.get_main_stats(date_from, date_to)
|
||
|
||
# Новые излучения
|
||
new_emissions = self.get_new_emissions(date_from, date_to)
|
||
|
||
# Статистика по спутникам
|
||
satellite_stats = self.get_satellite_stats(date_from, date_to)
|
||
|
||
# Статистика по месяцам
|
||
monthly_stats = self.get_monthly_stats(date_from, date_to)
|
||
|
||
# Статистика по дням недели
|
||
weekday_stats = self.get_weekday_stats(date_from, date_to)
|
||
|
||
# Статистика по часам
|
||
hourly_stats = self.get_hourly_stats(date_from, date_to)
|
||
|
||
# Топ объектов
|
||
top_objects = self.get_top_objects(date_from, date_to)
|
||
|
||
# Самый активный день
|
||
busiest_day = self.get_busiest_day(date_from, date_to)
|
||
|
||
# Доступные годы для выбора
|
||
years_with_data = ObjItem.objects.filter(
|
||
geo_obj__isnull=False,
|
||
geo_obj__timestamp__isnull=False
|
||
).dates('geo_obj__timestamp', 'year')
|
||
available_years = sorted([d.year for d in years_with_data], reverse=True)
|
||
|
||
# JSON данные для графиков
|
||
monthly_data_json = json.dumps([
|
||
{
|
||
'month': item['month'].strftime('%Y-%m') if item['month'] else None,
|
||
'month_name': item['month'].strftime('%B') if item['month'] else None,
|
||
'points': item['points'],
|
||
'sources': item['sources'],
|
||
}
|
||
for item in monthly_stats
|
||
])
|
||
|
||
satellite_stats_json = json.dumps(satellite_stats)
|
||
|
||
weekday_names = ['Вс', 'Пн', 'Вт', 'Ср', 'Чт', 'Пт', 'Сб']
|
||
weekday_data_json = json.dumps([
|
||
{
|
||
'weekday': item['weekday'],
|
||
'weekday_name': weekday_names[item['weekday'] - 1] if item['weekday'] else '',
|
||
'points': item['points'],
|
||
}
|
||
for item in weekday_stats
|
||
])
|
||
|
||
hourly_data_json = json.dumps([
|
||
{
|
||
'hour': item['hour'],
|
||
'points': item['points'],
|
||
}
|
||
for item in hourly_stats
|
||
])
|
||
|
||
top_objects_json = json.dumps(top_objects)
|
||
|
||
context.update({
|
||
'year': year,
|
||
'available_years': available_years,
|
||
'total_points': main_stats['total_points'],
|
||
'total_sources': main_stats['total_sources'],
|
||
'new_emissions_count': new_emissions['count'],
|
||
'new_emissions_sources': new_emissions['sources_count'],
|
||
'new_emission_objects': new_emissions['objects'],
|
||
'satellite_stats': satellite_stats[:10], # Топ-10
|
||
'satellite_count': len(satellite_stats),
|
||
'busiest_day': busiest_day,
|
||
'monthly_data_json': monthly_data_json,
|
||
'satellite_stats_json': satellite_stats_json,
|
||
'weekday_data_json': weekday_data_json,
|
||
'hourly_data_json': hourly_data_json,
|
||
'top_objects_json': top_objects_json,
|
||
})
|
||
|
||
return context
|