ai_notification

dtable_events/app/app.py

@@ -12,6 +12,7 @@
 from dtable_events.tasks.license_expiring_notices_sender import LicenseExpiringNoticesSender
 from dtable_events.notification_rules.handler import NotificationRuleHandler
 from dtable_events.notification_rules.dtable_notification_rules_scanner import DTableNofiticationRulesScanner
+from dtable_events.notification_rules.ai_notification_rules_scanner import AINotificationRulesScanner
 from dtable_events.automations.handler import AutomationRuleHandler
 from dtable_events.automations.dtable_automation_rules_scanner import DTableAutomationRulesScanner
 from dtable_events.webhook.webhook import Webhooker
@@ -46,6 +47,7 @@ def __init__(self, config, task_mode):
             self._dtables_cleaner = DTablesCleaner(config)
             self._dtable_updates_sender = DTableUpdatesSender(config)
             self._dtable_notification_rules_scanner = DTableNofiticationRulesScanner(config)
+            self._ai_notification_rules_scanner = AINotificationRulesScanner(config)
             self._dtable_automation_rules_scanner = DTableAutomationRulesScanner(config)
             self._ldap_syncer = LDAPSyncer(config)
             self._common_dataset_syncer = CommonDatasetSyncer(config)
@@ -77,6 +79,7 @@ def serve_forever(self):
             self._dtables_cleaner.start()                    # default True
             self._dtable_updates_sender.start()              # default True
             self._dtable_notification_rules_scanner.start()  # default True
+            self._ai_notification_rules_scanner.start()      # default True
             self._dtable_automation_rules_scanner.start()    # default True
             self._ldap_syncer.start()                        # default False
             self._common_dataset_syncer.start()              # default True

dtable_events/notification_rules/ai_notification_rules_scanner.py

@@ -0,0 +1,101 @@
+import json
+import os
+import logging
+from datetime import datetime
+from threading import Thread
+
+from sqlalchemy import text
+from apscheduler.schedulers.blocking import BlockingScheduler
+
+from dtable_events.app.config import TIME_ZONE, DTABLE_WEB_SERVICE_URL
+from dtable_events.db import init_db_session_class
+from dtable_events.utils.dtable_web_api import DTableWebAPI
+
+timezone = TIME_ZONE
+
+
+__all__ = [
+    'AINotificationRulesScanner',
+]
+
+
+class AINotificationRulesScanner(object):
+
+    def __init__(self, config):
+        self._enabled = True
+        self._logfile = None
+        self._prepare_logfile()
+        self._db_session_class = init_db_session_class(config)
+
+    def _prepare_logfile(self):
+        logdir = os.path.join(os.environ.get('LOG_DIR', ''))
+        self._logfile = os.path.join(logdir, 'ai_notification_rule_scanner.log')
+
+    def start(self):
+        if not self.is_enabled():
+            logging.warning('Can not start ai notification rules scanner: it is not enabled!')
+            return
+
+        logging.info('Start ai notification rules scanner')
+
+        AINotificationRulesScannerTimer(self._logfile, self._db_session_class).start()
+
+    def is_enabled(self):
+        return self._enabled
+
+
+def scan_ai_notification_rules(db_session):
+    sql = f'''
+            SELECT `id`, `assistant_uuid`, `detail` FROM ai_notification_rules
+        '''
+    rules = db_session.execute(text(sql))
+
+    for rule in rules:
+        try:
+            trigger_ai_notification_rule(rule)
+        except Exception as e:
+            logging.exception(e)
+            logging.error(f'check rule failed. {rule}, error: {e}')
+
+
+class AINotificationRulesScannerTimer(Thread):
+
+    def __init__(self, logfile, db_session_class):
+        super(AINotificationRulesScannerTimer, self).__init__()
+        self._logfile = logfile
+        self.db_session_class = db_session_class
+
+    def run(self):
+        sched = BlockingScheduler()
+
+        @sched.scheduled_job('cron', day_of_week='*', hour='*')
+        def timed_job():
+            logging.info('Starts to scan ai notification rules...')
+
+            db_session = self.db_session_class()
+            try:
+                scan_ai_notification_rules(db_session)
+            except Exception as e:
+                logging.exception('error when scanning ai notification rules: %s', e)
+            finally:
+                db_session.close()
+
+        sched.start()
+
+
+def trigger_ai_notification_rule(rule):
+    assistant_uuid = rule[1]
+    detail = rule[2]
+
+    detail = json.loads(detail)
+    notify_hour = detail.get('notify_hour', 0)
+    users = detail.get('users', [])
+
+    dtable_web_api = DTableWebAPI(DTABLE_WEB_SERVICE_URL)
+
+    cur_datetime = datetime.now()
+    cur_hour = int(cur_datetime.hour)
+    if notify_hour and int(notify_hour) != cur_hour:
+        return
+
+    dtable_web_api.add_issues_notification(users, assistant_uuid)

dtable_events/utils/dtable_web_api.py

@@ -130,3 +130,16 @@ def internal_add_notification(self, to_users, msg_type, detail):
             'type': msg_type
         }, headers=headers)
         return parse_response(resp)
+
+    def add_issues_notification(self, users, assistant_uuid):
+        logger.debug('add issues notification to users: %s assistant_uuid: %s', users, assistant_uuid)
+        url = '%(server_url)s/api/v2.1/ai/internal/issues-notification/?from=dtable_events' % {
+            'server_url': self.dtable_web_service_url
+        }
+        token = jwt.encode({}, DTABLE_PRIVATE_KEY, algorithm='HS256')
+        headers = {'Authorization': 'Token ' + token}
+        resp = requests.post(url, json={
+            'assistant_uuid': assistant_uuid,
+            'users': users,
+        }, headers=headers)
+        return parse_response(resp)