update db

eda/smart_rank/tag3_start.sh

@@ -2,6 +2,8 @@
 
 # 新画像
 # log
+mysql -u doris -po5gbA27hXHHm -h 172.16.30.136 -D doris_prod -P 3306 -e "delete from kafka_tag3_log where log_time < UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 60 day))"
+
 mysql -u st_user -paqpuBLYzEV7tML5RPsN1pntUzFy -h 172.16.40.158 -P 4000 -D jerry_test -e "delete from kafka_tag3_log where log_time < UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 60 day))"
 
 # portrait1

eda/smart_rank/tag3_update_user_portrait_offline.py

@@ -8,7 +8,8 @@ import redis
 
 from pyspark import SparkConf
 from pyspark.sql import SparkSession
-from tool import (get_jerry_test, get_tag3_user_log, send_email, write_user_portrait, write_user_portrait_by_event)
+from tool import (get_doris_prod, get_jerry_test, get_tag3_user_log, send_email, write_user_portrait,
+                  write_user_portrait_by_event)
 
 
 # [{'激光': 1.949194898204873}, {'手术': 1.949194898204873}, {'手术': 1.949194898204873}, {'手术': 1.949194898204873, '植发际线': 7.1}]
@@ -186,7 +187,7 @@ def update_tag3_user_portrait(cl_id):
 
 def consume_kafka():
     sql = "select distinct cl_id from kafka_tag3_log where log_time > UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 30 day))"
-    db, cursor = get_jerry_test()
+    db, cursor = get_doris_prod()
     cursor.execute(sql)
     device_ids_lst = [i[0] for i in cursor.fetchall()]
     db.close()

eda/smart_rank/tool.py

@@ -365,6 +365,16 @@ def get_jerry_test():
     return db, db.cursor()
 
 
+def get_doris_prod():
+    db = pymysql.connect(host="172.16.30.136",
+                         port=3306,
+                         user="doris",
+                         passwd="o5gbA27hXHHm",
+                         db="doris_prod",
+                         charset="utf8")
+    return db, db.cursor()
+
+
 def compute_tag3_score(x):
     if x.score_type == "henqiang":
         return compute_henqiang(x.days_diff_now, exponential=1)