实训项目-B站热门视频TOP20数据分析可视化系统源码(含实训报告+项目说明).zip

import json from typing import Iterator import uuid from flask import Flask from time import sleep from concurrent.futures import ThreadPoolExecutor import DataBase import re import string from flask import Flask, render_template from flask import * import json import requests from dm_pb2 import DmSegMobileReply from google.protobuf.json_format import MessageToJson, Parse import jieba from pyecharts.charts import Geo,Bar,Pie,WordCloud,Boxplot,Line,Map from pyecharts import options from pyecharts.globals import SymbolType class Data: DataBase=None data={}#弹幕与评论数据,请求ID为键，数据为值 # 创建线程池执行器 executor = ThreadPoolExecutor(10) app = Flask(__name__) ########################################################################## ###################### Flask接口 ########################### ########################################################################## #POST接口，提交任务——获取弹幕数据 @app.route('/submit_task_get_danmaku_data/<BVID>/',methods=["POST"]) def submit_task_get_danmaku_data(BVID): requestID=str(uuid.uuid4()) executor.submit(long_task_of_get_danmaku_data,requestID,BVID)#创建异步获取弹幕长任务 submit_task_record_to_SQLite('get_danmaku_data',requestID)#提交任务记录到数据库 return json.dumps([{ 'requestID':requestID , 'status': 'create_danmaku_task_success' }]) #GET接口，获取弹幕数据 @app.route('/get_danmaku_data/<requestID>/') def get_danmaku_data(requestID): if str(requestID) in Data.data: return Data.data[str(requestID)] else: return json.dumps([{ 'requestID':requestID , 'status':"danmaku_data_don't_exist" }]) ############################################################## #POST接口，提交任务获取TOP数据 @app.route('/submit_task_get_TOP20_data', methods=["POST"]) def submit_task_get_TOP20_data(): requestID=str(uuid.uuid4()) executor.submit(long_task_of_get_TOP20_data,requestID)#创建异步获取弹幕长任务 submit_task_record_to_SQLite('get_TOP20_data',requestID)#提交任务记录到数据库 return json.dumps([{ 'requestID':requestID , 'status': 'create_TOP20_task_success' }]) #GET接口，获取TOP数据 @app.route('/get_TOP20_data/<requestID>/') def get_TOP20_data(requestID): if str(requestID) in Data.data: return Data.data[requestID] else: return json.dumps([{ 'requestID':requestID , 'status':"TOP20_data_don't_exist" }]) ############################################################## #POST接口，获取评论数据 @app.route('/submit_task_get_reply_data/<aid>/', methods=["POST"]) def submit_task_get_reply_data(aid): requestID=str(uuid.uuid4()) executor.submit(long_task_of_get_reply_data,requestID,aid)#创建异步获取弹幕长任务 submit_task_record_to_SQLite('get_reply_data',requestID)#提交任务记录到数据库 return json.dumps([{ 'requestID':requestID , 'status': 'create_reply_task_success' }]) #GET接口，获取评论数据 @app.route('/get_reply_data/<requestID>/') def get_reply_data(requestID): if str(requestID) in Data.data: return Data.data[requestID] else: return json.dumps([{ 'requestID':requestID , 'status':"reply_data_don't_exist" }]) ############################################################## #POST接口，提交任务保存所有数据 @app.route('/submit_task_save_all_data/<data>/', methods=["POST"]) def submit_task_save_all_data(data): data=str(data).split("@"); bvid=data[0]; requestID_danmaku=data[1]; requestID_reply=data[2]; requestID=str(uuid.uuid4()); if str(requestID_danmaku) not in Data.data or str(requestID_reply) not in Data.data: return json.dumps([{ 'requestID':requestID, 'status': 'create_save_all_data_task_fail' }]) else: executor.submit(long_task_of_submit_task_save_all_data,requestID,bvid,requestID_danmaku,requestID_reply)#创建异步获取弹幕长任务 submit_task_record_to_SQLite('save_all_data',requestID)#提交任务记录到数据库 return json.dumps([{ 'requestID':requestID , 'status': 'create_save_all_data_task_success' }]) ############################################################## #GET接口，获取长任务执行状态 @app.route('/get_task_status/<requestID>/') def get_long_task_status(requestID): return json.dumps([{ 'requestID':requestID , 'status':query_task_record_to_SQLite(requestID) }]) ###################### 网页模板GET接口 ########################### ##################################################################### @app.route('/', methods=["GET"]) def home(): return render_template('index.html') @app.route('/TOP20', methods=["GET"]) def TOP20(): return render_template('TOP20.html') ############################################################## #GET接口，获取弹幕词云数据 @app.route('/get_danmaku_wordCloud/<requestID>/', methods=["GET"])#得到词云 def get_danmaku_wordCloud(requestID): if str(requestID) not in Data.data: return "<h1>警告！弹幕任务未完成，或数据已丢失！</h1>" ####开始处理#### dist="" for i in Data.data[requestID]: dist=dist+i["content"]+" " pattern = re.compile(u"([\u4e00-\u9fff]+)")#正则表达式去除非中文字符 dist= pattern.findall(dist) dist="".join(dist) # print(dist) seg_list =jieba.lcut(dist) # print(seg_list) wordcount = {} for word in seg_list: if len(word)>1: wordcount[word] = wordcount.get(word, 0)+1 sorted_words=sorted(wordcount.items(), key=lambda x: x[1], reverse=True)[:100]#排序后词语 #######生成词云图########## wordCloud=(WordCloud() .add("",sorted_words,word_size_range=[10,100],shape=SymbolType.DIAMOND) .set_global_opts(title_opts=options.TitleOpts(title="弹幕关键词TOP100"))) wordCloud.render(path="danmaku_wordCloud.html") #######生成词云图########## return open("danmaku_wordCloud.html", "r").read() ####结束处理#### ############################################################## #GET接口，获取评论词云数据 @app.route('/get_reply_wordCloud/<requestID>/', methods=["GET"])#得到词云 def get_reply_wordCloud(requestID): if str(requestID) not in Data.data: return "<h1>警告！评论任务未完成，或数据已丢失！</h1>" ####开始处理#### dist="" for i in Data.data[requestID]: dist=dist+i["评论"]+" " pattern = re.compile(u"([\u4e00-\u9fff]+)")#正则表达式去除非中文字符 dist= pattern.findall(dist) dist="".join(dist) # print(dist) seg_list =jieba.lcut(dist) # print(seg_list) wordcount = {} for word in seg_list: if len(word)>1: wordcount[word] = wordcount.get(word, 0)+1 sorted_words=sorted(wordcount.items(), key=lambda x: x[1], reverse=True)[:100]#排序后词语 #######生成词云图########## wordCloud=(WordCloud() .add("",sorted_words,word_size_range=[10,100],shape=SymbolType.DIAMOND) .set_global_opts(title_opts=options.TitleOpts(title="评论关键词TOP100"))) wordCloud.render(path="reply_wordCloud.html") #######生成词云图########## return open("reply_wordCloud.html", "r").read() ####结束处理#### ########################################################################## ###################### 数据库操作 ########################### ########################################################################## #提交记录写入到数据库 def submit_task_record_to_SQLite(task_name,requestID): SQL="insert into main.task_record(id,task_name,requestID,status) values('%s','%s','%s','%s')"%(str(uuid.uuid4()),task_name,requestID,'waiting') Data.DataB