factory/apps/develop/views.py

import json
from rest_framework.views import APIView
from rest_framework.permissions import IsAdminUser
from rest_framework.response import Response
from rest_framework.serializers import Serializer
from rest_framework.decorators import action
from apps.develop.serializers import GenerateVoiceSerializer, SendSmsSerializer, TestTaskSerializer
from apps.develop.tasks import backup_database, backup_media, reload_web_git, reload_server_git, reload_server_only
from rest_framework.exceptions import APIException, ParseError
from apps.system.models import User
from apps.system.tasks import show
from apps.utils.sms import send_sms
from apps.utils.speech import generate_voice
from django.core.cache import cache
import requests

from apps.utils.viewsets import CustomGenericViewSet
from apps.utils.wx import wxClient
# Create your views here.


class ReloadServerGit(APIView):
    authentication_classes = []
    permission_classes = []

    def post(self, request):
        """
        拉取后端代码并重启服务

        拉取后端代码并重启服务
        """
        reload_server_git.delay()
        return Response()
        # if completed.returncode == 0:
        #     return Response()
        # else:
        #     from server.settings import myLogger
        #     myLogger.error(completed)
        #     raise ParseError(completed.stderr)


class ReloadClientGit(APIView):
    authentication_classes = []
    permission_classes = []

    def post(self, request):
        """
        拉取前端代码并打包

        拉取前端代码并打包
        """
        reload_web_git.delay()
        return Response()
        # completed = reload_web_git()
        # if completed.returncode == 0:
        #     return Response()
        # else:
        #     raise APIException(completed.stdout)


class ReloadServerOnly(APIView):
    permission_classes = [IsAdminUser]

    def post(self, request):
        """
        仅重启服务

        仅重启服务
        """
        completed = reload_server_only()
        if completed.returncode == 0:
            return Response()
        else:
            raise APIException(completed.stdout)


class BackupDatabase(APIView):
    permission_classes = [IsAdminUser]

    def post(self, request):
        """
        备份数据库到指定位置

        备份数据库到指定位置
        """
        completed = backup_database()
        if completed.returncode == 0:
            return Response()
        else:
            raise APIException(completed.stdout)


class BackupMedia(APIView):
    permission_classes = [IsAdminUser]

    def post(self, request):
        """
        备份资源到指定位置

        备份资源到指定位置
        """
        completed = backup_media()
        if completed.returncode == 0:
            return Response()
        else:
            raise APIException(completed.stdout)


class TestViewSet(CustomGenericViewSet):
    authentication_classes = ()
    permission_classes = ()

    @action(methods=['post'], detail=False, serializer_class=SendSmsSerializer)
    def send_sms(self, request, pk=None):
        """发送短信测试

        发送短信测试
        """
        serializer = SendSmsSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        vdata = serializer.validated_data
        res = send_sms(**vdata)
        return Response(res)

    @action(methods=['post'], detail=False, serializer_class=GenerateVoiceSerializer)
    def generate_voice(self, request, pk=None):
        """文字转语音测试

        文字转语音测试
        """
        serializer = GenerateVoiceSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        vdata = serializer.validated_data
        res = generate_voice(**vdata)
        return Response(res)

    @action(methods=['post'], detail=False, serializer_class=TestTaskSerializer)
    def task(self, request, pk=None):
        """任务派发测试

        任务派发测试
        """
        serializer = TestTaskSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        vdata = serializer.validated_data
        ret = {}
        task = show.delay(*vdata.get('args', []), **vdata.get('kwargs', {}))
        ret['task_id'] = task.task_id
        return Response(ret)

    @action(methods=['post'], detail=False, serializer_class=Serializer)
    def send_wx_msg(self, request, pk=None):
        """微信模板消息发送测试

        微信模板消息发送测试
        """
        data = {
            "touser": "ocAnm5vab6QGOnf2dqpuUZGJ0xyw",
            "template_id": "wczl-hB_bs8CpxwO8_KZuogSTPFXqhBxf8C5HjKl2M4",
            "url": "http://weixin.qq.com/download",
            #    "miniprogram":{
            #      "appid":"xiaochengxuappid12345",
            #      "pagepath":"index?foo=bar"
            #    },
            "client_msg_id": "MSG_000001",
            "data": {
                "first": {
                   "value": "恭喜你购买成功！",
                   "color": "#173177"
                },
            }
        }
        # url = 'https://api.weixin.qq.com/cgi-bin/message/template/send?access_token'+wxClient.token
        # print(url)
        # rt = requests.post(url=url, json=data).text
        res = wxClient.send_tem_msg(data=data)
        return Response(res)