完成消息通知模块-邮件通知功能

apps/sched/models.py

@@ -131,6 +131,9 @@ class Meta:
 
     def to_alert_json(self):
         json_format = {
+            "job_name":self.job_name,
+            "job_cmd":self.job_command,
+            "job_node":self.job_node.sched_server.server_assets.ip,
             "notice_type":self.notice_type,
             "notice_number":self.notice_number,
             "notice_interval":self.notice_interval,
@@ -210,7 +213,7 @@ def to_json(self):
             'jid': self.job_id.job_id,
             'stime':self.stime,
             'etime':self.etime,
-            "status":self.status,
+            "status":int(self.status),
             "result":self.result,
         }
         return json_format                      

apps/tasks/celery_notice.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python  
+# _#_ coding:utf-8 _*_ 
+from celery import task
+from utils.notice import Notice
+
+@task
+def apsched_notice(jobs, jobLogs):
+    status = "失败"
+
+    if jobLogs.get("status") == 0:status = "成功"
+
+    message = """<strong>节点:</strong> {job_node}<br>
+                 <strong>命令:</strong> {job_cmd}<br>
+                 <strong>执行结果:</strong> {result}<br>
+                 <strong>执行状态:</strong> {status}""".format(job_node=jobs.get("job_node"),job_cmd=jobs.get("job_cmd"),
+                                                                                                          result=jobLogs.get("result"), status=status)
+
+    if jobs.get("notice_type") == 0:    
+        message = {
+                "e_content":message,
+                "e_sub":"[OpsManage-任务通知] {job_name}".format(job_name=jobs.get("job_name")),
+                "e_to":jobs.get("notice_number")
+            }
+
+    Notice(jobs.get("notice_type")).send(**message) 
+

apps/tasks/celery_sql.py

@@ -4,7 +4,8 @@
 from databases.models import (SQL_Execute_Histroy,Database_Detail)
 
 @task
-def record_exec_sql(exe_user,exe_db,exe_sql,exe_time,exec_status=None,exe_result=None):
+def record_exec_sql(exe_user,exe_db,exe_sql,exe_time,
+                    exe_effect_row=0,exec_status=None,exe_result=None):
     try:
         exe_db = Database_Detail.objects.get(id=exe_db)
     except Exception as ex:
@@ -16,6 +17,7 @@ def record_exec_sql(exe_user,exe_db,exe_sql,exe_time,exec_status=None,exe_result
                                   exe_sql = exe_sql,
                                   exec_status = exec_status,
                                   exe_result = exe_result,
+                                  exe_effect_row = exe_effect_row,
                                   exe_time = exe_time
                                   )
         return {"status":"success","msg":None}

dao/apsched.py

@@ -8,6 +8,7 @@
 from .assets import AssetsBase
 from utils.sched.rpc import sched_rpc
 from django.db.models import Q
+from apps.tasks.celery_notice import apsched_notice
 
 class ApschedBase(object):
     def __init__(self):
@@ -173,21 +174,22 @@ def insert_jobs_logs_by_jid(self,data):
             data["job_id"] = jobs
             data.pop("jid")
             jobLogs = Sched_Job_Logs.objects.create(**data)
-            self.judge_notice(jobs.to_alert_json(), jobLogs.to_json())
+            self.judge_notice(jobs, jobLogs)
             return jobLogs.to_json()
         except Exception as ex:
             msg = "record jobs logs error {ex}".format(ex=str(ex))
             logger.warn(msg)
             return msg               
 
-    def judge_notice(self,jobs,jobLogs):
+    def judge_notice(self,jobs, jobLogs):
         try:
-            atime = int(jobs.get('atime'))
-        except:
-            atime = 0
-        if jobs.get('is_alert') > 0 and int(time.time()) - atime > 0:
-            pass
-#             apsched_notice.apply_async(**{"jobs":jobs,"jobslog":jobLogs})
+            if jobs.is_alert > 0 and int(time.time()) - jobs.atime > jobs.notice_interval:
+                apsched_notice.apply_async((jobs.to_alert_json(), jobLogs.to_json()), queue='default', retry=True)
+                jobs.atime = int(time.time())
+                jobs.save()
+        except Exception as ex:
+            msg = "notice jobs status failed {ex}".format(ex=str(ex))
+            logger.warn(msg)            
 
     def create_jobs(self,request):   
         node = self.schedNode(request)

dao/database.py

@@ -519,11 +519,10 @@ def optimize_sql(self,request):
 
 
     def __record_operation(self,request,dbServer,time_consume,result):
-
         if isinstance(result, str):
-            record_exec_sql.apply_async((request.user.username,request.POST.get('db'),request.POST.get('sql'),time_consume,1,result),queue='default')
+            record_exec_sql.apply_async((request.user.username,request.POST.get('db'), request.POST.get('sql'), time_consume, 1,result), queue='default')
         else:
-            record_exec_sql.apply_async((request.user.username,request.POST.get('db'),request.POST.get('sql'),time_consume,0),queue='default')        
+            record_exec_sql.apply_async((request.user.username,request.POST.get('db'), request.POST.get('sql'), time_consume, result[0], 0),queue='default')        
 
     def __query_user_db_server(self,request=None):
         if request.user.is_superuser:

static/js/database/histroy.js

@@ -152,6 +152,7 @@ $(document).ready(function() {
 	    		            { "data": "exe_sql"},	    		            
 	    		            { "data": "exec_status","defaultContent": ''},
 	    		            { "data": "exe_time","defaultContent": ''},
+	    		            { "data": "exe_effect_row","defaultContent": ''},
 	    		            { "data": "exe_result","defaultContent": ''},
 	    		            { "data": "create_time"},   
 			        ]

static/js/database/query.js

@@ -269,6 +269,7 @@ function makeUserDBQueryHistory(db_id) {
 		            { "data": "exe_sql"},	    		            
 		            { "data": "exec_status","defaultContent": ''},
 		            { "data": "exe_time","defaultContent": ''},
+		            { "data": "exe_effect_row","defaultContent": ''},
 		            { "data": "exe_result","defaultContent": ''},
 		            { "data": "create_time"},   
 	        ]
@@ -298,7 +299,7 @@ function makeUserDBQueryHistory(db_id) {
 	   	                        }
 	   	    				},
 	   	    				"className": "text-center",
-		    		     },    
+		    		     },     		    		      		    		        		                 	    		     
     		        	{
 	   	    				targets: [6],
 	   	    				render: function(data, type, row, meta) {
@@ -309,13 +310,7 @@ function makeUserDBQueryHistory(db_id) {
 	   	                        }
 	   	    				},
 	   	    				"className": "text-center",
-		    		     },	    		    		      		    		        
-		                 {
-							targets: [5],
-							render: function(data, type, row, meta) {  
-								return row.exe_sql				              
-							},
-					     }, 
+		    		     } 
 	    		      ]	
     var buttons = [{
         text: '<span class="fa fa-plus"></span>',

templates/database/db_histroy.html

@@ -80,6 +80,7 @@ <h2>数据库列表 <small>Database Assets List</small></h2>
 					                <th>SQL</th>
 					                <th>执行状态</th>
 					                <th>耗时(sec)</th>
+					                <th>影响行数</th>
 					                <th>执行结果</th>
 					                <th>执行时间</th>
 		                        </tr>

templates/database/db_query.html

@@ -166,6 +166,7 @@ <h2><i class="fa fa-bars"></i> 查询结果 <small>SQL Query History</small></h2
 											                <th>SQL</th>
 											                <th>执行状态</th>
 											                <th>耗时(sec)</th>
+											                <th>影响行数</th>
 											                <th>执行结果</th>
 											                <th>执行时间</th>
 								                        </tr>

utils/notice/email.py

@@ -22,6 +22,7 @@ def send(self,**kwargs):
         msg['Subject'] = "%s " % kwargs.get('e_sub')
         msg['From'] = self.e_from
         if kwargs.get('e_to').find(',') == -1:
+            e_to = kwargs.get('e_to')
             msg['To'] = kwargs.get('e_to')
         else: 
             e_to = kwargs.get('e_to').split(',')