dingxin_toolbox_drf/ai/views.py

from django.http import JsonResponse
from django.db.models import Q
from django.views.decorators.csrf import csrf_exempt
import json
from ai.models import *
from ai.utils.show_process import show_main_process, show_manual_process
from ai.utils.show_func import *
from ai.utils.basic_func import *


# 手动触发ai排片
@csrf_exempt
def manual_general_show(request):
    zz_code = request.GET.dict().get('cinema_code')
    delta = request.GET.dict().get('delta')
    result = show_main_process(zz_code, delta)
    result_dict = {
        'status': 'success' if result else 'fail',
        'message': '生成成功' if result else '生成失败',
    }
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})


# 更新指定日期的排片
@csrf_exempt
def get_cinema_show_result(request):
    zz_code = request.GET.dict().get('cinema_code')
    show_date = request.GET.dict().get('show_date')
    print(zz_code, show_date)
    show, income = get_cinema_show_result_func(zz_code, show_date)
    result_dict = {
        'status': 'success',
        'message': '生成成功',
        'show': show,
        'income': income,
    }
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})


# 清除redis锁
@csrf_exempt
def clear_lock(request):
    clear_lock_func()
    result_dict = {
        'status': 'success',
        'message': '完成',
    }
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})


@csrf_exempt
def report(request):
    zz_code = request.GET.dict().get('cinema_code')
    show_date = request.GET.dict().get('show_date')
    force = request.GET.dict().get('force')
    test = request.GET.dict().get('test', None)
    print('report-request_params', zz_code, show_date, force, test)
    # cinema = TestCinema.objects.filter(zz_code=zz_code).first()
    # 获取真实排片和销售数据
    check_point = datetime.datetime.strftime(
        datetime.datetime.strptime(show_date, '%Y-%m-%d') + datetime.timedelta(hours=30), '%Y-%m-%d %H:%M:%S')
    last_real_data = AiShow.objects.filter(Q(is_ai_show=False) &
                                           Q(zz_code=zz_code) &
                                           Q(show_date=show_date) &
                                           Q(updated_at__gt=check_point)).order_by('-id').first()
    if (not last_real_data) or (force == '1'):
        get_cinema_show_result_func(zz_code, show_date)

    if test is None:
        last_ai_data = AiShow.objects.filter(Q(is_ai_show=True) &
                                             Q(is_manual_test=False) &
                                             Q(zz_code=zz_code) &
                                             Q(show_date=show_date)).order_by('-id').first()
    else:
        last_ai_data = AiShow.objects.filter(Q(is_ai_show=True) &
                                             Q(is_manual_test=True) &
                                             Q(zz_code=zz_code) &
                                             Q(show_date=show_date) &
                                             Q(prompt_version=test)).order_by('-id').first()
    print('last_ai_data.pk',last_ai_data.pk)
    if not last_ai_data:
        result_dict = {
            'status': 'Failure',
            'message': '没有查询到有效的智能排片数据，请调整日期和影院后再试！'}
        return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})
    last_real_data = AiShow.objects.filter(Q(is_ai_show=False) &
                                           Q(zz_code=zz_code) &
                                           Q(show_date=show_date)).order_by('-id').first()
    ai_dict = csv_to_obj(data_cleansing(last_ai_data.show))
    real_dict = csv_to_obj(last_real_data.show)
    temp_dict = csv_to_obj(last_ai_data.template)
    target_dict = csv_to_obj(last_ai_data.target)
    print(ai_dict, real_dict, temp_dict, target_dict)
    output_list_ai_real, output_csv_ai_real = format_show_data(zz_code, show_date, ai_dict, real_dict)
    output_list_ai_temp, output_csv_ai_temp = format_show_data(zz_code, show_date, ai_dict, temp_dict)
    output_list_ai_target, output_cdv_ai_target = format_show_data(zz_code, show_date, ai_dict, target_dict)
    # 获取影厅制式
    hall_type = get_hall_type(cinema_code=zz_code)
    # 计算排片占比
    ai_count, ai_total = count_show(ai_dict)
    real_count, real_total = count_show(real_dict)
    target_count, target_total = count_show(target_dict)
    template_count, template_total = count_show(temp_dict)
    # 生成时间
    create_time = datetime.datetime.strftime((last_ai_data.created_at + datetime.timedelta(hours=8)),
                                             '%Y-%m-%d %H:%M:%S')
    if last_ai_data.ai_model in ['lmstudio_qwen3.6_35b', 'company_qwen3.5_35b']:
        think = json.loads(last_ai_data.message)['choices'][0]['message']['reasoning']
    else:
        think = json.loads(last_ai_data.message)['choices'][0]['message']['reasoning_content']
    # 返回
    result_dict = {
        'status': 'success',
        'message': '成功',
        'data': {
            'cinema_name': last_ai_data.cinema,
            'zz_code': last_ai_data.zz_code,
            'hall_type': hall_type,
            'real_sales': last_real_data.sales,
            'ai_sales': last_ai_data.sales,
            'take_times': last_ai_data.take_times,
            'take_tokens': json.loads(last_ai_data.take_tokens)['total_tokens'],
            'csv': list_to_csv(output_csv_ai_real),
            'real_obj': output_list_ai_real,
            'template_obj': output_list_ai_temp,
            'template_date': last_ai_data.temp_date,
            'target_obj': output_list_ai_target,
            'prompt': last_ai_data.prompt,
            'ai_count': ai_count,
            'real_count': real_count,
            'target_count': target_count,
            'template_count': template_count,
            'ai_total': ai_total,
            'real_total': real_total,
            'target_total': target_total,
            'template_total': template_total,
            'result': last_ai_data.result,
            'think': think,
            'ai_model': last_ai_data.ai_model,
            'create_time': create_time
        }
    }
    print(result_dict)
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})


@csrf_exempt
def get_report_date_api(request):
    start, end = get_report_date()
    result_dict = {
        'status': 'success',
        'message': '成功',
        'start': start,
        'end': end
    }
    print(result_dict)
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})


@csrf_exempt
def insert_new_movie(request):
    start = request.GET.get('start')
    result = get_new_movie(start)
    if result is None:
        data = 'None'
    elif result is False:
        data = '参数start格式错误，示例2026-07-11'
    else:
        data = result
    result_dict = {
        'status': 'success',
        'message': '成功',
        'data': data
    }
    print(result_dict)
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})


@csrf_exempt
def manual_test(request):
    print(request)
    cinema_code = request.POST.get('cinema_code')
    show_date = request.POST.get('show_date')
    prompt = request.POST.get('prompt')
    print(cinema_code, show_date, prompt)

    result = show_manual_process(cinema_code, show_date, prompt)

    result_dict = {
        'status': 'success',
        'message': '成功',
        'data': result
    }
    print(result_dict)
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})


@csrf_exempt
def get_test_date(request):
    cinema_code = request.GET.get('cinema_code')
    show_date = request.GET.get('show_date')
    result_dict = {
        'status': 'success',
        'message': '成功',
        'date': get_test_date_from_db(cinema_code, show_date)
    }
    print(result_dict)
    return JsonResponse(result_dict, json_dumps_params={'ensure_ascii': False})