dingxin_toolbox_drf/ai/utils/show_database.py

import pymysql
from chinese_calendar import is_holiday
from pymysql.cursors import DictCursor
from ai.models import *
from ai.models import TestCinema
from ai.utils.sql import *
from ai.utils.basic_func import *
import datetime
import pandas as pd
import json


class GetData:
    def __init__(self, cinema):
        self.conn = pymysql.connect(**json.loads(cinema.db_config))
        self.cur = self.conn.cursor(cursor=DictCursor)
        self.zz_code = cinema.zz_code

    def exit(self):
        try:
            self.cur.close()
            self.conn.close()
        except Exception as e:
            print(e)

    # 获取指定日期范围的排片数据
    def get_show_data(self, start_datetime, end_datetime):
        # print(self.cur.mogrify(GET_SHOW_DATA, (start_datetime, end_datetime)))
        self.cur.execute(GET_SHOW_DATA, (start_datetime, end_datetime))
        show_data = self.cur.fetchall()
        if len(show_data) == 0:
            return False
        show_data_mapping = {
            'hall_name': '影厅别名',
            'hall_id': '影厅id',
            'movie_name': '影片别名',
            'movie_id': '本地影片id',
            'language': '语言',
            'show_date': '放映日期',
            'start_time': '开始时间',
            'end_time': '结束时间',
            'length': '片长',
            'duration': '场间',
        }
        return self.format_to_csv(show_data, show_data_mapping)

    # 获取指定日期的可放映影片
    def get_available_movie(self, date):
        start_datetime = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=29, minutes=59,
                                                                                           seconds=59)
        end_datetime = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=6)
        self.cur.execute(GET_AVAILABLE_MOVIE, (start_datetime, end_datetime))
        available_movie = self.cur.fetchall()
        available_movie_list = [
            f"\t{m['cinema_movie_alias']} - 影片id：{m['cinema_movie_id']}；"
            for m in available_movie]
        return '\n'.join(available_movie_list)

    # 获取指定日期的可用于排片的影片
    def get_handle_movie(self, date):
        if get_runtime_val('real_target') == '0':
            movie_date_start = datetime.datetime.today().replace(hour=6, minute=0, second=0, microsecond=0)
            movie_date_end = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=29, minutes=59,
                                                                                               seconds=59)
            show_date_limit = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=6, minutes=0,
                                                                                                seconds=0)
            bl_bool, bl_str, bl_list = get_black_list(self.zz_code)
            if bl_bool:
                sql = GET_HANDLE_MOVIE_TEST + bl_str + ');'
                self.cur.execute(sql, (movie_date_start, movie_date_end, show_date_limit, *bl_list))
                print(self.cur.mogrify(sql, (movie_date_start, movie_date_end, show_date_limit, *bl_list)))
            else:
                sql = GET_HANDLE_MOVIE_TEST + bl_str + ');'
                self.cur.execute(sql, (movie_date_start, movie_date_end, show_date_limit))
                print(self.cur.mogrify(sql, (movie_date_start, movie_date_end, show_date_limit)))
        else:
            start_datetime = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=29, minutes=59,
                                                                                               seconds=59)
            end_datetime = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=6)

            bl_bool, bl_str, bl_list = get_black_list(self.zz_code)
            if bl_bool:
                sql = GET_HANDLE_MOVIE + bl_str + ');'
                self.cur.execute(sql, (start_datetime, end_datetime, *bl_list))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime, *bl_list)))
            else:
                sql = GET_HANDLE_MOVIE + bl_str + ');'
                self.cur.execute(sql, (start_datetime, end_datetime))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime)))
        available_movie = self.cur.fetchall()
        available_movie_list = [
            f"\t{m['cinema_movie_alias']} - 影片id：{m['cinema_movie_id']}，影片时长：{m['cinema_movie_time']}，语言：{m['language']}，制式：{m['media_type']}， 最早排片开始时间：{m['cinema_movie_start_datetime']}， 最晚排片截止时间：{m['cinema_movie_end_datetime']}；"
            for m in available_movie]
        return '\n'.join(available_movie_list)

    # 获取新上映的影片
    def get_new_movie(self, date):
        start_datetime = datetime.datetime.strptime(date, '%Y-%m-%d')
        end_datetime = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=29, minutes=59,
                                                                                         seconds=59)
        self.cur.execute(GET_NEW_MOVIE, (start_datetime, end_datetime))
        new = self.cur.fetchall()
        new_list = [f"{m['cinema_movie_alias']}" for m in new]
        return new_list

    # 为获取影片热度功能提供影院数据
    def get_movie_info(self, date, show_type='all'):
        start_datetime = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=29, minutes=59,
                                                                                           seconds=59)
        end_datetime = datetime.datetime.strptime(date, '%Y-%m-%d') + datetime.timedelta(hours=6)
        bl_bool, bl_str, bl_list = get_black_list(self.zz_code)
        if bl_bool:
            if show_type == "new":
                sql = GET_NEW_MOVIE_INFO + bl_str + GET_MOVIE_INFO_END
                self.cur.execute(sql, (start_datetime, end_datetime, *bl_list))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime, *bl_list)))
            elif show_type == "old":
                sql = GET_OLD_MOVIE_INFO + bl_str + GET_MOVIE_INFO_END
                self.cur.execute(sql, (end_datetime, start_datetime, *bl_list))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime, *bl_list)))
            else:
                sql = GET_MOVIE_INFO + bl_str + GET_MOVIE_INFO_END
                self.cur.execute(sql, (start_datetime, end_datetime, *bl_list))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime, *bl_list)))

        else:
            if show_type == "new":
                sql = GET_NEW_MOVIE_INFO + bl_str + GET_MOVIE_INFO_END
                self.cur.execute(sql, (start_datetime, end_datetime))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime)))
            elif show_type == "old":
                sql = GET_OLD_MOVIE_INFO + bl_str + GET_MOVIE_INFO_END
                self.cur.execute(sql, (end_datetime, start_datetime))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime)))
            else:
                sql = GET_MOVIE_INFO + bl_str + GET_MOVIE_INFO
                self.cur.execute(sql, (start_datetime, end_datetime))
                print(self.cur.mogrify(sql, (start_datetime, end_datetime)))
        return self.cur.fetchall()

    # 按照场次获取销售数据
    def get_sell_data(self, start_datetime, end_datetime):
        # print(self.cur.mogrify(GET_SELL_DATA, (start_datetime, end_datetime)))
        self.cur.execute(GET_SELL_DATA, (start_datetime, end_datetime, start_datetime, end_datetime))
        sell_data = self.cur.fetchall()
        sell_data_mapping = {
            'movie_name': '影片别名',
            'movie_id': '本地影片id',
            'show_date': '放映日期',
            'start_time': '开始时间',
            'hall_name': '影厅别名',
            'hall_id': '影厅id',
            'total_count': '场次观影人数',
            'total_income': '场次收入',
            'seat_num': '影厅座位数',
            'average_price': '平均价',
            'rate': '上座率（单位%）',
        }
        return self.format_to_csv(sell_data, sell_data_mapping)

    # 按照影片获取销售数据
    def get_sell_data_by_movie(self, start_datetime, end_datetime):
        # 获取过去7天的数据
        new_start_datetime = datetime.datetime.strptime(start_datetime, '%Y-%m-%d %H:%M:%S') + datetime.timedelta(
            days=-7)
        start_datetime = datetime.datetime.strftime(new_start_datetime, '%Y-%m-%d %H:%M:%S')
        self.cur.execute(GET_SELL_DATA_BY_MOVIE, (start_datetime, end_datetime, start_datetime, end_datetime))
        sell_data = self.cur.fetchall()
        print('get_sell_data_by_movie', start_datetime, end_datetime, sell_data)
        sell_data_mapping = {
            'movie_alias_name': '影片别名',
            'movie_id': '本地影片id',
            'total_count': '观影总人数',
            'average_show_count': '场均人次',
            'average_seat_rate': '平均上座率（单位%）',
            'total_income': '票房总收入',
            'average_ticket_price': '平均票价',
            'average_show_ticket_price': '场均收入',
            'show_count': '总场次数',
            'show_rate': '排片占比（单位%）',
        }
        return self.format_to_csv(sell_data, sell_data_mapping)

    # 按工作日和节假日计算销售数据
    def get_sell_data_by_day_type(self, start_datetime, end_datetime):
        # 获取过去7天的数据
        print('get_sell_data_by_day_type_start_datetime', start_datetime)
        date_list = [datetime.datetime.strptime(start_datetime, '%Y-%m-%d %H:%M:%S') + datetime.timedelta(days=n) for n
                     in range(-7, 0, 1)]
        print('get_sell_data_by_day_type', date_list)
        sell_data_dict = {
            'holiday': [],
            'workday': [],
        }
        for date in date_list:
            start = datetime.datetime.strftime(date, '%Y-%m-%d %H:%M:%S')
            end = datetime.datetime.strftime(date + datetime.timedelta(days=1), '%Y-%m-%d %H:%M:%S')
            self.cur.execute(GET_SELL_DATA_SELF_CALCULATE, (start, end))
            sell_data = self.cur.fetchall()
            print('get_sell_data_by_movie', start, end, sell_data)
            if is_holiday(date):
                sell_data_dict['holiday'].append(sell_data)
            else:
                sell_data_dict['workday'].append(sell_data)

        movie_sell_data = dict()
        for key, val in sell_data_dict.items():
            print(key, val)
            for sell in val:
                print(sell)
                for m in sell:
                    print(m)
                    if m['movie_id'] not in movie_sell_data.keys():
                        movie_sell_data[m['movie_id']] = {
                            'movie_alias_name': m['movie_alias_name'],
                            'movie_id': m['movie_id'],
                            'viewer': {
                                'holiday': 0,
                                'workday': 0,
                            },
                            'income': {
                                'holiday': 0.0,
                                'workday': 0.0,
                            },
                            'show': {
                                'holiday': 0,
                                'workday': 0,
                            }
                        }
                    movie_sell_data[m['movie_id']]['viewer'][key] += int(m['total_viewers'])
                    movie_sell_data[m['movie_id']]['income'][key] += float(m['total_income'])
                    movie_sell_data[m['movie_id']]['show'][key] += int(m['total_show'])
        print('get_sell_data_by_day_type', movie_sell_data)
        movie_sell_data_list = []
        for m in movie_sell_data.values():
            movie_sell_data_list.append({
                'movie_alias_name': m['movie_alias_name'],
                'movie_id': m['movie_id'],
                'average_viewer_workday': round(
                    float(0 if m['show']['workday'] == 0 else m['viewer']['workday'] / m['show']['workday']), 2),
                'average_ticket_price_workday': round(
                    float(0 if m['viewer']['workday'] == 0 else m['income']['workday'] / m['viewer']['workday']), 2),
                'average_income_workday': round(
                    float(0 if m['show']['workday'] == 0 else m['income']['workday'] / m['show']['workday']), 2),
                'average_viewer_holiday': round(
                    float(0 if m['show']['holiday'] == 0 else m['viewer']['holiday'] / m['show']['holiday']), 2),
                'average_ticket_price_holiday': round(
                    float(0 if m['viewer']['holiday'] == 0 else m['income']['holiday'] / m['viewer']['holiday']), 2),
                'average_income_holiday': round(
                    float(0 if m['show']['holiday'] == 0 else m['income']['holiday'] / m['show']['holiday']), 2),
            })
        sell_data_mapping = {
            'movie_alias_name': '影片别名',
            'movie_id': '本地影片id',
            'average_viewer_workday': '场均人次（工作日）',
            'average_ticket_price_workday': '平均票价（工作日）',
            'average_income_workday': '场均收入（工作日）',
            'average_viewer_holiday': '场均人次（节假日）',
            'average_ticket_price_holiday': '平均票价（节假日）',
            'average_income_holiday': '场均收入（节假日）',
        }
        return self.format_to_csv(movie_sell_data_list, sell_data_mapping)

    # 获取指定范围的影片销售总额
    def get_total_income(self, start_datetime, end_datetime):
        print(self.cur.mogrify(GET_TOTAL_INCOME, (start_datetime, end_datetime, start_datetime, end_datetime)))
        self.cur.execute(GET_TOTAL_INCOME, (start_datetime, end_datetime, start_datetime, end_datetime))
        return self.cur.fetchone()['total_income']

    # 获取影厅支持的制式
    def get_hall_media_type(self):
        self.cur.execute(GET_HALL_MEDIA_TYPE)
        hall_data = self.cur.fetchall()
        # print(hall_data)
        hall_dict = dict()

        # 处理成对应的字典格式 {"影厅别名": {"hall_id"：影厅id，"media_type": 影片制式列表}}
        for hall in hall_data:
            if hall['cinema_hall_name'] not in hall_dict.keys():
                hall_dict[hall['cinema_hall_name']] = {
                    'hall_id': hall['cinema_hall_id'],
                    'media_type': [hall['media_type'], ]
                }
            else:
                hall_dict[hall['cinema_hall_name']]['media_type'].append(hall['media_type'])
        # 拼装字符串
        hall_str_list = []
        for hall, val in hall_dict.items():
            hall_str_list.append(f"\t{hall} - 影厅id：{val['hall_id']}，支持制式：{'，'.join(val['media_type'])}；")
        return '\n'.join(hall_str_list)

    # 获取模板日期
    def get_template_date(self, start_datetime, end_datetime):
        basic_val = 4
        self.cur.execute(GET_TEMPLATE_DATE, (start_datetime, end_datetime, basic_val))
        template_date = self.cur.fetchall()[0]['template_date']
        return template_date

    # 数据转csv格式字符串
    @staticmethod
    def format_to_csv(dict_data, mapping):
        # print(dict_data)
        new_dict = [{mapping[k]: v for k, v in d.items()} for d in dict_data]
        return pd.DataFrame(new_dict).to_csv(index=False)

    def get_media_type(self):
        self.cur.execute(GET_HALL_MEDIA_TYPE)