asrNew/text_analysis/bak/views.py0831

#coding:utf8
import os, sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf8')
cur_dir = os.path.dirname(os.path.abspath(__file__)) or os.getcwd()
par_dir = os.path.abspath(os.path.join(cur_dir, os.path.pardir))
sys.path.append(cur_dir)
sys.path.append(par_dir)
import json
from django.http import HttpResponse
from text_analysis.tools import to_kafka
from django.views.decorators.csrf import csrf_exempt
from log_util.set_logger import set_logger
logging=set_logger('logs/results.log')
import traceback
import queue
import requests
from text_analysis.tools.tool import get_data
import time
from datetime import datetime
import os
#任务队列
global task_queue
task_queue = queue.Queue()
#数据队列
global data_queue
data_queue = queue.Queue()

@csrf_exempt
def ASR(request):
    if request.method == 'POST':
        try:
            raw_data = json.loads(request.body)
            task_queue.put(raw_data)
            return HttpResponse(json.dumps({"code": 1, "msg": "请求正常！"}, ensure_ascii=False))
        except:
            logging.error(traceback.format_exc())
            return HttpResponse(json.dumps({"code": 0, "msg": "请求json格式不正确！"}, ensure_ascii=False))
    else:
        return HttpResponse(json.dumps({"code": 0, "msg": "请求方式错误，改为post请求"}, ensure_ascii=False))
    
def upload():
    while True:
        try:
            if task_queue.qsize() >0:
                logging.info("取任务队列长度{}".format(task_queue.qsize()))
                raw_data = task_queue.get()
                index=raw_data["metadata"]["index"]
                datasource=raw_data["metadata"]["admin"]["datasource"]
                if datasource not in raw_data["data"].keys():
                    logging.info("找不到相关数据源！—{}".format(raw_data))
                    continue
                allFile=raw_data["data"][datasource]
                currentFile=eval(allFile)[index]
                file=currentFile["fileUrl"]
                fileName=currentFile["fileName"]
                #从gofast获取视频
                myfile = requests.get(file)
                starttime = datetime.now().strftime('%Y-%m-%d')
                path='inputdata/'+starttime
                if not os.path.exists(path):
                    os.makedirs(path)
                with open(path+'/'+fileName, 'wb') as f:
                    f.write(myfile.content)
                logging.info("视频从gofast下载完毕，开始上传{}".format(fileName))
                #访问视频上传接口
                url = "https://realtime.pdeepmatrix.com/apis/media/analysis/upload"
                data = {'fromLanguage': 'zh'}
                files = {'file': open(path+'/'+fileName, 'rb')}
                response = requests.post(url, data=data, files=files)
                d = json.loads(response.text)
                if "code" in d.keys() and d["code"]==200:
                    #接口返回值data中存放视频获取结果的key
                    result = d["data"]
                    raw_data["result"] = {"successCode": "1", "errorLog": "", "results": "","dataKey":result}
                    data_queue.put(raw_data)
                    logging.info("视频上传成功{}".format(raw_data))
                    # to_kafka.send_kafka(raw_data,logging)
                else:
                    logging.info("视频上传失败,接口返回值{}".format(d))
            else:
                #暂无任务，进入休眠
                time.sleep(10)
        except:
            logging.error(traceback.format_exc())


def getResult():
    while True:
        #3秒钟结果获取一次
        time.sleep(3)
        try:
            if data_queue.qsize() >0:
                logging.info("取数据队列长度{}".format(data_queue.qsize()))
                raw_data = data_queue.get()
                print(raw_data)
                #根据视频key访问获取结果接口
                dataKey=raw_data["result"]["dataKey"]
                url = "https://realtime.pdeepmatrix.com/apis/media/analysis/getResult"
                params = {'taskId': dataKey}
                response = requests.get(url, params=params)
                # print(response.text)
                d = json.loads(response.text)
                if "code" in d.keys() and d["code"]==200:
                    results=""
                    if d["data"]["code"]=="1":
                        for sentence in d["data"]["sentences"]:
                            results+=sentence["text"]
                        raw_data["result"]["results"] =results
                        logging.info("视频解析获取结果成功{}".format(raw_data))
                        to_kafka.send_kafka(raw_data, logging)
                    elif d["data"]["code"]=="0":
                        #正在解析中，将任务再次放回数据队列
                        data_queue.put(raw_data)
                        logging.info("视频未解析完毕，放回队列等待{}-{}".format(raw_data,d))
                    else:
                        #解析失败
                        raw_data["result"]["successCode"] = "0"
                        raw_data["result"]["errorLog"] = response.text
                        logging.info("视频解析获取结果失败,数据{},接口返回值{}".format(raw_data, d))
                        to_kafka.send_kafka(raw_data, logging)
                else:
                    raw_data["result"]["successCode"] = "0"
                    raw_data["result"]["errorLog"]=response.text
                    logging.info("视频解析获取结果失败,数据{},接口返回值{}".format(raw_data,d))
                    to_kafka.send_kafka(raw_data, logging)

            else:
                #暂无任务，进入休眠
                time.sleep(10)
        except:
            raw_data["result"]["successCode"]="0"
            raw_data["result"]["errorLog"]=traceback.format_exc()
            logging.error(traceback.format_exc())
            to_kafka.send_kafka(raw_data, logging)