template_resolver.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162
  1. # core/utils/template_resolver.py
  2. import re
  3. from typing import Any, Dict
  4. PLACEHOLDER_PATTERN = re.compile(r"\{\{\s*([a-zA-Z0-9_.]+)(\|[^}]+)?\s*\}\}")
  5. def resolve_request_body_template(
  6. data: Any,
  7. variables: Dict[str, Any] = {},
  8. ) -> Any:
  9. """
  10. 仅解析请求参数中 {{var}} 模板
  11. Args:
  12. data: dict/list/str
  13. variables: {"next_cursor": "123", ...}
  14. Returns:
  15. 替换后的结构
  16. :rtype: Any
  17. """
  18. if isinstance(data, dict):
  19. return {k: resolve_request_body_template(v, variables) for k, v in data.items()}
  20. elif isinstance(data, list):
  21. return [resolve_request_body_template(item, variables) for item in data]
  22. elif isinstance(data, str):
  23. def replacer(match):
  24. var_name = match.group(1)
  25. default_value = match.group(2)[1:] if match.group(2) else None
  26. value = variables.get(var_name)
  27. if value is not None:
  28. return str(value)
  29. elif default_value is not None:
  30. return default_value
  31. else:
  32. return ""
  33. return PLACEHOLDER_PATTERN.sub(replacer, data)
  34. else:
  35. return data
  36. if __name__ == '__main__':
  37. data = {
  38. "cursor": "{{next_cursor}}",
  39. "page_size": 20,
  40. "filters": {
  41. "start_date": "{{start_date}}",
  42. "end_date": "{{end_date|2025-01-01}}"
  43. },
  44. "tags": ["{{tag1}}", "{{tag2|default_tag}}"]
  45. }
  46. variables = {
  47. # "dada":{"next_cursor":"1"},
  48. # "start_date": "2025-06-30",
  49. # "tag1": "news",
  50. # "tag3": "news",
  51. # "default_tag55": "1111",
  52. }
  53. result = resolve_request_body_template(data, variables)
  54. print(result)