""" 完整导入 daping 项目的所有模板 支持： 1. 提取 X11、Hplus、F2 Multi 的完整模板（包含关怀、健康、天气等） 2. 为 Ultra、Seeker、X11Ultra 创建复用关系（复用 X11 模板） """ import json import re from pathlib import Path from datetime import datetime DAPING_PRE_DATA = Path(__file__).parent.parent.parent / "daping_rec_online_api" / "app" / "pre_data" def extract_templates_from_py(py_file: Path) -> list: """从 Python 文件中提取所有模板""" content = py_file.read_text(encoding='utf-8') templates = [] # 匹配 value_xxx = """...""" 模式 pattern = r'value_(\w+)\s*=\s*"""(.*?)"""' matches = re.findall(pattern, content, re.DOTALL) for template_type, template_body in matches: # 跳过不需要的模板 if template_type in ['default', 'basic', 'w_basic', 'w_default', 'd_default']: continue # 提取 second_key key_pattern = rf'value_{template_type}_second_key\s*=\s*"([^"]+)"' key_match = re.search(key_pattern, content) redis_key = key_match.group(1) if key_match else "" # 清理模板类型名称 clean_type = template_type if template_type.startswith('w_'): clean_type = template_type[2:] elif template_type.startswith('d_'): clean_type = template_type[2:] templates.append({ "template_type": clean_type, "template_body": template_body.strip(), "redis_key": redis_key }) return templates def generate_all_templates() -> list: """生成所有模板记录（包括复用关系）""" records = [] template_id_counter = 1 # 模板类型标准化映射 TYPE_NORMALIZE = { "popular": "time_popular", "made": "time_custom", "custom": "time_custom", "season_in": "time_season", "season_storage": "time_season", "care": "time_care", "env": "time_environmental", "environmental": "time_environmental", "health": "time_health", "health2": "time_health_2", "weather": "time_weather", } # 模板类型描述 TYPE_DESCRIPTIONS = { "common": "通用", "time_popular": "人气", "time_custom": "定制", "time_season": "季节", "time_care": "关怀", "time_environmental": "环保", "time_health": "健康", "time_health_2": "健康2-杀菌", "time_weather": "天气", } # 机型配置（与 daping 项目一致） MODEL_CONFIGS = [ # X11 洗衣机 { "model_name": "x11", "display_name": "X11", "model_id": "x11", "device_type": "wash", "zone_id": "2", "zone_name": "洗衣机", "file": "write_jinja_x11_down.py", }, # X11 干衣机 { "model_name": "x11", "display_name": "X11", "model_id": "x11", "device_type": "dry", "zone_id": "0", "zone_name": "干衣机", "file": "write_jinja_x11_up.py", }, # Hplus 洗衣机 { "model_name": "hplus", "display_name": "Hplus", "model_id": "hplus", "device_type": "wash", "zone_id": "3", "zone_name": "洗衣机", "file": "write_jinja_h_down.py", }, # Hplus 干衣机 { "model_name": "hplus", "display_name": "Hplus", "model_id": "hplus", "device_type": "dry", "zone_id": "0", "zone_name": "干衣机", "file": "write_jinja_h_up.py", }, # F2 Multi 洗衣机 { "model_name": "f2_multi", "display_name": "F2 Multi", "model_id": "f2_multi", "device_type": "wash", "zone_id": "1", "zone_name": "洗衣机", "file": "write_jinja_multi_down.py", }, ] # 复用关系配置 REUSE_CONFIGS = [ # Ultra 洗衣机（复用 X11 洗涤） { "model_id": "ultra", "display_name": "Ultra", "device_type": "wash", "zone_id": "1", "zone_name": "洗衣机", "reuse_from_model": "x11", "reuse_from_device": "wash", }, # Seeker 洗衣机（复用 X11 洗涤） { "model_id": "seeker", "display_name": "Seeker", "device_type": "wash", "zone_id": "2", "zone_name": "洗衣机", "reuse_from_model": "x11", "reuse_from_device": "wash", }, # X11 Ultra 干衣机（复用 X11 干衣） { "model_id": "x11_ultra", "display_name": "X11 Ultra", "device_type": "dry", "zone_id": "24", "zone_name": "干衣机", "reuse_from_model": "x11", "reuse_from_device": "dry", }, ] # 1. 处理独立模板（X11、Hplus、F2 Multi） print("\n" + "="*60) print("提取独立模板...") print("="*60) source_template_map = {} # 用于复用关系查找 for config in MODEL_CONFIGS: file_path = DAPING_PRE_DATA / config["model_name"] / config["file"] if not file_path.exists(): print(f"⚠ 文件不存在: {file_path}") continue print(f"\n处理: {config['display_name']} {config['zone_name']}") templates = extract_templates_from_py(file_path) if not templates: print(f" ⚠ 未找到模板") continue print(f" ✓ 找到 {len(templates)} 个模板") for tpl in templates: template_type = tpl["template_type"] # 标准化模板类型 normalized_type = TYPE_NORMALIZE.get(template_type, template_type) if normalized_type.startswith("time_"): final_type = normalized_type else: final_type = f"time_{normalized_type}" if normalized_type != "common" else "common" description = TYPE_DESCRIPTIONS.get(final_type, final_type) # 生成模板ID tpl_id = f"{config['model_id']}_{config['device_type']}_{final_type}" template_id_counter += 1 # 记录源模板位置（用于复用） source_key = f"{config['model_id']}_{config['device_type']}_{final_type}" source_template_map[source_key] = tpl_id record = { "id": tpl_id, "name": f"{config['display_name']}{config['zone_name']}-{description}", "category": f"{'洗涤' if config['device_type'] == 'wash' else '干衣'}", "device_type": config["device_type"], "machine_model": config["model_id"], "template_type": final_type, "reusable_from": "", "reuse_targets": [], "description": f"{config['display_name']} {config['zone_name']} {description}模板", "output_description": "语音播报脚本", "call_method": "", "input_variables": [ { "name": "program_id", "label": "程序ID", "description": "洗涤/干衣程序ID", "data_type": "string", "required": True, "default": "15" } ], "template_body": tpl["template_body"], "created_at": datetime.utcnow().isoformat() + "Z", "updated_at": datetime.utcnow().isoformat() + "Z" } records.append(record) print(f" ✓ {record['name']} ({len(tpl['template_body'])} 字符)") # 2. 处理复用模板（Ultra、Seeker、X11Ultra） print("\n" + "="*60) print("创建复用模板...") print("="*60) for config in REUSE_CONFIGS: source_key = f"{config['reuse_from_model']}_{config['reuse_from_device']}" # 找出所有源模板 source_tpl_ids = [ tpl_id for key, tpl_id in source_template_map.items() if key.startswith(source_key) ] if not source_tpl_ids: print(f"⚠ 未找到 {config['display_name']} {config['zone_name']} 的源模板") continue print(f"\n处理: {config['display_name']} {config['zone_name']} (复用)") for source_tpl_id in source_tpl_ids: # 从源模板ID中提取模板类型 parts = source_tpl_id.split("_") template_type = "_".join(parts[2:]) # 去掉 model_id 和 device_type description = TYPE_DESCRIPTIONS.get(template_type, template_type) # 生成复用模板ID tpl_id = f"{config['model_id']}_{config['device_type']}_{template_type}" template_id_counter += 1 # 更新源模板的 reuse_targets for record in records: if record["id"] == source_tpl_id: if tpl_id not in record["reuse_targets"]: record["reuse_targets"].append(tpl_id) record = { "id": tpl_id, "name": f"{config['display_name']}{config['zone_name']}-{description}", "category": f"{'洗涤' if config['device_type'] == 'wash' else '干衣'}", "device_type": config["device_type"], "machine_model": config["model_id"], "template_type": template_type, "reusable_from": source_tpl_id, "reuse_targets": [], "description": f"{config['display_name']} {config['zone_name']} {description}模板（复用）", "output_description": "语音播报脚本", "call_method": "", "input_variables": [ { "name": "program_id", "label": "程序ID", "description": "洗涤/干衣程序ID", "data_type": "string", "required": True, "default": "15" } ], "template_body": "", # 复用模板不需要存储正文 "created_at": datetime.utcnow().isoformat() + "Z", "updated_at": datetime.utcnow().isoformat() + "Z" } records.append(record) print(f" ✓ {record['name']} (复用自 {source_tpl_id})") return records def main(): output_file = Path(__file__).parent.parent / "backend" / "data" / "templates.json" print("="*60) print("开始从 daping 项目提取所有模板...") print("="*60) # 生成所有模板记录 records = generate_all_templates() # 保存到 JSON output_file.parent.mkdir(parents=True, exist_ok=True) output_file.write_text( json.dumps(records, ensure_ascii=False, indent=2), encoding='utf-8' ) # 统计信息 type_stats = {} model_stats = {} reuse_count = sum(1 for r in records if r["reusable_from"]) for r in records: type_stats[r["template_type"]] = type_stats.get(r["template_type"], 0) + 1 model_stats[r["machine_model"]] = model_stats.get(r["machine_model"], 0) + 1 print(f"\n" + "="*60) print(f"✓ 成功导入 {len(records)} 个模板") print(f" - 独立模板: {len(records) - reuse_count} 个") print(f" - 复用模板: {reuse_count} 个") print(f"\n按机型统计:") for model, count in sorted(model_stats.items()): print(f" - {model}: {count} 个") print(f"\n按模板类型统计:") for tpl_type, count in sorted(type_stats.items()): print(f" - {tpl_type}: {count} 个") print(f"\n输出文件: {output_file}") print("="*60) if __name__ == "__main__": main()