""" 从 daping_rec_online_api 项目的 write_jinja 文件中提取模板信息 生成完整的模板数据（包含 type_id、zone_ids、card_type 等字段） 注意：此脚本仅读取 daping_rec 作为参考，不修改其任何文件 """ import json import re from pathlib import Path from datetime import datetime from typing import Optional # ==================== 机型硬件配置（与 backend/main.py 保持一致）==================== MACHINE_CONFIGS = { "x11": { "label": "X11", "type_ids": { "wash": "201c51890c31c308050100218008390000000000000000000000000000000040", "dry": "201c51890c31c308310100218003484748000000000000000000000000000040", }, "zone_ids": {"wash": ["2", "3"], "dry": ["0"]}, "card_type": "single_card", "has_template_body": True, # x11 有自己的模板正文 }, "hplus": { "label": "Hplus", "type_ids": { "wash": "201c51890c31c308050400218003484843000001000000000000000000000040", "dry": "201c51890c31c308050400218003484843000001000000000000000000000040", }, "zone_ids": {"wash": ["3"], "dry": ["0"]}, "card_type": "single_card", "has_template_body": True, }, "f2_multi": { "label": "F2 Multi", "type_ids": { "wash": "201c51890c31c30805040021800", }, "zone_ids": {"wash": ["1"]}, "card_type": "multi_card", "has_template_body": True, }, "ultra": { "label": "Ultra", "type_ids": { "wash": "201c51890c31c30805010021800805415000000000000000000000000000040", "dry": "201c51890c31c30805010021800805415000000000000000000000000000040", }, "zone_ids": {"wash": ["1", "3", "41", "43", "61", "63"], "dry": ["1", "3", "41", "43", "61", "63"]}, "card_type": "single_card", "has_template_body": False, # 复用 x11 }, "seeker": { "label": "Seeker", "type_ids": { "wash": "201c51890c31c308050100218008054251410000000000000000000000000040", "dry": "201c51890c31c308310100218008054251410003000000000000000000000040", }, "zone_ids": {"wash": ["2", "3", "4"], "dry": ["0"]}, "card_type": "single_card", "has_template_body": False, # 复用 x11 }, "seekerR2": { "label": "SeekerR2", "type_ids": { "wash": "201c51890c31c308050100218008054146000000000000000000000000000040", "dry": "201c51890c31c308310100218008054146000000000000000000000000000040", }, "zone_ids": {"wash": ["1", "41", "61"], "dry": ["24"]}, "card_type": "single_card", "has_template_body": False, # 复用 x11 }, "x11_ultra": { "label": "X11 Ultra", "type_ids": { "wash": "201c51890c31c308050400218003484446000000000000000000000000000040", "dry": "201c51890c31c308050400218003484446000000000000000000000000000040", }, "zone_ids": {"wash": ["1", "7"], "dry": ["24"]}, "card_type": "single_card", "has_template_body": False, # 复用 x11 }, } # 目录名到配置名的映射（处理大小写和下划线差异） DIR_TO_CONFIG_MAP = { "x11Ultra": "x11_ultra", } # 复用映射：哪些机型复用 x11 的模板 # 值表示：当前机型的 device_type 复用 x11 的哪个 device_type REUSE_MAP = { "ultra": {"wash": "wash", "dry": "dry"}, "seeker": {"wash": "wash", "dry": "dry"}, "seekerR2": {"wash": "wash", "dry": "dry"}, "x11_ultra": {"wash": "wash", "dry": "dry"}, } # 模板类型标准化映射 TEMPLATE_TYPE_MAP = { "popular": "time_popular", "made": "time_custom", "custom": "time_custom", "season": "time_season", "season_in": "time_season_in", # 季节上新（入季） "season_storage": "time_season_storage", "care": "care", "env": "environmental", "environmental": "environmental", "health": "health", "health2": "health2", "weather": "weather", "common": "common", } def extract_templates_from_py(py_file: Path, model_name: str) -> list: """从 Python write_jinja 文件中提取模板信息 Returns: list of dict: 每个 dict 包含 template_type、template_body、second_key、device_type """ content = py_file.read_text(encoding='utf-8') templates = [] # 提取 zone_ids（从 zone_ids = [...] 或 zone_id = "..."） zone_ids = [] zone_ids_match = re.search(r'zone_ids\s*=\s*\[([^\]]+)\]', content) if zone_ids_match: zone_ids = [z.strip().strip('"\'') for z in zone_ids_match.group(1).split(',')] else: zone_id_match = re.search(r'zone_id\s*=\s*["\']([^"\']+)["\']', content) if zone_id_match: zone_ids = [zone_id_match.group(1)] # 判断设备类型（通过文件名或内容推断） device_type = "wash" # 默认 file_name = py_file.name.lower() if any(kw in file_name for kw in ['_up', '_dry', 'dry']): device_type = "dry" elif any(kw in file_name for kw in ['_down', '_wash', 'wash']): device_type = "wash" # 提取模板正文 value_xxx = """...""" template_pattern = r'value_(\w+)\s*=\s*"""(.*?)"""' template_matches = re.findall(template_pattern, content, re.DOTALL) # 提取 second_key value_xxx_second_key = "..." second_key_pattern = r'value_(\w+)_second_key\s*=\s*"([^"]+)"' second_key_matches = dict(re.findall(second_key_pattern, content)) has_body = len(template_matches) > 0 if has_body: # 有模板正文的机型（x11/hplus/f2_multi） for template_type, template_body in template_matches: # 跳过基础模板 if template_type in ['default', 'basic', 'w_basic', 'w_default', 'd_default']: continue second_key = second_key_matches.get(template_type, "") templates.append({ "template_type": template_type, "template_body": template_body.strip(), "second_key": second_key, "device_type": device_type, "zone_ids": zone_ids, }) else: # 无模板正文的机型（ultra/seeker/seekerR2/x11Ultra），仅记录 second_key for template_type, second_key in second_key_matches.items(): if template_type in ['default', 'basic', 'w_basic', 'w_default', 'd_default']: continue templates.append({ "template_type": template_type, "template_body": "", # 复用 x11，无正文 "second_key": second_key, "device_type": device_type, "zone_ids": zone_ids, }) return templates def extract_all_templates(daping_rec_dir: Path) -> dict: """从 daping_rec_online_api 的 pre_data 目录提取所有机型模板 Returns: dict: {model_name: [template_info, ...]} """ all_templates = {} pre_data_dir = daping_rec_dir / "app" / "pre_data" if not pre_data_dir.exists(): print(f"❌ 未找到 daping_rec 目录: {pre_data_dir}") return all_templates print(f"\n📂 扫描 daping_rec 模板目录: {pre_data_dir}") # 遍历所有机型目录 for model_dir in sorted(pre_data_dir.iterdir()): if not model_dir.is_dir(): continue dir_name = model_dir.name # 转换目录名为配置名 model_name = DIR_TO_CONFIG_MAP.get(dir_name, dir_name) if model_name not in MACHINE_CONFIGS: print(f" ⏭ 跳过未知机型: {dir_name}") continue print(f"\n🔍 处理机型: {dir_name} -> {model_name} ({MACHINE_CONFIGS[model_name]['label']})") # 查找 write_*.py 文件（排除参数文件） write_files = sorted([ f for f in model_dir.glob("write_*.py") if not f.name.startswith("write_param") ]) if not write_files: print(f" ⚠️ 未找到 write_*.py 文件") continue model_templates = [] for py_file in write_files: print(f" 📄 {py_file.name}") templates = extract_templates_from_py(py_file, model_name) if templates: print(f" ✓ 提取 {len(templates)} 个模板") model_templates.extend(templates) if model_templates: all_templates[model_name] = model_templates print(f" 📊 机型 {model_name} 共 {len(model_templates)} 个模板") return all_templates def normalize_template_type(raw_type: str) -> str: """标准化模板类型名称 Examples: popular → time_popular made → time_custom w_popular → time_popular """ # 去除 w_/d_ 前缀 clean_type = raw_type if raw_type.startswith('w_'): clean_type = raw_type[2:] elif raw_type.startswith('d_'): clean_type = raw_type[2:] # 映射到标准类型 return TEMPLATE_TYPE_MAP.get(clean_type, clean_type) def generate_redis_keys(model_name: str, device_type: str, template_type: str, second_key: str) -> dict: """根据模板信息生成 Redis key Returns: dict: {second_key: str, first_keys: list} """ config = MACHINE_CONFIGS[model_name] card_type = config["card_type"] # 二级 key（如果未提供则自动生成） if not second_key: second_key = f"broadcast-template-jinja:{card_type}:{device_type}:{template_type}" # 一级 key（为每个 zone_id 生成） first_keys = [] type_id = config["type_ids"].get(device_type, "") zone_ids = config["zone_ids"].get(device_type, []) # 模板类型到 recommend_type 的映射 recommend_type_map = { "common": ("common", None), "time_popular": ("time_made_popularity", "1"), "time_custom": ("time_made_popularity", "2"), "time_season": ("time_season_storage", None), "time_season_new": ("time_season_storage", None), "time_season_storage": ("time_season_storage", None), "care": ("care", None), "environmental": ("environmental", None), "health": ("health", None), "health2": ("health", None), "weather": ("weather", None), } rec_type, tag = recommend_type_map.get(template_type, (template_type, None)) for zone_id in zone_ids: if tag: first_key = f"broadcast-template-jinja:{type_id}:{zone_id}:{card_type}:{rec_type}{tag}" else: first_key = f"broadcast-template-jinja:{type_id}:{zone_id}:{card_type}:{rec_type}" first_keys.append(first_key) return {"second": second_key, "first": first_keys} def generate_template_records(all_templates: dict) -> list: """生成完整的模板记录列表 Args: all_templates: {model_name: [template_info, ...]} Returns: list: 模板记录列表 """ records = [] seen_templates = set() # 去重：model_device_type # 模板类型描述 type_descriptions = { "common": "通用模板", "time_popular": "人气模板", "time_custom": "定制模板", "time_season": "季节模板", "time_season_new": "季节上新模板", "time_season_storage": "季节收纳模板", "care": "关怀模板", "environmental": "环保模板", "health": "健康模板", "health2": "健康2-杀菌模板", "weather": "天气模板", } # 处理有模板正文的机型（x11/hplus/f2_multi） for model_name in ["x11", "hplus", "f2_multi"]: if model_name not in all_templates: continue config = MACHINE_CONFIGS[model_name] templates = all_templates[model_name] for tpl in templates: raw_type = tpl["template_type"] template_type = normalize_template_type(raw_type) device_type = tpl["device_type"] # 去重 dedup_key = f"{model_name}_{device_type}_{template_type}" if dedup_key in seen_templates: continue seen_templates.add(dedup_key) # 生成 ID tpl_id = f"{model_name}_{device_type}_{template_type}" # 生成 Redis keys redis_keys = generate_redis_keys(model_name, device_type, template_type, tpl["second_key"]) # 描述 label = config["label"] device_label = "洗衣机" if device_type == "wash" else "干衣机" type_desc = type_descriptions.get(template_type, template_type) record = { "id": tpl_id, "name": f"{label}{device_label}-{type_desc}", "category": f"{'单卡' if config['card_type'] == 'single_card' else '多卡'}{device_label}", "device_type": device_type, "machine_model": model_name, "template_type": template_type, "type_id": config["type_ids"].get(device_type, ""), "zone_ids": config["zone_ids"].get(device_type, []), "card_type": config["card_type"], "reusable_from": "", "reuse_targets": [], "description": f"{label} {device_label} {type_desc}", "output_description": "语音播报脚本", "prerequisite": "", "call_method": "", "input_variables": [], "template_body": tpl["template_body"], "redis_key_second": redis_keys["second"], "redis_keys_first": redis_keys["first"], "created_at": datetime.now().astimezone().isoformat(), "updated_at": datetime.now().astimezone().isoformat(), } records.append(record) print(f" ✓ {record['name']} (ID: {tpl_id})") # 处理复用机型（ultra/seeker/seekerR2/x11_ultra） print(f"\n🔄 处理复用机型模板...") for model_name in ["ultra", "seeker", "seekerR2", "x11_ultra"]: if model_name not in REUSE_MAP: print(f" ⏭ {model_name} 不在 REUSE_MAP 中") continue print(f"\n 🔗 处理 {model_name} 复用模板...") config = MACHINE_CONFIGS[model_name] reuse_info = REUSE_MAP[model_name] source_model = "x11" # 所有复用机型都复用 x11 # 从已生成的 x11 模板中找出对应的模板 x11_templates = [r for r in records if r["machine_model"] == source_model] print(f" 找到 {len(x11_templates)} 个 x11 源模板") for device_type in ["wash", "dry"]: if device_type not in reuse_info: continue source_device = reuse_info[device_type] # 例如：ultra 的 wash 复用 x11 的 wash # 找到对应的 x11 模板 source_templates = [ r for r in x11_templates if r["device_type"] == source_device ] print(f" 设备类型 {device_type} -> 复用 x11 的 {source_device}: {len(source_templates)} 个模板") for source_tpl in source_templates: template_type = source_tpl["template_type"] tpl_id = f"{model_name}_{device_type}_{template_type}" # 去重 dedup_key = f"{model_name}_{device_type}_{template_type}" if dedup_key in seen_templates: print(f" ⏭ 跳过重复: {dedup_key}") continue seen_templates.add(dedup_key) # 生成 Redis keys redis_keys = generate_redis_keys(model_name, device_type, template_type, "") label = config["label"] device_label = "洗衣机" if device_type == "wash" else "干衣机" type_desc = type_descriptions.get(template_type, template_type) # 复用源模板 ID source_tpl_id = f"{source_model}_{source_device}_{template_type}" record = { "id": tpl_id, "name": f"{label}{device_label}-{type_desc} (复用{source_model.upper()})", "category": f"单卡{device_label}", "device_type": device_type, "machine_model": model_name, "template_type": template_type, "type_id": config["type_ids"].get(device_type, ""), "zone_ids": config["zone_ids"].get(device_type, []), "card_type": config["card_type"], "reusable_from": source_tpl_id, "reuse_targets": [], "description": f"{label} {device_label} {type_desc}，复用 {source_model.upper()} 模板", "output_description": "语音播报脚本", "prerequisite": "", "call_method": "", "input_variables": [], "template_body": "", # 复用模板无正文 "redis_key_second": redis_keys["second"], "redis_keys_first": redis_keys["first"], "created_at": datetime.now().astimezone().isoformat(), "updated_at": datetime.now().astimezone().isoformat(), } records.append(record) print(f" ✓ {record['name']} (ID: {tpl_id}, 复用: {source_tpl_id})") # 更新源模板的 reuse_targets for r in records: if r["id"] == source_tpl_id: if "reuse_targets" not in r: r["reuse_targets"] = [] r["reuse_targets"].append(tpl_id) return records def main(): # daping_rec 路径（仅读取，不修改） daping_rec_dir = Path(__file__).parent.parent.parent / "daping_rec_online_api" # 输出文件 output_file = Path(__file__).parent.parent / "backend" / "data" / "templates.json" print("=" * 80) print("🚀 语音播报模板导入工具") print("=" * 80) print(f"\n📖 参考项目: {daping_rec_dir}") print(f"📝 输出文件: {output_file}") if not daping_rec_dir.exists(): print(f"\n❌ 错误: 未找到 daping_rec_online_api 项目") print(f" 请确保 {daping_rec_dir} 存在") return # 提取所有模板 print("\n" + "=" * 80) print("📥 步骤 1: 从 daping_rec 提取模板信息") print("=" * 80) all_templates = extract_all_templates(daping_rec_dir) if not all_templates: print("\n❌ 未提取到任何模板，请检查 daping_rec 目录结构") return print(f"\n📊 提取完成:") for model, templates in all_templates.items(): print(f" • {model}: {len(templates)} 个模板") # 生成模板记录 print("\n" + "=" * 80) print("📦 步骤 2: 生成模板记录（含 Redis key 和复用关系）") print("=" * 80) records = generate_template_records(all_templates) # 保存到 JSON print("\n" + "=" * 80) print("💾 步骤 3: 保存到 templates.json") print("=" * 80) output_file.parent.mkdir(parents=True, exist_ok=True) output_file.write_text( json.dumps(records, ensure_ascii=False, indent=2), encoding='utf-8' ) print(f"\n✅ 导入成功!") print(f" • 模板总数: {len(records)}") print(f" • 输出文件: {output_file}") print(f"\n💡 提示: 运行后端服务后可通过 API 查看和管理这些模板") print("=" * 80) if __name__ == "__main__": main()