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 or test == '': 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() 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('manual_test-request-params', 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})