x11_template_vue2/templates/f2_multi/param_f2_program_test.py

import json
import sys 
import os

import pymysql

base_dir = os.path.dirname(os.path.abspath(__file__)) if '__file__' in globals() else os.getcwd()
project_root = os.path.abspath(os.path.join(base_dir, "..", ".."))
sys.path.append(project_root)

from utils.functions import convert_commands_to_md5
from pre_data.common.client_tools import  redis_client , write_hset


DB_CONFIG = {
    "host": "rm-2zenu092q42967j71.mysql.rds.aliyuncs.com",
    "port": 3306,
    "user": "u_ai_card",
    "password": "BHdfH6pMscM_sN3W",
    "database": "ai_card",
    "charset": "utf8mb4",
}

PROGRAM_NAME_REPLACE_RULES = {
    "'": "分钟",
}

PROGRAM_NAMES_INDEX_KEY = "program-names:index"
PROGRAM_NAMES_STREAM_KEY = "program-names:stream"
PROGRAM_NAMES_KEY_PREFIX = "program-names"


def get_db_connection() -> pymysql.Connection:
    return pymysql.connect(**DB_CONFIG)


def normalize_program_name(name: str) -> str:
    text = str(name).strip()
    for src, dst in PROGRAM_NAME_REPLACE_RULES.items():
        text = text.replace(src, dst)
    return text


def make_program_names_key(type_id: str, zone_id: str) -> str:
    return f"{PROGRAM_NAMES_KEY_PREFIX}:{type_id}:{zone_id}"


def extract_param_dict(param_json: dict, wanted_names: list[str]):
    """
    从 param_json 中提取指定参数名称的字典
    返回格式: {"夜间洗": "开", "风巡航": "护理"}
    """
    result: dict[str, str] = {}
    param_list = param_json.get("param", [])

    for item in param_list:
        para_name = item.get("paraName", "").strip()
        show_value = item.get("showValue")

        if (
            para_name
            and para_name in wanted_names
            and show_value is not None
            and str(show_value).strip() != ""
        ):
            result[item["paraName"]] = item["showValue"]

    return result


def build_mapping(param_json_list: list[dict], wanted_names: list[str]):
    """
    构建 Redis hset 的 mapping
    key: 提取的参数字典的JSON字符串
    value: 完整的 param_json JSON字符串
    """
    mapping: dict[str, str] = {}

    for param_json in param_json_list:
        param_dict = extract_param_dict(param_json, wanted_names)
        key_str = json.dumps(param_dict, ensure_ascii=False, sort_keys=True)
        value_str = json.dumps(param_json, ensure_ascii=False, sort_keys=True)
        mapping[key_str] = value_str

    return mapping


#写入程序名称及定义对应的stream
def write_program_names(type_id: str, zone_id: str, program_names: dict[str, str], clear_before_write: bool = True):
    redis_key = make_program_names_key(type_id, zone_id)
    index_value = f"{type_id}:{zone_id}"
    normalized = {str(k): str(v) for k, v in program_names.items()}

    if clear_before_write:
        redis_client.delete(redis_key)

    write_hset(redis_key, normalized)
    redis_client.sadd(PROGRAM_NAMES_INDEX_KEY, index_value)
    redis_client.xadd(
        PROGRAM_NAMES_STREAM_KEY,
        {
            "action": "upsert",
            "type_id": type_id,
            "zone_id": zone_id,
        },
    )

    print(f"[program_names] key={redis_key}, fields={len(normalized)}")

def clear_redis_key_pattern(pattern: str) :
    for key in redis_client.scan_iter(match=pattern, count=100):
        redis_client.delete(key)
def clear_default_commands(type_id: str, zone_id: str) :
    clear_redis_key_pattern(f"default-commands:{type_id}:{zone_id}:*")

#获取默认参数
def write_default_params(type_id: str ,test_type_id: str, wash_type: str, function_type: str, batch_num: str, wanted_names: list[str]):
    """
    1. 按当前范围写入一次 default-commands
    仅写入默认参数
    """
    conn = get_db_connection()
    try:
        cursor = conn.cursor()
        cursor.execute(
            """
            SELECT request_param
            FROM ai_card.t_screen_init_data
            WHERE type_id = %s AND  wash_type = %s AND function_type= %s AND batch_num = %s 
            ORDER BY id
            """,
            (type_id, wash_type, function_type, batch_num),
        )
        print(type_id, wash_type, function_type, batch_num)
        param_json_list: list[dict] = []

        index = 1
        for row in cursor.fetchall():
            
            outer_json = json.loads(row[0])
            param_json = outer_json.get("param_json", {})
            param_json_list.append(param_json)

            if not param_json_list:
                print(f"[skip] no rows for {wash_type}-{function_type}, batch_num={batch_num}")
                return

            # 原有默认参数写入逻辑：保持不变
            # type_id = str(outer_json.get("type_id", "")).strip()
            zone_id = str(param_json_list[-1].get("wash_type", "")).strip()
            program_id = str(param_json_list[-1].get("program_id", "")).strip()
            redis_key = f"default-commands:{test_type_id}:{zone_id}:{program_id}"
            mapping = build_mapping(param_json_list, wanted_names)
            print("============================================================================================")
            print(f"default_commands key: {redis_key}")
            print(f"index: {index}")
            index=index+1
            write_hset(redis_key, mapping)

        print(f"完成写入，总条数：{len(param_json_list)}")
    finally:
        conn.close()


def write_program_names_from_range(type_id:str,test_type_id:str, wash_type: str, function_type: str, batch_num: str):
    """
    仅从当前范围聚合 program_id / proName，并写入 program-names Redis + stream
    """
    conn = get_db_connection()
    program_names: dict[str, str] = {}
    try:
        cursor = conn.cursor()
        cursor.execute(
            """
            SELECT request_param
            FROM ai_card.t_screen_init_data
            WHERE type_id = %s AND  wash_type = %s AND function_type= %s AND batch_num = %s 
            ORDER BY id
            """,
            (type_id, wash_type, function_type, batch_num),
        )

        zone_id = wash_type

        for (request_param,) in cursor.fetchall():
            outer_json = json.loads(request_param)
            param_json = outer_json.get("param_json", {})
            # type_id = str(outer_json.get("type_id", "")).strip()
            # zone_id = str(param_json.get("wash_type", "")).strip()

            program_id = str(param_json.get("program_id", "")).strip()
            pro_name = str(param_json.get("proName", "")).strip()
            if program_id and pro_name:
                program_names[program_id] = normalize_program_name(pro_name)

        if type_id and zone_id and program_names:
            write_program_names(test_type_id, zone_id, program_names, clear_before_write=True)
            print("program_names written: "+test_type_id+":"+zone_id )
        else:
            print(f"[skip] no program_names for {wash_type}-{function_type}, batch_num={batch_num}")
    finally:
        conn.close()
    return program_names


def write_multi_program_names_from_range(type_id:str, wash_type: str, batch_num: str):
    """
    仅从当前范围聚合 program_id / proName，并写入 program-names Redis + stream
    """
    conn = get_db_connection()
    program_names: dict[str, str] = {}
    try:
        cursor = conn.cursor()
        cursor.execute(
            """
            SELECT program_id, program_name, program_desc
            FROM ai_card.ai_rec_v2_program_info
            WHERE type_id = %s AND  wash_type = %s AND batch_num = %s 
            ORDER BY id
            """,
            (type_id, wash_type, batch_num),
        )

        zone_id = wash_type

        for (program_id, program_name, program_desc) in cursor.fetchall():
            # 将每一行数据转换为指定格式的字符串作为value
            value_str = '{"name":"' + str(program_name) + '","desc":"' + str(program_desc) + '"}'
            program_names[str(program_id)] = value_str
        
        print(program_names)

        if type_id and zone_id and program_names:
            write_program_names(type_id, zone_id, program_names, clear_before_write=True)
            print("program_names written: "+type_id+":"+zone_id )
        else:
            print(f"[skip] no program_names for {wash_type}, batch_num={batch_num}")
    finally:
        conn.close()
    return program_names