chunyu_project/shorturl/views.py

import re

from django.http import HttpResponseRedirect, HttpResponseNotFound, HttpResponseGone
from django.utils import timezone
from rest_framework.permissions import AllowAny, IsAuthenticated
from adrf.views import APIView
from rest_framework.response import Response
from rest_framework import status
from drf_yasg.utils import swagger_auto_schema
from drf_yasg import openapi
from chunyu_project.common_schemas import success_response, error_response, unauthorized_response

from .models import ShortUrl


BASE62_ALPHABET = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'


def encode_base62(num):
    if num == 0:
        return BASE62_ALPHABET[0]
    result = []
    while num > 0:
        result.append(BASE62_ALPHABET[num % 62])
        num //= 62
    return ''.join(reversed(result))


def is_valid_url(url):
    if not url:
        return False
    return url.startswith('http://') or url.startswith('https://')


def is_valid_custom_code(code):
    return bool(re.match(r'^[a-zA-Z0-9_-]{3,20}$', code))


class ShortUrlShortenView(APIView):
    permission_classes = [AllowAny]

    @swagger_auto_schema(
        tags=['ShortUrl'],
        operation_summary='生成短链接',
        operation_description='将长URL转换为短链接，支持自定义短码和过期时间',
        request_body=openapi.Schema(
            type=openapi.TYPE_OBJECT,
            required=['url'],
            properties={
                'url': openapi.Schema(type=openapi.TYPE_STRING, description='需要缩短的长URL'),
                'custom_code': openapi.Schema(type=openapi.TYPE_STRING, description='自定义短码（3-20字符，仅限字母数字、连字符、下划线）'),
                'expire_days': openapi.Schema(type=openapi.TYPE_INTEGER, description='过期天数，为空则永不过期'),
            },
        ),
        responses={200: success_response, 400: error_response, 409: error_response}
    )
    async def post(self, request):
        url = request.data.get('url', '').strip()
        custom_code = request.data.get('custom_code', '').strip() or None
        expire_days = request.data.get('expire_days')

        if not is_valid_url(url):
            return Response(
                {"code": 400, "message": "请输入有效的URL（以 http:// 或 https:// 开头）"},
                status=status.HTTP_400_BAD_REQUEST
            )

        if custom_code:
            if not is_valid_custom_code(custom_code):
                return Response(
                    {"code": 400, "message": "自定义短码仅允许字母、数字、连字符、下划线，长度3-20字符"},
                    status=status.HTTP_400_BAD_REQUEST
                )
            if await ShortUrl.objects.filter(code=custom_code).aexists():
                return Response(
                    {"code": 409, "message": "该短码已被使用，请更换"},
                    status=status.HTTP_409_CONFLICT
                )
            code = custom_code
        else:
            last = await ShortUrl.objects.order_by('-id').afirst()
            next_id = (last.id + 1) if last else 1
            code = encode_base62(next_id)
            while await ShortUrl.objects.filter(code=code).aexists():
                next_id += 1
                code = encode_base62(next_id)

        expire_at = None
        if expire_days:
            try:
                days = int(expire_days)
                if days > 0:
                    expire_at = timezone.now() + timezone.timedelta(days=days)
            except (ValueError, TypeError):
                pass

        user = request.user if request.user.is_authenticated else None

        short_url = await ShortUrl.objects.acreate(
            code=code,
            original_url=url,
            custom_code=custom_code,
            creator=user,
            expire_at=expire_at,
        )

        short_url_base = request.build_absolute_uri('/s/')
        if not short_url_base.endswith('/'):
            short_url_base += '/'

        return Response({
            "code": 0,
            "data": {
                "code": short_url.code,
                "short_url": f"{short_url_base}{short_url.code}/",
                "original_url": short_url.original_url,
                "expire_at": short_url.expire_at.isoformat() if short_url.expire_at else None,
                "created_at": short_url.created_at.isoformat(),
            }
        }, status=status.HTTP_200_OK)


class ShortUrlInfoView(APIView):
    permission_classes = [AllowAny]

    @swagger_auto_schema(
        tags=['ShortUrl'],
        operation_summary='查询短链接信息',
        operation_description='通过短码查询短链接的详细信息',
        responses={200: success_response, 404: error_response}
    )
    async def get(self, request, code):
        try:
            short_url = await ShortUrl.objects.aget(code=code)
        except ShortUrl.DoesNotExist:
            return Response(
                {"code": 404, "message": "短链接不存在"},
                status=status.HTTP_404_NOT_FOUND
            )

        is_expired = short_url.expire_at and short_url.expire_at < timezone.now()

        return Response({
            "code": 0,
            "data": {
                "code": short_url.code,
                "original_url": short_url.original_url,
                "custom_code": short_url.custom_code,
                "created_at": short_url.created_at.isoformat(),
                "expire_at": short_url.expire_at.isoformat() if short_url.expire_at else None,
                "click_count": short_url.click_count,
                "is_expired": is_expired,
            }
        })


class ShortUrlRedirectView(APIView):
    permission_classes = [AllowAny]

    async def get(self, request, code):
        try:
            short_url = await ShortUrl.objects.aget(code=code)
        except ShortUrl.DoesNotExist:
            return HttpResponseNotFound('<h1>404 - 短链接不存在</h1>')

        if short_url.expire_at and short_url.expire_at < timezone.now():
            return HttpResponseGone('<h1>410 - 短链接已过期</h1>')

        await ShortUrl.objects.filter(pk=short_url.pk).aupdate(click_count=short_url.click_count + 1)

        return HttpResponseRedirect(short_url.original_url)


class ShortUrlListView(APIView):
    permission_classes = [IsAuthenticated]

    @swagger_auto_schema(
        tags=['ShortUrl'],
        operation_summary='获取当前用户的短链接列表',
        operation_description='分页返回当前登录用户创建的所有短链接',
        responses={200: success_response, 401: unauthorized_response}
    )
    async def get(self, request):
        queryset = ShortUrl.objects.filter(creator=request.user).order_by('-created_at')

        page = int(request.GET.get('page', 1))
        page_size = int(request.GET.get('page_size', 20))
        page_size = min(page_size, 100)

        start = (page - 1) * page_size
        end = start + page_size
        total = await queryset.acount()
        results = [item async for item in queryset[start:end]]

        data = [{
            "code": item.code,
            "short_url": request.build_absolute_uri(f'/s/{item.code}/'),
            "original_url": item.original_url,
            "click_count": item.click_count,
            "expire_at": item.expire_at.isoformat() if item.expire_at else None,
            "created_at": item.created_at.isoformat(),
        } for item in results]

        return Response({
            "code": 0,
            "data": {
                "total": total,
                "page": page,
                "page_size": page_size,
                "results": data,
            }
        })