用户画像分层影响

eda/smart_rank/dist_update_portrait_market.py

+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+import pymysql
+import smtplib
+from email.mime.text import MIMEText
+from email.utils import formataddr
+from email.mime.multipart import MIMEMultipart
+from email.mime.application import MIMEApplication
+import redis
+import datetime
+from pyspark import SparkConf
+import time
+from pyspark.sql import SparkSession
+import json
+import numpy as np
+import pandas as pd
+from pyspark.sql.functions import lit
+from pyspark.sql.functions import concat_ws
+
+
+def send_email(app,id,e):
+    # 第三方 SMTP 服务
+    mail_host = 'smtp.exmail.qq.com'  # 设置服务器
+    mail_user = "gaoyazhe@igengmei.com"  # 用户名
+    mail_pass = "VCrKTui99a7ALhiK"  # 口令
+
+    sender = 'gaoyazhe@igengmei.com'
+    receivers = ['gaoyazhe@igengmei.com']  # 接收邮件，可设置为你的QQ邮箱或者其他邮箱
+    e = str(e)
+    msg = MIMEMultipart()
+    part = MIMEText('app_id:'+id+':fail', 'plain', 'utf-8')
+    msg.attach(part)
+    msg['From'] = formataddr(["gaoyazhe", sender])
+    # 括号里的对应收件人邮箱昵称、收件人邮箱账号
+    msg['To'] = ";".join(receivers)
+    # message['Cc'] = ";".join(cc_reciver)
+
+    msg['Subject'] = 'spark streaming:app_name:'+app
+    with open('error.txt','w') as f:
+         f.write(e)
+         f.close()
+    part = MIMEApplication(open('error.txt', 'r').read())
+    part.add_header('Content-Disposition', 'attachment', filename="error.txt")
+    msg.attach(part)
+
+    try:
+        smtpObj = smtplib.SMTP_SSL(mail_host, 465)
+        smtpObj.login(mail_user, mail_pass)
+        smtpObj.sendmail(sender, receivers, msg.as_string())
+    except smtplib.SMTPException:
+        print('error')
+
+
+def get_data_by_mysql(host, port, user, passwd, db, sql):
+    try:
+        db = pymysql.connect(host=host, port=port, user=user, passwd=passwd, db=db, cursorclass=pymysql.cursors.DictCursor)
+        cursor = db.cursor()
+        cursor.execute(sql)
+        results = cursor.fetchall()
+        db.close()
+        return results
+    except Exception as e:
+        print(e)
+
+
+def get_all_search_word_and_synonym_tags():
+    """
+    :return:dict {"search_word1":[tag_list1],"search_word2":[tag_list2]...}
+    """
+    try:
+        sql = "select a.keyword , c.id from api_wordrel a  " \
+              "left join api_wordrelsynonym b on a.id = b.wordrel_id  " \
+              "left join api_tag c  on b.word=c.name " \
+              "where a.category in (1,13,10,11,12) and c.tag_type+0<'4'+0 and c.is_online=1"
+        mysql_results = get_data_by_mysql('172.16.30.141', 3306, 'work', 'BJQaT9VzDcuPBqkd', 'zhengxing', sql)
+        result_dict = dict()
+        for data in mysql_results:
+            if data['keyword'] not in result_dict:
+                result_dict[data['keyword']] = [data['id']]
+            else:
+                result_dict[data['keyword']].append(data['id'])
+        return result_dict
+    except Exception as e:
+        print(e)
+
+
+def get_all_synonym_tags():
+    """
+    :return:dict {"search_word1":[tag_list1],"search_word2":[tag_list2]...}
+    """
+    try:
+        sql = "select a.word, b.id from api_wordrelsynonym a left join api_tag b " \
+              "on a.word=b.name where b.tag_type+0<'4'+0 and b.is_online=1"
+        mysql_results = get_data_by_mysql('172.16.30.141', 3306, 'work', 'BJQaT9VzDcuPBqkd', 'zhengxing', sql)
+        result_dict = dict()
+        for data in mysql_results:
+            if data['word'] not in result_dict:
+                result_dict[data['word']] = [data['id']]
+            else:
+                result_dict[data['word']].append(data['id'])
+        return result_dict
+    except Exception as e:
+        print(e)
+
+
+def get_all_word_tags():
+    try:
+        search_word_and_synonym_tags = get_all_search_word_and_synonym_tags()
+        synonym_tags = get_all_synonym_tags()
+        if search_word_and_synonym_tags and synonym_tags:
+            return {**synonym_tags, **search_word_and_synonym_tags}
+    except Exception as e:
+        print(e)
+
+
+def compute_henqiang(x):
+    score = 15-x*((15-0.5)/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def compute_jiaoqiang(x):
+    score = 12-x*(12/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def compute_ruoyixiang(x):
+    score = 5-x*((5-0.5)/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def compute_validate(x):
+    score = 10-x*((10-0.5)/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def compute_ai_scan(x):
+    score = 2 - x * ((2 - 0.5) / 180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+
+
+def get_user_tag_score(cl_id, all_log_df, all_word_tags, size=10):
+    try:
+        db_jerry_test = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC',
+                                        db='jerry_test', charset='utf8')
+        cur_jerry_test = db_jerry_test.cursor()
+
+        user_log_df = all_log_df.loc[(all_log_df['cl_id'] == cl_id) & (all_log_df['action'] != 'do_search')]
+        user_df_search = all_log_df.loc[(all_log_df['cl_id'] == cl_id) & (all_log_df['action'] == 'do_search')]
+        # 搜索词转成tag
+        for index, row in user_df_search.iterrows():
+            if row['tag_referrer'] in all_word_tags:
+                for search_tag in all_word_tags[row['tag_referrer']]:
+                    row['tag_id'] = int(search_tag)
+                    user_log_df = user_log_df.append(row, ignore_index=True)
+                    break
+        if not user_log_df.empty:
+            user_log_df["days_diff_now"] = round((int(time.time())-user_log_df["time"]) / (24*60*60))
+            user_log_df["score"] = user_log_df.apply(
+                lambda x: compute_henqiang(x.days_diff_now) if x.score_type == "henqiang" else (
+                    compute_jiaoqiang(x.days_diff_now) if x.score_type == "jiaoqiang" else (
+                        compute_ai_scan(x.days_diff_now) if x.score_type == "ai_scan" else (
+                            compute_ruoyixiang(x.days_diff_now) if x.score_type == "ruoyixiang" else compute_validate(x.days_diff_now)))), axis=1)
+            finally_score = user_log_df.sort_values(by=["score","time"],ascending=False)
+            finally_score.drop_duplicates(subset="tag_id", inplace=True)
+            finally_score["weight"] = finally_score['score'] / finally_score['score'].sum()
+            finally_score["pay_type"] = finally_score.apply(
+                lambda x: 3 if x.action == "api/order/validate" else (
+                    2 if x.action == "api/settlement/alipay_callback" else 1
+                ), axis=1
+            )
+            score_result = finally_score[["tag_id", "cl_id", "score", "weight", "pay_type"]]
+            score_result.rename(columns={"cl_id": "device_id"}, inplace=True)
+
+            # 写tidb
+            delete_sql = "delete from api_market_personas where device_id='{}'".format(cl_id)
+            cur_jerry_test.execute(delete_sql)
+            db_jerry_test.commit()
+
+            for index, row in score_result.iterrows():
+                insert_sql = "insert into api_market_personas values (null, {}, '{}', {}, {}, {})".format(
+                    row['tag_id'], row['device_id'], row['score'], row['weight'], row['pay_type'])
+                cur_jerry_test.execute(insert_sql)
+                db_jerry_test.commit()
+            db_jerry_test.close()
+            return "sucess"
+    except Exception as e:
+        return 'pass'
+
+
+if __name__ == '__main__':
+    try:
+        db_jerry_test = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC',
+                                        db='jerry_test', charset='utf8')
+        cur_jerry_test = db_jerry_test.cursor()
+
+        # 获取所有用户的设备id
+        # sql_device_ids = "select distinct cl_id from user_new_tag_log"
+        # 获取最近30天内的用户设备id
+        sql_device_ids = "select distinct cl_id from user_new_tag_log " \
+                         "where time > UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 30 day))"
+        cur_jerry_test.execute(sql_device_ids)
+        device_ids_lst = [i[0] for i in cur_jerry_test.fetchall()]
+
+        # 获取所有用户的行为日志
+        # sql_all_log = "select time,cl_id,score_type,tag_id,tag_referrer,action from user_new_tag_log"
+        # 获取最近30天内的用户的所有行为
+        sql_all_log = "select time,cl_id,score_type,tag_id,tag_referrer,action from user_new_tag_log where cl_id in " \
+                      "(select distinct cl_id from user_new_tag_log " \
+                      "where time > UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 30 day)))"
+        cur_jerry_test.execute(sql_all_log)
+        all_log = cur_jerry_test.fetchall()
+        db_jerry_test.close()
+        all_log_df = pd.DataFrame(list(all_log))
+        all_log_df.columns = ["time", "cl_id", "score_type","tag_id","tag_referrer","action"]
+        stat_date = datetime.datetime.today().strftime('%Y-%m-%d')
+        #搜索词及其同义词匹配tag
+        all_word_tags = get_all_word_tags()
+        # rdd
+        sparkConf = SparkConf().set("spark.hive.mapred.supports.subdirectories", "true") \
+            .set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", "true") \
+            .set("spark.tispark.plan.allow_index_double_read", "false") \
+            .set("spark.tispark.plan.allow_index_read", "true") \
+            .set("spark.sql.extensions", "org.apache.spark.sql.TiExtensions") \
+            .set("spark.tispark.pd.addresses", "172.16.40.158:2379").set("spark.io.compression.codec", "lzf") \
+            .set("spark.driver.maxResultSize", "8g").set("spark.sql.avro.compression.codec", "snappy")
+
+        spark = SparkSession.builder.config(conf=sparkConf).enableHiveSupport().getOrCreate()
+        spark.sparkContext.setLogLevel("WARN")
+        device_ids_lst_rdd = spark.sparkContext.parallelize(device_ids_lst)
+        result = device_ids_lst_rdd.repartition(100).map(lambda x: get_user_tag_score(x, all_log_df, all_word_tags))
+        result.collect()
+
+        # result_last = result_rename.withColumn("stat_date", lit(stat_date))
+        # result_last.show()
+        # df = result_last.select("stat_date", "cl_id", concat_ws(',', 'tag_list').alias("tag_list"))
+        # df.show()
+        # df.write.jdbc(
+        #     mode="overwrite",
+        #     url="jdbc:mysql://172.16.40.158:4000/jerry_test?user=root&password=3SYz54LS9#^9sBvC&useSSL=true",
+        #     table="user_portrait_tags",
+        #     properties={"driver": 'com.mysql.jdbc.Driver'})
+    except Exception as e:
+        send_email("dist_update_portrait_market", "dist_update_portrait_market", "dist_update_portrait_market")
+
+
+
+
+
+

eda/smart_rank/dist_update_user_portrait.py

@@ -76,6 +76,12 @@ def compute_validate(x):
         return score
     else:
         return 0.5
+def compute_ai_scan(x):
+    score = 2 - x * ((2 - 0.5) / 180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
 def tag_list2dict(lst, size):
     result = []
     if lst:
@@ -104,7 +110,8 @@ def get_user_tag_score(cl_id, all_log_df, size=10):
             user_log_df["tag_score"] = user_log_df.apply(
                 lambda x: compute_henqiang(x.days_diff_now) if x.score_type == "henqiang" else (
                     compute_jiaoqiang(x.days_diff_now) if x.score_type == "jiaoqiang" else (
-                        compute_ruoyixiang(x.days_diff_now) if x.score_type == "ruoyixiang" else compute_validate(x.days_diff_now))), axis=1)
+                        compute_ai_scan(x.days_diff_now) if x.score_type == "ai_scan" else (
+                        compute_ruoyixiang(x.days_diff_now) if x.score_type == "ruoyixiang" else compute_validate(x.days_diff_now)))), axis=1)
             finally_score = user_log_df.sort_values(by=["tag_score","time"],ascending=False)
             finally_score.drop_duplicates(subset="tag_id", inplace=True)
             finally_score_lst = finally_score[["tag_id","tag_score"]].to_dict('record')