Test User
256 lines
9.4 KiB
Python
256 lines
9.4 KiB
Python
"""
|
|
Self-contained rules processor
|
|
Processes JSON-based rules without hardcoded logic
|
|
"""
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import List, Tuple, Dict, Any
|
|
|
|
|
|
class RulesProcessor:
|
|
def __init__(self, rules_dir: Path):
|
|
self.rules_dir = rules_dir
|
|
self.rules = self._load_rules()
|
|
|
|
def _load_rules(self) -> Dict[str, List[dict]]:
|
|
"""Load all rule files and organize by field"""
|
|
rules_by_field = {}
|
|
|
|
if self.rules_dir.exists():
|
|
for rule_file in self.rules_dir.glob('*.json'):
|
|
with open(rule_file, 'r', encoding='utf-8') as f:
|
|
rule = json.load(f)
|
|
field = rule.get('field', '*')
|
|
|
|
if field not in rules_by_field:
|
|
rules_by_field[field] = []
|
|
|
|
rules_by_field[field].append(rule)
|
|
|
|
# Sort rules by priority (lower number = higher priority)
|
|
for field in rules_by_field:
|
|
rules_by_field[field].sort(key=lambda r: r.get('priority', 100))
|
|
|
|
return rules_by_field
|
|
|
|
def apply_rules(self, frontmatter: dict) -> Tuple[dict, bool, List[str]]:
|
|
"""
|
|
Apply all rules to frontmatter
|
|
Returns: (modified_fm, is_conformant, violations)
|
|
"""
|
|
if not frontmatter:
|
|
return frontmatter, True, []
|
|
|
|
modified = dict(frontmatter)
|
|
violations = []
|
|
|
|
# Process global rules (field: "*") first
|
|
if '*' in self.rules:
|
|
for rule in self.rules['*']:
|
|
modified, rule_violations = self._apply_rule(modified, rule, None)
|
|
violations.extend(rule_violations)
|
|
|
|
# Process rename rules to create new fields
|
|
for field in list(modified.keys()):
|
|
if field in self.rules:
|
|
for rule in self.rules[field]:
|
|
if rule.get('action') == 'rename_field':
|
|
modified, rule_violations = self._apply_rule(modified, rule, field)
|
|
violations.extend(rule_violations)
|
|
|
|
# Process other field-specific rules on current fields
|
|
for field in list(modified.keys()):
|
|
if field in self.rules:
|
|
for rule in self.rules[field]:
|
|
if rule.get('action') != 'rename_field': # Skip renames, already done
|
|
modified, rule_violations = self._apply_rule(modified, rule, field)
|
|
violations.extend(rule_violations)
|
|
|
|
is_conformant = len(violations) == 0
|
|
return modified, is_conformant, violations
|
|
|
|
def _apply_rule(self, frontmatter: dict, rule: dict, field: str = None) -> Tuple[dict, List[str]]:
|
|
"""Apply a single rule to frontmatter"""
|
|
modified = dict(frontmatter)
|
|
violations = []
|
|
action = rule.get('action', 'validate')
|
|
|
|
if action == 'normalize_keys':
|
|
modified, violated = self._action_normalize_keys(modified, rule)
|
|
if violated:
|
|
violations.append(rule.get('name', 'unknown'))
|
|
|
|
elif action == 'rename_field':
|
|
modified, violated = self._action_rename_field(modified, rule)
|
|
if violated:
|
|
violations.append(rule.get('name', 'unknown'))
|
|
|
|
elif action == 'normalize_value':
|
|
if field and field in modified:
|
|
modified[field], violated = self._action_normalize_value(modified[field], rule)
|
|
if violated:
|
|
violations.append(rule.get('name', 'unknown'))
|
|
|
|
elif action == 'validate':
|
|
if field and field in modified:
|
|
violated = not self._action_validate(modified[field], rule)
|
|
if violated:
|
|
violations.append(rule.get('name', 'unknown'))
|
|
|
|
return modified, violations
|
|
|
|
def _action_normalize_keys(self, frontmatter: dict, rule: dict) -> Tuple[dict, bool]:
|
|
"""Normalize all keys using regex pattern or transform"""
|
|
pattern = rule.get('pattern', '')
|
|
replacement = rule.get('replacement', '')
|
|
transform = rule.get('transform', '')
|
|
|
|
modified = {}
|
|
violated = False
|
|
|
|
for key, value in frontmatter.items():
|
|
new_key = key
|
|
|
|
# Apply transform (e.g., lower, upper)
|
|
if transform == 'lower':
|
|
new_key = key.lower()
|
|
elif transform == 'upper':
|
|
new_key = key.upper()
|
|
elif pattern:
|
|
# Apply regex pattern
|
|
new_key = re.sub(pattern, replacement, key, flags=re.MULTILINE if rule.get('multiline') else 0)
|
|
|
|
if new_key != key:
|
|
violated = True
|
|
modified[new_key] = value
|
|
|
|
return modified, violated
|
|
|
|
def _action_rename_field(self, frontmatter: dict, rule: dict) -> Tuple[dict, bool]:
|
|
"""Rename a specific field"""
|
|
from_field = rule.get('from')
|
|
to_field = rule.get('to')
|
|
|
|
if not from_field or not to_field:
|
|
return frontmatter, False
|
|
|
|
modified = dict(frontmatter)
|
|
|
|
if from_field in modified:
|
|
modified[to_field] = modified.pop(from_field)
|
|
return modified, True
|
|
|
|
return modified, False
|
|
|
|
def _action_normalize_value(self, value: Any, rule: dict) -> Tuple[Any, bool]:
|
|
"""Normalize a value using regex pattern(s) or transform"""
|
|
pattern = rule.get('pattern', '')
|
|
replacement = rule.get('replacement', '')
|
|
transform = rule.get('transform', '')
|
|
split_on = rule.get('split_on', '')
|
|
|
|
violated = False
|
|
|
|
# Handle comma-separated string first (convert to list)
|
|
if split_on and isinstance(value, str):
|
|
items = [item.strip() for item in value.split(split_on)]
|
|
normalized = []
|
|
for item in items:
|
|
if transform == 'lower':
|
|
normalized_item = item.lower()
|
|
elif transform == 'upper':
|
|
normalized_item = item.upper()
|
|
elif pattern:
|
|
normalized_item = re.sub(
|
|
pattern,
|
|
replacement,
|
|
item,
|
|
flags=re.MULTILINE if rule.get('multiline') else 0
|
|
)
|
|
else:
|
|
normalized_item = item
|
|
|
|
if normalized_item:
|
|
normalized.append(normalized_item)
|
|
violated = True # Format changed from string to list
|
|
return normalized, violated
|
|
|
|
# Handle list values (like tags)
|
|
elif isinstance(value, list):
|
|
normalized = []
|
|
for item in value:
|
|
item_str = str(item)
|
|
|
|
# Apply transform first
|
|
if transform == 'lower':
|
|
normalized_item = item_str.lower()
|
|
elif transform == 'upper':
|
|
normalized_item = item_str.upper()
|
|
elif pattern:
|
|
normalized_item = re.sub(
|
|
pattern,
|
|
replacement,
|
|
item_str,
|
|
flags=re.MULTILINE if rule.get('multiline') else 0
|
|
)
|
|
else:
|
|
normalized_item = item_str
|
|
|
|
if normalized_item != item_str:
|
|
violated = True
|
|
if normalized_item: # Only keep non-empty
|
|
normalized.append(normalized_item)
|
|
return normalized, violated
|
|
|
|
# Handle string values
|
|
elif isinstance(value, str):
|
|
if transform == 'lower':
|
|
normalized = value.lower()
|
|
elif transform == 'upper':
|
|
normalized = value.upper()
|
|
elif pattern:
|
|
normalized = re.sub(
|
|
pattern,
|
|
replacement,
|
|
value,
|
|
flags=re.MULTILINE if rule.get('multiline') else 0
|
|
)
|
|
else:
|
|
normalized = value
|
|
|
|
violated = normalized != value
|
|
return normalized, violated
|
|
|
|
return value, False
|
|
|
|
def _action_validate(self, value: Any, rule: dict) -> bool:
|
|
"""Validate a value against regex pattern"""
|
|
pattern = rule.get('pattern', '')
|
|
|
|
if not pattern:
|
|
return True
|
|
|
|
# Handle list values
|
|
if isinstance(value, list):
|
|
for item in value:
|
|
if not re.match(pattern, str(item), flags=re.MULTILINE if rule.get('multiline') else 0):
|
|
return False
|
|
return True
|
|
|
|
# Handle string values
|
|
if isinstance(value, str):
|
|
return bool(re.match(pattern, value, flags=re.MULTILINE if rule.get('multiline') else 0))
|
|
|
|
return True
|
|
|
|
def get_llm_prompt_for_field(self, field: str) -> str:
|
|
"""Get LLM prompt text for a field if defined in rules"""
|
|
if field in self.rules:
|
|
for rule in self.rules[field]:
|
|
llm_prompt = rule.get('llm_prompt')
|
|
if llm_prompt:
|
|
return llm_prompt
|
|
|
|
return ""
|