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 str(get_runtime_val('real_target')) == '1':
            target_show = get_target_show(cinema, show_date)
        else:
            target_show = '影厅别名,影厅id,影片别名,本地影片id,语言,放映日期,开始时间,结束时间,片长,场间'
        prompt = show_ai.general_prompt()
        if str(get_runtime_val('test')) == '0':
            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


def get_prompt_process(zz_code=None, show_date=None):
    print('zz_code:', zz_code)
    print('show_date:', show_date)
    if zz_code == '' or show_date == '' :
        return '请检查影院编码和排片日期'
    # 获取影院
    cinema = TestCinema.objects.filter(Q(zz_code=zz_code) & Q(is_active=True)).first()
    # 获取提示词版本
    prompt_ver = PromptTemplate.objects.filter(
        Q(del_flag=False) & Q(prompt_type='version') & Q(prompt_key='ShowPromptVersion')).first().prompt_val
    # 开始处理提示词
    # 按照影院生成排片
    print(cinema.name)
    show_ai = ShowAI(cinema, show_date)
    prompt = show_ai.general_prompt()
    return prompt