zyc/rtc_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
zyc 2026-01-29 10:02:15 +08:00
commit c0deacd79c
55 changed files with 3827 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
.env.example Normal file
View File

@ -0,0 +1,24 @@
# Django Settings
DJANGO_SECRET_KEY=your-secret-key-here
DJANGO_DEBUG=True
DJANGO_ALLOWED_HOSTS=*
# MySQL Database
DB_NAME=rtc_demo
DB_USER=root
DB_PASSWORD=your-password
DB_HOST=127.0.0.1
DB_PORT=3306
# Redis
REDIS_URL=redis://127.0.0.1:6379/0
# Aliyun OSS
OSS_ACCESS_KEY_ID=your-access-key-id
OSS_ACCESS_KEY_SECRET=your-access-key-secret
OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com
OSS_BUCKET_NAME=your-bucket-name
OSS_CUSTOM_DOMAIN=
# CORS (production only)
CORS_ALLOWED_ORIGINS=https://your-domain.com

32
.gitignore vendored Normal file
View File

@ -0,0 +1,32 @@
# Python
__pycache__/
*.py[cod]
*.so
*.egg-info/
dist/
build/
# Virtual environment
venv/
.venv/
# Environment variables
.env
# IDE
.vscode/
.idea/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
# Django
*.sqlite3
media/
staticfiles/
# Logs
*.log

0
apps/__init__.py Normal file
View File

1
apps/admins/__init__.py Normal file
View File

@ -0,0 +1 @@
default_app_config = 'apps.admins.apps.AdminsConfig'

7
apps/admins/apps.py Normal file
View File

@ -0,0 +1,7 @@
from django.apps import AppConfig
class AdminsConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'apps.admins'
verbose_name = '管理员'

82
apps/admins/authentication.py Normal file
View File

@ -0,0 +1,82 @@
"""
管理员模块自定义JWT认证
"""
from rest_framework_simplejwt.tokens import RefreshToken
from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework_simplejwt.exceptions import InvalidToken, AuthenticationFailed
from django.conf import settings
def get_admin_tokens(admin_user):
"""
为管理员生成JWT Token
在token中添加 user_type='admin' 以区分App用户
"""
refresh = RefreshToken.for_user(admin_user)
# 添加自定义声明
refresh['user_type'] = 'admin'
refresh['username'] = admin_user.username
refresh['role'] = admin_user.role
return {
'access': str(refresh.access_token),
'refresh': str(refresh),
}
class AdminJWTAuthentication(JWTAuthentication):
"""
管理员专用JWT认证
验证token中的user_type必须为'admin'
"""
def get_user(self, validated_token):
"""
重写get_user方法，从AdminUser模型获取用户
"""
from apps.admins.models import AdminUser
# 验证user_type
user_type = validated_token.get('user_type')
if user_type != 'admin':
raise AuthenticationFailed('无效的管理员Token')
try:
user_id = validated_token.get('user_id')
user = AdminUser.objects.get(id=user_id)
except AdminUser.DoesNotExist:
raise AuthenticationFailed('管理员用户不存在')
if not user.is_active:
raise AuthenticationFailed('管理员账户已被禁用')
return user
class AppJWTAuthentication(JWTAuthentication):
"""
App端专用JWT认证
验证token中的user_type必须为'app'或不存在（兼容旧token）
"""
def get_user(self, validated_token):
"""
重写get_user方法，从User模型获取用户
"""
from apps.users.models import User
# 验证user_type（兼容旧token，默认为app）
user_type = validated_token.get('user_type', 'app')
if user_type not in ['app', None]:
raise AuthenticationFailed('无效的用户Token')
try:
user_id = validated_token.get('user_id')
user = User.objects.get(id=user_id)
except User.DoesNotExist:
raise AuthenticationFailed('用户不存在')
if not user.is_active:
raise AuthenticationFailed('用户账户已被禁用')
return user

84
apps/admins/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,84 @@
# Generated by Django 6.0.1 on 2026-01-28 10:18
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = []
operations = [
migrations.CreateModel(
name="AdminUser",
fields=[
("password", models.CharField(max_length=128, verbose_name="password")),
(
"last_login",
models.DateTimeField(
blank=True, null=True, verbose_name="last login"
),
),
("id", models.BigAutoField(primary_key=True, serialize=False)),
(
"username",
models.CharField(max_length=50, unique=True, verbose_name="用户名"),
),
(
"name",
models.CharField(
blank=True, default="", max_length=50, verbose_name="姓名"
),
),
(
"email",
models.EmailField(
blank=True, default="", max_length=254, verbose_name="邮箱"
),
),
(
"phone",
models.CharField(
blank=True, default="", max_length=20, verbose_name="手机号"
),
),
(
"role",
models.CharField(
choices=[
("super_admin", "超级管理员"),
("admin", "管理员"),
("operator", "操作员"),
],
default="operator",
max_length=20,
verbose_name="角色",
),
),
(
"is_active",
models.BooleanField(default=True, verbose_name="是否启用"),
),
(
"last_login_ip",
models.GenericIPAddressField(
blank=True, null=True, verbose_name="最后登录IP"
),
),
(
"created_at",
models.DateTimeField(auto_now_add=True, verbose_name="创建时间"),
),
(
"updated_at",
models.DateTimeField(auto_now=True, verbose_name="更新时间"),
),
],
options={
"verbose_name": "管理员",
"verbose_name_plural": "管理员",
"db_table": "admin_user",
},
),
]

0
apps/admins/migrations/__init__.py Normal file
View File

77
apps/admins/models.py Normal file
View File

@ -0,0 +1,77 @@
"""
管理员模块模型 - 独立于App用户的账户体系
"""
from django.contrib.auth.models import AbstractBaseUser, BaseUserManager
from django.db import models
class AdminUserManager(BaseUserManager):
"""管理员用户管理器"""
def create_user(self, username, password=None, **extra_fields):
if not username:
raise ValueError('用户名不能为空')
user = self.model(username=username, **extra_fields)
user.set_password(password)
user.save(using=self._db)
return user
def create_superuser(self, username, password=None, **extra_fields):
extra_fields.setdefault('role', 'super_admin')
return self.create_user(username, password, **extra_fields)
class AdminUser(AbstractBaseUser):
"""
管理员用户模型 - Web后台使用
与App端User模型完全独立
不使用PermissionsMixin以避免与User模型冲突
"""
ROLE_CHOICES = [
('super_admin', '超级管理员'),
('admin', '管理员'),
('operator', '操作员'),
]
id = models.BigAutoField(primary_key=True)
username = models.CharField('用户名', max_length=50, unique=True)
name = models.CharField('姓名', max_length=50, blank=True, default='')
email = models.EmailField('邮箱', blank=True, default='')
phone = models.CharField('手机号', max_length=20, blank=True, default='')
role = models.CharField('角色', max_length=20, choices=ROLE_CHOICES, default='operator')
is_active = models.BooleanField('是否启用', default=True)
last_login_ip = models.GenericIPAddressField('最后登录IP', null=True, blank=True)
created_at = models.DateTimeField('创建时间', auto_now_add=True)
updated_at = models.DateTimeField('更新时间', auto_now=True)
objects = AdminUserManager()
USERNAME_FIELD = 'username'
REQUIRED_FIELDS = []
class Meta:
db_table = 'admin_user'
verbose_name = '管理员'
verbose_name_plural = '管理员'
@property
def is_staff(self):
"""Django admin兼容"""
return self.role in ['super_admin', 'admin']
@property
def is_superuser(self):
"""超级管理员检查"""
return self.role == 'super_admin'
def has_perm(self, perm, obj=None):
"""权限检查"""
return self.role in ['super_admin', 'admin']
def has_module_perms(self, app_label):
"""模块权限检查"""
return self.role in ['super_admin', 'admin']
def __str__(self):
return f"{self.username} ({self.get_role_display()})"

46
apps/admins/permissions.py Normal file
View File

@ -0,0 +1,46 @@
"""
管理员模块权限
"""
from rest_framework.permissions import BasePermission
class IsAdminUser(BasePermission):
"""
验证是否为管理员用户（来自AdminUser模型）
"""
def has_permission(self, request, view):
from apps.admins.models import AdminUser
return (
request.user and
request.user.is_authenticated and
isinstance(request.user, AdminUser) and
request.user.is_active
)
class IsSuperAdmin(BasePermission):
"""
验证是否为超级管理员
"""
def has_permission(self, request, view):
from apps.admins.models import AdminUser
return (
request.user and
request.user.is_authenticated and
isinstance(request.user, AdminUser) and
request.user.role == 'super_admin'
)
class IsAdminOrOperator(BasePermission):
"""
验证是否为管理员或操作员
"""
def has_permission(self, request, view):
from apps.admins.models import AdminUser
return (
request.user and
request.user.is_authenticated and
isinstance(request.user, AdminUser) and
request.user.role in ['super_admin', 'admin', 'operator']
)

53
apps/admins/serializers.py Normal file
View File

@ -0,0 +1,53 @@
"""
管理员模块序列化器
"""
from rest_framework import serializers
from .models import AdminUser
class AdminUserSerializer(serializers.ModelSerializer):
"""管理员用户序列化器"""
role_display = serializers.CharField(source='get_role_display', read_only=True)
class Meta:
model = AdminUser
fields = ['id', 'username', 'name', 'email', 'phone', 'role', 'role_display',
'is_active', 'last_login_ip', 'created_at']
read_only_fields = ['id', 'last_login_ip', 'created_at']
class AdminLoginSerializer(serializers.Serializer):
"""管理员登录序列化器"""
username = serializers.CharField(max_length=50, help_text='用户名')
password = serializers.CharField(max_length=128, write_only=True, help_text='密码')
class AdminChangePasswordSerializer(serializers.Serializer):
"""管理员修改密码序列化器"""
old_password = serializers.CharField(max_length=128, write_only=True)
new_password = serializers.CharField(max_length=128, write_only=True)
def validate_new_password(self, value):
if len(value) < 6:
raise serializers.ValidationError('密码长度不能少于6位')
return value
class CreateAdminUserSerializer(serializers.ModelSerializer):
"""创建管理员序列化器"""
password = serializers.CharField(max_length=128, write_only=True)
class Meta:
model = AdminUser
fields = ['username', 'password', 'name', 'email', 'phone', 'role']
def create(self, validated_data):
password = validated_data.pop('password')
admin = AdminUser(**validated_data)
admin.set_password(password)
admin.save()
return admin

15
apps/admins/urls.py Normal file
View File

@ -0,0 +1,15 @@
"""
管理员模块URL配置
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import AdminAuthViewSet, AdminProfileViewSet, AdminUserManageViewSet
router = DefaultRouter()
router.register('auth', AdminAuthViewSet, basename='admin-auth')
router.register('profile', AdminProfileViewSet, basename='admin-profile')
router.register('admins', AdminUserManageViewSet, basename='admin-users')
urlpatterns = [
path('', include(router.urls)),
]

229
apps/admins/views.py Normal file
View File

@ -0,0 +1,229 @@
"""
管理员模块视图
"""
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.permissions import AllowAny
from rest_framework_simplejwt.tokens import RefreshToken
from utils.response import success, error
from utils.exceptions import ErrorCode
from .models import AdminUser
from .serializers import (
AdminUserSerializer,
AdminLoginSerializer,
AdminChangePasswordSerializer,
CreateAdminUserSerializer
)
from .authentication import get_admin_tokens, AdminJWTAuthentication
from .permissions import IsAdminUser, IsSuperAdmin
class AdminAuthViewSet(viewsets.ViewSet):
"""管理员认证视图集"""
permission_classes = [AllowAny]
@action(detail=False, methods=['post'])
def login(self, request):
"""
管理员登录
POST /api/admin/auth/login
"""
serializer = AdminLoginSerializer(data=request.data)
if not serializer.is_valid():
return error(message=str(serializer.errors))
username = serializer.validated_data['username']
password = serializer.validated_data['password']
try:
admin = AdminUser.objects.get(username=username)
except AdminUser.DoesNotExist:
return error(code=ErrorCode.USER_NOT_FOUND, message='用户名或密码错误')
if not admin.check_password(password):
return error(code=ErrorCode.USER_NOT_FOUND, message='用户名或密码错误')
if not admin.is_active:
return error(code=ErrorCode.USER_DISABLED, message='账户已被禁用')
# 记录登录IP
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0].strip()
else:
ip = request.META.get('REMOTE_ADDR')
admin.last_login_ip = ip
admin.save(update_fields=['last_login_ip'])
# 生成Token
tokens = get_admin_tokens(admin)
return success(data={
'token': tokens,
'admin': AdminUserSerializer(admin).data
}, message='登录成功')
@action(detail=False, methods=['post'])
def refresh(self, request):
"""
刷新Token
POST /api/admin/auth/refresh
"""
refresh_token = request.data.get('refresh')
if not refresh_token:
return error(message='refresh token不能为空')
try:
refresh = RefreshToken(refresh_token)
# 验证是否为admin token
if refresh.get('user_type') != 'admin':
return error(code=ErrorCode.TOKEN_EXPIRED, message='无效的管理员Token')
return success(data={
'access': str(refresh.access_token),
'refresh': str(refresh)
})
except Exception as e:
return error(code=ErrorCode.TOKEN_EXPIRED, message='Token已过期或无效')
class AdminProfileViewSet(viewsets.ViewSet):
"""管理员个人信息视图集"""
authentication_classes = [AdminJWTAuthentication]
permission_classes = [IsAdminUser]
@action(detail=False, methods=['get'], url_path='me')
def me(self, request):
"""
获取当前管理员信息
GET /api/admin/profile/me
"""
serializer = AdminUserSerializer(request.user)
return success(data=serializer.data)
@action(detail=False, methods=['put'], url_path='change-password')
def change_password(self, request):
"""
修改密码
PUT /api/admin/profile/change-password
"""
serializer = AdminChangePasswordSerializer(data=request.data)
if not serializer.is_valid():
return error(message=str(serializer.errors))
if not request.user.check_password(serializer.validated_data['old_password']):
return error(message='原密码错误')
request.user.set_password(serializer.validated_data['new_password'])
request.user.save()
return success(message='密码修改成功')
class AdminUserManageViewSet(viewsets.ModelViewSet):
"""管理员用户管理视图集（仅超级管理员可用）"""
queryset = AdminUser.objects.all().order_by('-created_at')
authentication_classes = [AdminJWTAuthentication]
permission_classes = [IsSuperAdmin]
def get_serializer_class(self):
if self.action == 'create':
return CreateAdminUserSerializer
return AdminUserSerializer
def list(self, request, *args, **kwargs):
"""
管理员列表
GET /api/admin/admins
"""
queryset = self.filter_queryset(self.get_queryset())
# 搜索
username = request.query_params.get('username')
role = request.query_params.get('role')
if username:
queryset = queryset.filter(username__contains=username)
if role:
queryset = queryset.filter(role=role)
page = self.paginate_queryset(queryset)
if page is not None:
serializer = AdminUserSerializer(page, many=True)
return success(data={
'total': self.paginator.page.paginator.count,
'items': serializer.data
})
serializer = AdminUserSerializer(queryset, many=True)
return success(data={'items': serializer.data})
def create(self, request, *args, **kwargs):
"""
创建管理员
POST /api/admin/admins
"""
serializer = CreateAdminUserSerializer(data=request.data)
if serializer.is_valid():
admin = serializer.save()
return success(data=AdminUserSerializer(admin).data, message='创建成功')
return error(message=str(serializer.errors))
def retrieve(self, request, *args, **kwargs):
"""
管理员详情
GET /api/admin/admins/{id}
"""
instance = self.get_object()
serializer = AdminUserSerializer(instance)
return success(data=serializer.data)
def update(self, request, *args, **kwargs):
"""
更新管理员
PUT /api/admin/admins/{id}
"""
instance = self.get_object()
serializer = AdminUserSerializer(instance, data=request.data, partial=True)
if serializer.is_valid():
serializer.save()
return success(data=serializer.data, message='更新成功')
return error(message=str(serializer.errors))
@action(detail=True, methods=['post'], url_path='toggle-status')
def toggle_status(self, request, pk=None):
"""
启用/禁用管理员
POST /api/admin/admins/{id}/toggle-status
"""
admin = self.get_object()
# 不能禁用自己
if admin.id == request.user.id:
return error(message='不能禁用自己的账户')
admin.is_active = not admin.is_active
admin.save()
return success(
data=AdminUserSerializer(admin).data,
message='已启用' if admin.is_active else '已禁用'
)
@action(detail=True, methods=['post'], url_path='reset-password')
def reset_password(self, request, pk=None):
"""
重置管理员密码
POST /api/admin/admins/{id}/reset-password
"""
admin = self.get_object()
new_password = request.data.get('new_password', '123456')
admin.set_password(new_password)
admin.save()
return success(message=f'密码已重置为: {new_password}')

0
apps/devices/__init__.py Normal file
View File

182
apps/devices/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,182 @@
# Generated by Django 6.0.1 on 2026-01-28 10:03
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = []
operations = [
migrations.CreateModel(
name="Device",
fields=[
("id", models.BigAutoField(primary_key=True, serialize=False)),
(
"sn",
models.CharField(max_length=50, unique=True, verbose_name="SN码"),
),
(
"mac_address",
models.CharField(
blank=True,
max_length=20,
null=True,
unique=True,
verbose_name="MAC地址",
),
),
(
"name",
models.CharField(
blank=True,
default="",
max_length=100,
verbose_name="自定义名称",
),
),
(
"status",
models.CharField(
choices=[
("in_stock", "库存中"),
("out_stock", "已出库"),
("bound", "已绑定"),
],
default="in_stock",
max_length=20,
verbose_name="状态",
),
),
(
"firmware_version",
models.CharField(
blank=True, default="", max_length=20, verbose_name="固件版本"
),
),
(
"last_online_at",
models.DateTimeField(
blank=True, null=True, verbose_name="最后在线时间"
),
),
(
"created_at",
models.DateTimeField(auto_now_add=True, verbose_name="创建时间"),
),
(
"updated_at",
models.DateTimeField(auto_now=True, verbose_name="更新时间"),
),
],
options={
"verbose_name": "设备",
"verbose_name_plural": "设备",
"db_table": "device",
},
),
migrations.CreateModel(
name="DeviceBatch",
fields=[
("id", models.BigAutoField(primary_key=True, serialize=False)),
("batch_no", models.CharField(max_length=50, verbose_name="批次号")),
("production_date", models.DateField(verbose_name="生产日期")),
(
"production_week",
models.CharField(blank=True, max_length=10, verbose_name="生产周"),
),
("quantity", models.IntegerField(verbose_name="数量")),
(
"remark",
models.TextField(blank=True, default="", verbose_name="备注"),
),
(
"status",
models.CharField(
choices=[("pending", "待生成"), ("generated", "已生成")],
default="pending",
max_length=20,
verbose_name="状态",
),
),
(
"created_at",
models.DateTimeField(auto_now_add=True, verbose_name="创建时间"),
),
(
"updated_at",
models.DateTimeField(auto_now=True, verbose_name="更新时间"),
),
],
options={
"verbose_name": "设备批次",
"verbose_name_plural": "设备批次",
"db_table": "device_batch",
},
),
migrations.CreateModel(
name="DeviceType",
fields=[
("id", models.BigAutoField(primary_key=True, serialize=False)),
("brand", models.CharField(max_length=50, verbose_name="品牌")),
(
"product_code",
models.CharField(
max_length=50, unique=True, verbose_name="产品代号"
),
),
("name", models.CharField(max_length=100, verbose_name="名称")),
(
"is_network_required",
models.BooleanField(default=True, verbose_name="是否需要联网"),
),
(
"is_active",
models.BooleanField(default=True, verbose_name="是否启用"),
),
(
"created_at",
models.DateTimeField(auto_now_add=True, verbose_name="创建时间"),
),
(
"updated_at",
models.DateTimeField(auto_now=True, verbose_name="更新时间"),
),
],
options={
"verbose_name": "设备类型",
"verbose_name_plural": "设备类型",
"db_table": "device_type",
},
),
migrations.CreateModel(
name="UserDevice",
fields=[
("id", models.BigAutoField(primary_key=True, serialize=False)),
(
"bind_type",
models.CharField(
choices=[("owner", "所有者"), ("share", "分享")],
default="owner",
max_length=20,
verbose_name="绑定类型",
),
),
(
"bind_time",
models.DateTimeField(auto_now_add=True, verbose_name="绑定时间"),
),
(
"is_active",
models.BooleanField(default=True, verbose_name="是否有效"),
),
],
options={
"verbose_name": "用户设备绑定",
"verbose_name_plural": "用户设备绑定",
"db_table": "user_device",
},
),
]

111
apps/devices/migrations/0002_initial.py Normal file
View File

@ -0,0 +1,111 @@
# Generated by Django 6.0.1 on 2026-01-28 10:03
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
("devices", "0001_initial"),
("spirits", "0001_initial"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AddField(
model_name="devicebatch",
name="created_by",
field=models.ForeignKey(
null=True,
on_delete=django.db.models.deletion.SET_NULL,
to=settings.AUTH_USER_MODEL,
verbose_name="创建人",
),
),
migrations.AddField(
model_name="device",
name="batch",
field=models.ForeignKey(
null=True,
on_delete=django.db.models.deletion.SET_NULL,
related_name="devices",
to="devices.devicebatch",
verbose_name="所属批次",
),
),
migrations.AddField(
model_name="devicebatch",
name="device_type",
field=models.ForeignKey(
on_delete=django.db.models.deletion.PROTECT,
related_name="batches",
to="devices.devicetype",
verbose_name="设备类型",
),
),
migrations.AddField(
model_name="device",
name="device_type",
field=models.ForeignKey(
null=True,
on_delete=django.db.models.deletion.PROTECT,
related_name="devices",
to="devices.devicetype",
verbose_name="设备类型",
),
),
migrations.AddField(
model_name="userdevice",
name="device",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="user_devices",
to="devices.device",
verbose_name="设备",
),
),
migrations.AddField(
model_name="userdevice",
name="spirit",
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.SET_NULL,
related_name="user_devices",
to="spirits.spirit",
verbose_name="绑定的智能体",
),
),
migrations.AddField(
model_name="userdevice",
name="user",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="user_devices",
to=settings.AUTH_USER_MODEL,
verbose_name="用户",
),
),
migrations.AlterUniqueTogether(
name="devicebatch",
unique_together={("device_type", "batch_no", "production_date")},
),
migrations.AddIndex(
model_name="device",
index=models.Index(
fields=["mac_address"], name="device_mac_add_09f85c_idx"
),
),
migrations.AddIndex(
model_name="device",
index=models.Index(fields=["status"], name="device_status_6321a6_idx"),
),
migrations.AlterUniqueTogether(
name="userdevice",
unique_together={("user", "device")},
),
]

0
apps/devices/migrations/__init__.py Normal file
View File

131
apps/devices/models.py Normal file
View File

@ -0,0 +1,131 @@
"""
设备模块模型
"""
from django.db import models
from apps.users.models import User
from apps.spirits.models import Spirit
class DeviceType(models.Model):
"""设备类型表"""
id = models.BigAutoField(primary_key=True)
brand = models.CharField('品牌', max_length=50)
product_code = models.CharField('产品代号', max_length=50, unique=True)
name = models.CharField('名称', max_length=100)
is_network_required = models.BooleanField('是否需要联网', default=True)
is_active = models.BooleanField('是否启用', default=True)
created_at = models.DateTimeField('创建时间', auto_now_add=True)
updated_at = models.DateTimeField('更新时间', auto_now=True)
class Meta:
db_table = 'device_type'
verbose_name = '设备类型'
verbose_name_plural = '设备类型'
def save(self, *args, **kwargs):
# 根据产品代号后缀自动判断是否需要联网
if self.product_code:
self.is_network_required = not self.product_code.endswith('-OFF')
super().save(*args, **kwargs)
def __str__(self):
return f"{self.brand} - {self.name}"
class DeviceBatch(models.Model):
"""设备批次表"""
STATUS_CHOICES = [
('pending', '待生成'),
('generated', '已生成'),
]
id = models.BigAutoField(primary_key=True)
device_type = models.ForeignKey(DeviceType, on_delete=models.PROTECT, related_name='batches', verbose_name='设备类型')
batch_no = models.CharField('批次号', max_length=50)
production_date = models.DateField('生产日期')
production_week = models.CharField('生产周', max_length=10, blank=True) # 格式: 25W45
quantity = models.IntegerField('数量')
remark = models.TextField('备注', blank=True, default='')
status = models.CharField('状态', max_length=20, choices=STATUS_CHOICES, default='pending')
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, verbose_name='创建人')
created_at = models.DateTimeField('创建时间', auto_now_add=True)
updated_at = models.DateTimeField('更新时间', auto_now=True)
class Meta:
db_table = 'device_batch'
verbose_name = '设备批次'
verbose_name_plural = '设备批次'
unique_together = ['device_type', 'batch_no', 'production_date']
def save(self, *args, **kwargs):
# 计算生产周
if self.production_date and not self.production_week:
year = self.production_date.year
week = self.production_date.isocalendar()[1]
self.production_week = f"{year % 100}W{week:02d}"
super().save(*args, **kwargs)
def __str__(self):
return f"{self.device_type.brand}-{self.batch_no}"
class Device(models.Model):
"""设备表"""
STATUS_CHOICES = [
('in_stock', '库存中'),
('out_stock', '已出库'),
('bound', '已绑定'),
]
id = models.BigAutoField(primary_key=True)
sn = models.CharField('SN码', max_length=50, unique=True)
device_type = models.ForeignKey(DeviceType, on_delete=models.PROTECT, null=True, related_name='devices', verbose_name='设备类型')
batch = models.ForeignKey(DeviceBatch, on_delete=models.SET_NULL, null=True, related_name='devices', verbose_name='所属批次')
mac_address = models.CharField('MAC地址', max_length=20, blank=True, null=True, unique=True)
name = models.CharField('自定义名称', max_length=100, blank=True, default='')
status = models.CharField('状态', max_length=20, choices=STATUS_CHOICES, default='in_stock')
firmware_version = models.CharField('固件版本', max_length=20, blank=True, default='')
last_online_at = models.DateTimeField('最后在线时间', null=True, blank=True)
created_at = models.DateTimeField('创建时间', auto_now_add=True)
updated_at = models.DateTimeField('更新时间', auto_now=True)
class Meta:
db_table = 'device'
verbose_name = '设备'
verbose_name_plural = '设备'
indexes = [
models.Index(fields=['mac_address']),
models.Index(fields=['status']),
]
def __str__(self):
return self.sn
class UserDevice(models.Model):
"""用户设备绑定表"""
BIND_TYPE_CHOICES = [
('owner', '所有者'),
('share', '分享'),
]
id = models.BigAutoField(primary_key=True)
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='user_devices', verbose_name='用户')
device = models.ForeignKey(Device, on_delete=models.CASCADE, related_name='user_devices', verbose_name='设备')
spirit = models.ForeignKey(Spirit, on_delete=models.SET_NULL, null=True, blank=True, related_name='user_devices', verbose_name='绑定的智能体')
bind_type = models.CharField('绑定类型', max_length=20, choices=BIND_TYPE_CHOICES, default='owner')
bind_time = models.DateTimeField('绑定时间', auto_now_add=True)
is_active = models.BooleanField('是否有效', default=True)
class Meta:
db_table = 'user_device'
verbose_name = '用户设备绑定'
verbose_name_plural = '用户设备绑定'
unique_together = ['user', 'device']
def __str__(self):
return f"{self.user.phone} - {self.device.sn}"

88
apps/devices/serializers.py Normal file
View File

@ -0,0 +1,88 @@
"""
设备模块序列化器
"""
from rest_framework import serializers
from .models import DeviceType, DeviceBatch, Device, UserDevice
class DeviceTypeSerializer(serializers.ModelSerializer):
"""设备类型序列化器"""
class Meta:
model = DeviceType
fields = ['id', 'brand', 'product_code', 'name', 'is_network_required', 'is_active', 'created_at']
read_only_fields = ['id', 'is_network_required', 'created_at']
class DeviceBatchSerializer(serializers.ModelSerializer):
"""设备批次序列化器"""
device_type_info = DeviceTypeSerializer(source='device_type', read_only=True)
class Meta:
model = DeviceBatch
fields = ['id', 'device_type', 'device_type_info', 'batch_no', 'production_date',
'production_week', 'quantity', 'remark', 'status', 'created_at']
read_only_fields = ['id', 'production_week', 'status', 'created_at']
class DeviceBatchCreateSerializer(serializers.ModelSerializer):
"""创建设备批次序列化器"""
class Meta:
model = DeviceBatch
fields = ['device_type', 'batch_no', 'production_date', 'quantity', 'remark']
class DeviceSerializer(serializers.ModelSerializer):
"""设备序列化器"""
device_type_info = DeviceTypeSerializer(source='device_type', read_only=True)
class Meta:
model = Device
fields = ['id', 'sn', 'device_type', 'device_type_info', 'mac_address',
'name', 'status', 'firmware_version', 'last_online_at', 'created_at']
class DeviceSimpleSerializer(serializers.ModelSerializer):
"""设备简单序列化器"""
class Meta:
model = Device
fields = ['id', 'sn', 'mac_address', 'status', 'created_at']
class UserDeviceSerializer(serializers.ModelSerializer):
"""用户设备绑定序列化器"""
device = DeviceSerializer(read_only=True)
spirit_name = serializers.CharField(source='spirit.name', read_only=True, allow_null=True)
class Meta:
model = UserDevice
fields = ['id', 'device', 'spirit', 'spirit_name', 'bind_type', 'bind_time', 'is_active']
class BindDeviceSerializer(serializers.Serializer):
"""绑定设备序列化器"""
sn = serializers.CharField(max_length=50, help_text='设备SN码')
spirit_id = serializers.IntegerField(required=False, allow_null=True, help_text='智能体ID')
class DeviceVerifySerializer(serializers.Serializer):
"""验证设备SN序列化器"""
sn = serializers.CharField(max_length=50, help_text='设备SN码')
class QueryByMacSerializer(serializers.Serializer):
"""通过MAC查询SN序列化器"""
mac = serializers.CharField(max_length=20, help_text='MAC地址')
def validate_mac(self, value):
# 统一MAC地址格式
value = value.upper().replace('-', ':')
return value

15
apps/devices/urls.py Normal file
View File

@ -0,0 +1,15 @@
"""
设备模块URL配置
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import DeviceViewSet
router = DefaultRouter()
router.register('devices', DeviceViewSet, basename='devices')
urlpatterns = [
path('', include(router.urls)),
]
admin_urlpatterns = []

179
apps/devices/views.py Normal file
View File

@ -0,0 +1,179 @@
"""
设备模块视图 - App端
"""
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated, AllowAny
from utils.response import success, error
from utils.exceptions import ErrorCode
from apps.admins.authentication import AppJWTAuthentication
from .models import Device, UserDevice, DeviceType
from .serializers import (
DeviceSerializer,
UserDeviceSerializer,
BindDeviceSerializer,
DeviceVerifySerializer,
DeviceTypeSerializer
)
class DeviceViewSet(viewsets.ViewSet):
"""设备视图集（App端）"""
authentication_classes = [AppJWTAuthentication]
permission_classes = [IsAuthenticated]
@action(detail=False, methods=['get'], url_path='query-by-mac',
authentication_classes=[], permission_classes=[AllowAny])
def query_by_mac(self, request):
"""
通过MAC地址查询SN码（无需登录）
GET /api/v1/devices/query-by-mac?mac=AA:BB:CC:DD:EE:FF
"""
mac = request.query_params.get('mac', '')
if not mac:
return error(message='MAC地址不能为空')
# 统一格式
mac = mac.upper().replace('-', ':')
try:
device = Device.objects.select_related('device_type').get(mac_address=mac)
return success(data={
'sn': device.sn,
'mac_address': device.mac_address,
'device_type': DeviceTypeSerializer(device.device_type).data if device.device_type else None,
'status': device.status,
'is_bound': device.status == 'bound'
})
except Device.DoesNotExist:
return error(
code=404,
message='未找到对应的设备，请检查MAC地址是否正确或设备是否已完成入库',
status_code=status.HTTP_404_NOT_FOUND
)
@action(detail=False, methods=['post'])
def verify(self, request):
"""
验证设备SN
POST /api/v1/devices/verify
"""
serializer = DeviceVerifySerializer(data=request.data)
if not serializer.is_valid():
return error(message=str(serializer.errors))
sn = serializer.validated_data['sn']
try:
device = Device.objects.select_related('device_type').get(sn=sn)
except Device.DoesNotExist:
return error(code=ErrorCode.DEVICE_NOT_FOUND, message='设备不存在')
# 检查是否已被绑定
is_bindable = device.status != 'bound'
return success(data={
'sn': device.sn,
'is_bindable': is_bindable,
'device_type': DeviceTypeSerializer(device.device_type).data if device.device_type else None
})
@action(detail=False, methods=['post'])
def bind(self, request):
"""
绑定设备
POST /api/v1/devices/bind
"""
serializer = BindDeviceSerializer(data=request.data)
if not serializer.is_valid():
return error(message=str(serializer.errors))
sn = serializer.validated_data['sn']
spirit_id = serializer.validated_data.get('spirit_id')
try:
device = Device.objects.get(sn=sn)
except Device.DoesNotExist:
return error(code=ErrorCode.DEVICE_NOT_FOUND, message='设备不存在')
# 检查是否已被绑定
if device.status == 'bound':
# 检查是否是当前用户绑定的
existing = UserDevice.objects.filter(device=device, is_active=True).first()
if existing and existing.user != request.user:
return error(code=ErrorCode.DEVICE_ALREADY_BOUND, message='设备已被其他用户绑定')
# 创建绑定关系
user_device, created = UserDevice.objects.update_or_create(
user=request.user,
device=device,
defaults={
'spirit_id': spirit_id,
'is_active': True
}
)
# 更新设备状态
device.status = 'bound'
device.save()
return success(
data=UserDeviceSerializer(user_device).data,
message='绑定成功' if created else '更新绑定成功'
)
@action(detail=False, methods=['get'])
def my_devices(self, request):
"""
我的设备列表
GET /api/v1/devices/my_devices
"""
user_devices = UserDevice.objects.filter(
user=request.user,
is_active=True
).select_related('device', 'device__device_type', 'spirit')
serializer = UserDeviceSerializer(user_devices, many=True)
return success(data=serializer.data)
@action(detail=True, methods=['delete'])
def unbind(self, request, pk=None):
"""
解绑设备
DELETE /api/v1/devices/{id}/unbind
"""
try:
user_device = UserDevice.objects.get(id=pk, user=request.user)
except UserDevice.DoesNotExist:
return error(code=ErrorCode.DEVICE_NOT_FOUND, message='绑定记录不存在')
# 更新绑定状态
user_device.is_active = False
user_device.save()
# 检查设备是否还有其他活跃绑定
active_bindings = UserDevice.objects.filter(device=user_device.device, is_active=True).count()
if active_bindings == 0:
user_device.device.status = 'out_stock'
user_device.device.save()
return success(message='解绑成功')
@action(detail=True, methods=['put'], url_path='update-spirit')
def update_spirit(self, request, pk=None):
"""
更新设备绑定的智能体
PUT /api/v1/devices/{id}/update-spirit
"""
try:
user_device = UserDevice.objects.get(id=pk, user=request.user, is_active=True)
except UserDevice.DoesNotExist:
return error(code=ErrorCode.DEVICE_NOT_FOUND, message='绑定记录不存在')
spirit_id = request.data.get('spirit_id')
user_device.spirit_id = spirit_id
user_device.save()
return success(data=UserDeviceSerializer(user_device).data, message='更新成功')

0
apps/inventory/__init__.py Normal file
View File

1
apps/inventory/models.py Normal file
View File

@ -0,0 +1 @@
# 出入库模块使用devices模块的模型，无需额外模型

224
apps/inventory/services.py Normal file
View File

@ -0,0 +1,224 @@
"""
出入库模块服务层 - SN码生成、Excel导入导出
"""
import io
from datetime import datetime
from django.db import transaction
from openpyxl import Workbook, load_workbook
from openpyxl.drawing.image import Image as XLImage
import qrcode
from io import BytesIO
from apps.devices.models import DeviceType, DeviceBatch, Device
def generate_production_week(date):
"""
生成生产周格式: YYWXX
:param date: 日期对象
:return: 如 25W45
"""
year = date.year % 100
week = date.isocalendar()[1]
return f"{year}W{week:02d}"
def generate_sn_code(brand, product_code, production_week, batch_no, sequence):
"""
生成SN码
格式: {品牌}-{产品代号}-{生产周}-{批次号}-{5位序列号}
示例: AL-DZBJ-ON-25W45-A01-00001
"""
return f"{brand}-{product_code}-{production_week}-{batch_no}-{sequence:05d}"
def generate_batch_devices(batch):
"""
为批次生成设备和SN码
:param batch: DeviceBatch对象
:return: 生成的Device列表
"""
device_type = batch.device_type
brand = device_type.brand
product_code = device_type.product_code
production_week = batch.production_week or generate_production_week(batch.production_date)
batch_no = batch.batch_no
# 更新生产周
if not batch.production_week:
batch.production_week = production_week
batch.save(update_fields=['production_week'])
devices = []
for i in range(1, batch.quantity + 1):
sn = generate_sn_code(brand, product_code, production_week, batch_no, i)
device = Device(
sn=sn,
device_type=device_type,
batch=batch,
status='in_stock'
)
devices.append(device)
# 批量创建
Device.objects.bulk_create(devices)
# 更新批次状态
batch.status = 'generated'
batch.save(update_fields=['status'])
return devices
def generate_qrcode_image(data, size=100):
"""
生成二维码图片
:param data: 二维码内容
:param size: 尺寸
:return: BytesIO对象
"""
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=4,
border=1,
)
qr.add_data(data)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img_io = BytesIO()
img.save(img_io, format='PNG')
img_io.seek(0)
return img_io
def export_batch_excel(batch):
"""
导出批次SN码Excel
包含四列: SN码 / MAC地址 / 创建时间 / SN码二维码
:param batch: DeviceBatch对象
:return: BytesIO对象（Excel文件）
"""
wb = Workbook()
ws = wb.active
ws.title = f"批次{batch.batch_no}"
# 设置表头
headers = ['SN码', 'MAC地址', '创建时间', 'SN码二维码']
for col, header in enumerate(headers, 1):
ws.cell(row=1, column=col, value=header)
# 设置列宽
ws.column_dimensions['A'].width = 35
ws.column_dimensions['B'].width = 20
ws.column_dimensions['C'].width = 22
ws.column_dimensions['D'].width = 15
# 获取批次设备
devices = Device.objects.filter(batch=batch).order_by('sn')
for row, device in enumerate(devices, 2):
ws.cell(row=row, column=1, value=device.sn)
ws.cell(row=row, column=2, value=device.mac_address or '')
ws.cell(row=row, column=3, value=device.created_at.strftime('%Y-%m-%d %H:%M:%S'))
# 生成二维码并嵌入
try:
qr_img = generate_qrcode_image(device.sn)
img = XLImage(qr_img)
img.width = 60
img.height = 60
ws.add_image(img, f'D{row}')
ws.row_dimensions[row].height = 50
except Exception:
ws.cell(row=row, column=4, value='[QR]')
# 保存到内存
output = BytesIO()
wb.save(output)
output.seek(0)
return output
def import_mac_addresses(batch, file_obj):
"""
导入MAC地址
:param batch: DeviceBatch对象
:param file_obj: Excel文件对象
:return: 导入结果字典
"""
result = {
'total': 0,
'success': 0,
'failed': 0,
'errors': []
}
try:
wb = load_workbook(file_obj)
ws = wb.active
except Exception as e:
result['errors'].append({'row': 0, 'error': f'文件格式错误: {str(e)}'})
return result
# 跳过表头
rows = list(ws.iter_rows(min_row=2, values_only=True))
result['total'] = len(rows)
for row_num, row in enumerate(rows, 2):
if len(row) < 2:
continue
sn = row[0]
mac = row[1]
if not sn or not mac:
continue
# 统一MAC地址格式
mac = str(mac).upper().replace('-', ':').strip()
# 验证MAC地址格式
import re
if not re.match(r'^([0-9A-F]{2}:){5}[0-9A-F]{2}$', mac):
result['failed'] += 1
result['errors'].append({
'row': row_num,
'sn': sn,
'mac': mac,
'error': 'MAC地址格式不正确'
})
continue
try:
device = Device.objects.get(sn=sn, batch=batch)
except Device.DoesNotExist:
result['failed'] += 1
result['errors'].append({
'row': row_num,
'sn': sn,
'mac': mac,
'error': 'SN码不存在或不属于当前批次'
})
continue
# 检查MAC地址是否已被使用
existing = Device.objects.filter(mac_address=mac).exclude(id=device.id).first()
if existing:
result['failed'] += 1
result['errors'].append({
'row': row_num,
'sn': sn,
'mac': mac,
'error': f'MAC地址已被设备 {existing.sn} 使用'
})
continue
# 更新MAC地址
device.mac_address = mac
device.save(update_fields=['mac_address'])
result['success'] += 1
return result

16
apps/inventory/urls.py Normal file
View File

@ -0,0 +1,16 @@
"""
出入库模块URL配置
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import DeviceTypeViewSet, DeviceBatchViewSet
admin_router = DefaultRouter()
admin_router.register('device-types', DeviceTypeViewSet, basename='admin-device-types')
admin_router.register('device-batches', DeviceBatchViewSet, basename='admin-device-batches')
urlpatterns = []
admin_urlpatterns = [
path('', include(admin_router.urls)),
]

259
apps/inventory/views.py Normal file
View File

@ -0,0 +1,259 @@
"""
出入库模块视图 - 管理端
使用AdminJWTAuthentication进行认证
"""
from django.http import HttpResponse
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.parsers import MultiPartParser
from utils.response import success, error
from utils.exceptions import ErrorCode
from apps.admins.authentication import AdminJWTAuthentication
from apps.admins.permissions import IsAdminUser
from apps.devices.models import DeviceType, DeviceBatch, Device
from apps.devices.serializers import (
DeviceTypeSerializer,
DeviceBatchSerializer,
DeviceBatchCreateSerializer,
DeviceSimpleSerializer
)
from .services import generate_batch_devices, export_batch_excel, import_mac_addresses
class DeviceTypeViewSet(viewsets.ModelViewSet):
"""设备类型管理视图集 - 管理端"""
queryset = DeviceType.objects.all().order_by('-created_at')
serializer_class = DeviceTypeSerializer
authentication_classes = [AdminJWTAuthentication]
permission_classes = [IsAdminUser]
def list(self, request, *args, **kwargs):
"""
设备类型列表
GET /api/admin/device-types
"""
queryset = self.filter_queryset(self.get_queryset())
# 过滤
brand = request.query_params.get('brand')
is_active = request.query_params.get('is_active')
if brand:
queryset = queryset.filter(brand__contains=brand)
if is_active is not None:
queryset = queryset.filter(is_active=is_active.lower() == 'true')
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return success(data={
'total': self.paginator.page.paginator.count,
'items': serializer.data
})
serializer = self.get_serializer(queryset, many=True)
return success(data={'items': serializer.data})
def create(self, request, *args, **kwargs):
"""
创建设备类型
POST /api/admin/device-types
"""
serializer = self.get_serializer(data=request.data)
if serializer.is_valid():
serializer.save()
return success(data=serializer.data, message='创建成功')
return error(message=str(serializer.errors))
def retrieve(self, request, *args, **kwargs):
"""
设备类型详情
GET /api/admin/device-types/{id}
"""
instance = self.get_object()
serializer = self.get_serializer(instance)
return success(data=serializer.data)
def update(self, request, *args, **kwargs):
"""
更新设备类型
PUT /api/admin/device-types/{id}
"""
instance = self.get_object()
serializer = self.get_serializer(instance, data=request.data, partial=True)
if serializer.is_valid():
serializer.save()
return success(data=serializer.data, message='更新成功')
return error(message=str(serializer.errors))
class DeviceBatchViewSet(viewsets.ModelViewSet):
"""设备批次管理视图集 - 管理端"""
queryset = DeviceBatch.objects.all().select_related('device_type').order_by('-created_at')
authentication_classes = [AdminJWTAuthentication]
permission_classes = [IsAdminUser]
def get_serializer_class(self):
if self.action == 'create':
return DeviceBatchCreateSerializer
return DeviceBatchSerializer
def list(self, request, *args, **kwargs):
"""
批次列表
GET /api/admin/device-batches
"""
queryset = self.filter_queryset(self.get_queryset())
# 过滤
device_type_id = request.query_params.get('device_type_id')
brand = request.query_params.get('brand')
batch_no = request.query_params.get('batch_no')
status_filter = request.query_params.get('status')
if device_type_id:
queryset = queryset.filter(device_type_id=device_type_id)
if brand:
queryset = queryset.filter(device_type__brand__contains=brand)
if batch_no:
queryset = queryset.filter(batch_no__contains=batch_no)
if status_filter:
queryset = queryset.filter(status=status_filter)
page = self.paginate_queryset(queryset)
if page is not None:
serializer = DeviceBatchSerializer(page, many=True)
return success(data={
'total': self.paginator.page.paginator.count,
'items': serializer.data
})
serializer = DeviceBatchSerializer(queryset, many=True)
return success(data={'items': serializer.data})
def create(self, request, *args, **kwargs):
"""
创建批次并生成SN码
POST /api/admin/device-batches
"""
serializer = DeviceBatchCreateSerializer(data=request.data)
if not serializer.is_valid():
return error(message=str(serializer.errors))
# 验证设备类型
device_type_id = serializer.validated_data['device_type'].id
try:
device_type = DeviceType.objects.get(id=device_type_id, is_active=True)
except DeviceType.DoesNotExist:
return error(code=ErrorCode.DEVICE_TYPE_NOT_FOUND, message='设备类型不存在或已禁用')
# 创建批次（注意：这里created_by需要存AdminUser的引用，暂时设为None）
batch = serializer.save(created_by=None)
# 生成SN码和设备
try:
devices = generate_batch_devices(batch)
except Exception as e:
batch.delete()
return error(message=f'生成SN码失败: {str(e)}')
# 返回结果
batch.refresh_from_db()
return success(data={
**DeviceBatchSerializer(batch).data,
'sn_range': {
'start': devices[0].sn if devices else '',
'end': devices[-1].sn if devices else ''
}
}, message='批次创建成功')
def retrieve(self, request, *args, **kwargs):
"""
批次详情
GET /api/admin/device-batches/{id}
"""
instance = self.get_object()
# 获取设备统计
devices = Device.objects.filter(batch=instance)
stats = {
'total': devices.count(),
'in_stock': devices.filter(status='in_stock').count(),
'out_stock': devices.filter(status='out_stock').count(),
'bound': devices.filter(status='bound').count(),
'with_mac': devices.exclude(mac_address__isnull=True).exclude(mac_address='').count()
}
return success(data={
**DeviceBatchSerializer(instance).data,
'statistics': stats
})
@action(detail=True, methods=['get'])
def devices(self, request, pk=None):
"""
批次设备列表
GET /api/admin/device-batches/{id}/devices
"""
batch = self.get_object()
devices = Device.objects.filter(batch=batch).order_by('sn')
page = self.paginate_queryset(devices)
if page is not None:
serializer = DeviceSimpleSerializer(page, many=True)
return success(data={
'total': self.paginator.page.paginator.count,
'items': serializer.data
})
serializer = DeviceSimpleSerializer(devices, many=True)
return success(data={'items': serializer.data})
@action(detail=True, methods=['get'])
def export(self, request, pk=None):
"""
导出批次SN码Excel
GET /api/admin/device-batches/{id}/export
"""
batch = self.get_object()
# 生成Excel
excel_file = export_batch_excel(batch)
# 返回文件
response = HttpResponse(
excel_file.getvalue(),
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
)
filename = f"batch_{batch.batch_no}_{batch.production_week}.xlsx"
response['Content-Disposition'] = f'attachment; filename="{filename}"'
return response
@action(detail=True, methods=['post'], parser_classes=[MultiPartParser])
def import_mac(self, request, pk=None):
"""
导入MAC地址
POST /api/admin/device-batches/{id}/import_mac
"""
batch = self.get_object()
file_obj = request.FILES.get('file')
if not file_obj:
return error(message='请上传Excel文件')
# 验证文件类型
if not file_obj.name.endswith(('.xlsx', '.xls')):
return error(message='仅支持Excel文件格式(.xlsx, .xls)')
# 导入MAC地址
result = import_mac_addresses(batch, file_obj)
message = f"导入完成，成功{result['success']}条"
if result['failed'] > 0:
message += f"，失败{result['failed']}条"
return success(data=result, message=message)

0
apps/spirits/__init__.py Normal file
View File

58
apps/spirits/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,58 @@
# Generated by Django 6.0.1 on 2026-01-28 10:03
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = []
operations = [
migrations.CreateModel(
name="Spirit",
fields=[
("id", models.BigAutoField(primary_key=True, serialize=False)),
("name", models.CharField(max_length=100, verbose_name="名称")),
(
"avatar",
models.URLField(
blank=True, default="", max_length=500, verbose_name="头像"
),
),
(
"prompt",
models.TextField(blank=True, default="", verbose_name="提示词"),
),
(
"memory",
models.TextField(blank=True, default="", verbose_name="记忆"),
),
(
"voice_id",
models.CharField(
blank=True, default="", max_length=100, verbose_name="音色ID"
),
),
(
"is_active",
models.BooleanField(default=True, verbose_name="是否启用"),
),
(
"created_at",
models.DateTimeField(auto_now_add=True, verbose_name="创建时间"),
),
(
"updated_at",
models.DateTimeField(auto_now=True, verbose_name="更新时间"),
),
],
options={
"verbose_name": "智能体",
"verbose_name_plural": "智能体",
"db_table": "spirit",
"ordering": ["-created_at"],
},
),
]

28
apps/spirits/migrations/0002_initial.py Normal file
View File

@ -0,0 +1,28 @@
# Generated by Django 6.0.1 on 2026-01-28 10:03
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
("spirits", "0001_initial"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AddField(
model_name="spirit",
name="user",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="spirits",
to=settings.AUTH_USER_MODEL,
verbose_name="所属用户",
),
),
]

0
apps/spirits/migrations/__init__.py Normal file
View File

29
apps/spirits/models.py Normal file
View File

@ -0,0 +1,29 @@
"""
智能体模块模型
"""
from django.db import models
from apps.users.models import User
class Spirit(models.Model):
"""智能体/心灵模型"""
id = models.BigAutoField(primary_key=True)
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='spirits', verbose_name='所属用户')
name = models.CharField('名称', max_length=100)
avatar = models.URLField('头像', max_length=500, blank=True, default='')
prompt = models.TextField('提示词', blank=True, default='')
memory = models.TextField('记忆', blank=True, default='')
voice_id = models.CharField('音色ID', max_length=100, blank=True, default='')
is_active = models.BooleanField('是否启用', default=True)
created_at = models.DateTimeField('创建时间', auto_now_add=True)
updated_at = models.DateTimeField('更新时间', auto_now=True)
class Meta:
db_table = 'spirit'
verbose_name = '智能体'
verbose_name_plural = '智能体'
ordering = ['-created_at']
def __str__(self):
return f"{self.name} - {self.user.phone}"

41
apps/spirits/serializers.py Normal file
View File

@ -0,0 +1,41 @@
"""
智能体模块序列化器
"""
from rest_framework import serializers
from .models import Spirit
class SpiritSerializer(serializers.ModelSerializer):
"""智能体序列化器"""
class Meta:
model = Spirit
fields = ['id', 'name', 'avatar', 'prompt', 'memory', 'voice_id', 'is_active', 'created_at', 'updated_at']
read_only_fields = ['id', 'created_at', 'updated_at']
class CreateSpiritSerializer(serializers.ModelSerializer):
"""创建智能体序列化器"""
class Meta:
model = Spirit
fields = ['name', 'avatar', 'prompt', 'memory', 'voice_id']
class SpiritListSerializer(serializers.ModelSerializer):
"""智能体列表序列化器（不含memory等大字段）"""
class Meta:
model = Spirit
fields = ['id', 'name', 'avatar', 'is_active', 'created_at']
class AdminSpiritSerializer(serializers.ModelSerializer):
"""管理端智能体序列化器"""
user_phone = serializers.CharField(source='user.phone', read_only=True)
class Meta:
model = Spirit
fields = ['id', 'user', 'user_phone', 'name', 'avatar', 'prompt', 'memory', 'voice_id', 'is_active', 'created_at', 'updated_at']
read_only_fields = ['id', 'created_at', 'updated_at']

13
apps/spirits/urls.py Normal file
View File

@ -0,0 +1,13 @@
"""
智能体模块URL配置 - 仅App端接口
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import SpiritViewSet
router = DefaultRouter()
router.register('spirits', SpiritViewSet, basename='spirits')
urlpatterns = [
path('', include(router.urls)),
]

85
apps/spirits/views.py Normal file
View File

@ -0,0 +1,85 @@
"""
智能体模块视图 - App端
"""
from rest_framework import viewsets, status
from rest_framework.permissions import IsAuthenticated
from utils.response import success, error
from utils.exceptions import ErrorCode
from apps.admins.authentication import AppJWTAuthentication
from .models import Spirit
from .serializers import (
SpiritSerializer,
SpiritListSerializer,
CreateSpiritSerializer,
)
class SpiritViewSet(viewsets.ModelViewSet):
"""智能体视图集（App端）"""
authentication_classes = [AppJWTAuthentication]
permission_classes = [IsAuthenticated]
def get_queryset(self):
return Spirit.objects.filter(user=self.request.user)
def get_serializer_class(self):
if self.action == 'list':
return SpiritListSerializer
if self.action == 'create':
return CreateSpiritSerializer
return SpiritSerializer
def list(self, request, *args, **kwargs):
"""
获取我的智能体列表
GET /api/v1/spirits
"""
queryset = self.get_queryset()
serializer = SpiritListSerializer(queryset, many=True)
return success(data=serializer.data)
def create(self, request, *args, **kwargs):
"""
创建智能体
POST /api/v1/spirits
"""
serializer = CreateSpiritSerializer(data=request.data)
if serializer.is_valid():
spirit = serializer.save(user=request.user)
return success(data=SpiritSerializer(spirit).data, message='创建成功')
return error(message=str(serializer.errors))
def retrieve(self, request, *args, **kwargs):
"""
获取智能体详情
GET /api/v1/spirits/{id}
"""
try:
instance = self.get_object()
serializer = SpiritSerializer(instance)
return success(data=serializer.data)
except Spirit.DoesNotExist:
return error(code=ErrorCode.SPIRIT_NOT_FOUND, message='智能体不存在')
def update(self, request, *args, **kwargs):
"""
更新智能体
PUT /api/v1/spirits/{id}
"""
instance = self.get_object()
serializer = SpiritSerializer(instance, data=request.data, partial=True)
if serializer.is_valid():
serializer.save()
return success(data=serializer.data, message='更新成功')
return error(message=str(serializer.errors))
def destroy(self, request, *args, **kwargs):
"""
删除智能体
DELETE /api/v1/spirits/{id}
"""
instance = self.get_object()
instance.delete()
return success(message='删除成功')

0
apps/users/__init__.py Normal file
View File

95
apps/users/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,95 @@
# Generated by Django 6.0.1 on 2026-01-28 10:03
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
("auth", "0012_alter_user_first_name_max_length"),
]
operations = [
migrations.CreateModel(
name="User",
fields=[
("password", models.CharField(max_length=128, verbose_name="password")),
(
"last_login",
models.DateTimeField(
blank=True, null=True, verbose_name="last login"
),
),
(
"is_superuser",
models.BooleanField(
default=False,
help_text="Designates that this user has all permissions without explicitly assigning them.",
verbose_name="superuser status",
),
),
("id", models.BigAutoField(primary_key=True, serialize=False)),
(
"phone",
models.CharField(max_length=20, unique=True, verbose_name="手机号"),
),
(
"nickname",
models.CharField(
blank=True, default="", max_length=50, verbose_name="昵称"
),
),
(
"avatar",
models.URLField(
blank=True, default="", max_length=500, verbose_name="头像"
),
),
(
"is_active",
models.BooleanField(default=True, verbose_name="是否激活"),
),
(
"is_staff",
models.BooleanField(default=False, verbose_name="是否管理员"),
),
(
"created_at",
models.DateTimeField(auto_now_add=True, verbose_name="创建时间"),
),
(
"updated_at",
models.DateTimeField(auto_now=True, verbose_name="更新时间"),
),
(
"groups",
models.ManyToManyField(
blank=True,
help_text="The groups this user belongs to. A user will get all permissions granted to each of their groups.",
related_name="user_set",
related_query_name="user",
to="auth.group",
verbose_name="groups",
),
),
(
"user_permissions",
models.ManyToManyField(
blank=True,
help_text="Specific permissions for this user.",
related_name="user_set",
related_query_name="user",
to="auth.permission",
verbose_name="user permissions",
),
),
],
options={
"verbose_name": "用户",
"verbose_name_plural": "用户",
"db_table": "user",
},
),
]

0
apps/users/migrations/__init__.py Normal file
View File

49
apps/users/models.py Normal file
View File

@ -0,0 +1,49 @@
"""
用户模块模型
"""
from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin
from django.db import models
class UserManager(BaseUserManager):
"""用户管理器"""
def create_user(self, phone, password=None, **extra_fields):
if not phone:
raise ValueError('手机号不能为空')
user = self.model(phone=phone, **extra_fields)
if password:
user.set_password(password)
user.save(using=self._db)
return user
def create_superuser(self, phone, password=None, **extra_fields):
extra_fields.setdefault('is_staff', True)
extra_fields.setdefault('is_superuser', True)
return self.create_user(phone, password, **extra_fields)
class User(AbstractBaseUser, PermissionsMixin):
"""用户模型"""
id = models.BigAutoField(primary_key=True)
phone = models.CharField('手机号', max_length=20, unique=True)
nickname = models.CharField('昵称', max_length=50, blank=True, default='')
avatar = models.URLField('头像', max_length=500, blank=True, default='')
is_active = models.BooleanField('是否激活', default=True)
is_staff = models.BooleanField('是否管理员', default=False)
created_at = models.DateTimeField('创建时间', auto_now_add=True)
updated_at = models.DateTimeField('更新时间', auto_now=True)
objects = UserManager()
USERNAME_FIELD = 'phone'
REQUIRED_FIELDS = []
class Meta:
db_table = 'user'
verbose_name = '用户'
verbose_name_plural = '用户'
def __str__(self):
return self.phone

44
apps/users/serializers.py Normal file
View File

@ -0,0 +1,44 @@
"""
用户模块序列化器
"""
from rest_framework import serializers
from .models import User
class UserSerializer(serializers.ModelSerializer):
"""用户序列化器"""
class Meta:
model = User
fields = ['id', 'phone', 'nickname', 'avatar', 'created_at']
read_only_fields = ['id', 'phone', 'created_at']
class UserDetailSerializer(serializers.ModelSerializer):
"""用户详情序列化器（管理端）"""
class Meta:
model = User
fields = ['id', 'phone', 'nickname', 'avatar', 'is_active', 'is_staff', 'created_at', 'updated_at']
class PhoneLoginSerializer(serializers.Serializer):
"""手机号一键登录序列化器"""
phone = serializers.CharField(max_length=20, help_text='手机号')
# 实际项目中应该有验证码或token
# code = serializers.CharField(max_length=10, help_text='验证码')
def validate_phone(self, value):
# 简单验证手机号格式
if not value.isdigit() or len(value) != 11:
raise serializers.ValidationError('手机号格式不正确')
return value
class UpdateUserSerializer(serializers.ModelSerializer):
"""更新用户信息序列化器"""
class Meta:
model = User
fields = ['nickname', 'avatar']

14
apps/users/urls.py Normal file
View File

@ -0,0 +1,14 @@
"""
用户模块URL配置 - 仅App端接口
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import AuthViewSet, UserViewSet
router = DefaultRouter()
router.register('auth', AuthViewSet, basename='auth')
router.register('users', UserViewSet, basename='users')
urlpatterns = [
path('', include(router.urls)),
]

120
apps/users/views.py Normal file
View File

@ -0,0 +1,120 @@
"""
用户模块视图 - App端用户
"""
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework_simplejwt.tokens import RefreshToken
from utils.response import success, error
from apps.admins.authentication import AppJWTAuthentication
from .models import User
from .serializers import (
UserSerializer,
UserDetailSerializer,
PhoneLoginSerializer,
UpdateUserSerializer
)
def get_app_tokens(user):
"""
为App用户生成JWT Token
在token中添加 user_type='app' 以区分管理员
"""
refresh = RefreshToken.for_user(user)
# 添加自定义声明
refresh['user_type'] = 'app'
refresh['phone'] = user.phone
return {
'access': str(refresh.access_token),
'refresh': str(refresh),
}
class AuthViewSet(viewsets.ViewSet):
"""认证视图集 - App端"""
permission_classes = [AllowAny]
@action(detail=False, methods=['post'], url_path='phone-login')
def phone_login(self, request):
"""
手机号一键登录
POST /api/v1/auth/phone-login
"""
serializer = PhoneLoginSerializer(data=request.data)
if not serializer.is_valid():
return error(message=str(serializer.errors))
phone = serializer.validated_data['phone']
# 获取或创建用户
user, created = User.objects.get_or_create(
phone=phone,
defaults={'nickname': f'用户{phone[-4:]}'}
)
if not user.is_active:
return error(code=101, message='账号已被禁用')
# 生成JWT Token（带user_type='app'标识）
tokens = get_app_tokens(user)
return success(data={
'user': UserSerializer(user).data,
'token': tokens,
'is_new_user': created
}, message='登录成功')
@action(detail=False, methods=['post'], url_path='refresh')
def refresh_token(self, request):
"""
刷新Token
POST /api/v1/auth/refresh
"""
refresh_token = request.data.get('refresh')
if not refresh_token:
return error(message='refresh token不能为空')
try:
refresh = RefreshToken(refresh_token)
# 验证是否为app token
user_type = refresh.get('user_type', 'app')
if user_type not in ['app', None]:
return error(code=103, message='无效的用户Token')
return success(data={
'access': str(refresh.access_token),
'refresh': str(refresh),
})
except Exception as e:
return error(code=103, message='Token已过期或无效')
class UserViewSet(viewsets.ViewSet):
"""用户视图集 - App端"""
authentication_classes = [AppJWTAuthentication]
permission_classes = [IsAuthenticated]
@action(detail=False, methods=['get'])
def me(self, request):
"""
获取当前用户信息
GET /api/v1/users/me
"""
return success(data=UserSerializer(request.user).data)
@action(detail=False, methods=['put'])
def update_me(self, request):
"""
更新当前用户信息
PUT /api/v1/users/update_me
"""
serializer = UpdateUserSerializer(request.user, data=request.data, partial=True)
if serializer.is_valid():
serializer.save()
return success(data=UserSerializer(request.user).data, message='更新成功')
return error(message=str(serializer.errors))

0
config/__init__.py Normal file
View File

16
config/asgi.py Normal file
View File

@ -0,0 +1,16 @@
"""
ASGI config for config project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/6.0/howto/deployment/asgi/
"""
import os
from django.core.asgi import get_asgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
application = get_asgi_application()

196
config/settings.py Normal file
View File

@ -0,0 +1,196 @@
"""
Django settings for RTC_DEMO project.
"""
import os
from pathlib import Path
from datetime import timedelta
# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = os.environ.get('DJANGO_SECRET_KEY', 'django-insecure-dev-key-change-in-production')
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = os.environ.get('DJANGO_DEBUG', 'True').lower() == 'true'
ALLOWED_HOSTS = os.environ.get('DJANGO_ALLOWED_HOSTS', '*').split(',')
# Application definition
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
# Third party
'rest_framework',
'rest_framework_simplejwt',
'corsheaders',
'drf_spectacular',
# Local apps
'apps.users',
'apps.spirits',
'apps.devices',
'apps.inventory',
'apps.admins',
]
MIDDLEWARE = [
'corsheaders.middleware.CorsMiddleware',
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
ROOT_URLCONF = 'config.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
WSGI_APPLICATION = 'config.wsgi.application'
# Database - MySQL via PyMySQL
import pymysql
pymysql.version_info = (2, 2, 1, 'final', 0) # Django 6.x requires mysqlclient >= 2.2.1
pymysql.install_as_MySQLdb()
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.mysql',
'NAME': os.environ.get('DB_NAME', 'rtc'),
'USER': os.environ.get('DB_USER', 'rtc'),
'PASSWORD': os.environ.get('DB_PASSWORD', 'JogNQdtrd3WY8CBCAiYfYEGx'),
'HOST': os.environ.get('DB_HOST', 'rm-7xv1uaw910558p1788o.mysql.rds.aliyuncs.com'),
'PORT': os.environ.get('DB_PORT', '3306'),
'OPTIONS': {
'charset': 'utf8mb4',
},
'TEST': {
'NAME': 'rtc', # 使用现有数据库进行测试
},
}
}
# Redis Cache
REDIS_PASSWORD = os.environ.get('REDIS_PASSWORD', 'vAhRnAA6VMco')
REDIS_URL = os.environ.get('REDIS_URL', f'redis://:{REDIS_PASSWORD}@r-7xvat0vez5clwbzk5vpd.redis.rds.aliyuncs.com:6379/8')
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.redis.RedisCache',
'LOCATION': REDIS_URL,
}
}
# Password validation
AUTH_PASSWORD_VALIDATORS = [
{'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'},
{'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator'},
{'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
{'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
]
# Custom User Model
AUTH_USER_MODEL = 'users.User'
# Internationalization
LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True
USE_TZ = True
# Static files
STATIC_URL = 'static/'
# Default primary key field type
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
# CORS Settings
CORS_ALLOW_ALL_ORIGINS = DEBUG
CORS_ALLOWED_ORIGINS = os.environ.get('CORS_ALLOWED_ORIGINS', '').split(',') if not DEBUG else []
# REST Framework
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework_simplejwt.authentication.JWTAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
],
'DEFAULT_RENDERER_CLASSES': [
'rest_framework.renderers.JSONRenderer',
],
'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema',
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 20,
'EXCEPTION_HANDLER': 'utils.exceptions.custom_exception_handler',
}
# JWT Settings
SIMPLE_JWT = {
'ACCESS_TOKEN_LIFETIME': timedelta(days=7),
'REFRESH_TOKEN_LIFETIME': timedelta(days=30),
'ROTATE_REFRESH_TOKENS': True,
'ALGORITHM': 'HS256',
'AUTH_HEADER_TYPES': ('Bearer',),
}
# Aliyun OSS Settings
ALIYUN_OSS = {
'ACCESS_KEY_ID': os.environ.get('OSS_ACCESS_KEY_ID', 'LTAI5tBGAkR2rra2prTAX9yc'),
'ACCESS_KEY_SECRET': os.environ.get('OSS_ACCESS_KEY_SECRET', 'U1z3d0p5saPRD5sCxVooJYSjxSAmKB'),
'ENDPOINT': os.environ.get('OSS_ENDPOINT', 'oss-cn-beijing.aliyuncs.com'),
'BUCKET_NAME': os.environ.get('OSS_BUCKET_NAME', 'qy-rtc'),
'CUSTOM_DOMAIN': os.environ.get('OSS_CUSTOM_DOMAIN', ''),
}
# Swagger/OpenAPI Settings
SPECTACULAR_SETTINGS = {
'TITLE': 'RTC_DEMO API',
'DESCRIPTION': 'RTC物联网设备管理平台API文档',
'VERSION': '1.0.0',
'SERVE_INCLUDE_SCHEMA': False,
'COMPONENT_SPLIT_REQUEST': True,
'SWAGGER_UI_SETTINGS': {
'persistAuthorization': True,
},
}
# Logging
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {message}',
'style': '{',
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'verbose',
},
},
'root': {
'handlers': ['console'],
'level': 'INFO',
},
}

25
config/settings_test.py Normal file
View File

@ -0,0 +1,25 @@
"""
测试配置文件
使用SQLite运行测试以避免云数据库权限问题
"""
from config.settings import *
# 使用SQLite进行测试
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': BASE_DIR / 'test_db.sqlite3',
}
}
# 禁用Redis缓存，使用本地内存缓存
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
}
}
# 加速密码哈希
PASSWORD_HASHERS = [
'django.contrib.auth.hashers.MD5PasswordHasher',
]

43
config/urls.py Normal file
View File

@ -0,0 +1,43 @@
"""
URL configuration for RTC_DEMO project.
路由完全分离：
- /api/v1/ - App端接口（使用AppJWTAuthentication，User模型）
- /api/admin/ - Web管理端接口（使用AdminJWTAuthentication，AdminUser模型）
"""
from django.contrib import admin
from django.urls import path, include
from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView, SpectacularRedocView
# ============ App端路由 (普通用户，手机一键登录) ============
app_api_patterns = [
path('', include('apps.users.urls')),
path('', include('apps.spirits.urls')),
path('', include('apps.devices.urls')),
]
# ============ Web管理端路由 (管理员，用户名密码登录) ============
from apps.inventory.urls import admin_urlpatterns as inventory_admin_urls
admin_api_patterns = [
# 管理员认证和个人信息
path('', include('apps.admins.urls')),
# 业务管理接口
path('', include(inventory_admin_urls)),
]
urlpatterns = [
# Django Admin
path('django-admin/', admin.site.urls),
# Swagger/OpenAPI文档
path('api/schema/', SpectacularAPIView.as_view(), name='schema'),
path('api/docs/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'),
path('api/redoc/', SpectacularRedocView.as_view(url_name='schema'), name='redoc'),
# App端API - 普通用户
path('api/v1/', include(app_api_patterns)),
# Web管理端API - 管理员
path('api/admin/', include(admin_api_patterns)),
]

16
config/wsgi.py Normal file
View File

@ -0,0 +1,16 @@
"""
WSGI config for config project.
It exposes the WSGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/6.0/howto/deployment/wsgi/
"""
import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
application = get_wsgi_application()

22
manage.py Executable file
View File

@ -0,0 +1,22 @@
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main():
"""Run administrative tasks."""
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == "__main__":
main()

28
requirements.txt Normal file
View File

@ -0,0 +1,28 @@
aliyun-python-sdk-core==2.16.0
aliyun-python-sdk-kms==2.16.5
asgiref==3.11.0
certifi==2026.1.4
cffi==2.0.0
charset-normalizer==3.4.4
crcmod==1.7
cryptography==46.0.4
Django==6.0.1
django-cors-headers==4.9.0
djangorestframework==3.16.1
djangorestframework_simplejwt==5.5.1
et_xmlfile==2.0.0
idna==3.11
jmespath==0.10.0
openpyxl==3.1.5
oss2==2.19.1
pillow==12.1.0
pycparser==3.0
pycryptodome==3.23.0
PyJWT==2.10.1
PyMySQL==1.1.2
qrcode==8.2
redis==7.1.0
requests==2.32.5
six==1.17.0
sqlparse==0.5.5
urllib3==2.6.3

787
tests.py Normal file
View File

@ -0,0 +1,787 @@
"""
RTC_DEMO API 完整测试用例
覆盖所有API接口
运行方式: python manage.py test tests --verbosity=2 --keepdb
"""
import json
from datetime import date
from django.test import TestCase
from django.urls import reverse
from rest_framework.test import APITestCase, APIClient
from rest_framework import status
from apps.users.models import User
from apps.admins.models import AdminUser
from apps.spirits.models import Spirit
from apps.devices.models import DeviceType, DeviceBatch, Device, UserDevice
# ==================== App端测试 ====================
class UserAuthTests(APITestCase):
"""App端用户认证测试"""
def test_phone_login_new_user(self):
"""测试手机号一键登录 - 新用户"""
url = '/api/v1/auth/phone-login/'
data = {'phone': '13800138001'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertIn('token', response.data['data'])
self.assertTrue(response.data['data']['is_new_user'])
self.assertEqual(response.data['data']['user']['phone'], '13800138001')
def test_phone_login_existing_user(self):
"""测试手机号一键登录 - 已有用户"""
User.objects.create_user(phone='13800138002', nickname='测试用户')
url = '/api/v1/auth/phone-login/'
data = {'phone': '13800138002'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertFalse(response.data['data']['is_new_user'])
def test_phone_login_invalid_phone(self):
"""测试手机号格式验证"""
url = '/api/v1/auth/phone-login/'
data = {'phone': '1234567'}
response = self.client.post(url, data, format='json')
self.assertNotEqual(response.data['code'], 0)
def test_phone_login_disabled_user(self):
"""测试禁用用户登录"""
User.objects.create_user(phone='13800138003', is_active=False)
url = '/api/v1/auth/phone-login/'
data = {'phone': '13800138003'}
response = self.client.post(url, data, format='json')
self.assertNotEqual(response.data['code'], 0)
def test_token_refresh(self):
"""测试Token刷新"""
login_url = '/api/v1/auth/phone-login/'
login_response = self.client.post(login_url, {'phone': '13800138004'}, format='json')
refresh_token = login_response.data['data']['token']['refresh']
refresh_url = '/api/v1/auth/refresh/'
response = self.client.post(refresh_url, {'refresh': refresh_token}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertIn('access', response.data['data'])
def test_token_refresh_invalid(self):
"""测试无效Token刷新"""
url = '/api/v1/auth/refresh/'
response = self.client.post(url, {'refresh': 'invalid_token'}, format='json')
self.assertNotEqual(response.data['code'], 0)
def test_token_refresh_empty(self):
"""测试空Token刷新"""
url = '/api/v1/auth/refresh/'
response = self.client.post(url, {}, format='json')
self.assertNotEqual(response.data['code'], 0)
class UserProfileTests(APITestCase):
"""App端用户信息测试"""
def setUp(self):
self.user = User.objects.create_user(phone='13800138010', nickname='测试用户')
login_url = '/api/v1/auth/phone-login/'
response = self.client.post(login_url, {'phone': '13800138010'}, format='json')
self.access_token = response.data['data']['token']['access']
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
def test_get_user_info(self):
"""测试获取用户信息"""
url = '/api/v1/users/me/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertEqual(response.data['data']['phone'], '13800138010')
def test_get_user_info_unauthorized(self):
"""测试未登录获取用户信息"""
self.client.credentials()
url = '/api/v1/users/me/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_update_user_info(self):
"""测试更新用户信息"""
url = '/api/v1/users/update_me/'
data = {'nickname': '新昵称'}
response = self.client.put(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertEqual(response.data['data']['nickname'], '新昵称')
def test_update_user_avatar(self):
"""测试更新用户头像"""
url = '/api/v1/users/update_me/'
data = {'avatar': 'https://example.com/avatar.jpg'}
response = self.client.put(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
class SpiritTests(APITestCase):
"""App端智能体测试"""
def setUp(self):
self.user = User.objects.create_user(phone='13800138020', nickname='测试用户')
login_url = '/api/v1/auth/phone-login/'
response = self.client.post(login_url, {'phone': '13800138020'}, format='json')
self.access_token = response.data['data']['token']['access']
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
def test_create_spirit(self):
"""测试创建智能体"""
url = '/api/v1/spirits/'
data = {
'name': '测试心灵',
'prompt': '你是一个友好的助手',
'voice_id': 'voice_001'
}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertEqual(response.data['data']['name'], '测试心灵')
def test_create_spirit_minimal(self):
"""测试创建智能体 - 最少字段"""
url = '/api/v1/spirits/'
data = {'name': '简单心灵'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
def test_list_spirits(self):
"""测试获取智能体列表"""
Spirit.objects.create(user=self.user, name='心灵1')
Spirit.objects.create(user=self.user, name='心灵2')
url = '/api/v1/spirits/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertEqual(len(response.data['data']), 2)
def test_list_spirits_empty(self):
"""测试获取空智能体列表"""
url = '/api/v1/spirits/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['data']), 0)
def test_get_spirit_detail(self):
"""测试获取智能体详情"""
spirit = Spirit.objects.create(user=self.user, name='详情测试')
url = f'/api/v1/spirits/{spirit.id}/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['data']['name'], '详情测试')
def test_update_spirit(self):
"""测试更新智能体"""
spirit = Spirit.objects.create(user=self.user, name='原始名称')
url = f'/api/v1/spirits/{spirit.id}/'
data = {'name': '新名称', 'prompt': '新的提示词'}
response = self.client.put(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['data']['name'], '新名称')
def test_delete_spirit(self):
"""测试删除智能体"""
spirit = Spirit.objects.create(user=self.user, name='待删除')
url = f'/api/v1/spirits/{spirit.id}/'
response = self.client.delete(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertFalse(Spirit.objects.filter(id=spirit.id).exists())
def test_spirit_isolation(self):
"""测试智能体用户隔离 - 不能访问其他用户的智能体"""
other_user = User.objects.create_user(phone='13800138021')
other_spirit = Spirit.objects.create(user=other_user, name='其他用户的心灵')
url = f'/api/v1/spirits/{other_spirit.id}/'
response = self.client.get(url)
# 应该返回404或权限错误
self.assertNotEqual(response.status_code, status.HTTP_200_OK)
class DeviceTests(APITestCase):
"""App端设备测试"""
def setUp(self):
self.user = User.objects.create_user(phone='13800138050', nickname='测试用户')
login_url = '/api/v1/auth/phone-login/'
response = self.client.post(login_url, {'phone': '13800138050'}, format='json')
self.access_token = response.data['data']['token']['access']
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
self.device_type = DeviceType.objects.create(
brand='AL',
product_code='DZBJ-ON',
name='电子吧唧-联网版'
)
self.device = Device.objects.create(
sn='AL-DZBJ-ON-25W45-A01-00001',
device_type=self.device_type,
mac_address='AA:BB:CC:DD:EE:FF',
status='in_stock'
)
def test_query_by_mac(self):
"""测试通过MAC地址查询SN码（无需登录）"""
self.client.credentials() # 清除认证
url = '/api/v1/devices/query-by-mac/?mac=AA:BB:CC:DD:EE:FF'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertEqual(response.data['data']['sn'], 'AL-DZBJ-ON-25W45-A01-00001')
def test_query_by_mac_lowercase(self):
"""测试MAC地址查询 - 小写格式"""
self.client.credentials()
url = '/api/v1/devices/query-by-mac/?mac=aa:bb:cc:dd:ee:ff'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
def test_query_by_mac_dash_format(self):
"""测试MAC地址查询 - 横杠格式"""
self.client.credentials()
url = '/api/v1/devices/query-by-mac/?mac=AA-BB-CC-DD-EE-FF'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_query_by_mac_not_found(self):
"""测试MAC地址查询 - 设备不存在"""
self.client.credentials()
url = '/api/v1/devices/query-by-mac/?mac=11:22:33:44:55:66'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_query_by_mac_empty(self):
"""测试MAC地址查询 - 空参数"""
self.client.credentials()
url = '/api/v1/devices/query-by-mac/'
response = self.client.get(url)
self.assertNotEqual(response.data['code'], 0)
def test_verify_device(self):
"""测试验证设备SN"""
url = '/api/v1/devices/verify/'
data = {'sn': 'AL-DZBJ-ON-25W45-A01-00001'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertTrue(response.data['data']['is_bindable'])
def test_verify_device_not_found(self):
"""测试验证设备SN - 不存在"""
url = '/api/v1/devices/verify/'
data = {'sn': 'NOT-EXIST-SN'}
response = self.client.post(url, data, format='json')
self.assertNotEqual(response.data['code'], 0)
def test_bind_device(self):
"""测试绑定设备"""
spirit = Spirit.objects.create(user=self.user, name='测试心灵')
url = '/api/v1/devices/bind/'
data = {
'sn': 'AL-DZBJ-ON-25W45-A01-00001',
'spirit_id': spirit.id
}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
# 验证设备状态
self.device.refresh_from_db()
self.assertEqual(self.device.status, 'bound')
def test_bind_device_without_spirit(self):
"""测试绑定设备 - 不绑定智能体"""
device2 = Device.objects.create(
sn='AL-DZBJ-ON-25W45-A01-00002',
device_type=self.device_type,
status='in_stock'
)
url = '/api/v1/devices/bind/'
data = {'sn': 'AL-DZBJ-ON-25W45-A01-00002'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_my_devices(self):
"""测试我的设备列表"""
UserDevice.objects.create(user=self.user, device=self.device, is_active=True)
url = '/api/v1/devices/my_devices/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertEqual(len(response.data['data']), 1)
def test_my_devices_empty(self):
"""测试我的设备列表 - 空列表"""
url = '/api/v1/devices/my_devices/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['data']), 0)
def test_unbind_device(self):
"""测试解绑设备"""
user_device = UserDevice.objects.create(user=self.user, device=self.device, is_active=True)
self.device.status = 'bound'
self.device.save()
url = f'/api/v1/devices/{user_device.id}/unbind/'
response = self.client.delete(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
user_device.refresh_from_db()
self.assertFalse(user_device.is_active)
def test_update_spirit_on_device(self):
"""测试更新设备绑定的智能体"""
spirit1 = Spirit.objects.create(user=self.user, name='心灵1')
spirit2 = Spirit.objects.create(user=self.user, name='心灵2')
user_device = UserDevice.objects.create(
user=self.user, device=self.device, spirit=spirit1, is_active=True
)
url = f'/api/v1/devices/{user_device.id}/update-spirit/'
data = {'spirit_id': spirit2.id}
response = self.client.put(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
user_device.refresh_from_db()
self.assertEqual(user_device.spirit_id, spirit2.id)
# ==================== 管理端测试 ====================
class AdminAuthTests(APITestCase):
"""管理端认证测试"""
def setUp(self):
self.admin = AdminUser.objects.create_user(
username='admin',
password='admin123',
role='super_admin'
)
def test_admin_login(self):
"""测试管理员登录"""
url = '/api/admin/auth/login/'
data = {'username': 'admin', 'password': 'admin123'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertIn('token', response.data['data'])
self.assertEqual(response.data['data']['admin']['username'], 'admin')
def test_admin_login_wrong_password(self):
"""测试管理员登录 - 密码错误"""
url = '/api/admin/auth/login/'
data = {'username': 'admin', 'password': 'wrong'}
response = self.client.post(url, data, format='json')
self.assertNotEqual(response.data['code'], 0)
def test_admin_login_user_not_exist(self):
"""测试管理员登录 - 用户不存在"""
url = '/api/admin/auth/login/'
data = {'username': 'notexist', 'password': 'password'}
response = self.client.post(url, data, format='json')
self.assertNotEqual(response.data['code'], 0)
def test_admin_login_disabled(self):
"""测试管理员登录 - 账户禁用"""
disabled_admin = AdminUser.objects.create_user(
username='disabled', password='pass123', is_active=False
)
url = '/api/admin/auth/login/'
data = {'username': 'disabled', 'password': 'pass123'}
response = self.client.post(url, data, format='json')
self.assertNotEqual(response.data['code'], 0)
def test_admin_token_refresh(self):
"""测试管理员Token刷新"""
login_url = '/api/admin/auth/login/'
login_response = self.client.post(
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
)
refresh_token = login_response.data['data']['token']['refresh']
refresh_url = '/api/admin/auth/refresh/'
response = self.client.post(refresh_url, {'refresh': refresh_token}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
class AdminProfileTests(APITestCase):
"""管理端个人信息测试"""
def setUp(self):
self.admin = AdminUser.objects.create_user(
username='admin', password='admin123', role='admin'
)
login_url = '/api/admin/auth/login/'
response = self.client.post(
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
)
self.access_token = response.data['data']['token']['access']
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
def test_get_admin_profile(self):
"""测试获取管理员信息"""
url = '/api/admin/profile/me/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['data']['username'], 'admin')
def test_change_password(self):
"""测试修改密码"""
url = '/api/admin/profile/change-password/'
data = {'old_password': 'admin123', 'new_password': 'newpass123'}
response = self.client.put(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
# 验证新密码可以登录
self.client.credentials()
login_url = '/api/admin/auth/login/'
login_response = self.client.post(
login_url, {'username': 'admin', 'password': 'newpass123'}, format='json'
)
self.assertEqual(login_response.data['code'], 0)
def test_change_password_wrong_old(self):
"""测试修改密码 - 原密码错误"""
url = '/api/admin/profile/change-password/'
data = {'old_password': 'wrongpass', 'new_password': 'newpass123'}
response = self.client.put(url, data, format='json')
self.assertNotEqual(response.data['code'], 0)
class AdminUserManageTests(APITestCase):
"""管理员用户管理测试"""
def setUp(self):
self.super_admin = AdminUser.objects.create_user(
username='superadmin', password='super123', role='super_admin'
)
login_url = '/api/admin/auth/login/'
response = self.client.post(
login_url, {'username': 'superadmin', 'password': 'super123'}, format='json'
)
self.access_token = response.data['data']['token']['access']
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
def test_list_admins(self):
"""测试管理员列表"""
url = '/api/admin/admins/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
def test_create_admin(self):
"""测试创建管理员"""
url = '/api/admin/admins/'
data = {
'username': 'newadmin',
'password': 'newpass123',
'name': '新管理员',
'role': 'operator'
}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['data']['username'], 'newadmin')
def test_get_admin_detail(self):
"""测试获取管理员详情"""
admin = AdminUser.objects.create_user(username='testadmin', password='test123')
url = f'/api/admin/admins/{admin.id}/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['data']['username'], 'testadmin')
def test_toggle_admin_status(self):
"""测试启用/禁用管理员"""
admin = AdminUser.objects.create_user(username='testadmin2', password='test123')
url = f'/api/admin/admins/{admin.id}/toggle-status/'
response = self.client.post(url, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
admin.refresh_from_db()
self.assertFalse(admin.is_active)
def test_reset_admin_password(self):
"""测试重置管理员密码"""
admin = AdminUser.objects.create_user(username='testadmin3', password='oldpass')
url = f'/api/admin/admins/{admin.id}/reset-password/'
data = {'new_password': 'resetpass123'}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
class AdminDeviceTypeTests(APITestCase):
"""管理端设备类型测试"""
def setUp(self):
self.admin = AdminUser.objects.create_user(
username='admin', password='admin123', role='super_admin'
)
login_url = '/api/admin/auth/login/'
response = self.client.post(
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
)
self.access_token = response.data['data']['token']['access']
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
def test_create_device_type_network(self):
"""测试创建联网设备类型"""
url = '/api/admin/device-types/'
data = {
'brand': 'AL',
'product_code': 'DZBJ-ON',
'name': '电子吧唧-联网版'
}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertTrue(response.data['data']['is_network_required'])
def test_create_device_type_offline(self):
"""测试创建非联网设备类型"""
url = '/api/admin/device-types/'
data = {
'brand': 'AL',
'product_code': 'DZBJ-OFF',
'name': '电子吧唧-离线版'
}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertFalse(response.data['data']['is_network_required'])
def test_list_device_types(self):
"""测试设备类型列表"""
DeviceType.objects.create(brand='AL', product_code='TEST-ON', name='测试设备')
url = '/api/admin/device-types/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
def test_get_device_type_detail(self):
"""测试设备类型详情"""
dt = DeviceType.objects.create(brand='AL', product_code='TEST-DT', name='测试类型')
url = f'/api/admin/device-types/{dt.id}/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['data']['name'], '测试类型')
def test_update_device_type(self):
"""测试更新设备类型"""
dt = DeviceType.objects.create(brand='AL', product_code='TEST-UP', name='原名称')
url = f'/api/admin/device-types/{dt.id}/'
data = {'name': '新名称'}
response = self.client.put(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['data']['name'], '新名称')
class AdminBatchTests(APITestCase):
"""管理端批次测试"""
def setUp(self):
self.admin = AdminUser.objects.create_user(
username='admin', password='admin123', role='super_admin'
)
login_url = '/api/admin/auth/login/'
response = self.client.post(
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
)
self.access_token = response.data['data']['token']['access']
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
self.device_type = DeviceType.objects.create(
brand='AL', product_code='DZBJ-ON', name='电子吧唧-联网版'
)
def test_create_batch_and_generate_sn(self):
"""测试创建批次并生成SN码"""
url = '/api/admin/device-batches/'
data = {
'device_type': self.device_type.id,
'batch_no': 'A01',
'production_date': '2026-01-28',
'quantity': 10,
'remark': '测试批次'
}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
self.assertEqual(response.data['data']['quantity'], 10)
self.assertIn('sn_range', response.data['data'])
# 验证SN码格式
sn_start = response.data['data']['sn_range']['start']
self.assertTrue(sn_start.startswith('AL-DZBJ-ON-'))
# 验证设备数量
device_count = Device.objects.filter(batch__batch_no='A01').count()
self.assertEqual(device_count, 10)
def test_list_batches(self):
"""测试批次列表"""
url = '/api/admin/device-batches/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['code'], 0)
def test_get_batch_detail(self):
"""测试批次详情"""
batch = DeviceBatch.objects.create(
device_type=self.device_type,
batch_no='B01',
production_date=date(2026, 1, 28),
quantity=5
)
url = f'/api/admin/device-batches/{batch.id}/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('statistics', response.data['data'])
def test_get_batch_devices(self):
"""测试批次设备列表"""
batch = DeviceBatch.objects.create(
device_type=self.device_type,
batch_no='C01',
production_date=date(2026, 1, 28),
quantity=3
)
Device.objects.create(sn='TEST-001', batch=batch, device_type=self.device_type)
Device.objects.create(sn='TEST-002', batch=batch, device_type=self.device_type)
url = f'/api/admin/device-batches/{batch.id}/devices/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['data']['items']), 2)
class AuthSeparationTests(APITestCase):
"""验证App和Admin认证分离"""
def setUp(self):
# 创建App用户
self.app_user = User.objects.create_user(phone='13800138099')
login_url = '/api/v1/auth/phone-login/'
response = self.client.post(login_url, {'phone': '13800138099'}, format='json')
self.app_token = response.data['data']['token']['access']
# 创建Admin用户
self.admin = AdminUser.objects.create_user(
username='admin', password='admin123', role='admin'
)
login_url = '/api/admin/auth/login/'
response = self.client.post(
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
)
self.admin_token = response.data['data']['token']['access']
def test_app_token_cannot_access_admin_api(self):
"""测试App Token无法访问管理端API"""
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.app_token}')
url = '/api/admin/device-types/'
response = self.client.get(url)
self.assertNotEqual(response.data.get('code', 0), 0)
def test_admin_token_cannot_access_app_api(self):
"""测试Admin Token无法访问App端API"""
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.admin_token}')
url = '/api/v1/users/me/'
response = self.client.get(url)
self.assertNotEqual(response.data.get('code', 0), 0)
def test_no_token_cannot_access_protected_api(self):
"""测试无Token无法访问受保护API"""
self.client.credentials()
url = '/api/v1/users/me/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
url = '/api/admin/device-types/'
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

0
utils/__init__.py Normal file
View File

91
utils/exceptions.py Normal file
View File

@ -0,0 +1,91 @@
"""
自定义异常处理
"""
from rest_framework.views import exception_handler
from rest_framework import status
from utils.response import APIResponse
class BusinessException(Exception):
"""业务异常"""
def __init__(self, code=1, message='业务错误', status_code=status.HTTP_400_BAD_REQUEST):
self.code = code
self.message = message
self.status_code = status_code
super().__init__(message)
class ErrorCode:
"""错误码定义"""
# 通用错误 1-99
SUCCESS = 0
UNKNOWN_ERROR = 1
PARAM_ERROR = 2
NOT_FOUND = 3
PERMISSION_DENIED = 4
# 用户模块 100-199
USER_NOT_FOUND = 100
USER_DISABLED = 101
PHONE_INVALID = 102
TOKEN_EXPIRED = 103
# 设备模块 200-299
DEVICE_NOT_FOUND = 200
DEVICE_ALREADY_BOUND = 201
DEVICE_MAC_EXISTS = 202
DEVICE_SN_INVALID = 203
# 智能体模块 300-399
SPIRIT_NOT_FOUND = 300
SPIRIT_LIMIT_EXCEEDED = 301
# 批次模块 400-499
BATCH_NOT_FOUND = 400
BATCH_SN_EXISTS = 401
DEVICE_TYPE_NOT_FOUND = 402
def custom_exception_handler(exc, context):
"""自定义异常处理器"""
# 处理业务异常
if isinstance(exc, BusinessException):
return APIResponse(
code=exc.code,
message=exc.message,
status_code=exc.status_code
)
# 调用DRF默认异常处理
response = exception_handler(exc, context)
if response is not None:
# 统一格式化DRF异常
error_message = ''
if isinstance(response.data, dict):
if 'detail' in response.data:
error_message = str(response.data['detail'])
else:
# 处理字段验证错误
errors = []
for field, messages in response.data.items():
if isinstance(messages, list):
errors.append(f"{field}: {', '.join(str(m) for m in messages)}")
else:
errors.append(f"{field}: {messages}")
error_message = '; '.join(errors)
else:
error_message = str(response.data)
return APIResponse(
code=ErrorCode.UNKNOWN_ERROR,
message=error_message,
status_code=response.status_code
)
# 处理未知异常
return APIResponse(
code=ErrorCode.UNKNOWN_ERROR,
message=str(exc),
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
)

92
utils/oss.py Normal file
View File

@ -0,0 +1,92 @@
"""
阿里云OSS工具类
"""
import os
import uuid
from datetime import datetime
from django.conf import settings
try:
import oss2
OSS_AVAILABLE = True
except ImportError:
OSS_AVAILABLE = False
class OSSClient:
"""阿里云OSS客户端"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
if self._initialized:
return
if not OSS_AVAILABLE:
self.bucket = None
self._initialized = True
return
oss_config = settings.ALIYUN_OSS
if not oss_config.get('ACCESS_KEY_ID'):
self.bucket = None
self._initialized = True
return
auth = oss2.Auth(
oss_config['ACCESS_KEY_ID'],
oss_config['ACCESS_KEY_SECRET']
)
self.bucket = oss2.Bucket(
auth,
oss_config['ENDPOINT'],
oss_config['BUCKET_NAME']
)
self.custom_domain = oss_config.get('CUSTOM_DOMAIN', '')
self._initialized = True
def upload_file(self, file_obj, folder='uploads'):
"""
上传文件
:param file_obj: 文件对象
:param folder: 存储文件夹
:return: 文件URL
"""
if not self.bucket:
raise Exception('OSS未配置')
# 生成唯一文件名
ext = os.path.splitext(file_obj.name)[1] if hasattr(file_obj, 'name') else ''
filename = f"{datetime.now().strftime('%Y%m%d')}/{uuid.uuid4().hex}{ext}"
key = f"{folder}/{filename}"
# 上传文件
self.bucket.put_object(key, file_obj)
# 返回URL
if self.custom_domain:
return f"https://{self.custom_domain}/{key}"
return f"https://{settings.ALIYUN_OSS['BUCKET_NAME']}.{settings.ALIYUN_OSS['ENDPOINT']}/{key}"
def delete_file(self, file_url):
"""
删除文件
:param file_url: 文件URL
"""
if not self.bucket:
return
# 从URL提取key
key = file_url.split('/')[-2] + '/' + file_url.split('/')[-1]
self.bucket.delete_object(key)
def get_oss_client():
"""获取OSS客户端单例"""
return OSSClient()

27
utils/permissions.py Normal file
View File

@ -0,0 +1,27 @@
"""
权限控制
"""
from rest_framework.permissions import BasePermission
class IsAdmin(BasePermission):
"""管理员权限"""
def has_permission(self, request, view):
return request.user and request.user.is_authenticated and request.user.is_staff
class IsOwner(BasePermission):
"""资源所有者权限"""
def has_object_permission(self, request, view, obj):
# 假设对象有user字段
if hasattr(obj, 'user'):
return obj.user == request.user
if hasattr(obj, 'owner'):
return obj.owner == request.user
return False
class AllowAny(BasePermission):
"""允许任何访问"""
def has_permission(self, request, view):
return True

52
utils/response.py Normal file
View File

@ -0,0 +1,52 @@
"""
统一响应格式工具类
"""
from rest_framework.response import Response
from rest_framework import status
class APIResponse(Response):
"""
统一API响应格式
{
"code": 0, # 0表示成功，非0表示错误码
"message": "success", # 响应消息
"data": {} # 响应数据
}
"""
def __init__(self, data=None, code=0, message='success',
status_code=status.HTTP_200_OK, **kwargs):
response_data = {
'code': code,
'message': message,
'data': data
}
super().__init__(data=response_data, status=status_code, **kwargs)
def success(data=None, message='success'):
"""成功响应"""
return APIResponse(data=data, code=0, message=message)
def error(code=1, message='error', status_code=status.HTTP_400_BAD_REQUEST, data=None):
"""错误响应"""
return APIResponse(data=data, code=code, message=message, status_code=status_code)
def paginated_response(paginator, serializer_class, queryset, request):
"""分页响应"""
page = paginator.paginate_queryset(queryset, request)
if page is not None:
serializer = serializer_class(page, many=True)
return APIResponse(
data={
'total': paginator.page.paginator.count,
'page': paginator.page.number,
'page_size': paginator.page_size,
'items': serializer.data
}
)
serializer = serializer_class(queryset, many=True)
return APIResponse(data={'items': serializer.data})