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