dingxin_toolbox_drf/ai/utils/show_process.py

import json
from ai.models import *
from ai.utils.show_prompt import *
from ai.utils.show_func import *
from ai.utils.basic_func import *
import datetime
from django_redis import get_redis_connection
from django.db.models import Q
from ai.models import TestCinema
import re


# ai排片主流程
def show_main_process(zz_code=None, day_delta=None):
    # 查看状态
    redis_conn = get_redis_connection()
    redis_key = f'ai_show{datetime.date.today().strftime("%Y%m%d")}'
    if redis_conn.exists(redis_key):
        return False
    # 获取影院列表
    redis_conn.set(redis_key, 1, ex=60 * 60 * 20)
    if not zz_code:
        test_cinema_list = TestCinema.objects.filter(is_active=True).all()
    else:
        test_cinema_list = TestCinema.objects.filter(Q(zz_code=zz_code) & Q(is_active=True)).all()
    day_delta = int(get_runtime_val('day_delta')) if day_delta is None else int(day_delta)
    # 获取提示词版本
    prompt_ver = PromptTemplate.objects.filter(
        Q(del_flag=False) & Q(prompt_type='version') & Q(prompt_key='ShowPromptVersion')).first().prompt_val
    # 开始处理提示词
    # 生成目标日期
    show_date = datetime.date.strftime(datetime.date.today() + datetime.timedelta(days=day_delta), '%Y-%m-%d')
    # 获取当前模型
    model = AiModel.objects.filter(current_used=True).first()

    # 按照影院生成排片
    for cinema in test_cinema_list:
        print(cinema.name)
        show_ai = ShowAI(cinema, show_date)
        # template_show, template_date = get_template_show(cinema, show_date)
        template_show, template_date = get_template_show_tomorrow(cinema)
        if get_runtime_val('real_target') == 1:
            target_show = get_target_show(cinema, show_date)
        else:
            target_show = '影厅别名,影厅id,影片别名,本地影片id,语言,放映日期,开始时间,结束时间,片长,场间'
        prompt = show_ai.general_prompt()
        # start = datetime.datetime.now()  # 开始计时
        # result, message, tokens = show_ai.get_show_result_ai()
        # end = datetime.datetime.now()  # 结束计时
        # print('prompt:', prompt)
        # print('result:', result)
        # print('message:', message)
        # print('tokens:', tokens)
        #
        # # 获取排片数据
        # print('result:', result)
        # if model.config_name in ['lmstudio_qwen3.6_35b', 'company_qwen3.5_35b']:
        #     result_obj = json.loads(result.replace('```json', '').replace('}\n```', '}').strip())
        # else:
        #     result_obj = json.loads(result.strip())
        #
        # # 预测排片数据
        # _show = result_obj['show'].replace('```', '').replace('csv', '').strip()
        #
        # # 预测销售数据
        # _sales = str(result_obj['income']).strip()
        #
        # # 处理返回结果
        # try:
        #     AiShow.objects.create(
        #         cinema=cinema.name,
        #         zz_code=cinema.zz_code,
        #         show_date=show_date,
        #         is_ai_show=True,
        #         is_manual_test=False,
        #         ai_model=model.config_name,
        #         template=template_show,
        #         temp_date=template_date,
        #         target=target_show,
        #         show=data_cleansing(_show),
        #         sales=_sales,
        #         prompt=prompt,
        #         result=result,
        #         message=message,
        #         take_times=int((end - start).seconds),
        #         take_tokens=tokens,
        #         prompt_version=prompt_ver,
        #     )
        # except Exception as e:
        #     print(e)
    return True


def show_manual_process(zz_code: str, show_date: str, prompt: str):
    cinema = TestCinema.objects.filter(zz_code=zz_code).first()
    ai = ShowAI(cinema, show_date)

    start = datetime.datetime.now()  # 开始计时
    result, message, tokens = ai.get_show_result_ai(prompt)
    end = datetime.datetime.now()  # 结束计时
    print('prompt:', prompt)
    print('result:', result)
    print('message:', message)
    print('tokens:', tokens)

    # 获取排片数据
    print('result:', result)
    # 获取当前模型
    model = AiModel.objects.filter(current_used=True).first()
    if model.config_name in ['lmstudio_qwen3.6_35b', 'company_qwen3.5_35b']:
        result_obj = json.loads(result.replace('```json', '').replace('}\n```', '}').strip())
    else:
        result_obj = json.loads(result.strip())

    # 预测排片数据
    _show = result_obj['show'].replace('```', '').replace('csv', '').strip()

    # 预测销售数据
    _sales = str(result_obj['income']).strip()

    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()
    print(ai_data)
    # 处理返回结果
    try:
        AiShow.objects.create(
            cinema=cinema.name,
            zz_code=cinema.zz_code,
            show_date=show_date,
            is_ai_show=True,
            is_manual_test=True,
            ai_model=model.config_name,
            template=ai_data.template if ai_data is not None else '',
            temp_date=ai_data.temp_date if ai_data is not None else '',
            target=ai_data.target if ai_data is not None else '',
            show=data_cleansing(_show),
            sales=_sales,
            prompt=prompt,
            result=result,
            message=message,
            take_times=int((end - start).seconds),
            take_tokens=tokens,
            prompt_version=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        )
    except Exception as e:
        print(e)
        return e
    return result

# # 方式一
# _show = next((s for s in result.split('------') if s.startswith('\n影厅别名,影厅id')), '')
# _show = _show.strip()
# # 方式二
# if _show == '':
#     _show = next((s for s in result.split('------') if s.startswith('\n```csv\n影厅别名,')), '')
#     _show = _show.replace('```', '').replace('csv', '').strip()
# # 方式三
# if _show == '':
#     # 正则提取中间内容
#     csv_pattern = r"```csv\s*([\s\S]*?)```"
#     match = re.search(csv_pattern, result)
#     if match:
#         _show = match.group(1)
# print('_show:', _show)
# # 获取销售额
# _sales = next((s for s in result.split('\n') if re.search(r'\d{3,6}元', s)), '')
# _sales = str(re.findall(r'\d{3,6}元', _sales)[-1])
# print('_sales:', _sales)