import os import uuid from datetime import datetime from typing import Optional, List from django.conf import settings from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from ninja import Router from base.models import FriendLink, LinkCategory from base.schema import FriendLinkCreateSchema, FriendLinkSchema, LinkCategoryCreateSchema, LinkCategorySchema # Create your views here. router = Router(tags=['base']) @csrf_exempt @require_http_methods(["POST"]) def custom_upload_file(request): """ 自定义 CKEditor 5 上传视图 支持:图片 + PDF / Word / Excel / ZIP 等所有文件 """ file = request.FILES.get('upload') if not file: return JsonResponse({"error": "请选择文件"}, status=400) # 安全处理文件名(避免中文/特殊字符报错) ext = file.name.split('.')[-1].lower() filename = f"{uuid.uuid4().hex}_{datetime.now().strftime('%Y%m%d%H%M%S')}.{ext}" # 保存路径 upload_dir = os.path.join(settings.MEDIA_ROOT, 'uploads') os.makedirs(upload_dir, exist_ok=True) file_path = os.path.join(upload_dir, filename) # 保存文件 with open(file_path, 'wb+') as f: for chunk in file.chunks(): f.write(chunk) # 返回文件 URL(CKEditor 规范格式) file_url = f"{settings.MEDIA_URL}uploads/{filename}" return JsonResponse({ "url": file_url, "default": file_url }) # ========================= # 分类 CRUD # ========================= @router.get("/link-categories", response=List[LinkCategorySchema]) def list_categories(request): return LinkCategory.objects.all() @router.post("/link-categories", response=LinkCategorySchema) def create_category(request, data: LinkCategoryCreateSchema): return LinkCategory.objects.create(**data.dict()) @router.put("/link-categories/{id}", response=LinkCategorySchema) def update_category(request, id: int, data: LinkCategoryCreateSchema): obj = LinkCategory.objects.get(id=id) for k, v in data.dict().items(): setattr(obj, k, v) obj.save() return obj @router.delete("/link-categories/{id}") def delete_category(request, id: int): LinkCategory.objects.get(id=id).delete() return {"success": True} # ========================= # 友情链接 CRUD # ========================= @router.get("/friend-links", response=List[FriendLinkSchema]) def list_links(request, category_id: Optional[int] = None): query = FriendLink.objects.all() if category_id: query = query.filter(category_id=category_id) return query @router.post("/friend-links", response=FriendLinkSchema) def create_link(request, data: FriendLinkCreateSchema): return FriendLink.objects.create(**data.dict()) @router.put("/friend-links/{id}", response=FriendLinkSchema) def update_link(request, id: int, data: FriendLinkCreateSchema): obj = FriendLink.objects.get(id=id) for k, v in data.dict().items(): setattr(obj, k, v) obj.save() return obj @router.delete("/friend-links/{id}") def delete_link(request, id: int): FriendLink.objects.get(id=id).delete() return {"success": True}