183 lines
13 KiB
Python
183 lines
13 KiB
Python
"""Pure deterministic evaluator. This module performs no I/O."""
|
|
from collections import defaultdict
|
|
from decimal import Decimal, ROUND_CEILING
|
|
from datetime import date, datetime
|
|
from .registry import FIELDS, OPERATORS, TRIGGERS, ACTIONS, METHODS
|
|
|
|
def coverage_allows(resource, context):
|
|
"""Empty legacy coverage is unrestricted; exclusions always win."""
|
|
coverage = resource.get('workflow_coverage') or {}
|
|
if not isinstance(coverage,dict): return False
|
|
tags = set(context.get('tag_ids',context.get('tags',[]))) if coverage.get('include_descendants',False) else set(context.get('concrete_tag_ids',context.get('tag_ids',context.get('tags',[]))))
|
|
if tags.intersection(coverage.get('excluded_tag_ids') or []): return False
|
|
required = coverage.get('tag_ids') or []
|
|
return not required or bool(tags.intersection(required))
|
|
|
|
def automatic_coverage_choices(context):
|
|
"""Compile saved resource coverage into billing choices without duplicate rules."""
|
|
choices = []
|
|
for pool, method, reference in (('agreements', 'subscription', 'agreement_id'), ('cards', 'prepaid', 'card_id')):
|
|
for resource in context.get(pool, []):
|
|
coverage = resource.get('workflow_coverage') or {}
|
|
if not isinstance(coverage, dict) or coverage.get('auto_billing') is False:
|
|
continue
|
|
configured = coverage.get('auto_billing') is True or bool(coverage.get('tag_ids') or coverage.get('excluded_tag_ids'))
|
|
if configured and coverage_allows(resource, context):
|
|
choices.append({'type': 'set_billing', 'method': method, reference: resource['id'],
|
|
'source': 'resource_coverage', 'resource_type': 'agreement' if pool == 'agreements' else 'card'})
|
|
return choices
|
|
|
|
def validate_definition(definition):
|
|
if not isinstance(definition, dict) or definition.get('trigger') not in TRIGGERS:
|
|
raise ValueError('Unknown workflow trigger')
|
|
branches = definition.get('branches')
|
|
if 'stop_after_match' in definition and not isinstance(definition['stop_after_match'],bool):
|
|
raise ValueError('Stop after match requires a boolean')
|
|
if not isinstance(branches, list) or not branches or len(branches) > 100:
|
|
raise ValueError('Workflow requires 1-100 branches')
|
|
def validate_condition(node, depth=0):
|
|
if depth > 30 or not isinstance(node, dict):
|
|
raise ValueError('Invalid or excessively nested condition')
|
|
groups = [key for key in ('all', 'any') if key in node]
|
|
if groups:
|
|
if len(groups) != 1 or len(node) != 1 or not isinstance(node[groups[0]], list) or not node[groups[0]]:
|
|
raise ValueError('Condition group requires children')
|
|
for child in node[groups[0]]: validate_condition(child, depth + 1)
|
|
elif node.get('field') not in FIELDS or node.get('operator') not in OPERATORS:
|
|
raise ValueError('Unknown condition field or operator')
|
|
elif node['operator'] in ('in','not_in') and not isinstance(node.get('value'),list):
|
|
raise ValueError('Membership comparison requires a list')
|
|
elif 'include_descendants' in node and not isinstance(node['include_descendants'],bool):
|
|
raise ValueError('Include descendants requires a boolean')
|
|
elif node['operator'] in ('gt','gte','lt','lte') and node['field']=='worked_date':
|
|
try: date.fromisoformat(node['value'])
|
|
except (TypeError,ValueError,KeyError): raise ValueError('Date comparison requires an ISO date')
|
|
elif node['operator'] in ('gt','gte','lt','lte') and (not isinstance(node.get('value'),(float,int)) or isinstance(node.get('value'),bool)):
|
|
raise ValueError('Numeric comparison requires a number')
|
|
for index, branch in enumerate(branches):
|
|
if not isinstance(branch, dict): raise ValueError('Invalid branch')
|
|
condition = branch.get('condition')
|
|
if condition is None:
|
|
if index != len(branches) - 1: raise ValueError('ELSE must be last')
|
|
else: validate_condition(condition)
|
|
actions = branch.get('actions')
|
|
if not isinstance(actions, list) or not actions: raise ValueError('Branch requires actions')
|
|
for action in actions:
|
|
if not isinstance(action, dict) or action.get('type') not in ACTIONS: raise ValueError('Unknown action')
|
|
if action['type'] == 'set_billing' and action.get('method') not in METHODS: raise ValueError('Unknown billing method')
|
|
if action['type'] == 'set_field' and action.get('field') not in ('priority','status'): raise ValueError('Only case priority/status may be changed')
|
|
if action['type'] == 'set_field' and action.get('field') == 'priority' and action.get('value') not in ('low','normal','high','urgent'): raise ValueError('Invalid case priority')
|
|
if action['type'] == 'set_field' and action.get('field') == 'status' and (not isinstance(action.get('value'),str) or not action['value'].strip()): raise ValueError('Invalid status')
|
|
for kind,key in (('assign_user','user_id'),('assign_group','group_id'),('start_workflow','workflow_id'),('add_tag','tag_id'),('remove_tag','tag_id'),('suggest_tag','tag_id')):
|
|
if action['type']==kind and (not isinstance(action.get(key),int) or isinstance(action.get(key),bool) or action[key]<=0): raise ValueError('Action requires a positive '+key)
|
|
if action['type'] in ('create_task','create_reminder','notify') and (not isinstance(action.get('title'),str) or not action['title'].strip()): raise ValueError('Action requires a title')
|
|
if action['type'] in ('create_reminder','notify') and (not isinstance(action.get('recipient_user_ids'),list) or not action['recipient_user_ids'] or any(not isinstance(v,int) or isinstance(v,bool) or v<=0 for v in action['recipient_user_ids'])): raise ValueError('Notification requires recipients')
|
|
return definition
|
|
|
|
def matches(node, context):
|
|
if node is None: return True
|
|
if 'all' in node: return all(matches(child, context) for child in node['all'])
|
|
if 'any' in node: return any(matches(child, context) for child in node['any'])
|
|
actual, expected, op = context.get(node['field']), node.get('value'), node['operator']
|
|
if node['field'] in ('tags','tag_ids') and not node.get('include_descendants',False):
|
|
actual=context.get('concrete_tag_ids',actual)
|
|
if node['field']=='customer_id':
|
|
actual=str(actual) if actual is not None else None
|
|
if op in ('in','not_in'):
|
|
expected=[str(value) if value is not None else None for value in expected]
|
|
elif op in ('eq','ne'):
|
|
expected=str(expected) if expected is not None else None
|
|
try:
|
|
if op == 'exists': return (actual is not None) == bool(expected if expected is not None else True)
|
|
if op == 'eq': return actual == expected
|
|
if op == 'ne': return actual != expected
|
|
if op in ('contains', 'not_contains'):
|
|
found = expected.casefold() in actual.casefold() if isinstance(actual, str) and isinstance(expected, str) else expected in (actual or [])
|
|
return found if op == 'contains' else not found
|
|
if op in ('in', 'not_in'):
|
|
found = actual in expected
|
|
return found if op == 'in' else not found
|
|
if actual is None or expected is None: return False
|
|
return {'gt': lambda: actual > expected, 'gte': lambda: actual >= expected, 'lt': lambda: actual < expected, 'lte': lambda: actual <= expected}[op]()
|
|
except (TypeError, ValueError, KeyError): return False
|
|
|
|
def explain_condition(node, context):
|
|
if node is None: return {'kind':'else','matched':True}
|
|
if 'all' in node or 'any' in node:
|
|
key = 'all' if 'all' in node else 'any'
|
|
children = [explain_condition(child,context) for child in node[key]]
|
|
matched = all(child['matched'] for child in children) if key=='all' else any(child['matched'] for child in children)
|
|
return {'kind':key,'matched':matched,'children':children}
|
|
actual=context.get(node['field'])
|
|
if node['field'] in ('tags','tag_ids') and not node.get('include_descendants',False): actual=context.get('concrete_tag_ids',actual)
|
|
return {'kind':'comparison','field':node['field'],'operator':node['operator'],'expected':node.get('value'),'actual':actual,'matched':matches(node,context)}
|
|
|
|
def evaluate(workflows, context, trigger):
|
|
if context.get('customer_mapping_missing'):
|
|
return {'status':'manual_review','method':None,'reason':'missing_customer_mapping','actions':[],'trace':[]}
|
|
trace, actions, selected = [], [], None
|
|
stopped = False
|
|
groups = defaultdict(list)
|
|
for workflow in workflows:
|
|
if context.get('target_workflow_id') and str(workflow.get('id')) != str(context['target_workflow_id']): continue
|
|
scope = workflow.get('customer_id')
|
|
if scope is not None and str(scope) != str(context.get('customer_id')): continue
|
|
if workflow.get('definition', {}).get('trigger') != trigger: continue
|
|
groups[(0 if scope is not None else 1, int(workflow.get('priority', 100)))].append(workflow)
|
|
for group in sorted(groups):
|
|
candidates, reviews, stop = [], [], False
|
|
for workflow in sorted(groups[group], key=lambda row: row.get('id', 0)):
|
|
definition = validate_definition(workflow['definition'])
|
|
for index, branch in enumerate(definition['branches']):
|
|
condition_trace = explain_condition(branch.get('condition'),context)
|
|
matched = condition_trace['matched']
|
|
trace.append({'workflow_id': workflow.get('id'), 'version_id': workflow.get('version_id'), 'branch': index, 'matched': matched,'condition':condition_trace})
|
|
if matched:
|
|
for action in branch['actions']:
|
|
entry = dict(action,source_workflow_id=workflow.get('id'),version_id=workflow.get('version_id'))
|
|
if action['type']!='start_workflow': entry['workflow_id']=workflow.get('id')
|
|
actions.append(entry)
|
|
if action['type'] == 'set_billing': candidates.append(entry)
|
|
if action['type'] == 'send_to_review': reviews.append(entry)
|
|
stop = stop or definition.get('stop_after_match', False)
|
|
break
|
|
if reviews:
|
|
return {'status':'manual_review','method':None,'reason':'explicit_manual_review','review_reason':reviews[0].get('reason'),'actions':actions,'trace':trace}
|
|
if candidates:
|
|
keys = {(a.get('method'), str(a.get('agreement_id')), str(a.get('card_id')), a.get('product_number')) for a in candidates}
|
|
if len(keys) > 1: return {'status': 'manual_review', 'method': None, 'reason': 'conflicting_rules', 'actions': actions, 'trace': trace}
|
|
selected = candidates[0]
|
|
break
|
|
if stop:
|
|
stopped = True
|
|
break
|
|
if context.get('pending_billing_tag_suggestions') or any(action['type']=='suggest_tag' for action in actions):
|
|
return {'status': 'manual_review', 'method': None, 'reason': 'pending_tag_approval', 'actions': actions, 'trace': trace}
|
|
if selected is None and not stopped and not context.get('target_workflow_id') and trigger in ('time.created', 'time.updated'):
|
|
automatic = automatic_coverage_choices(context)
|
|
trace.append({'source': 'resource_coverage', 'matched': bool(automatic), 'choices': automatic})
|
|
if len(automatic) > 1:
|
|
return {'status': 'manual_review', 'method': None, 'reason': 'ambiguous_resource_coverage', 'actions': actions, 'trace': trace}
|
|
if automatic:
|
|
selected = automatic[0]
|
|
actions.append(selected)
|
|
if selected:
|
|
if selected['method'] in ('subscription','prepaid'):
|
|
pool_key, ref_key = ('agreements','agreement_id') if selected['method']=='subscription' else ('cards','card_id')
|
|
candidates = [item for item in context.get(pool_key,[]) if coverage_allows(item,context)]
|
|
reference = selected.get(ref_key)
|
|
eligible = [item for item in candidates if reference is None or str(item['id'])==str(reference)]
|
|
if len(eligible)!=1:
|
|
return {'status':'manual_review','method':None,'reason':'ambiguous_or_missing_'+pool_key,'actions':actions,'trace':trace}
|
|
selected = dict(selected,**{ref_key:eligible[0]['id']})
|
|
if selected['method']=='prepaid':
|
|
hours=Decimal(str(context.get('hours') or 0))
|
|
rounding=int(eligible[0].get('rounding_minutes') or 0)
|
|
if rounding>0:
|
|
interval=Decimal(rounding)/Decimal(60)
|
|
hours=(hours/interval).to_integral_value(rounding=ROUND_CEILING)*interval
|
|
if Decimal(str(eligible[0].get('remaining_hours') or 0))<hours:
|
|
return {'status':'manual_review','method':None,'reason':'insufficient_card_balance','actions':actions,'trace':trace}
|
|
return {'status': 'decided', 'method': selected['method'], 'reason': 'resource_coverage' if selected.get('source') == 'resource_coverage' else 'matched_rule', 'decision': selected, 'actions': actions, 'trace': trace}
|
|
return {'status': 'manual_review', 'method': None, 'reason': 'no_matching_rule', 'actions': actions, 'trace': trace}
|