send msg to ding talk

eda/smart_rank/tag3_update_user_portrait_offline.py

@@ -9,7 +9,7 @@ from pyspark import SparkConf
 from pyspark.sql import SparkSession
 
 from tool import (get_doris_prod, get_redis_client, get_tag3_user_log, get_tag3_user_order_log, get_user_portrait_tag3_with_score,
-                  send_email, write_user_portrait, send_msg_to_dingtalk)
+                  send_msg_to_dingtalk, write_user_portrait)
 
 
 # [{'激光': 1.949194898204873}, {'手术': 1.949194898204873}, {'手术': 1.949194898204873}, {'手术': 1.949194898204873, '植发际线': 7.1}]
@@ -327,6 +327,7 @@ def update_tag3_user_portrait(cl_id):
 
 
 def consume_kafka():
+    start = datetime.datetime.now()
     sql = "select distinct cl_id from kafka_tag3_log where log_time > UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 30 day))"
     db, cursor = get_doris_prod()
     cursor.execute(sql)
@@ -360,28 +361,27 @@ def consume_kafka():
         # result2.foreach(print)
         # result2.collect()
 
+        end = datetime.datetime.now()
+        msg_dict = {"start": str(start), "end": str(end), "device_number": len(device_ids_lst)}
+        msg_res = ""
+        for (k, v) in msg_dict.items():
+            msg_res += str(k)
+            msg_res += ": "
+            msg_res += str(v)
+            msg_res += "\n"
+        send_msg_to_dingtalk(msg_res)
         spark.stop()
     except Exception as e:
-        send_email("tag3_update_user_portrait_offline", "tag3_update_user_portrait_offline", e)
+        send_msg_to_dingtalk("tag3_update_user_portrait_offline\n" + str(e))
 
 
 if __name__ == "__main__":
     start = datetime.datetime.now()
 
-    # update_tag3_user_portrait("862460044588666")
-    # update_tag3_user_portrait("androidid_a25a1129c0b38f7b")
-
-    # cl_id = "864350041167473"
-    # df = get_tag3_user_log(cl_id)
-    # df[["projects", "business_tags"]]
-    # update_tag3_user_portrait(cl_id)
-
     # cl_id = "867617044159377"
     # print(update_tag3_user_portrait(cl_id))
 
-    # consume_kafka()
-
-    send_msg_to_dingtalk("test")
+    consume_kafka()
     end = datetime.datetime.now()
     print(end - start)
     print("done")