360 lines
10 KiB
Python
360 lines
10 KiB
Python
import hashlib
|
||
import os
|
||
import uuid
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional, List
|
||
|
||
from django.conf import settings
|
||
from django.http import JsonResponse
|
||
from django.shortcuts import get_object_or_404
|
||
from django.views.decorators.csrf import csrf_exempt
|
||
from django.views.decorators.http import require_http_methods
|
||
from loguru import logger
|
||
from ninja import Router, UploadedFile, File
|
||
from ninja.pagination import paginate, PageNumberPagination
|
||
|
||
from base.models import FriendLink, LinkCategory, PictureLink, MainMenu, BaseInfo
|
||
from base.schema import *
|
||
|
||
# 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
|
||
})
|
||
|
||
|
||
UPLOAD_DIR = Path("media/uploads")
|
||
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
@router.post("/upload/{category}")
|
||
def upload_file(request, category: str, file: UploadedFile = File(...), ):
|
||
try:
|
||
# 计算 md5
|
||
md5 = hashlib.md5(file.read()).hexdigest()
|
||
file.seek(0) # 重置文件指针
|
||
|
||
ext = file.name.split('.')[-1].lower()
|
||
filename = f"{md5}.{ext}"
|
||
|
||
upload_dir = os.path.join(settings.MEDIA_ROOT, f"uploads/{category}/")
|
||
os.makedirs(upload_dir, exist_ok=True)
|
||
file_path = os.path.join(upload_dir, filename)
|
||
|
||
with open(file_path, "wb") as f:
|
||
f.write(file.read())
|
||
|
||
url = f"{settings.MEDIA_URL}uploads/{category}/{filename}"
|
||
return {"url": url}
|
||
except Exception as e:
|
||
logger.exception("Upload failed")
|
||
return {"error": str(e)}
|
||
|
||
|
||
# =========================
|
||
# 分类 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}
|
||
|
||
|
||
# ================= 列表与创建 =================
|
||
@router.get("/picture-link/", response=List[PictureLinkSchema])
|
||
def get_picture_links(request):
|
||
"""
|
||
获取所有图片链接
|
||
GET /api/picture-link/
|
||
"""
|
||
# 这里的排序与 models.py 中的 Meta.ordering 保持一致
|
||
return PictureLink.objects.all().order_by('order', '-created_at')
|
||
|
||
|
||
@router.post("/picture-link/", response=PictureLinkSchema)
|
||
def create_picture_link(request, payload: PictureLinkCreateSchema):
|
||
"""
|
||
新增图片链接
|
||
POST /api/picture-link/
|
||
"""
|
||
# 直接利用 payload 创建对象
|
||
# 注意：这里调用 save() 会触发你 models.py 中定义的 "最多4个启用" 的逻辑
|
||
link = PictureLink.objects.create(
|
||
name=payload.name,
|
||
picture=payload.picture,
|
||
url=payload.url,
|
||
order=payload.order,
|
||
is_active=payload.is_active
|
||
)
|
||
return link
|
||
|
||
|
||
# ================= 单个对象操作 =================
|
||
@router.get("/picture-link/{link_id}", response=PictureLinkSchema)
|
||
def get_picture_link(request, link_id: int):
|
||
"""
|
||
获取单个图片链接详情
|
||
GET /api/picture-link/{id}/
|
||
"""
|
||
return get_object_or_404(PictureLink, id=link_id)
|
||
|
||
|
||
@router.put("/picture-link/{link_id}", response=PictureLinkSchema)
|
||
def update_picture_link(request, link_id: int, payload: PictureLinkUpdateSchema):
|
||
"""
|
||
更新图片链接
|
||
PUT /api/picture-link/{id}/
|
||
"""
|
||
link = get_object_or_404(PictureLink, id=link_id)
|
||
|
||
# 更新字段
|
||
if payload.name is not None: link.name = payload.name
|
||
if payload.picture is not None: link.picture = payload.picture
|
||
if payload.url is not None: link.url = payload.url
|
||
if payload.order is not None: link.order = payload.order
|
||
if payload.is_active is not None: link.is_active = payload.is_active
|
||
|
||
# 保存，同样会触发 models.py 中的限制逻辑
|
||
link.save()
|
||
return link
|
||
|
||
|
||
@router.delete("/picture-link/{link_id}")
|
||
def delete_picture_link(request, link_id: int):
|
||
"""
|
||
删除图片链接
|
||
DELETE /api/picture-link/{id}/
|
||
"""
|
||
link = get_object_or_404(PictureLink, id=link_id)
|
||
link.delete()
|
||
return {"success": True}
|
||
|
||
|
||
def build_menu_tree(menus: List[MainMenu]) -> List[MainMenuOut]:
|
||
"""
|
||
将平铺的菜单列表转换为嵌套的树形结构。
|
||
这是一个高效的 O(n) 算法。
|
||
"""
|
||
menu_map = {}
|
||
root_menus = []
|
||
|
||
# 第一步：将所有菜单放入字典，并初始化 children 列表
|
||
for menu in menus:
|
||
menu_data = MainMenuOut(
|
||
id=menu.id,
|
||
parent_id=menu.parent_id,
|
||
title=menu.title,
|
||
code=menu.code,
|
||
url=menu.url,
|
||
is_external=menu.is_external,
|
||
is_carousel=menu.is_carousel,
|
||
order=menu.order,
|
||
visible=menu.visible,
|
||
children=[]
|
||
)
|
||
menu_map[menu.id] = menu_data
|
||
if menu.parent_id is None:
|
||
root_menus.append(menu_data)
|
||
|
||
# 第二步：遍历字典，将子菜单挂载到父菜单上
|
||
for menu_id, menu_obj in menu_map.items():
|
||
if menu_obj.parent_id is not None and menu_obj.parent_id in menu_map:
|
||
parent = menu_map[menu_obj.parent_id]
|
||
parent.children.append(menu_obj)
|
||
|
||
# 第三步：对每个层级的菜单按 'order' 排序
|
||
def sort_children(menu_list):
|
||
menu_list.sort(key=lambda x: (x.order, x.title))
|
||
for menu in menu_list:
|
||
sort_children(menu.children)
|
||
|
||
sort_children(root_menus)
|
||
return root_menus
|
||
|
||
|
||
@router.get("main-menu/", response=List[MainMenuOut])
|
||
def get_menu_tree(request):
|
||
"""
|
||
获取完整的导航菜单树。
|
||
GET /api/main-menu/
|
||
"""
|
||
all_menus = MainMenu.objects.all().select_related('parent')
|
||
return build_menu_tree(all_menus)
|
||
|
||
|
||
@router.post("/main-menu/", response=MainMenuOut)
|
||
def create_menu(request, payload: MainMenuCreate):
|
||
"""
|
||
创建一个新的菜单项。
|
||
POST /api/main-menu/
|
||
"""
|
||
parent = None
|
||
if payload.parent_id:
|
||
parent = get_object_or_404(MainMenu, id=payload.parent_id)
|
||
|
||
menu = MainMenu.objects.create(
|
||
parent=parent,
|
||
**payload.dict(exclude={'parent_id'})
|
||
)
|
||
return menu
|
||
|
||
|
||
@router.put("/main-menu/{menu_id}", response=MainMenuOut)
|
||
def update_menu(request, menu_id: int, payload: MainMenuUpdate):
|
||
"""
|
||
更新一个菜单项。
|
||
PUT /api/main-menu/{menu_id}/
|
||
"""
|
||
menu = get_object_or_404(MainMenu, id=menu_id)
|
||
|
||
# 处理 parent_id
|
||
if payload.parent_id is not None:
|
||
if payload.parent_id != menu.parent_id:
|
||
new_parent = get_object_or_404(MainMenu, id=payload.parent_id)
|
||
menu.parent = new_parent
|
||
else:
|
||
menu.parent = None
|
||
|
||
# 更新其他字段
|
||
for attr, value in payload.dict(exclude={'parent_id'}).items():
|
||
setattr(menu, attr, value)
|
||
|
||
menu.save()
|
||
return menu
|
||
|
||
|
||
@router.delete("/main-menu/{menu_id}")
|
||
def delete_menu(request, menu_id: int):
|
||
"""
|
||
删除一个菜单项（及其所有子项）。
|
||
DELETE /api/main-menu/{menu_id}/
|
||
"""
|
||
menu = get_object_or_404(MainMenu, id=menu_id)
|
||
menu.delete() # Django 的级联删除会自动处理子项
|
||
return {"success": True}
|
||
|
||
|
||
# 查询列表
|
||
@router.get("/base-info/", response=List[BaseInfoOut])
|
||
@paginate(PageNumberPagination, page_size=20)
|
||
def list_base_info(request):
|
||
return BaseInfo.objects.all().order_by("-id")
|
||
|
||
|
||
# 获取详情
|
||
@router.get("/base-info/{pk}", response=BaseInfoOut)
|
||
def get_base_info(request, pk: int):
|
||
return get_object_or_404(BaseInfo, pk=pk)
|
||
|
||
|
||
# 创建
|
||
@router.post("/base-info/", response=BaseInfoOut)
|
||
def create_base_info(request, data: BaseInfoIn, ):
|
||
obj = BaseInfo.objects.create(**data.dict())
|
||
return obj
|
||
|
||
|
||
# 更新
|
||
@router.put("/base-info/{pk}", response=BaseInfoOut)
|
||
def update_base_info(
|
||
request,
|
||
pk: int,
|
||
data: BaseInfoIn,
|
||
):
|
||
obj = get_object_or_404(BaseInfo, pk=pk)
|
||
|
||
for attr, value in data.dict().items():
|
||
setattr(obj, attr, value)
|
||
|
||
obj.save()
|
||
return obj
|
||
|
||
|
||
# 删除
|
||
@router.delete("/base-info/{pk}")
|
||
def delete_base_info(request, pk: int):
|
||
obj = get_object_or_404(BaseInfo, pk=pk)
|
||
obj.delete()
|
||
return {"success": True}
|