import abc import functools import json import logging import dotmap from src import suspicions logger = logging.getLogger("oncall_master.verifier") class Verifier(abc.ABC): def __init__(self): self.solutions: list[suspicions.Solution] = [] self.info = {} def verify(self, ctx: dotmap.DotMap, suspicion_types: list[type(suspicions.SuspicionABC)]) -> bool: found = False for tp in suspicion_types or []: if self.verify_one_suspicion(ctx, tp): found = True return found def verify_and_output(self, ctx: dotmap.DotMap, suspicion_types: list[type(suspicions.SuspicionABC)]): self.verify(ctx, suspicion_types) self.output_result() def verify_one_suspicion(self, ctx: dotmap.DotMap, suspicion_cls: type(suspicions.SuspicionABC)): if isinstance(suspicion_cls, type): # class suspicion = suspicion_cls(ctx=ctx) elif issubclass(type(suspicion_cls), suspicions.SuspicionABC): # instance suspicion = suspicion_cls elif isinstance(suspicion_cls, functools.partial): # partial func suspicion = suspicion_cls(ctx=ctx) else: raise ValueError(f"Invalid type {suspicion_cls}") if not suspicion.verify(): logger.info(f"{suspicion.__class__.__name__} PASS") self.record_info(suspicion) return self.record_solution(suspicion) found_in_children = False for cls in suspicion.next_suspicion_clss or []: if self.verify_one_suspicion(ctx, cls): found_in_children = True # Found return True def record_solution(self, suspicion: suspicions.SuspicionABC): sol = suspicions.Solution(suspicion.__class__.__name__, suspicion.solutions) self.solutions.append(sol) self.info.update(suspicion.info) def record_info(self, suspicion: suspicions.SuspicionABC): self.info.update(suspicion.info) def output_result(self): if len(self.solutions) == 0: logger.info("No solution found") return logger.info(f"Found {len(self.solutions)} solutions") removed_cr = {k: v for k, v in self.info.items() if not k.startswith("cr.")} cr = {k: v for k, v in self.info.items() if k.startswith("cr.")} # logger.info(f"cr: {json.dumps(cr, indent=2)}") for solution in self.solutions: logger.info(f"{solution.suspicion} -> {solution.solutions}") logger.info(f"info: {json.dumps(removed_cr, indent=2)}")