feat: add pricing display API for customer-facing pricing data

- generate_formula_from_factors(): auto-generate formula from price_factors array
- get_pricing_display(ppid): return structured human-readable pricing data
- wwwroot/api/get_pricing_display.dspy: API endpoint for frontend consumption
- Supports price_factors display layer (label, unit_price, unit_label)
- Backward compatible: old YAML without pricing_type/price_factors works
- Registered via load_pricing() with ServerEnv

pricing/__init__.py

@@ -0,0 +1,6 @@
+from pricing.pricing import (
+    PricingProgram,
+    test_pricing,
+    generate_formula_from_factors,
+    get_pricing_display,
+)

pricing/init.py

@@ -2,7 +2,9 @@ from appPublic.log import debug
 from sqlor.dbpools import DBPools
 from pricing.pricing import (
 	PricingProgram,
-	test_pricing
+	test_pricing,
+	generate_formula_from_factors,
+	get_pricing_display
 )
 from ahserver.serverenv import ServerEnv
 
@@ -33,6 +35,8 @@ def load_pricing():
 	env.load_pricing_data = PricingProgram.load_pricing_data
 	env.get_pricing_program = PricingProgram.get_pricing_program
 	env.test_pricing = test_pricing
+	env.generate_formula_from_factors = generate_formula_from_factors
+	env.get_pricing_display = get_pricing_display
 	# Bind hot_reload event — only when running in ahserver (event_dispatcher available)
 	if getattr(env, 'event_dispatcher', None) is not None:
 		env.event_dispatcher.bind('hot_reload', PricingProgram.on_hot_reload)

pricing/pricing.py

@@ -606,6 +606,110 @@ order by b.enabled_date desc"""
 			raise Exception(e)
 		return ret_items
 
+def generate_formula_from_factors(price_factors):
+	"""从 price_factors 数组自动生成 formula 字符串。
+
+	price_factors 格式:
+	[
+		{"factor": "prompt_tokens", "unit_price": 3.2, "unit_label": "元/百万Token"},
+		{"factor": "completion_tokens", "unit_price": 16.0, "unit_label": "元/百万Token"}
+	]
+	返回: "(3.2 * float(prompt_tokens) + 16.0 * float(completion_tokens)) / 1000000.0"
+	"""
+	if not price_factors:
+		return None
+	parts = []
+	has_million = False
+	for f in price_factors:
+		factor = f.get('factor', '')
+		unit_price = f.get('unit_price', 0)
+		unit_label = f.get('unit_label', '')
+		# 判断单位是否是"百万"级别（元/百万Token 等）
+		if '百万' in unit_label:
+			has_million = True
+		parts.append(f'{unit_price} * float({factor})')
+	formula = ' + '.join(parts)
+	if has_million:
+		formula = f'({formula}) / 1000000.0'
+	return formula
+
+
+async def get_pricing_display(ppid):
+	"""获取定价项目的可读定价数据，供客户/前端展示。
+
+	返回结构:
+	{
+		"ppid": "...",
+		"name": "...",
+		"pricing_type": "per_use",
+		"items": [
+			{
+				"filters": {"model": "doubao-seed-2-0-pro"},
+				"filter_labels": {"模型": "doubao-seed-2-0-pro"},
+				"price_factors": [...],
+				"formula": "..."
+			}
+		]
+	}
+	"""
+	r = await PricingProgram.get_ppid_pricing(ppid)
+	pricing_data_str = r.pricing_data
+	d = yaml.safe_load(pricing_data_str)
+	if not d:
+		raise Exception(f'{ppid} pricing_data 为空')
+
+	fields = d.get('fields', {})
+	pricings = d.get('pricings', [])
+	pricing_type = d.get('pricing_type', 'per_use')
+
+	items = []
+	for p in pricings:
+		# 提取过滤条件（role=filter 或无 role 的字段）
+		filters = {}
+		filter_labels = {}
+		for k, v in p.items():
+			if k in ('formula', 'price_factors', 'name'):
+				continue
+			fdef = fields.get(k, {})
+			role = fdef.get('role', 'filter') if isinstance(fdef, dict) else 'filter'
+			if role == 'filter':
+				filters[k] = v
+				label = fdef.get('label', k) if isinstance(fdef, dict) else k
+				filter_labels[label] = v
+
+		# 提取 price_factors（展示层）
+		price_factors = p.get('price_factors', None)
+		if not price_factors:
+			# fallback: 从 fields 中构建展示信息
+			price_factors = []
+			for k, fdef in fields.items():
+				if not isinstance(fdef, dict):
+					continue
+				role = fdef.get('role', 'filter')
+				if role == 'factor' or fdef.get('type') == 'float':
+					label = fdef.get('label', k)
+					price_factors.append({
+						'factor': k,
+						'label': label,
+						'unit_price': None,
+						'unit_label': ''
+					})
+
+		items.append({
+			'filters': filters,
+			'filter_labels': filter_labels,
+			'price_factors': price_factors,
+			'formula': p.get('formula', '')
+		})
+
+	return {
+		'ppid': ppid,
+		'name': getattr(r, 'name', ''),
+		'pricing_type': pricing_type,
+		'items': items
+	}
+
+
 async def get_pricing_program_timeing(pptid):
 	env = ServerEnv()
 	async with get_sor_context(env, 'pricing') as sor:

wwwroot/api/get_pricing_display.dspy

@@ -0,0 +1,10 @@
+ppid = params_kw.get('ppid')
+if not ppid:
+    return json.dumps({"status": "error", "message": "ppid parameter required"}, ensure_ascii=False)
+
+try:
+    result = await get_pricing_display(ppid)
+    return json.dumps({"status": "ok", "data": result}, ensure_ascii=False, default=str)
+except Exception as e:
+    exception(f'get_pricing_display({ppid}) failed: {e}\n{format_exc()}')
+    return json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False)