from datetime import date, timedelta import requests from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.permissions import IsAuthenticated, AllowAny from rest_framework.response import Response from rest_framework.views import APIView from django.shortcuts import redirect from drf_spectacular.utils import extend_schema from apps.analytics.utils import client_ip from .models import URL, URLClick, QRCode, ImageToolUsage, FileConversion, CurrencyRate from .serializers import ( URLSerializer, URLCreateSerializer, URLClickSerializer, QRCodeSerializer, QRCodeCreateSerializer, ImageToolUsageSerializer, ImageToolUsageCreateSerializer, FileConversionSerializer, FileConversionCreateSerializer ) class URLViewSet(viewsets.ModelViewSet): """ ViewSet for managing shortened URLs. """ queryset = URL.objects.select_related("created_by") def get_serializer_class(self): if self.action in ['create', 'update', 'partial_update']: return URLCreateSerializer return URLSerializer def get_queryset(self): user = self.request.user if user.is_authenticated: return URL.objects.filter(created_by=user) return URL.objects.none() def get_permissions(self): # Short links must be followable by anonymous visitors (links shared on # social, email, etc). Creation/management stays authenticated. if self.action == "redirect_to_original": return [AllowAny()] return super().get_permissions() def perform_create(self, serializer): serializer.save(created_by=self.request.user) @extend_schema( description="Redirect to the original URL using short code" ) @action(detail=False, methods=['get'], url_path=r'(?P[^/]+)') def redirect_to_original(self, request, short_code=None): try: url = URL.objects.get(short_code=short_code) if url.is_expired(): return Response( {'error': 'This URL has expired'}, status=status.HTTP_410_GONE ) url.increment_clicks() URLClick.objects.create( url=url, ip_address=client_ip(request) or None, user_agent=request.META.get('HTTP_USER_AGENT', '')[:512], referer=request.META.get('HTTP_REFERER', '') ) return redirect(url.original_url) except URL.DoesNotExist: return Response( {'error': 'URL not found'}, status=status.HTTP_404_NOT_FOUND ) class QRCodeViewSet(viewsets.ModelViewSet): """ ViewSet for generating and managing QR codes. """ queryset = QRCode.objects.all() def get_serializer_class(self): if self.action in ['create', 'update', 'partial_update']: return QRCodeCreateSerializer return QRCodeSerializer def get_queryset(self): user = self.request.user if user.is_authenticated: return QRCode.objects.filter(created_by=user).select_related("created_by") return QRCode.objects.none() def perform_create(self, serializer): serializer.save(created_by=self.request.user) @extend_schema( description="Download QR code as image" ) @action(detail=False, methods=['get'], url_path=r'(?P[^/]+)/download') def download(self, request, short_code=None): import io import qrcode from django.http import HttpResponse try: qr = QRCode.objects.get(short_code=short_code) except QRCode.DoesNotExist: return Response( {'error': 'QR code not found'}, status=status.HTTP_404_NOT_FOUND ) qr_image = qrcode.make(qr.content) buffer = io.BytesIO() qr_image.save(buffer, format=qr.format.upper()) buffer.seek(0) qr.downloads += 1 qr.save(update_fields=['downloads']) return HttpResponse(buffer.getvalue(), content_type=f'image/{qr.format}') class ImageToolUsageViewSet(viewsets.ModelViewSet): """ ViewSet for tracking image tool usage. """ queryset = ImageToolUsage.objects.all() serializer_class = ImageToolUsageSerializer def get_serializer_class(self): if self.action in ['create', 'update', 'partial_update']: return ImageToolUsageCreateSerializer return ImageToolUsageSerializer def get_queryset(self): user = self.request.user if user.is_authenticated: return ImageToolUsage.objects.filter(created_by=user).select_related("created_by") return ImageToolUsage.objects.none() def perform_create(self, serializer): serializer.save(created_by=self.request.user) class FileConversionViewSet(viewsets.ModelViewSet): """ ViewSet for tracking file conversions. """ queryset = FileConversion.objects.all() serializer_class = FileConversionSerializer def get_serializer_class(self): if self.action in ['create', 'update', 'partial_update']: return FileConversionCreateSerializer return FileConversionSerializer def get_queryset(self): user = self.request.user if user.is_authenticated: return FileConversion.objects.filter(created_by=user).select_related("created_by") return FileConversion.objects.none() def perform_create(self, serializer): serializer.save(created_by=self.request.user) class CurrencyHistoryView(APIView): """1-year exchange rate history for a base→quote pair, cached in DB.""" permission_classes = [IsAuthenticated] FRANKFURTER = "https://api.frankfurter.app" SUPPORTED = {"USD","EUR","GBP","JPY","AUD","CAD","CHF","CNY","HKD","SGD","IDR","MYR","THB","PHP","INR","KRW","NZD","SEK","NOK","DKK","ZAR","TRY","BRL","MXN","PLN","CZK","HUF","ILS","BGN","RON","ISK"} def get(self, request): base = (request.query_params.get("from") or "").upper().strip() quote = (request.query_params.get("to") or "").upper().strip() if base not in self.SUPPORTED or quote not in self.SUPPORTED: return Response({"error": "Unsupported or missing currency code."}, status=400) if base == quote: return Response({"base": base, "quote": quote, "series": []}) end = date.today() start = end - timedelta(days=365) qs = CurrencyRate.objects.filter(base=base, quote=quote, date__gte=start, date__lte=end) latest = qs.order_by("-date").values_list("date", flat=True).first() if latest is None or latest < end - timedelta(days=1): self._refresh(base, quote, start, end) qs = CurrencyRate.objects.filter(base=base, quote=quote, date__gte=start, date__lte=end) series = [{"date": r.date.isoformat(), "rate": float(r.rate)} for r in qs.order_by("date")] return Response({"base": base, "quote": quote, "series": series}) def _refresh(self, base, quote, start, end): url = f"{self.FRANKFURTER}/{start.isoformat()}..{end.isoformat()}" try: r = requests.get(url, params={"from": base, "to": quote}, timeout=15) r.raise_for_status() data = r.json() except Exception: return rates = data.get("rates") or {} objs = [] for d_iso, kv in rates.items(): v = kv.get(quote) if v is None: continue objs.append(CurrencyRate(base=base, quote=quote, date=d_iso, rate=v)) if objs: CurrencyRate.objects.bulk_create(objs, ignore_conflicts=True)