基于Django的豆瓣电影数据可视化系统.zip

# coding=UTF-8 import json import os.path from django.conf import settings import jieba from jieba.analyse import tfidf from rest_framework.generics import GenericAPIView from wordcloud import WordCloud import pandas as pd from sqlalchemy import create_engine from app.models import MovieDetail, MovieComment from user.models import UserToken, UserFavorites, UserRating from utils.auth import LoginAuth from utils.permission import MyPermission from rest_framework import serializers, mixins from rest_framework.viewsets import ModelViewSet from rest_framework.decorators import action, api_view # 路由 from rest_framework.response import Response # Json返回 from rest_framework.pagination import PageNumberPagination # 分页 from rest_framework.filters import SearchFilter # 模糊搜索(过滤) from django_pandas.io import read_frame # pandas.DataFrame from app.recommend_base_content import RecommendBaseContent # mysql驱动引擎 engine = create_engine('mysql+pymysql://root:123456@localhost:3306/douban_movie') sql1 = 'select * from user_favorites;' user_favor = pd.read_sql(sql1, engine) sql2 = 'select * from user_rating;' user_rating = pd.read_sql(sql2, engine) def getMovieTypeDict(df): """统计每种类型的电影个数""" typeList = [] # 存放电影种类的列表 for types in df['types'].values.tolist(): typeList += types.split(',') # 统计每一类电影的个数 type_dict = dict() for type in typeList: if type in type_dict: type_dict[type] += 1 else: type_dict[type] = 1 return type_dict def getMovieCountryDict(df): """统计每个制片国家的电影个数""" countryList = [] # 存放电影种类的列表 for countries in df['countries'].values.tolist(): countryList += countries.split(',') # 统计每一类电影的个数 country_dict = dict() for country in countryList: if country in country_dict: country_dict[country] += 1 else: country_dict[country] = 1 return country_dict def getMovieLangDict(df): """统计每种语言的电影个数""" langList = [] # 存放电影种类的列表 for lang in df['lang'].values.tolist(): langList += lang.split(',') # 统计每一类电影的个数 lang_dict = dict() for lang in langList: if lang in lang_dict: lang_dict[lang] += 1 else: lang_dict[lang] = 1 return lang_dict class MovieInfoSerializer(serializers.ModelSerializer): class Meta: model = MovieDetail fields = '__all__' class MovieInfoView(ModelViewSet): """ movie信息 """ queryset = MovieDetail.objects.all() serializer_class = MovieInfoSerializer authentication_classes = [LoginAuth] # 登录认证 # permission_classes = [MyPermission] # 权限 pagination_class = None # 配置自定义的分页器 filter_backends = [SearchFilter] search_fields = ['title'] # coerce_float=True: 尝试将值转换为非字符串，将非数字对象（如decimal.Decimal）转化为浮点类型 df = read_frame(qs=queryset, coerce_float=True) df['release_date'] = pd.to_datetime(df['release_date']) type_dict = getMovieTypeDict(df) country_dict = getMovieCountryDict(df) lang_dict = getMovieLangDict(df) # 推荐类 recommendContent = RecommendBaseContent(df.copy(), user_favor, user_rating) # 127.0.0.1:8000/api/movie/info/homeInfo/ ---->get请求 @action(methods=['GET'], detail=False) def homeInfo(self, request): columns = ['id', 'title', 'rate', 'release_date', 'comment_len', 'cover'] df = self.df.copy()[columns] # 《天空之城》 home_movie = df.loc[df['title'] == '天空之城'] home_cover = home_movie['cover'] # 评分最高电影 # 查找rate最高的那一行数据 highest_movie = df.loc[df['rate'].idxmax()] highest_movie = highest_movie.to_dict() # 最新电影 # 查找上映日期最新的那一行数据 latest_movie = df.loc[df['release_date'].idxmax()] latest_movie = latest_movie.to_dict() # 近期热门电影 new_date = df['release_date'].max() # 最新电影上映日期 old_date = new_date - pd.Timedelta(60, unit='d') # 最新日期往前60天 recent_df = df[df['release_date'].between(old_date, new_date)] # 取这区间的电影 hot_movie = recent_df.loc[recent_df['comment_len'].idxmax()] # 用评论数来定义热门电影 hot_movie = hot_movie.to_dict() return Response( {'msg': 'OK', 'home_cover': home_cover, 'highest': highest_movie, 'latest': latest_movie, 'hot': hot_movie}) # 127.0.0.1:8000/api/movie/info/getSearchTitles/ ---->get请求 @action(methods=['GET'], detail=False) def getSearchTitles(self, request): """根据输入的关键词搜索相关电影名称""" keyword = request.GET.get('keyword') if not keyword: return Response({'msg': '请输入关键字', 'res_code': 0, 'titles': []}) titles = self.df[self.df['title'].str.contains(keyword)]['title'] if titles.empty: return Response({'msg': '没有相关电影', 'res_code': 0, 'titles': []}) titles = titles.values.tolist() return Response({'msg': 'OK', 'res_code': 1, 'titles': titles}) # 127.0.0.1:8000/api/movie/info/allMovieTypes/ ---->get请求 @action(methods=['GET'], detail=False) def allMovieTypes(self, request): # 按每种类型的个数从多到少排序 type_dict = sorted(self.type_dict.items(), key=lambda x: x[1], reverse=True) # 取出全部类型名称 types = map(lambda x: x[0], type_dict) if types: return Response({'msg': 'OK', 'res_code': 1, 'types': types}) else: return Response({'msg': 'Fail', 'res_code': 0, 'types': []}) # 127.0.0.1:8000/api/movie/info/allMovieYears/ ---->get请求 @action(methods=['GET'], detail=False) def allMovieYears(self, request): years = self.df['year'].unique().tolist() years = sorted(years, reverse=True) if years: return Response({'msg': 'OK', 'res_code': 1, 'years': years}) else: return Response({'msg': '未查到数据', 'res_code': 0, 'years': []}) # 127.0.0.1:8000/api/movie/info/allMovieCountries/ ---->get请求 @action(methods=['GET'], detail=False) def allMovieCountries(self, request): """获取全部制片国家""" # 按每种制片国家的个数从多到少排序 country_dict = sorted(self.country_dict.items(), key=lambda x: x[1], reverse=True) # 取出全部制片国家名称 movie_countries = map(lambda x: x[0], country_dict) if movie_countries: return Response({'msg': 'OK', 'res_code': 1, 'countries': movie_countries}) else: return Response({'msg': '未查到数据', 'res_code': 0, 'countries': []}) # 127.0.0.1:8000/api/movie/info/allMovieLang/ ---->get请求 @action(methods=['GET'], detail=False) def allMovieLang(self, request): """获取全部语言种类""" # 按每种语言的个数从多到少排序 lang_dict = sorted(self.lang_dict.items(), key=lambda x: x[1], reverse=True) # 取出全部语言名称 movie_langs = map(lambda x: x[0], lang_dict) if movie_langs: return Response({'msg': 'OK', 'res_code': 1, 'langs': movie_langs}) else: return Response({'msg': '未查到数据', 'res_code': 0, 'langs': []}) # 127.0.0.1:8000/api/movie/info/recommendBaseContent/ ---->GET请求 @action(methods=['GET']