add compare kafka and portrait

eda/smart_rank/test_divided_portrait.py

@@ -101,8 +101,7 @@ def write_user_portrait_action_divided(cl_id,event_cn,first_solutions, second_so
         print("write_user_portrait_action_divided error")
         print(e)
 
-if __name__ == '__main__':
-# def get_user_portrait_action_divided(cl_ids):
+def get_user_portrait_action_divided(cl_ids):
     first_solutions = []
     first_positions = []
     first_demands = []
@@ -144,3 +143,76 @@ if __name__ == '__main__':
                                                   projects=set(projects))
 
 
+def get_jerry_test():
+    db = pymysql.connect(host="172.16.40.170",
+                         port=4000,
+                         user="st_user",
+                         passwd="aqpuBLYzEV7tML5RPsN1pntUzFy",
+                         db="jerry_test",
+                         charset="utf8")
+    return db, db.cursor()
+
+def connect_doris_table():
+    db = pymysql.connect(host="172.16.30.136",
+                         port=3306,
+                         user="doris",
+                         password="o5gbA27hXHHm",
+                         db="doris_prod",
+                         charset="utf8")
+    return db, db.cursor()
+
+def get_kafka(cl_id):
+    # try:
+    sql = """select  cl_id,event_cn, first_solutions, second_solutions, first_demands,
+                    second_demands, first_positions, second_positions, projects
+                    from kafka_tag3_log where cl_id = '{}' """.format(cl_id)
+    db, cursor = get_jerry_test()
+    cursor.execute(sql)
+    data = list(cursor.fetchall())
+    db.close()
+    cursor.close()
+    return data
+
+def get_portrait(cl_id):
+    # try:
+    sql = """select  cl_id,event_cn, first_solutions, second_solutions, first_demands,
+                       second_demands, first_positions, second_positions, projects
+                       from user_portrait_action_divided where cl_id = '{}' """.format(cl_id)
+    db, cursor = connect_doris_table()
+    cursor.execute(sql)
+    data = list(cursor.fetchall())
+    db.close()
+    cursor.close()
+    return data
+
+def get_portrait_device_ids_lst():
+    sql = "select cl_id from user_portrait_action_divided limit 10"
+    db, cursor = get_jerry_test()
+    cursor.execute(sql)
+    device_ids_lst = [i[0] for i in cursor.fetchall()]
+    db.close()
+    cursor.close()
+    return device_ids_lst
+
+def show_kafka_and_portrait():
+    try:
+        device_ids_lst = get_portrait_device_ids_lst()
+        for cl_id in device_ids_lst:
+            print('*' * 100)
+            kafka_data = get_kafka(cl_id)
+            doris_data = get_doris(cl_id)
+            for key,val in kafka_data.items:
+               print('kafka:',cl_id,key,val)
+               print('portrait:',cl_id,key,doris_data[key])
+    except Exception as e:
+        print(e)
+        return None
+
+
+
+
+if __name__ == '__main__':
+    show_kafka_and_portrait()
+    # cl_ids = ['']
+    # get_user_portrait_action_divided(cl_ids)
+