增加美购的用户画像

eda/smart_rank/dist_update_user_portrait_service.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 get_all_tag_tag_type():
+    """
+    :return:dict {tag_id1:tag_type1,tag_id2:tag_type2...}
+    """
+    try:
+        sql = "select id,tag_type from api_tag where tag_type+0<'4'+0 and 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:
+            result_dict[data['id']] = data['tag_type']
+        return result_dict
+    except Exception as e:
+        print(e)
+
+
+def get_all_3tag_2tag():
+    try:
+        sql = "select a.child_id,a.parent_id from api_tagrelation a" \
+              " left join api_tag b on a.parent_id=b.id " \
+              "where a.child_id in (select id from api_tag where tag_type='3' and is_online=1) " \
+              "and b.tag_type='2'"
+        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['child_id'] not in result_dict:
+                result_dict[data['child_id']] = [data['parent_id']]
+            else:
+                result_dict[data['child_id']].append(data['parent_id'])
+        return result_dict
+    except Exception as e:
+        print(e)
+
+
+def get_tag2_from_tag3(tag3, all_3tag_2tag, user_log_df_tag2_list):
+    try:
+        tag2s = []
+        if tag3 in all_3tag_2tag:
+            tag2s = all_3tag_2tag[tag3]
+        for tag2 in tag2s:
+            if tag2 in user_log_df_tag2_list:
+                return tag2
+        return tag3
+    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_action_tag_count(df, action_time):
+    try:
+        if not df[df['time'] == action_time].empty:
+            return len(df[df['time'] == action_time])
+        else:
+            return 1
+    except Exception as e:
+        print(e)
+
+
+def get_user_service_portrait(cl_id, all_word_tags, all_tag_tag_type, all_3tag_2tag):
+    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_df_service_sql = "select time,cl_id,score_type,tag_id,tag_referrer,action from user_new_tag_log " \
+                              "where cl_id ='{}' and action not in " \
+                              "('api/order/validate','api/settlement/alipay_callback','do_search')".format(cl_id)
+        cur_jerry_test.execute(user_df_service_sql)
+        user_df_service = pd.DataFrame(list(cur_jerry_test.fetchall()))
+        user_df_service.columns = ["time", "cl_id", "score_type", "tag_id", "tag_referrer", "action"]
+        # 用户的搜索行为
+        user_df_search_sql = "select time,cl_id,score_type,tag_id,tag_referrer,action from user_new_tag_log " \
+                             "where cl_id ='{}' and action = 'do_search'".format(cl_id)
+        cur_jerry_test.execute(user_df_search_sql)
+        user_df_search = pd.DataFrame(list(cur_jerry_test.fetchall()))
+        user_df_search.columns = ["time", "cl_id", "score_type", "tag_id", "tag_referrer", "action"]
+        db_jerry_test.close()
+
+        # 搜索词转成tag
+        # user_df_search_2_tag = pd.DataFrame(columns=list(user_df_service.columns))
+        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_df_service = user_df_service.append(row, ignore_index=True)
+                    break
+
+        # 增加df字段(days_diff_now, tag_type, tag2)
+        if not user_df_service.empty:
+            user_df_service["days_diff_now"] = round((int(time.time()) - user_df_service["time"].astype(float)) / (24 * 60 * 60))
+            user_df_service["tag_type"] = user_df_service.apply(lambda x: all_tag_tag_type.get(x["tag_id"]), axis=1)
+            user_df_service = user_df_service[user_df_service['tag_type'].isin(['2','3'])]
+            user_log_df_tag2_list = user_df_service[user_df_service['tag_type'] == '2']['tag_id'].unique().tolist()
+            user_df_service["tag2"] = user_df_service.apply(lambda x:
+                                                    get_tag2_from_tag3(x.tag_id, all_3tag_2tag, user_log_df_tag2_list)
+                                                    if x.tag_type == '3' else x.tag_id, axis=1)
+            user_df_service["tag2_type"] = user_df_service.apply(lambda x: all_tag_tag_type.get(x["tag2"]), axis=1)
+            # 算分及比例
+            user_df_service["tag_score"] = user_df_service.apply(
+                lambda x: compute_henqiang(x.days_diff_now)/get_action_tag_count(user_df_service, x.time) if x.score_type == "henqiang" else (
+                    compute_jiaoqiang(x.days_diff_now)/get_action_tag_count(user_df_service, x.time) if x.score_type == "jiaoqiang" else (
+                        compute_ai_scan(x.days_diff_now)/get_action_tag_count(user_df_service, x.time) if x.score_type == "ai_scan" else (
+                        compute_ruoyixiang(x.days_diff_now)/get_action_tag_count(user_df_service, x.time) if x.score_type == "ruoyixiang" else
+                        compute_validate(x.days_diff_now)/get_action_tag_count(user_df_service, x.time)))), axis=1)
+            tag_score_sum = user_df_service.groupby(by=["tag2", "tag2_type"]).agg({'tag_score': sum}).reset_index().sort_values(by=["tag_score"],ascending=False)
+            tag_score_sum['weight'] = 100 * tag_score_sum['tag_score'] / tag_score_sum['tag_score'].sum()
+            gmkv_tag_score_sum = tag_score_sum[["tag2", "tag_score", "weight"]].to_dict('record')
+            # 写gmkv
+            gm_kv_cli = redis.Redis(host="172.16.40.135", port=5379, db=2, socket_timeout=2000)
+            cl_id_portrait_key = "user:service_portrait_tags:cl_id:" + str(cl_id)
+            tag_id_list_json = json.dumps(gmkv_tag_score_sum)
+            gm_kv_cli.set(cl_id_portrait_key, tag_id_list_json)
+            gm_kv_cli.expire(cl_id_portrait_key, time=30 * 24 * 60 * 60)
+            return gmkv_tag_score_sum
+    except Exception as e:
+        print(e)
+
+
+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()
+
+        # 获取最近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()]
+        db_jerry_test.close()
+
+        # 获取搜索词及其近义词对应的tag
+        all_word_tags = get_all_word_tags()
+        all_tag_tag_type = get_all_tag_tag_type()
+
+        # 3级tag对应的2级tag
+        all_3tag_2tag = get_all_3tag_2tag()
+
+        # 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_service_portrait(x, all_word_tags, all_tag_tag_type, all_3tag_2tag))
+        result.collect()
+
+    except Exception as e:
+        send_email("dist_update_user_portrait_service", "dist_update_user_portrait_service", "dist_update_user_portrait_service")
\ No newline at end of file