import logging from datetime import timedelta from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated, AllowAny from django.utils.translation import gettext as _ from django.utils import timezone from django.db.models import Count from django.utils.decorators import method_decorator from django.views.decorators.cache import cache_page from pingcap_sso.api.base import Endpoint from pingcap_sso.discourse.lib import badge_list from pingcap_sso.discourse.models import DiscourseUserBadge from pingcap_sso.models import User, AskTUGSiteConfig logger = logging.getLogger(__name__) class BadgeListEndpoint(Endpoint): permission_classes = (IsAuthenticated, ) def get(self, request, *args, **kwargs): return Response({ 'detail': _('success'), 'data': badge_list(request.user) }) class BadgeTopListEndpoint(Endpoint): permission_classes = (AllowAny, ) def get(self, request, *args, **kwargs): now = timezone.now() period_map = { 'all': now.replace(year=2000), 'weekly': now - timedelta(days=7), 'monthly': now - timedelta(days=31), 'yearly': now - timedelta(days=365), } try: limit = int(request.GET.get('limit', 50)) period = request.GET.get('period', 'all') if limit > 100 or limit < 1: raise Exception('limit should between 1 ~ 100') if period not in period_map: raise Exception('invalid period, it could be {}'.format( ', '.join(period_map.keys()))) except Exception as e: return Response({ 'detail': str(e) }, 400) qs = DiscourseUserBadge.objects\ .filter(granted_at__gte=period_map[period], badge__enabled=True)\ .values('user__username')\ .annotate(count=Count('id'))\ .order_by('-count')[:limit] usernames = [d['user__username'] for d in qs] users = User.objects.filter(username__in=usernames).values('username', 'avatar_url') user_map = {u['username']: u for u in users} data = [{ 'ranking': i + 1, 'user': { 'username': d['user__username'], 'avatar_url': user_map.get(d['user__username'], {}).get('avatar_url', ''), }, 'count': d['count'], } for i, d in enumerate(qs)] return Response({ 'data': data, 'limit': limit, 'period': period }) class SiteConfigEndpoint(Endpoint): permission_classes = (AllowAny, ) throttle_classes = () @method_decorator(cache_page(60 * 3)) def get(self, request, *args, **kwargs): try: banner_image_url = AskTUGSiteConfig.objects.get(name='banner_image_url').typed_value banner_link = AskTUGSiteConfig.objects.get(name='banner_link').typed_value banner_title = AskTUGSiteConfig.objects.get(name='banner_title').typed_value banner = { "image_url": banner_image_url, "link": banner_link, "title": banner_title } except AskTUGSiteConfig.DoesNotExist: banner = None return Response({ 'detail': _('success'), 'data': { "banner": banner } })