update

main.py

@@ -1,6 +1,10 @@
 import json
 import os
 import sys
+import time
+import traceback
+
+from utils import logger, ddsend_msg
 from multiprocessing import Pool
 
 from importlib import import_module
@@ -15,12 +19,20 @@ def get_game() -> list:
 
 
 def run_task(kwargs):
+    start_ts = int(time.time())
+    try:
         module_name = kwargs.get('task_name')
         class_name = ''.join([s.capitalize() for s in module_name.split('_')])
         module = import_module(f'.{module_name}', package='task')
         c_obj = getattr(module, class_name)
         obj = c_obj(**kwargs)
         obj.run()
+    except Exception as e:
+        msg = traceback.format_exc()
+        ddsend_msg(f'异常退出 {msg}')
+        logger.error(repr(e))
+    finally:
+        ddsend_msg(f'正常完成任务 {kwargs.get("task_name")} 耗时 {int(time.time()) - start_ts}')
 
 
 if __name__ == '__main__':

task/summary3.py

@@ -343,13 +343,13 @@ class Summary3(Task):
 
     @HandlerWrapper
     def handler_now(self, role_data, device_data, account_data, df, group, cdate):
-        if df.get('_game_role_id'):
+        if df.get('_game_role_id') is not None:
             role_data['new_role_list'] = list(df['_game_role_id'].unique())
             role_data['new_role'] = len(role_data['new_role_list'])
-        if df.get('is_new_device'):
+        if df.get('is_new_device') is not None:
             device_data['new_device_list'] = list(df[df['is_new_device'] == 1]['_device_id'].unique())
             device_data['new_device'] = len(device_data['new_device_list'])
-        if df.get('is_new_channel_uid'):
+        if df.get('is_new_channel_uid') is not None:
             account_data['new_account_list'] = list(df[df['is_new_channel_uid'] == 1]['_channel_uid'].unique())
             account_data['new_account'] = len(account_data['new_account_list'])
 

task/task.py

@@ -50,12 +50,13 @@ class Task(metaclass=abc.ABCMeta):
             return True
         elif int(time.time()) - last_ts > time_out:
             # 任务超时
-            # todo 钉钉通知
+            ddsend_msg(f'{self.game_name} {self.task_name} 任务超时')
             logger.info(f'{self.game_name} 钉钉通知')
             return False
         else:
             # 正在运行没超时
             logger.info(f'{self.game_name} 正在运行没超时')
+            ddsend_msg(f'{self.game_name} {self.task_name} 已运行 {time.time()-last_ts}s')
             return False
 
     def set_run_ts(self):

utils/__init__.py

@@ -1,4 +1,5 @@
 from loguru import logger
+from .dd_msg import ddsend_msg
 
 logger.add('/data/log/data_cleaning/log.log', format="{time} {level} {name}:{line} {message}", level="INFO",
            rotation="100 MB", retention='7 days',

utils/dd_msg.py

@@ -0,0 +1,34 @@
+import time
+import hmac
+import hashlib
+import urllib.parse
+import base64
+
+import requests
+
+
+def get_sign():
+    timestamp = str(round(time.time() * 1000))
+    secret = 'SEC8de0abcb83a3c387fb168995be58913f6e5fcc3612a02686fcc9107d25809930'
+    secret_enc = secret.encode('utf-8')
+    string_to_sign = '{}\n{}'.format(timestamp, secret)
+    string_to_sign_enc = string_to_sign.encode('utf-8')
+    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+    return timestamp, sign
+
+
+def ddsend_msg(msg):
+    data = {
+        'msgtype': 'text',
+        'text': {
+            'content': msg
+        }
+    }
+    timestamp, sign = get_sign()
+    url = f'https://oapi.dingtalk.com/robot/send?access_token=9d620df1653de39f28bf6b059e0fff9978ede3185645e4917c52d754a050b458&timestamp={timestamp}&sign={sign}'
+    resp = requests.post(url, json=data)
+
+
+if __name__ == '__main__':
+    ddsend_msg('大家好，我是叉叉,一个冒的感情的机器人')