bmc_hub/app/workflows/engine.py

"""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}
    time_coverage_trigger = trigger in ('time.created', 'time.updated') or (
        context.get('entity_type') in ('time', 'time_entry') and trigger in ('tags.added', 'tags.removed', 'tags.changed'))
    if selected is None and not stopped and not context.get('target_workflow_id') and time_coverage_trigger:
        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}