Allow using webhooks for notifications

core/lib/rules.py

@@ -83,29 +83,63 @@ class NotificationRuleData(object):
         self.object.save()
         log.debug(f"Stored match: {index} - {match}")
 
+    def get_match(self, index):
+        """
+        Get a match result for an index.
+        """
+        if self.object.match is None:
+            return None
+        if not isinstance(self.object.match, dict):
+            return None
+
+        return self.object.match.get(index)
+
+    def format_aggs(self, aggs):
+        """
+        Format aggregations for the query.
+        We have self.aggs, which contains:
+        {"avg_sentiment": (">", 0.5)}
+        and aggs, which contains:
+        {"avg_sentiment": {"value": 0.6}}
+        It's matched already, we just need to format it like so:
+        {"avg_sentiment": "0.06>0.5"}
+        """
+        new_aggs = {}
+        for agg_name, agg in aggs.items():
+            # Already checked membership below
+            op, value = self.aggs[agg_name]
+            new_aggs[agg_name] = f"{agg['value']}{op}{value}"
+
+        return new_aggs
+
     async def run_schedule(self):
         """
         Run the schedule query.
         """
-        if self.db:
-            response = await self.db.schedule_query_results(self)
-            for index, (aggs, results) in response.items():
-                if not results:
-                    self.store_match(index, False)
-
-                aggs_for_index = []
-                for agg_name in self.aggs.keys():
-                    if agg_name in aggs:
-                        if "match" in aggs[agg_name]:
-                            aggs_for_index.append(aggs[agg_name]["match"])
-
-                # All required aggs are present
-                if len(aggs_for_index) == len(self.aggs.keys()):
-                    if all(aggs_for_index):
-                        self.store_match(index, True)
-                        continue
+        response = await self.db.schedule_query_results(self)
+        for index, (aggs, results) in response.items():
+            if not results:
                 self.store_match(index, False)
 
+            aggs_for_index = []
+            for agg_name in self.aggs.keys():
+                if agg_name in aggs:
+                    if "match" in aggs[agg_name]:
+                        aggs_for_index.append(aggs[agg_name]["match"])
+
+            # All required aggs are present
+            if len(aggs_for_index) == len(self.aggs.keys()):
+                if all(aggs_for_index):
+                    # Ensure we only send notifications when the previous run
+                    # did not return any matches
+                    current_match = self.get_match(index)
+                    if current_match is False:
+                        formatted_aggs = self.format_aggs(aggs)
+                        rule_matched(self.object, results[:5], formatted_aggs)
+                    self.store_match(index, True)
+                    continue
+            self.store_match(index, False)
+
     def test_schedule(self):
         """
         Test the schedule query to ensure it is valid.