fakeNewIdentification/text_analysis/bak/views_20240807.py

#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,tool
from text_analysis.tools import pred
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
from text_analysis.cusException import userFile_Exception,chainFile_Exception
import requests
import time
from datetime import datetime
import os
import joblib
#任务队列
global task_queue
task_queue = queue.Queue()


@csrf_exempt
def fakeNewIdentification(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 predict_news(dbConfig):
    while True:
        if task_queue.qsize() > 0:
            try:
                logging.info("取任务队列长度{}".format(task_queue.qsize()))
                raw_data = task_queue.get()
                logging.info("原始数据-{}".format(raw_data))
                res = {"successCode": "1", "errorLog": "", "results": {},"status":1,"message":"成功"}
                # 账号数据
                userData = tool.mysqlData(raw_data, logging,"1",dbConfig)
                # if not userData:
                #     raise userFile_Exception
                logging.info("账号数据获取完毕!-长度{}".format(len(userData)))
                # 传播链数据
                postChain=tool.mysqlData(raw_data, logging,"0",dbConfig)
                if not postChain:
                    raise chainFile_Exception
                logging.info("传播链数据获取完毕!-长度{}".format(len(postChain)))
                news=pred.predict_news(userData,postChain,logging)
                # 结束标识
                res['isLast'] = True
                for i in range(len(news)):
                    row_dict = news.iloc[i].to_dict()
                    row_dict['pageType'] = 'fakeNewsPage'
                    # postId
                    row_dict['postId'] = userData[0]['postId']
                    if i == len(news) - 1:
                        row_dict["isLast"]=1
                    res["results"] = json.dumps(row_dict,ensure_ascii=False)
                    res["status"] = 1
                    res["message"] = "成功"
                    raw_data["result"] = res
                    logging.info("共{}条数据，第{}条数据输出-{}".format(len(news),i+1,raw_data))
                    to_kafka.send_kafka(raw_data, logging)
            except userFile_Exception:
                res = {"successCode": "0", "errorLog": "用户数据为空！", "results": {}, "status": 2,"message": "异常"}
                results={}
                results['pageType'] = 'fakeNewsPage'
                results['recognitionResult'] = '用户数据为空'
                res['results'] = json.dumps(results)
                res["status"] = 2
                res["message"] = "用户数据为空"
                raw_data["result"] = res
                logging.info("该条请求用户数据为空-{}".format(raw_data))
                to_kafka.send_kafka(raw_data, logging)
            except chainFile_Exception:
                res = {"successCode": "0", "errorLog": "关系链数据为空！", "results": {}, "status": 2,"message": "异常"}
                results={}
                results['pageType'] = 'fakeNewsPage'
                results['recognitionResult'] = '关系链数据为空'
                res['results'] = json.dumps(results)
                res["status"] = 2
                res["message"] = "关系链数据为空"
                raw_data["result"] = res
                logging.info("该条请求关系链数据为空-{}".format(raw_data))
                to_kafka.send_kafka(raw_data, logging)
            except:
                res = {"successCode": "0", "errorLog": "", "results": {}, "status": 2,"message": "异常"}
                results={}
                results['pageType'] = 'fakeNewsPage'
                results['recognitionResult'] = ""
                res['results'] = json.dumps(results)
                res["status"] = 2
                res["message"] = "异常"
                raw_data["result"] = res
                raw_data["result"]["errorLog"] = traceback.format_exc()
                logging.info(traceback.format_exc())
                to_kafka.send_kafka(raw_data, logging)
        else:
            # 暂无任务，进入休眠
            time.sleep(10)