dingxin_toolbox_drf/mock/views.py

import requests
from django.shortcuts import render
from django.http.response import JsonResponse
from mock.utils import mock_service
from mock.models import ZZMockModel
from update.models import Cinema
from django.views.decorators.csrf import csrf_exempt
from pymysql.cursors import DictCursor
import pymysql
from mock.mock_templates import *

ZZ_URL = 'https://zzcs.yinghezhong.com'


# 数据上报
# 1、票房数据上报接口 POST /report/reportTicket
@csrf_exempt
def report_ticket(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.reportTicket is True:
        mock_service.bypass(request)
        return mock_service.mock(ip, 'reportTicket')
    else:
        return mock_service.bypass(request)


# 2、排片数据上报接口 POST /report/reportFilmSchedule
@csrf_exempt
def report_film_schedule(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.reportFilmSchedule is True:
        mock_service.bypass(request)
        return mock_service.mock(ip, 'reportFilmSchedule')
    else:
        return mock_service.bypass(request)


# 3、座位信息上报接口 POST /report/uploadScreenSeatInfo
@csrf_exempt
def upload_screen_seat_info(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.uploadScreenSeatInfo is True:
        mock_service.bypass(request)
        return mock_service.mock(ip, 'uploadScreenSeatInfo')
    else:
        return mock_service.bypass(request)


# 数据下载
# 1、影片信息下载接口 GET /data/downloadFilmInfo
def download_film_info(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.downloadFilmInfo is True:
        return mock_service.mock(ip, 'downloadFilmInfo')
    else:
        return mock_service.bypass(request)


# 2、影院信息下载接口 GET /data/getCinemaInfo
def get_cinema_info(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.getCinemaInfo is True:
        return mock_service.mock(ip, 'getCinemaInfo')
    else:
        return mock_service.bypass(request)


# 3、影厅信息下载接口 GET /data/getScreenInfo
def get_screen_info(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.getScreenInfo is True:
        return mock_service.mock(ip, 'getScreenInfo')
    else:
        return mock_service.bypass(request)


# 超时票务审批
# 1、超时票务受理情况查询接口 GET /data/getOvertimeTicketStatus
# 参数 场次id 审核状态 0不通过 1通过 2待审核 99待提交
def get_overtime_ticket_status(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.getOvertimeTicketStatus is True:
        return mock_service.mock(ip, 'getOvertimeTicketStatus', request=request)
    else:
        return mock_service.bypass(request)


# 数据查询
# 2、数据清洗错误查询接口 POST /query/validError
def valid_error(request):
    ip = request.META.get('REMOTE_ADDR')
    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
    if mock_service_switch.validError is True:
        return mock_service.mock(ip, 'validError')
    else:
        return mock_service.bypass(request)


# 获取过场信息
def get_overtime_show(request):
    query_params = request.query_params.dict()
    ip = query_params.get('ip')
    cinema_data = Cinema.objects.filter(ip=ip).first()
    db_config = {
        'host': cinema_data.ip,
        'user': cinema_data.db_user,
        'password': cinema_data.db_pwd,
        'database': 'cine',
        'connect_timeout': 5,
    }
    # 获取过场售票
    sql_str = '''
        SELECT csa.cinema_sell_add_showid       AS show_id,
               cms.cinema_movie_name            AS name,
               cms.cinema_movie_show_start_time AS start_time,
               chi.cinema_hall_name             AS hall_name
        FROM cinema_sell_add csa
                 LEFT JOIN cinema_hall_info chi ON csa.cinema_sell_add_hall = chi.cinema_hall_id
                 LEFT JOIN cinema_movie_show cms ON csa.cinema_sell_add_showid = cms.cinema_movie_show_id
        GROUP BY csa.cinema_sell_add_showid ORDER BY cms.cinema_movie_show_start_time  DESC;
        '''
    db_conn = pymysql.Connect(**db_config)
    db_cursor = db_conn.cursor(cursor=DictCursor)
    db_cursor.execute(sql_str)
    show_data = db_cursor.fetchall()
    sell_show_list = []
    for show in show_data:
        sell_show_list.append({'id': show['show_id'],
                               'show': show['name'] + ' ' + show['start_time'] + ' ' + show[
                                   'hall_name'] if show['hall_name'] is not None else ''
                               })
    # 获取过场退票场次
    sql_str = '''
        SELECT csl.cinema_movie_show_id         AS show_id,
               csl.cinema_movie_name            AS name,
               csl.cinema_movie_show_start_time AS start_time,
               chi.cinema_hall_name             AS hall_name
        FROM cinema_sell_log csl
                 LEFT JOIN cinema_hall_info chi ON csl.cinema_hall_id = chi.cinema_hall_id
        WHERE csl.cinema_sell_id IN (SELECT cinema_sell_id FROM cinema_refund_order_detail)
        GROUP BY cinema_movie_id ORDER BY csl.cinema_movie_show_start_time DESC;
    '''
    db_conn = pymysql.Connect(**db_config)
    db_cursor = db_conn.cursor(cursor=DictCursor)
    db_cursor.execute(sql_str)
    show_data = db_cursor.fetchall()
    refund_show_list = []
    for show in show_data:
        refund_show_list.append({'id': show['show_id'],
                                 'show': show['name'] + ' ' + show['start_time'] + ' ' + show[
                                     'hall_name'] if show['hall_name'] is not None else ''
                                 })
    return JsonResponse({'sell': sell_show_list, 'refund': refund_show_list}, json_dumps_params={'ensure_ascii': False})


# 获取mock数据的json
def get_response(request, _api):
    _ip = request.query_params.dict().get('ip')
    return mock_service.mock(_ip, _api)