File size: 2,779 Bytes
921d377 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """
Personalization rule DSL.
Rules are persisted in ``ix_personalization_rules`` with two JSON
fields:
condition = a dict describing the match criteria
action = a dict describing what the rule does
Supported condition keys (phase 1):
role str viewer role must equal
level str viewer level must equal
language str viewer language must equal
country str viewer country must equal
has_tag str viewer tags must contain
mood str character_mood must equal
min_affinity float character affinity_score >= value
max_affinity float character affinity_score <= value
metric dict { scheme, key, min?, max? } — progress metric in range
Supported action keys (phase 1):
route_to_node str override the next_node_id with this
prefer_tone str override the tone for this turn
bump_affinity float delta to apply to affinity_score
A rule is 'applicable' if ALL conditions pass. The evaluator picks
the highest-priority applicable rule (lowest priority wins ties).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
_ALLOWED_CONDITION_KEYS = {
"role", "level", "language", "country", "has_tag",
"mood", "min_affinity", "max_affinity", "metric",
}
_ALLOWED_ACTION_KEYS = {
"route_to_node", "prefer_tone", "bump_affinity",
}
@dataclass(frozen=True)
class RuleCondition:
"""Resolved condition object."""
raw: Dict[str, Any]
def get(self, key: str, default: Any = None) -> Any:
return self.raw.get(key, default)
@dataclass(frozen=True)
class Rule:
"""A personalization rule, ready for the evaluator."""
id: str
name: str
condition: RuleCondition
action: Dict[str, Any]
priority: int = 100
enabled: bool = True
def validate_rule(condition: Dict[str, Any], action: Dict[str, Any]) -> List[str]:
"""Return a list of problems (empty list = valid)."""
problems: List[str] = []
for k in condition:
if k not in _ALLOWED_CONDITION_KEYS:
problems.append(f"unknown condition key: {k}")
for k in action:
if k not in _ALLOWED_ACTION_KEYS:
problems.append(f"unknown action key: {k}")
metric = condition.get("metric") if isinstance(condition.get("metric"), dict) else None
if metric is not None:
if not metric.get("scheme") or not metric.get("key"):
problems.append("metric condition requires 'scheme' and 'key'")
if "bump_affinity" in action:
try:
float(action["bump_affinity"])
except (TypeError, ValueError):
problems.append("bump_affinity must be numeric")
return problems
|