sync from local backup
This commit is contained in:
chunyu
@@ -0,0 +1,212 @@
|
||||
import re
|
||||
|
||||
from django.http import HttpResponseRedirect, HttpResponseNotFound, HttpResponseGone
|
||||
from django.utils import timezone
|
||||
from rest_framework.permissions import AllowAny, IsAuthenticated
|
||||
from rest_framework.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}
|
||||
)
|
||||
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 ShortUrl.objects.filter(code=custom_code).exists():
|
||||
return Response(
|
||||
{"code": 409, "message": "该短码已被使用，请更换"},
|
||||
status=status.HTTP_409_CONFLICT
|
||||
)
|
||||
code = custom_code
|
||||
else:
|
||||
last = ShortUrl.objects.order_by('-id').first()
|
||||
next_id = (last.id + 1) if last else 1
|
||||
code = encode_base62(next_id)
|
||||
while ShortUrl.objects.filter(code=code).exists():
|
||||
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 = ShortUrl.objects.create(
|
||||
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}
|
||||
)
|
||||
def get(self, request, code):
|
||||
try:
|
||||
short_url = ShortUrl.objects.get(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]
|
||||
|
||||
def get(self, request, code):
|
||||
try:
|
||||
short_url = ShortUrl.objects.get(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>')
|
||||
|
||||
ShortUrl.objects.filter(pk=short_url.pk).update(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}
|
||||
)
|
||||
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 = queryset.count()
|
||||
results = 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,
|
||||
}
|
||||
})
|
||||
Reference in New Issue
Block a user