fin

api_core/__init__.py

@@ -0,0 +1 @@
+

api_core/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class ApiCoreConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "api_core"
+

api_core/permissions.py

@@ -0,0 +1,26 @@
+from rest_framework.permissions import BasePermission
+
+
+class IsCoreSuperAdmin(BasePermission):
+    """
+    Core APIs are reserved for Django superusers with a bound Employee merchant.
+    """
+
+    message = "需要 superadmin 权限并绑定有效员工商户"
+
+    def has_permission(self, request, view):
+        user = request.user
+        if not user or not user.is_authenticated:
+            return False
+        if not user.is_superuser:
+            return False
+
+        employee = getattr(user, "employee", None)
+        if employee is None:
+            return False
+        if getattr(employee, "merchant_id", None) is None:
+            return False
+
+        request.core_merchant = employee.merchant
+        return True
+

api_core/serializers.py

@@ -0,0 +1,171 @@
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group, Permission
+from django.db import transaction
+from rest_framework import serializers
+from drf_spectacular.utils import extend_schema_field
+
+from basic_info.models import UserProfile
+
+
+User = get_user_model()
+
+
+class CorePermissionSerializer(serializers.ModelSerializer):
+    app_label = serializers.CharField(source="content_type.app_label", read_only=True)
+    model = serializers.CharField(source="content_type.model", read_only=True)
+    full_code = serializers.SerializerMethodField()
+
+    class Meta:
+        model = Permission
+        fields = [
+            "id",
+            "name",
+            "codename",
+            "content_type",
+            "app_label",
+            "model",
+            "full_code",
+        ]
+        read_only_fields = fields
+
+    def get_full_code(self, obj) -> str:
+        return f"{obj.content_type.app_label}.{obj.codename}"
+
+
+class CoreGroupSerializer(serializers.ModelSerializer):
+    permissions = serializers.PrimaryKeyRelatedField(
+        many=True,
+        queryset=Permission.objects.select_related("content_type").all(),
+        required=False,
+    )
+    permission_details = CorePermissionSerializer(
+        source="permissions",
+        many=True,
+        read_only=True,
+    )
+
+    class Meta:
+        model = Group
+        fields = ["id", "name", "permissions", "permission_details"]
+
+
+class CoreUserSerializer(serializers.ModelSerializer):
+    groups = serializers.PrimaryKeyRelatedField(
+        many=True,
+        queryset=Group.objects.all(),
+        required=False,
+    )
+    user_permissions = serializers.PrimaryKeyRelatedField(
+        many=True,
+        queryset=Permission.objects.select_related("content_type").all(),
+        required=False,
+    )
+    password = serializers.CharField(write_only=True, required=False, min_length=6)
+    merchant_id = serializers.IntegerField(source="profile.merchant_id", read_only=True)
+    employee_id = serializers.IntegerField(source="employee.id", read_only=True)
+    employee_name = serializers.CharField(source="employee.name", read_only=True)
+    group_details = serializers.SerializerMethodField()
+    permission_details = serializers.SerializerMethodField()
+
+    class Meta:
+        model = User
+        fields = [
+            "id",
+            "username",
+            "email",
+            "first_name",
+            "last_name",
+            "is_active",
+            "is_staff",
+            "is_superuser",
+            "last_login",
+            "date_joined",
+            "merchant_id",
+            "employee_id",
+            "employee_name",
+            "groups",
+            "group_details",
+            "user_permissions",
+            "permission_details",
+            "password",
+        ]
+        read_only_fields = [
+            "id",
+            "last_login",
+            "date_joined",
+            "merchant_id",
+            "employee_id",
+            "employee_name",
+            "group_details",
+            "permission_details",
+        ]
+
+    @extend_schema_field(serializers.ListField(child=serializers.DictField()))
+    def get_group_details(self, obj) -> list[dict[str, object]]:
+        return [{"id": group.id, "name": group.name} for group in obj.groups.all()]
+
+    @extend_schema_field(CorePermissionSerializer(many=True))
+    def get_permission_details(self, obj) -> list[dict[str, object]]:
+        permissions = obj.user_permissions.select_related("content_type").all()
+        return CorePermissionSerializer(permissions, many=True).data
+
+    def validate_username(self, value):
+        queryset = User.objects.filter(username=value)
+        if self.instance is not None:
+            queryset = queryset.exclude(pk=self.instance.pk)
+        if queryset.exists():
+            raise serializers.ValidationError("用户名已存在")
+        return value
+
+    def validate(self, attrs):
+        request = self.context.get("request")
+        current_user = getattr(request, "user", None)
+        target_user = self.instance
+
+        if current_user is not None and target_user is not None and current_user.pk == target_user.pk:
+            if attrs.get("is_active") is False:
+                raise serializers.ValidationError({"is_active": "不能停用当前登录用户"})
+            if attrs.get("is_superuser") is False:
+                raise serializers.ValidationError({"is_superuser": "不能取消当前登录用户的 superuser 权限"})
+
+        return attrs
+
+    def create(self, validated_data):
+        groups = validated_data.pop("groups", [])
+        user_permissions = validated_data.pop("user_permissions", [])
+        password = validated_data.pop("password", None)
+        merchant = self.context["merchant"]
+
+        with transaction.atomic():
+            user = User(**validated_data)
+            if password:
+                user.set_password(password)
+            else:
+                user.set_unusable_password()
+            user.save()
+            if groups:
+                user.groups.set(groups)
+            if user_permissions:
+                user.user_permissions.set(user_permissions)
+            UserProfile.objects.create(user=user, merchant=merchant)
+
+        return user
+
+    def update(self, instance, validated_data):
+        groups = validated_data.pop("groups", None)
+        user_permissions = validated_data.pop("user_permissions", None)
+        password = validated_data.pop("password", None)
+
+        with transaction.atomic():
+            for attr, value in validated_data.items():
+                setattr(instance, attr, value)
+            if password:
+                instance.set_password(password)
+            instance.save()
+            if groups is not None:
+                instance.groups.set(groups)
+            if user_permissions is not None:
+                instance.user_permissions.set(user_permissions)
+
+        return instance
+

api_core/tests.py

@@ -0,0 +1,174 @@
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group, Permission
+from django.contrib.contenttypes.models import ContentType
+from django.test import TestCase
+from rest_framework.test import APIClient
+
+from basic_info import models as basic_models
+
+
+User = get_user_model()
+
+
+class CoreAPITestCase(TestCase):
+    def setUp(self):
+        self.client = APIClient()
+        self.merchant = basic_models.Merchant.objects.create(
+            name="核心商户",
+            type=basic_models.MerchantTypeEnum.STORE,
+        )
+        self.other_merchant = basic_models.Merchant.objects.create(
+            name="其他商户",
+            type=basic_models.MerchantTypeEnum.STORE,
+        )
+        self.super_user = User.objects.create_superuser(
+            username="core_admin",
+            password="pass12345",
+            email="core@example.com",
+        )
+        self.employee = basic_models.Employee.objects.create(
+            merchant=self.merchant,
+            sys_user=self.super_user,
+            name="核心管理员",
+        )
+        self.normal_user = User.objects.create_user(username="normal", password="pass12345")
+        self.no_employee_super_user = User.objects.create_superuser(
+            username="no_employee_admin",
+            password="pass12345",
+        )
+
+    def authenticate_super_admin(self):
+        self.client.force_authenticate(user=self.super_user)
+
+    def test_core_api_requires_superuser_and_employee_merchant(self):
+        response = self.client.get("/api/core/users/")
+        self.assertEqual(response.status_code, 401)
+
+        self.client.force_authenticate(user=self.normal_user)
+        response = self.client.get("/api/core/users/")
+        self.assertEqual(response.status_code, 403)
+
+        self.client.force_authenticate(user=self.no_employee_super_user)
+        response = self.client.get("/api/core/users/")
+        self.assertEqual(response.status_code, 403)
+
+    def test_user_list_is_merchant_scoped(self):
+        same_user = User.objects.create_user(username="same_merchant", password="pass12345")
+        basic_models.UserProfile.objects.create(user=same_user, merchant=self.merchant)
+
+        other_user = User.objects.create_user(username="other_merchant", password="pass12345")
+        basic_models.UserProfile.objects.create(user=other_user, merchant=self.other_merchant)
+
+        self.authenticate_super_admin()
+        response = self.client.get("/api/core/users/", {"limit": 100})
+        self.assertEqual(response.status_code, 200)
+        usernames = {item["username"] for item in response.data["results"]}
+        self.assertIn("core_admin", usernames)
+        self.assertIn("same_merchant", usernames)
+        self.assertNotIn("other_merchant", usernames)
+
+        detail_response = self.client.get(f"/api/core/users/{other_user.id}/")
+        self.assertEqual(detail_response.status_code, 404)
+
+    def test_create_user_creates_profile_but_not_employee(self):
+        group = Group.objects.create(name="核心角色")
+        self.authenticate_super_admin()
+        response = self.client.post(
+            "/api/core/users/",
+            {
+                "username": "core_created",
+                "password": "pass12345",
+                "email": "created@example.com",
+                "first_name": "Core",
+                "is_active": True,
+                "groups": [group.id],
+            },
+            format="json",
+        )
+        self.assertEqual(response.status_code, 201)
+
+        user = User.objects.get(username="core_created")
+        self.assertEqual(user.profile.merchant, self.merchant)
+        self.assertFalse(hasattr(user, "employee"))
+        self.assertIn(group, user.groups.all())
+
+    def test_user_cannot_be_deleted_and_self_cannot_be_deactivated_or_demoted(self):
+        self.authenticate_super_admin()
+        response = self.client.delete(f"/api/core/users/{self.super_user.id}/")
+        self.assertEqual(response.status_code, 405)
+
+        response = self.client.patch(
+            f"/api/core/users/{self.super_user.id}/",
+            {"is_active": False},
+            format="json",
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("is_active", response.data)
+
+        response = self.client.patch(
+            f"/api/core/users/{self.super_user.id}/",
+            {"is_superuser": False},
+            format="json",
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("is_superuser", response.data)
+
+    def test_groups_are_global_full_crud(self):
+        content_type = ContentType.objects.get_for_model(User)
+        permission = Permission.objects.get(content_type=content_type, codename="view_user")
+        self.authenticate_super_admin()
+
+        create_response = self.client.post(
+            "/api/core/groups/",
+            {"name": "全局核心组", "permissions": [permission.id]},
+            format="json",
+        )
+        self.assertEqual(create_response.status_code, 201)
+        group_id = create_response.data["id"]
+
+        detail_response = self.client.get(f"/api/core/groups/{group_id}/")
+        self.assertEqual(detail_response.status_code, 200)
+        self.assertEqual(detail_response.data["name"], "全局核心组")
+        self.assertEqual(detail_response.data["permissions"], [permission.id])
+
+        patch_response = self.client.patch(
+            f"/api/core/groups/{group_id}/",
+            {"name": "已更新核心组"},
+            format="json",
+        )
+        self.assertEqual(patch_response.status_code, 200)
+        self.assertEqual(patch_response.data["name"], "已更新核心组")
+
+        delete_response = self.client.delete(f"/api/core/groups/{group_id}/")
+        self.assertEqual(delete_response.status_code, 204)
+        self.assertFalse(Group.objects.filter(id=group_id).exists())
+
+    def test_permissions_are_read_only(self):
+        content_type = ContentType.objects.get_for_model(User)
+        permission = Permission.objects.get(content_type=content_type, codename="view_user")
+        self.authenticate_super_admin()
+
+        list_response = self.client.get("/api/core/permissions/", {"limit": 100})
+        self.assertEqual(list_response.status_code, 200)
+        self.assertTrue(any(item["id"] == permission.id for item in list_response.data["results"]))
+
+        detail_response = self.client.get(f"/api/core/permissions/{permission.id}/")
+        self.assertEqual(detail_response.status_code, 200)
+        self.assertEqual(detail_response.data["full_code"], "auth.view_user")
+
+        create_response = self.client.post(
+            "/api/core/permissions/",
+            {"name": "fake", "codename": "fake_permission", "content_type": content_type.id},
+            format="json",
+        )
+        self.assertEqual(create_response.status_code, 405)
+
+        patch_response = self.client.patch(
+            f"/api/core/permissions/{permission.id}/",
+            {"name": "renamed"},
+            format="json",
+        )
+        self.assertEqual(patch_response.status_code, 405)
+
+        delete_response = self.client.delete(f"/api/core/permissions/{permission.id}/")
+        self.assertEqual(delete_response.status_code, 405)

api_core/urls.py

@@ -0,0 +1,12 @@
+from rest_framework.routers import DefaultRouter
+
+from api_core.views import CoreGroupViewSet, CorePermissionViewSet, CoreUserViewSet
+
+
+router = DefaultRouter()
+router.register("users", CoreUserViewSet, basename="core-user")
+router.register("groups", CoreGroupViewSet, basename="core-group")
+router.register("permissions", CorePermissionViewSet, basename="core-permission")
+
+urlpatterns = router.urls
+

api_core/views.py

@@ -0,0 +1,75 @@
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group, Permission
+from django.db.models import Q
+from rest_framework import filters
+from rest_framework import viewsets
+from rest_framework.exceptions import MethodNotAllowed
+
+from api_core.permissions import IsCoreSuperAdmin
+from api_core.serializers import (
+    CoreGroupSerializer,
+    CorePermissionSerializer,
+    CoreUserSerializer,
+)
+from flower.viewsets import LimitedLimitOffsetPagination
+
+
+User = get_user_model()
+
+
+class CoreUserViewSet(viewsets.ModelViewSet):
+    serializer_class = CoreUserSerializer
+    permission_classes = [IsCoreSuperAdmin]
+    pagination_class = LimitedLimitOffsetPagination
+    filter_backends = [filters.SearchFilter, filters.OrderingFilter]
+    search_fields = ["username", "email", "first_name", "last_name", "employee__name"]
+    ordering_fields = ["id", "username", "date_joined", "last_login"]
+    ordering = ["id"]
+
+    def get_queryset(self):
+        merchant = getattr(self.request, "core_merchant", None)
+        if merchant is None:
+            return User.objects.none()
+        return (
+            User.objects.filter(
+                Q(employee__merchant=merchant) | Q(profile__merchant=merchant)
+            )
+            .distinct()
+            .prefetch_related("groups", "user_permissions__content_type")
+            .select_related("employee", "profile")
+            .order_by("id")
+        )
+
+    def get_serializer_context(self):
+        context = super().get_serializer_context()
+        context["merchant"] = self.request.core_merchant
+        return context
+
+    def destroy(self, request, *args, **kwargs):
+        raise MethodNotAllowed("DELETE", detail="Core users 不允许删除，请通过 is_active=false 停用")
+
+
+class CoreGroupViewSet(viewsets.ModelViewSet):
+    queryset = Group.objects.prefetch_related("permissions__content_type").order_by("id")
+    serializer_class = CoreGroupSerializer
+    permission_classes = [IsCoreSuperAdmin]
+    pagination_class = LimitedLimitOffsetPagination
+    filter_backends = [filters.SearchFilter, filters.OrderingFilter]
+    search_fields = ["name"]
+    ordering_fields = ["id", "name"]
+    ordering = ["id"]
+
+
+class CorePermissionViewSet(viewsets.ReadOnlyModelViewSet):
+    queryset = Permission.objects.select_related("content_type").order_by(
+        "content_type__app_label",
+        "content_type__model",
+        "codename",
+    )
+    serializer_class = CorePermissionSerializer
+    permission_classes = [IsCoreSuperAdmin]
+    pagination_class = LimitedLimitOffsetPagination
+    filter_backends = [filters.SearchFilter, filters.OrderingFilter]
+    search_fields = ["name", "codename", "content_type__app_label", "content_type__model"]
+    ordering_fields = ["id", "name", "codename"]
+