Merge branch 'dev2' of ssh://119.23.72.7:22611/lic/tas_monitor

 Conflicts:
	utils/mongo.py

crawler_process.py

@@ -21,9 +21,9 @@ ichunt_elk = Elasticsearch(host="127.0.0.1", port=9200, timeout=20)
 
 
 def get_process_task(task_code):
-    keys = task_redis.keys(task_code + '*')
+    keys = task_redis.keys(task_code + '_*')
     keys_len = sum([task_redis.llen(k) for k in keys])
-    abnormal_keys = task_redis.keys('abnormal_' + task_code + '*')
+    abnormal_keys = task_redis.keys('abnormal_' + task_code + '_*')
     abnormal_keys_len = sum([task_redis.llen(k) for k in abnormal_keys])
     return keys_len, abnormal_keys_len
 
@@ -76,7 +76,7 @@ def get_crawler_process_info():
         info_dict = dict()
         info_dict['pid'] = pid.strip()
         info_dict['task_code'] = info['cmdline']['-r']
-        info_dict['platform'] = info['cmdline']['-mq'].split("_")[0]
+        info_dict['platform'] = info['cmdline']['-mq'].split("_")[0] if info['cmdline'].get('-mq', '') else 'data_rank'
         info_dict['concurrency'] = int(info['cmdline']['-c'])
         info_dict['queue'] = info['cmdline']['-mq'] if '-mq' in info['cmdline'] else ''
         unic = ''.join([str(info['create_time']), info_dict['task_code'], str(info_dict['concurrency'])]).encode()

monitor.py

@@ -58,10 +58,11 @@ def task_running_flag_monitor():
             dd_send_msg(traceback.format_exc())
             time.sleep(60 * 5)
 
+
 def mongo_none_data_rate_monitor():
     while True:
         try:
-            msg = mongo.get_platform_none_data_rate ()
+            msg = mongo.get_platform_none_data_rate()
             if msg:
                 logger.info(msg)
                 dd_send_msg(msg)
@@ -71,6 +72,7 @@ def mongo_none_data_rate_monitor():
             dd_send_msg(traceback.format_exc())
             time.sleep(60 * 5)
 
+
 def main():
     thread_monitor1 = threading.Thread(target=task_fail_rate_monitor)
     thread_monitor1.start()

utils/glo.py

@@ -5,14 +5,13 @@ class Glo():
         global _global_dict
         _global_dict = {}
 
-    def set_value(self,key,value):
+    def set_value(self, key, value):
 
         _global_dict[key] = value
 
-
-    def get_value(self,key):
+    def get_value(self, key):
 
         try:
             return _global_dict[key]
         except KeyError:
-            return None
\ No newline at end of file
+            return None

utils/mysql.py

@@ -38,7 +38,7 @@ class MySqlOperator:
     def update_process(self, p):
         with self.db.cursor() as cursor:
             sql = """
-                UPDATE lie_spider_board SET run_time=%(run_time)s, remain_task=%(remain_task)s, 
+                UPDATE lie_spider_board SET run_time=%(run_time)s, remain_task=%(remain_task)s, concurrency=%(concurrency)s,
                 remain_wrong_task=%(remain_wrong_task)s, wrong_radio=%(wrong_radio)s WHERE unique_code=%(unique_code)s
               """
             cursor.execute(sql, p)
@@ -58,3 +58,11 @@ class MySqlOperator:
               """
             cursor.execute(sql, (s, u))
             self.db.commit()
+
+    def insert_expired_data(self, time_stamp, platform, expired_rate):
+        with self.db.cursor() as cursor:
+            sql = """
+                         INSERT INTO lie_spider_expired_rate(time_stamp, platform,expired_rate) VALUES (%s,%s,%s)
+                         """
+            cursor.execute(sql, (time_stamp, platform, expired_rate))
+            self.db.commit()

utils/robots.py

 #!/usr/bin/env python
 # -*- coding:utf-8 -*-
-
+import hashlib
+import time
 import requests
+import re
 
 dd_robot_api = ' https://oapi.dingtalk.com/robot/send?access_token='
 tom = 'c19beeab837f5c6e019e7fc602d470704d634ae33510f40fb0e903cde215fb24'
+filter_plat_kw = ['pc', 'bkc']
+wrong_ratio_pat = re.compile(r'\d\d\.\d\d')
+msgapi_url = 'http://api.ichunt.com/msg/sendMessageByAuto'
+
+
+def parse_msg(msg):
+    lines = msg.split('\n')
+    filter_lines = []
+    for line in lines:
+        for filter_kw in filter_plat_kw:
+            if filter_kw in line:
+                break
+            filter_lines.append(line)
+            break
+
+    msgs = []
+    for line in filter_lines:
+        if '错误率' in line and float(wrong_ratio_pat.findall(line)[0]) >= 80:
+            msgs.append(line)
+    return msgs
 
 
 def dd_send_msg(msg, robot=tom):
@@ -15,7 +37,27 @@ def dd_send_msg(msg, robot=tom):
         }
     }
     requests.post(dd_robot_api + robot, json=data)
+    time.sleep(5)
+
+    phone_msgs = parse_msg(msg)
+    if not phone_msgs:
+        return
+
+    content = '\\n'.join(phone_msgs)
+    cur_time = int(time.time())
+    m = hashlib.md5()
+    d = hashlib.md5()
+    m.update(str(cur_time).encode())
+    d.update((m.hexdigest() + 'fh6y5t4rr351d2c3bryi').encode())
+
+    data = {'data': "{\"content\": \"%s\"}" % (content), 'touser': "[\"15113123231\",\"18594239421\"]", 'keyword': 'monitoring-py',
+            'fromuser': '', 'wechat_data': '', 'is_oversea': '', 'channel_type': '', 'is_ignore': '', 'ex_int': '',
+            'ex_str': '', 'k2': d.hexdigest(), 'k1': str(cur_time)}
+    requests.post(msgapi_url, data=data)
 
 
 if __name__ == "__main__":
-    dd_send_msg('喂喂 你发的太多了！')
+    dd_send_msg('''TAS任务监控预警 过去90分钟内:
+任务:muguf 未知 平均错误率为64.30%
+任务:pc 平台比价 平均错误率为97.99%
+任务:bkc buerklin新品采集 平均错误率为72.97%''')