import math from decimal import Decimal from .analysis import METRICS RESERVED_DATA_HEADERS = { "period", "customertoken", "revenue", "cost", "grossinput", "marginrate", "grosssource", "listamount", "selfoperated", "customerpresent", "rowcount", "grosscalc", } def available_metrics(ds): roles = {column["role"] for column in ds.mapping["columns"]} dimensions = {column.get("key") for column in ds.mapping["columns"] if column["role"] == "dimension"} products = bool(ds.mapping.get("self_operated_products")) and "product" in dimensions gross = "gross" in roles or (bool(ds.mapping.get("margin_rates")) and "product" in dimensions) customer, list_amount, cost = "customer" in roles, "list_amount" in roles, "cost" in roles available = {"revenue"} if cost: available.add("cost") if gross: available.update(("gross", "margin")) if customer: available.update(("customers", "basket")) if list_amount: available.add("discount_rate") if customer and list_amount: available.add("list_basket") if cost and (ds.mapping.get("roi_formula") == "profit_over_cost" or gross): available.add("roi") if products: available.add("self_revenue") if customer: available.add("upgrade_rate") return {key: value for key, value in METRICS.items() if key in available} OPERATORS = {"add": "+", "subtract": "-", "multiply": "*", "divide": "/"} def validate_column(value): if not isinstance(value, dict) or set(value) != {"name", "meaning", "scope", "missing", "expression"}: raise ValueError("自定义列定义字段无效") for key, limit in (("name", 80), ("meaning", 500), ("scope", 200), ("missing", 300)): text = value[key] if not isinstance(text, str) or not text.strip() or len(text) > limit: raise ValueError(f"自定义列{key}说明不完整") if value["missing"] != "结果不可用(#N/A);零分母不可用": raise ValueError("暂不支持自定义缺失值策略;请选择结果不可用(#N/A)及零分母不可用") _validate_expression(value["expression"]) return value def _validate_expression(node, depth=0): if depth > 8 or not isinstance(node, dict): raise ValueError("自定义计算公式结构无效") if set(node) == {"metric"}: if not isinstance(node["metric"], str) or node["metric"] not in METRICS: raise ValueError("公式引用了不支持的指标") return if set(node) == {"number"}: if ( type(node["number"]) not in (int, float) or not math.isfinite(node["number"]) or abs(node["number"]) > 1e12 ): raise ValueError("公式常数无效") return if ( set(node) == {"operator", "left", "right"} and isinstance(node["operator"], str) and node["operator"] in OPERATORS ): _validate_expression(node["left"], depth + 1) _validate_expression(node["right"], depth + 1) return raise ValueError("公式包含不支持的字段或运算") def metrics_used(node): if "metric" in node: return {node["metric"]} if "number" in node: return set() return metrics_used(node["left"]) | metrics_used(node["right"]) def ensure_available_metrics(ds, column): unavailable = metrics_used(column["expression"]) - available_metrics(ds).keys() if unavailable: names = "、".join(METRICS[key] for key in sorted(unavailable)) raise ValueError(f"当前数据集未映射计算“{names}”所需字段,不能生成此自定义列") def to_excel(node, references): if "metric" in node: return references[node["metric"]] if "number" in node: return str(node["number"]) return f"({to_excel(node['left'], references)}{OPERATORS[node['operator']]}{to_excel(node['right'], references)})" def to_decimal(node, values): """按已校验的白名单表达式树用 Decimal 计算服务器参考值。 只遍历 metric/number/四则运算节点,不执行任何代码;依赖指标缺失或除零时 返回 None,与 Excel 侧 #N/A 的语义一致。 """ if "metric" in node: return values[node["metric"]] if "number" in node: return Decimal(str(node["number"])) left, right = to_decimal(node["left"], values), to_decimal(node["right"], values) if left is None or right is None: return None if node["operator"] == "add": return left + right if node["operator"] == "subtract": return left - right if node["operator"] == "multiply": return left * right if right == 0: return None return left / right