139 lines
5.3 KiB
Python
139 lines
5.3 KiB
Python
import re
|
|
from datetime import datetime
|
|
from django.db import models
|
|
|
|
def resolve_value(expr, context):
|
|
"""
|
|
Giải quyết các placeholder động (dynamic placeholder):
|
|
- Literal (số, boolean)
|
|
- Key trực tiếp (ví dụ: "customer_id")
|
|
- Đường dẫn (ví dụ: "transaction.id")
|
|
- Template chuỗi (ví dụ: "{customer_id}", "URL/{product_id}")
|
|
- Hàm hệ thống: $add(a, b), $sub(a, b), $now, $now_iso
|
|
"""
|
|
if expr is None:
|
|
return None
|
|
|
|
# Direct literal (int, float, bool)
|
|
if isinstance(expr, (int, float, bool)):
|
|
return expr
|
|
|
|
if not isinstance(expr, str):
|
|
return expr
|
|
|
|
expr = expr.strip()
|
|
|
|
# =============================================
|
|
# 1. Hỗ trợ thời gian hiện tại từ server
|
|
# =============================================
|
|
if expr == "$now":
|
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Format phù hợp với DateTimeField Django
|
|
|
|
if expr == "$today":
|
|
return datetime.now().strftime("%Y-%m-%d") # Chỉ lấy ngày, đúng format DateField
|
|
|
|
if expr == "$now_iso":
|
|
return datetime.now().isoformat(timespec='seconds') # 2025-12-21T14:30:45
|
|
|
|
# =============================================
|
|
# 2. Hàm toán học: $add(a, b), $sub(a, b)
|
|
# =============================================
|
|
func_match = re.match(r"^\$(add|sub)\(([^,]+),\s*([^)]+)\)$", expr)
|
|
if func_match:
|
|
func_name = func_match.group(1)
|
|
arg1_val = resolve_value(func_match.group(2).strip(), context)
|
|
arg2_val = resolve_value(func_match.group(3).strip(), context)
|
|
|
|
try:
|
|
num1 = float(arg1_val or 0)
|
|
num2 = float(arg2_val or 0)
|
|
if func_name == "add":
|
|
return num1 + num2
|
|
if func_name == "sub":
|
|
return num1 - num2
|
|
except (ValueError, TypeError):
|
|
print(f" [ERROR] Math function {func_name} failed with values: {arg1_val}, {arg2_val}")
|
|
return 0
|
|
|
|
# =============================================
|
|
# 3. Helper: Lấy giá trị từ context theo đường dẫn dotted
|
|
# =============================================
|
|
def get_context_value(key_path):
|
|
if "." not in key_path:
|
|
return context.get(key_path)
|
|
|
|
root, *rest = key_path.split(".")
|
|
val = context.get(root)
|
|
|
|
for r in rest:
|
|
if val is None:
|
|
return None
|
|
|
|
# 1. Xử lý truy cập index mảng, ví dụ: payment_plan[0]
|
|
array_match = re.match(r"(\w+)\[(\d+)\]", r)
|
|
if array_match:
|
|
attr_name = array_match.group(1)
|
|
index = int(array_match.group(2))
|
|
# Lấy list/queryset
|
|
val = getattr(val, attr_name, None) if not isinstance(val, dict) else val.get(attr_name)
|
|
try:
|
|
if hasattr(val, 'all'): # Django QuerySet/Manager
|
|
val = val[index]
|
|
else: # List thông thường
|
|
val = val[index]
|
|
except (IndexError, TypeError, KeyError):
|
|
return None
|
|
else:
|
|
# 2. Xử lý truy cập thuộc tính hoặc dict key
|
|
if isinstance(val, dict):
|
|
val = val.get(r)
|
|
else:
|
|
val = getattr(val, r, None)
|
|
|
|
# 3. Hỗ trợ tự động lấy bản ghi đầu tiên nếu là Manager (1-n)
|
|
if hasattr(val, 'all') and not isinstance(val, models.Model):
|
|
val = val.first()
|
|
|
|
return val
|
|
|
|
# =============================================
|
|
# 4. Xử lý placeholder kiểu {key} hoặc {obj.field}
|
|
# =============================================
|
|
pattern = re.compile(r"\{(\w+(\.\w+)*)\}")
|
|
if pattern.search(expr):
|
|
# Trường hợp toàn bộ expr là một placeholder duy nhất: {customer_id}
|
|
single_match = pattern.fullmatch(expr)
|
|
if single_match:
|
|
key = single_match.group(1)
|
|
return get_context_value(key)
|
|
|
|
# Trường hợp chuỗi có nhiều placeholder: "Hello {customer.name}"
|
|
def replace_match(match):
|
|
key = match.group(1)
|
|
val = get_context_value(key)
|
|
return str(val) if val is not None else ""
|
|
|
|
return pattern.sub(replace_match, expr)
|
|
|
|
# =============================================
|
|
# 5. Hỗ trợ $last_result và $last_result.field
|
|
# =============================================
|
|
if expr.startswith("$last_result"):
|
|
_, _, field = expr.partition(".")
|
|
last_res = context.get("last_result")
|
|
if not field:
|
|
return last_res
|
|
if last_res is None:
|
|
return None
|
|
return getattr(last_res, field, None) if not isinstance(last_res, dict) else last_res.get(field)
|
|
|
|
# =============================================
|
|
# 6. Dotted path trực tiếp: customer.name hoặc obj in context
|
|
# =============================================
|
|
if "." in expr or expr in context:
|
|
return get_context_value(expr)
|
|
|
|
# =============================================
|
|
# 7. Trả về nguyên expr nếu không match gì
|
|
# =============================================
|
|
return expr |