import json
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil, DateTypes
from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F, DataFrame
from pyspark.sql.types import MapType, StringType, DecimalType, IntegerType, DoubleType, ArrayType
from yswg_utils.common_udf import parse_asin_volume_str, sort_volume, get_Fba_Fee
"""
搜索词 计算=>平均长宽高 => 计算基本利润率相关数据
依赖 dwd_st_asin_measure 表 dim_asin_detail 表
输出为 Dwd_st_volume_fba
支持所有站点 日期类型
# 更新利润率
update us_aba_last_30_day
set gross_profit_fee_sea = tb_2.gross_profit_fee_sea,
gross_profit_fee_air=tb_2.gross_profit_fee_air
from us_aba_profit_gross_last30day tb_2
where tb_2.search_term_id = id
"""
class DwdStVolumeFba(object):
def __init__(self, site_name, date_type, date_info):
self.site_name = site_name
self.date_info = date_info
self.date_type = date_type
app_name = f"{self.__class__.__name__}:{site_name}:{date_type}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
# # 注册本地静态方法 udf 返回新函数
# self.udf_parse_volume_reg = F.udf(self.udf_parse_volume, MapType(StringType(), DecimalType(10, 4)))
self.spark.udf.register("udf_sort_volume", sort_volume, ArrayType(DoubleType()))
self.udf_transfer_weight_reg = F.udf(self.udf_transfer_weight, DoubleType())
self.udf_transfer_volume_val_reg = F.udf(self.udf_transfer_volume_val, DoubleType())
self.udf_calc_avg_reg = F.udf(self.udf_calc_avg, DecimalType(10, 4))
self.udf_calc_profit_reg = F.udf(self.udf_calc_profit, MapType(StringType(), StringType()))
self.hive_tb = "dwd_st_volume_fba"
@staticmethod
def udf_transfer_weight(weight, weight_type):
"""
重量及重量类型转为 克
:param asin_volume:
:return:
"""
if weight is None:
return None
if weight_type == 'pounds':
rate = 453.59237
elif weight_type == 'grams':
rate = 1
pass
else:
raise Exception("weight_type 异常请检查！！")
return round(weight * rate, 4)
@staticmethod
def udf_transfer_volume_val(val, asin_volume_type):
"""
长度单位转为 cm
:param asin_volume:
:return:
"""
if val is None:
return None
if asin_volume_type == 'inches':
rate = 2.54
elif asin_volume_type == 'cm':
rate = 1
elif asin_volume_type == 'mm':
rate = 10
elif asin_volume_type == 'm':
rate = 100
else:
rate = 1
return round(val * rate, 4)
@staticmethod
def udf_calc_avg(
val1_col,
val2_col,
val3_col,
val1_count,
val2_count,
val3_count
):
val1_col = val1_col or 0
val2_col = val2_col or 0
val3_col = val3_col or 0
val1_count = val1_count or 0
val2_count = val2_count or 0
val3_count = val3_count or 0
arr = [(val1_count, val1_col), (val2_count, val2_col), (val3_count, val3_col)]
# 从小到大排序
arr.sort(key=lambda it: it[0], reverse=False)
val1, count1 = arr[0]
val2, count2 = arr[1]
val3, count3 = arr[2]
if (count1 == count2 and count2 == count3):
return round((val1 + val2 + val3) / 3, 2)
if (count2 / count3 >= 0.3):
return round((val2 + val3) / 2, 2)
else:
return val3
@staticmethod
def get_fba_calc_dict(df_profit_config: DataFrame):
"""
获取配置表
:param df_profit_config:
:return:
"""
calc_profit_list_df = df_profit_config.where("calc_type is not null") \
.groupby("category_first_id").agg(
F.first(F.col("calc_type")).alias("calc_type"),
F.first(F.col("fba_config_json")).alias("fba_config_json"),
)
calc_profit_dict = {}
for row in calc_profit_list_df.collect():
tmp = row.asDict()
calc_profit_dict[str(tmp['category_first_id'])] = {
"calc_type": row['calc_type'],
"fba_config_json": json.loads(row['fba_config_json'])
}
return calc_profit_dict
@staticmethod
def udf_calc_fba_reg(config_dict: dict):
"""
根据配置表计算fba费用
:param config_dict:
:return:
"""
def udf_calc_fba(categoy_first_id: str, price):
if price is None:
return None
config_row = config_dict.get(str(categoy_first_id))
if config_row is None:
# 默认佣金比例是0.15
return price * 0.15
else:
calc_type = config_row['calc_type']
calc_json = config_row['fba_config_json']
if calc_type == '价格分段':
min = calc_json['min']
rate_1 = calc_json['rate_1']
rate_2 = calc_json['rate_2']
break_val = calc_json['break_val']
if price <= break_val:
return max(min, price * rate_1)
else:
return price * rate_2
elif calc_type == '佣金分段':
min = calc_json['min']
rate_1 = calc_json['rate_1']
rate_2 = calc_json['rate_2']
break_val = calc_json['break_val']
if price * rate_1 <= break_val:
return max(min, price * rate_1)
else:
return price * rate_2
elif calc_type == '最小限制':
rate = calc_json['rate']
min = calc_json['min']
return max(min, price * rate)
pass
elif calc_type == '固定比率':
rate = calc_json['rate']
return price * rate
pass
return F.udf(udf_calc_fba, DoubleType())
@staticmethod
def udf_calc_profit(long, width, high, weight, tmp_cost_all_sea, price):
longVal_before = long or 0
width_before = width or 0
high_before = high or 0
weight_before = weight or 0
tmpCost = tmp_cost_all_sea or 0
price_val = price or 0
fee_type_before, fba_fee_before = get_Fba_Fee(longVal_before, width_before, high_before, weight_before)
cost_sum = tmpCost + fba_fee_before
count = 0
long_result = longVal_before
width_result = width_before
high_result = high_before
fee_type_result = fee_type_before
fba_fee_result = fba_fee_before
breakFlag = False
val_43 = 43
if cost_sum > price_val:
while (long_result >= val_43 and count < 4):
tmpVal1 = long_result / 2
tmpVal2 = high_result * 2
breakFlag = tmpVal1 <= tmpVal2
tmp_list = [tmpVal1, width_result, tmpVal2]
tmp_list.sort(reverse=True)
long_result = tmp_list[0]
width_result = tmp_list[1]
high_result = tmp_list[2]
count = count + 1
fee_type_result, fba_fee_result = get_Fba_Fee(long_result, width_result, high_result, weight_before)
if breakFlag:
break
if breakFlag or (count == 4 and long_result > val_43):
long_result = longVal_before
width_result = width_before
high_result = high_before
fee_type_result = fee_type_before
fba_fee_result = fba_fee_before
return {
"long": long_result,
"width": width_result,
"high": high_result,
"fee_type": fee_type_result,
"fba_fee": fba_fee_result,
"long_before": longVal_before,
"width_before": width_before,
"high_before": high_before
}
def get_repartition_num(self):
"""
根据 date_type 设置文件块数
:return:
"""
if self.date_type == DateTypes.day.name:
return 1
if self.date_type == DateTypes.week.name:
return 2
if self.date_type == DateTypes.month.name:
return 2
if self.date_type == DateTypes.last30day.name:
return 2
return 10
def run(self):
# 获取 利润率相关配置表
profit_config_sql = f"""
select category_id,
category_first_id,
cost,
avg_cost,
calc_type,
fba_config_json,
adv,
return_ratio / 100 as return_ratio
from dim_profit_config
where site_name = 'us'
"""
df_profit_config = self.spark.sql(profit_config_sql).cache()
calc_profit_dict = self.get_fba_calc_dict(df_profit_config)
# 获取搜索词相关信息
sql = f"""
select dsam.search_term,
osk.st_key as search_term_id,
dsam.asin,
sorted_volume[0] as long,
sorted_volume[1] as width,
sorted_volume[2] as height,
asin_volume_type,
asin_weight,
asin_weight_type,
st_bsr_cate_1_id_new as category_first_id,
st_bsr_cate_current_id_new as category_id,
asin_price
from (
select search_term,
asin
from dwd_st_asin_measure
where date_type = '{CommonUtil.get_rel_date_type('dwd_st_asin_measure', self.date_type)}'
and date_info = '{self.date_info}'
and site_name = '{self.site_name}'
) dsam
left join
(
select search_term,
st_bsr_cate_1_id_new,
st_bsr_cate_current_id_new
from dim_st_detail
where date_type = '{CommonUtil.get_rel_date_type('dim_st_detail', self.date_type)}'
and date_info = '{self.date_info}'
and site_name = '{self.site_name}'
) dsd on dsd.search_term = dsam.search_term
left join
-- 使用 事实表的四分位价格
(
select search_term,
round(st_price_avg, 6 ) as asin_price
from dwd_st_measure
where date_type = '{CommonUtil.get_rel_date_type('dwd_st_measure', self.date_type)}'
and date_info = '{self.date_info}'
and site_name = '{self.site_name}'
) dad on dad.search_term = dsam.search_term
left join
(
select asin,
asin_weight,
asin_weight_type,
udf_sort_volume(asin_length,asin_width,asin_height) sorted_volume,
asin_volume_type
from dim_asin_stable_info
where site_name = '{self.site_name}'
) dssi on dssi.asin = dsam.asin
inner join
(
select search_term,
st_key
from ods_st_key
where site_name = '{self.site_name}'
) osk on osk.search_term = dsam.search_term
"""
print("======================查询sql如下======================")
print(sql)
df_all = self.spark.sql(sql)
# 长宽高转换
df_all = df_all.withColumn("long", self.udf_transfer_volume_val_reg(F.col("long"), F.col("asin_volume_type")))
df_all = df_all.withColumn("width", self.udf_transfer_volume_val_reg(F.col("width"), F.col("asin_volume_type")))
df_all = df_all.withColumn("height", self.udf_transfer_volume_val_reg(F.col("height"), F.col("asin_volume_type")))
df_all = df_all.withColumn("weight", self.udf_transfer_weight_reg(F.col("asin_weight"), F.col("asin_weight_type")))
st_volume_info = df_all.groupBy("search_term_id") \
.agg(
F.first("search_term").alias("search_term"),
F.first("category_first_id").alias("category_first_id"),
F.first("category_id").alias("category_id"),
F.avg("asin_price").alias("price"),
F.avg(F.col('weight')).alias("weight"),
F.avg(F.col('long')).alias("long"),
F.avg(F.col('width')).alias("width"),
F.avg(F.col('height')).alias("high"),
)
# 平均成本比例
def_cost_rate = df_profit_config.select(F.avg("cost").cast(DecimalType(10, 3)).alias("cost_rate")).first()[0]
# 平均广告费
def_adv = df_profit_config.select(F.avg("adv").cast(DecimalType(10, 3)).alias("adv")).first()[0]
# 平均退款率
def_return_ratio = df_profit_config.select(F.avg("return_ratio").cast(DecimalType(10, 3)).alias("return_ratio")).first()[0]
# 头程空运 运费比例
freight_air_rate = 8.55
# 头程海运 运费比例
freight_sea_rate = 2.06
# 用一级分类进行关联的数据
df_category_first_id_config = df_profit_config.groupby("category_first_id").agg(
F.max("adv").alias("adv"),
F.max("return_ratio").alias("return_ratio"),
)
# 用 categoy_id 进行配置计算的数据
df_category_id_config = df_profit_config.groupby("category_id").agg(
F.max("cost").alias("cost_rate")
)
df_save = st_volume_info \
.join(df_category_first_id_config, ['category_first_id'], "left") \
.join(df_category_id_config, ['category_id'], "left")
df_save = df_save.select(
st_volume_info["search_term"],
st_volume_info["search_term_id"],
st_volume_info["category_first_id"],
st_volume_info["category_id"],
st_volume_info["price"],
st_volume_info["long"],
st_volume_info["width"],
st_volume_info["high"],
st_volume_info["weight"],
df_category_first_id_config["return_ratio"],
df_category_first_id_config["adv"],
df_category_id_config['cost_rate']
)
df_save = df_save.fillna({
# 无分类 填充退款率 佣金
"return_ratio": float(def_return_ratio),
"cost_rate": float(def_cost_rate),
"adv": float(def_adv),
}).cache()
# 计算公式
df_save = df_save.withColumn("referral_fee",
F.round(self.udf_calc_fba_reg(calc_profit_dict)(F.col("category_first_id"), F.col("price")), 4))
# 头程 海运
df_save = df_save.withColumn("ocean_freight", F.expr(f"weight * {freight_sea_rate} /1000"))
# 头程 空运
df_save = df_save.withColumn("air_delivery_fee", F.expr(f"weight * {freight_air_rate} /1000"))
# 运营费固定 平均售价 * 5%
df_save = df_save.withColumn("operating_costs", F.expr(f"price * 0.05"))
# 成本
df_save = df_save.withColumn("costs", F.expr(f" price * cost_rate "))
# 广告占比 adv
df_save = df_save.withColumn("advertise", F.expr(f" price * adv "))
# 退款率 * 价格 即为退款额
df_save = df_save.withColumn("return_ratio", F.expr(f" price * return_ratio "))
# 除了fba之外的所有的费用
df_save = df_save.withColumn("tmp_cost_all_sea",
F.expr("ocean_freight + referral_fee + return_ratio + costs + advertise + operating_costs"))
# 计算利润率
df_save = df_save.withColumn("tmp_row", self.udf_calc_profit_reg(
F.col("long"),
F.col("width"),
F.col("high"),
F.col("weight"),
F.col("tmp_cost_all_sea"),
F.col("price"),
))
df_save = df_save.withColumn("fba_fee", F.col("tmp_row").getField("fba_fee"))
df_save = df_save.withColumn("gross_profit_fee_sea",
F.expr(
"(price-(ocean_freight +referral_fee+return_ratio +costs+advertise+operating_costs + fba_fee))/price")
.cast(DecimalType(10, 3)))
df_save = df_save.withColumn("gross_profit_fee_air",
F.expr(
"(price-(air_delivery_fee +referral_fee+return_ratio +costs+advertise+operating_costs + fba_fee))/price")
.cast(DecimalType(10, 3)))
df_save = df_save.select(
F.col("search_term"),
F.col("search_term_id"),
st_volume_info["category_first_id"],
st_volume_info["category_id"],
F.col("weight").cast(DecimalType(10, 3)),
F.col("price").cast(DecimalType(10, 3)),
F.col("referral_fee").cast(DecimalType(10, 3)),
# 长宽高之前
F.col("tmp_row").getField("long_before").cast(DecimalType(10, 3)).alias("long_before"),
F.col("tmp_row").getField("width_before").cast(DecimalType(10, 3)).alias("width_before"),
F.col("tmp_row").getField("high_before").cast(DecimalType(10, 3)).alias("high_before"),
# 计算后
F.col("tmp_row").getField("long").cast(DecimalType(10, 3)).alias("longs"),
F.col("tmp_row").getField("width").cast(DecimalType(10, 3)).alias("width"),
F.col("tmp_row").getField("high").cast(DecimalType(10, 3)).alias("high"),
F.col("tmp_row").getField("fba_fee").cast(DecimalType(10, 3)).alias("fba_fee"),
F.col("tmp_row").getField("fee_type").cast(IntegerType()).alias("fee_type"),
F.col("return_ratio"),
F.col("ocean_freight"),
F.col("air_delivery_fee"),
F.col("operating_costs"),
F.col("costs"),
F.col("advertise"),
F.col("gross_profit_fee_sea"),
F.col("gross_profit_fee_air"),
F.col("cost_rate"),
F.lit(self.site_name).alias("site_name"),
F.lit(self.date_type).alias("date_type"),
F.lit(self.date_info).alias("date_info")
).where("weight < 10000000")
# 分区数量调整
df_save = df_save.repartition(self.get_repartition_num())
partition_dict = {
"site_name": self.site_name,
"date_type": self.date_type,
"date_info": self.date_info,
}
df_save = CommonUtil.format_df_with_template(self.spark, df_save, self.hive_tb, True)
hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict)
HdfsUtils.delete_hdfs_file(hdfs_path)
partition_by = list(partition_dict.keys())
print(f"当前存储的表名为：:self.hive_tb,分区为{partition_by}", )
df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=partition_by)
print("success")
if __name__ == '__main__':
site_name = CommonUtil.get_sys_arg(1, None)
date_type = CommonUtil.get_sys_arg(2, None)
date_info = CommonUtil.get_sys_arg(3, None)
obj = DwdStVolumeFba(site_name, date_type, date_info)
obj.run()