""" Jinja2 模板渲染服务 - 自动变量识别与条件分支预览版 从 daping_rec 模板逻辑中提取所有变量，支持条件分支控制预览 """ from __future__ import annotations import copy import random as random_module import re from typing import Dict, Iterable, List, Optional from jinja2 import BaseLoader, Environment, StrictUndefined, meta from jinja2.exceptions import TemplateError from fastapi import HTTPException from ..schemas import ( TemplatePreviewRequest, TemplateRecord, TemplateRenderRequest, VariableDefinition, ) # === 从 daping_rec 模板中识别的完整变量配置 === VARIABLE_DEFAULTS = { # 基础变量 'city': '青岛市', 'period': '早上', 'season': '夏天', # 程序相关 'program_id': '1', # 推荐类型 - 条件分支控制变量 'recommend_type': 'POPULAR', # POPULAR / CUSTOM 'rerank_reason': '', # PIN_CONSECUTIVE / PIN_WEEKLY / ""（空=走推荐类型分支） 'feature_name': '最喜欢用', # 人气统计 'count': 150, 'user_count': 50, 'user_proportion': 60, # 天气 'weather': '小雪', 'wd_link': '1', 'washdry': '_wd', # 季节 'time_term': '立春', 'filter_rule_code': 'time_season_in', # 关怀（同时包含洗衣机和干衣机的关怀功能，避免模板找不到键） 'care': { 'night_wash': {'active': 1, 'display_name': '夜间洗'}, # 洗衣机 'night_dry': {'active': 1, 'display_name': '夜间烘'}, # 干衣机 'wind_cruise': {'active': 1, 'display_name': '风巡航'} # 洗衣机/干衣机共用 }, # 干衣机参数 'changed_params_dry': { 'DryLevel': '3', 'DryTemp': '2', 'FanLevel': '2' }, # 洗衣机参数 'changed_params_wash': { 'setTemp': 40, 'speed': 12, 'spinTime': 3, 'washTime': 45 }, # 简写变量映射（模板中可能直接使用 changed_params） # 合并洗衣机和干衣机参数，避免干衣机模板访问 DryLevel 等键时报错 'changed_params': { 'setTemp': 40, 'speed': 12, 'spinTime': 3, 'washTime': 45, 'DryLevel': '3', 'DryTemp': '2', 'FanLevel': '2' }, # 多卡模板变量 'washNight': '1', 'WFreeCn': '1', } # === 条件分支变量定义 === CONDITION_VARIABLES = { 'rerank_reason': { 'label': '推荐原因', 'type': 'select', 'options': [ {'value': 'PIN_CONSECUTIVE', 'label': '连续使用习惯'}, {'value': 'PIN_WEEKLY', 'label': '每周习惯'}, {'value': '', 'label': '无（走推荐类型分支）'}, ], 'description': '控制推荐原因分支', }, 'recommend_type': { 'label': '推荐类型', 'type': 'select', 'options': [ {'value': 'POPULAR', 'label': '人气推荐'}, {'value': 'CUSTOM', 'label': '定制推荐'}, ], 'description': '控制推荐类型分支', }, 'filter_rule_code': { 'label': '季节规则', 'type': 'select', 'options': [ {'value': 'time_season_in', 'label': '入季/上新'}, {'value': 'time_season_storage', 'label': '收纳'}, ], 'description': '控制季节模板分支', }, 'feature_name': { 'label': '特征名称', 'type': 'select', 'options': [ {'value': '最喜欢用', 'label': '最喜欢用'}, {'value': '最近很喜欢用', 'label': '最近很喜欢用'}, {'value': '最近一周常用', 'label': '最近一周常用'}, {'value': '上周这个时间喜欢用', 'label': '上周这个时间喜欢用'}, {'value': '', 'label': '空（走兜底播报）'}, ], 'description': '控制特征分支（空字符串走else分支）', }, 'program_id': { 'label': '程序ID', 'type': 'text', 'placeholder': '1-131', 'description': '程序ID（1=棉麻, 15=速洗, etc）', }, } # === 数据变量定义 === DATA_VARIABLES = { 'city': {'label': '城市', 'placeholder': '青岛市'}, 'period': {'label': '时段', 'placeholder': '早上/中午/晚上'}, 'season': {'label': '季节', 'placeholder': '夏天'}, 'weather': {'label': '天气', 'placeholder': '小雪'}, 'count': {'label': '使用次数', 'placeholder': '150'}, 'user_count': {'label': '用户数', 'placeholder': '50'}, 'user_proportion': {'label': '用户比例', 'placeholder': '60'}, 'time_term': {'label': '节气', 'placeholder': '立春'}, } class TemplateRenderer: """智能模板渲染器 - 支持条件分支预览""" def __init__(self) -> None: self.env = Environment( loader=BaseLoader(), undefined=StrictUndefined, trim_blocks=True, lstrip_blocks=True, autoescape=False, ) # 注册 random filter - 解决 'str' object is not callable self.env.filters['random'] = self._jinja_random # 注册全局变量 self.env.globals['program_name_map'] = self._default_program_name_map() def _jinja_random(self, seq): """Jinja2 random filter - 从序列中随机选一个元素""" if not seq: return "" try: return random_module.choice(list(seq)) except: return "" def _default_program_name_map(self) -> dict: """程序名称映射""" programs = { "1": ("棉麻", ""), "2": ("化纤", ""), "3": ("羽绒", ""), "6": ("衬衣", ""), "7": ("童装", ""), "9": ("烫烫净", ""), "10": ("羊毛", ""), "13": ("筒自洁", "清洁洗衣机专用"), "14": ("内衣", ""), "15": ("速洗", "快速洗涤"), "17": ("大件", ""), "18": ("运动服", ""), "19": ("超柔", ""), "22": ("丝绸", ""), "27": ("漂+脱", ""), "28": ("单脱水", ""), "30": ("除毛烘", ""), "33": ("定时烘", ""), "34": ("静筒烘", ""), "39": ("云程序", ""), "40": ("强力洗", ""), "43": ("定时烘", ""), "45": ("云程序", ""), "52": ("冲锋衣", ""), "56": ("智慧烘", ""), "58": ("浴巾", ""), "61": ("智慧洗", "智能识别衣物"), "79": ("快洗15", ""), "107": ("除菌螨", ""), "118": ("除菌螨", ""), "131": ("夏凉被", ""), } class Program: def __init__(self, name, desc=""): self.name = name self.desc = desc def __str__(self): return self.name def __format__(self, fmt): return self.name.__format__(fmt) def get(self, key, default=None): if key in programs: return programs[key][0] return default return { pid: Program(name, desc) for pid, (name, desc) in programs.items() } def _extract_common_support(self, common_body: str) -> str: """ 从common模板中提取宏定义和列表变量定义 返回需要合并到子模板前面的代码片段 """ import re if not common_body: return "" parts = [] # 1. 提取 {% macro name(...) %}...{% endmacro %} # 支持 {%- macro ... -%} 和 {%- endmacro -%} 等 trim 语法变体 macro_pattern = r'{%-?\s*macro\s+\w+\s*\([^)]*\)\s*-%?%}[\s\S]*?{%-?\s*endmacro\s*-%?%}' for match in re.finditer(macro_pattern, common_body): parts.append(match.group(0)) # 2. 提取 {% set var_name = ["..."] %} 形式的列表变量 # 匹配 greeting/assistant1/assistant2/user1/user2 等列表变量 list_set_pattern = r'{%\s*set\s+(greeting|assistant1|assistant2|user1|user2|period_list|city_list)\s*=\s*\[[^\]]*\]\s*%}' for match in re.finditer(list_set_pattern, common_body, re.IGNORECASE): parts.append(match.group(0)) # 3. 也提取 {% set var_name = ("...") %} 或 {% set var_name = "..." %} # 简单的字符串/数字变量 simple_set_pattern = r'{%\s*set\s+(feature_week_name|feature_name_default|default_program)\s*=\s*[^%]+\s*%}' for match in re.finditer(simple_set_pattern, common_body, re.IGNORECASE): parts.append(match.group(0)) return '\n\n'.join(parts) if parts else "" def _find_macro_calls(self, template_body: str) -> set: """从模板AST中提取被当作函数调用的变量名（即宏调用）""" try: ast = self.env.parse(template_body) macro_calls = set() # 遍历AST查找Call节点 import jinja2.nodes as nodes for node in ast.find_all(nodes.Call): if isinstance(node.node, nodes.Name): macro_calls.add(node.node.name) return macro_calls except: return set() def preview(self, payload: TemplatePreviewRequest) -> dict: """ 预览模板 - 基于真实模板正文渲染 自动检测 {% set xxx %} 定义的变量，如果没有对应的 {{ xxx }} 输出则自动添加 如果有common_body，自动合并宏定义和列表变量 返回：{success, result, error, condition_vars, data_vars, template_body} """ try: # 先对原始模板处理 auto_add_output body = self._auto_add_output(payload.template_body) # 合并common模板的宏和变量定义（如果提供） if payload.common_body: body = payload.common_body + '\n\n' + body # 分析变量 condition_vars, data_vars = self._analyze_variables(body) # 构建上下文 - 使用默认值 + 用户自定义值 context = self._build_context(body, payload.parameters) # 渲染 template = self.env.from_string(body) rendered = template.render(**context) # 清理输出 paragraphs = [p.strip() for p in rendered.split("\n\n") if p.strip()] result_text = "\n\n".join(paragraphs) return { "success": True, "result": result_text, "error": "", "condition_vars": condition_vars, "data_vars": data_vars, "template_body": payload.template_body, "context_used": {k: str(v)[:100] for k, v in context.items() if k != 'program_name_map'} } except TemplateError as exc: return { "success": False, "result": "", "error": f"模板语法错误: {exc}", "condition_vars": [], "data_vars": [], "template_body": payload.template_body, "context_used": {} } except Exception as exc: import traceback return { "success": False, "result": "", "error": f"渲染失败: {exc}\n{traceback.format_exc()}", "condition_vars": [], "data_vars": [], "template_body": payload.template_body, "context_used": {} } def _auto_add_output(self, template_body: str) -> str: """ 自动检测 {% set xxx %}...{% endset %} 块级定义的变量， 如果没有对应的 {{ xxx }} 输出则自动添加。 只处理块级set（内容块），不处理赋值型set（如 {% set count = 1 %}） """ import re body = template_body # 只匹配块级set: {% set name %}...{% endset %} set_block_pattern = r'{%\s*set\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*%}' set_vars = set() for match in re.finditer(set_block_pattern, body): var_name = match.group(1) # 排除宏定义 if var_name not in ('macro', 'endmacro'): set_vars.add(var_name) # 过滤掉宏名 set_vars = {v for v in set_vars if v not in ('macro', 'endmacro')} # 检查每个块级set变量是否有输出 added_outputs = [] for var in set_vars: # 检查是否有 {{ var }} 输出 output_pattern = r'{{\s*' + var + r'\s*}}' if not re.search(output_pattern, body): added_outputs.append(f'{{{{ {var} }}}}') if added_outputs: body = body.rstrip() + '\n\n' + '\n'.join(added_outputs) return body def _analyze_variables(self, template_body: str) -> tuple: """ 分析模板中的变量，分离条件变量和数据变量 返回: (condition_vars, data_vars) """ try: ast = self.env.parse(template_body) undeclared = meta.find_undeclared_variables(ast) except: return [], [] condition_vars = [] data_vars = [] for var in sorted(undeclared): if var == 'program_name_map': continue # 检查是否是条件变量 if var in CONDITION_VARIABLES: cfg = CONDITION_VARIABLES[var] var_def = { 'name': var, 'label': cfg['label'], 'type': cfg.get('type', 'text'), 'value': str(VARIABLE_DEFAULTS.get(var, '')), 'default': str(VARIABLE_DEFAULTS.get(var, '')), 'options': cfg.get('options', []), 'placeholder': cfg.get('placeholder', ''), 'description': cfg.get('description', ''), } condition_vars.append(var_def) elif var in DATA_VARIABLES: cfg = DATA_VARIABLES[var] var_def = { 'name': var, 'label': cfg['label'], 'type': 'text', 'value': str(VARIABLE_DEFAULTS.get(var, '')), 'default': str(VARIABLE_DEFAULTS.get(var, '')), 'placeholder': cfg.get('placeholder', ''), 'description': cfg.get('description', ''), } data_vars.append(var_def) else: # 未知变量，作为数据变量处理 var_def = { 'name': var, 'label': var, 'type': 'text', 'value': str(VARIABLE_DEFAULTS.get(var, '')), 'default': str(VARIABLE_DEFAULTS.get(var, '')), 'placeholder': '', 'description': '', } data_vars.append(var_def) return condition_vars, data_vars def _build_context(self, template_body: str, custom_params: Optional[dict] = None) -> dict: """ 智能构建渲染上下文 """ context = {} # 提取模板变量 try: ast = self.env.parse(template_body) undeclared = meta.find_undeclared_variables(ast) except: undeclared = set() # 检测模板中的宏调用，不为宏名填充空字符串 macro_calls = self._find_macro_calls(template_body) # 为每个变量填充默认值 for var in undeclared: if var == 'program_name_map': continue # 跳过宏调用（这些应该由模板中的 {% macro %} 定义提供） if var in macro_calls: continue if var in VARIABLE_DEFAULTS: context[var] = copy.deepcopy(VARIABLE_DEFAULTS[var]) else: context[var] = "" # 合并自定义参数（覆盖默认值） if custom_params: for key, value in custom_params.items(): # 尝试类型转换 if key in ('count', 'user_count', 'user_proportion'): try: context[key] = int(value) except: context[key] = value elif key in ('changed_params', 'changed_params_wash', 'changed_params_dry', 'care'): # 如果是JSON字符串，先解析为对象 if isinstance(value, str): import json try: context[key] = json.loads(value) except (json.JSONDecodeError, ValueError): # 解析失败，使用默认值 context[key] = copy.deepcopy(VARIABLE_DEFAULTS.get(key, {})) else: context[key] = value else: context[key] = value return context def render_from_record(self, record: TemplateRecord, parameters: dict) -> str: """渲染模板记录（保留旧接口兼容）""" normalized = self._normalize_parameters(parameters, record.input_variables) normalized = self._fill_missing_variables(record.template_body, normalized) return self._render(record.template_body, normalized) def _fill_missing_variables(self, template_body: str, parameters: dict) -> dict: """扫描模板，为未提供的变量填充默认值""" try: ast = self.env.parse(template_body) undeclared = meta.find_undeclared_variables(ast) for var in undeclared: if var not in parameters and var != 'program_name_map': parameters[var] = copy.deepcopy(VARIABLE_DEFAULTS.get(var, "")) except Exception: pass return parameters def _render(self, template_body: str, parameters: dict) -> str: """执行渲染""" try: template = self.env.from_string(template_body) rendered = template.render(**parameters) paragraphs = [p.strip() for p in rendered.split("\n\n") if p.strip()] return "\n\n".join(paragraphs) except TemplateError as exc: raise HTTPException(status_code=400, detail=f"模板渲染出错: {exc}") from exc def _normalize_parameters( self, parameters: Dict[str, object], definitions: Optional[Iterable[VariableDefinition]] = None, ) -> Dict[str, object]: """参数类型转换""" type_map = {} for definition in definitions or []: if isinstance(definition, VariableDefinition): name = (definition.name or "").strip() data_type = (definition.data_type or "").lower() elif isinstance(definition, dict): name = (definition.get("name") or "").strip() data_type = (definition.get("data_type") or "").lower() else: continue if name: type_map[name] = data_type return { key: self._convert_value(value, type_map.get(key)) for key, value in parameters.items() } def _convert_value(self, value: object, data_type: Optional[str]) -> object: """值类型转换""" if value is None or isinstance(value, (int, float, bool)): return value if isinstance(value, str): stripped = value.strip() if stripped == "": return None if (data_type or "").startswith("bool"): lowered = stripped.lower() if lowered in {"true", "1", "yes", "y", "on"}: return True if lowered in {"false", "0", "no", "n", "off"}: return False if (data_type or "") in {"number", "integer"} or self._looks_numeric(stripped): try: return int(stripped) if stripped.isdigit() else float(stripped) except ValueError: return value return value def _looks_numeric(self, value: str) -> bool: return bool(re.fullmatch(r"[+-]?\d+(\.\d+)?", value)) renderer = TemplateRenderer()