dwd_st_volume_fba_old.py 25.8 KB
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680
import json
import os
import re
import sys

sys.path.append(os.path.dirname(sys.path[0]))
from utils.db_util import DBUtil
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
from pyspark.sql.types import MapType, StringType, DecimalType, IntegerType, Row, DoubleType

"""
搜索词 计算=>平均长宽高 => 计算基本利润率相关数据
依赖 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
"""


def get_Fba_Fee(longVal: float,
                width: float,
                high: float,
                weight: float,
                ):
    fee_type = 0
    fba_fee = 0
    if (longVal <= 36 and width <= 28 and high <= 1.6 and weight <= 113.5):
        fee_type = 1
        fba_fee = 3.22
    elif (longVal <= 36 and width <= 28 and high <= 1.6 and weight > 113.5 and weight <= 227):
        fee_type = 2
        fba_fee = 3.4
    elif (longVal <= 36 and width <= 28 and high <= 1.6 and weight > 227 and weight <= 340.5):
        fee_type = 3
        fba_fee = 3.58
    elif (longVal <= 36 and width <= 28 and high <= 1.6 and weight > 340.5 and weight <= 454):
        fee_type = 4
        fba_fee = 3.77
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight <= 113.5):
        fee_type = 5
        fba_fee = 3.86
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 113.5 and weight <= 227):
        fee_type = 6
        fba_fee = 4.08
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 227 and weight <= 340.5):
        fee_type = 7
        fba_fee = 4.24
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 340.5 and weight <= 454):
        fee_type = 8
        fba_fee = 4.75
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 454 and weight <= 681):
        fee_type = 9
        fba_fee = 5.4
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 681 and weight <= 908):
        fee_type = 10
        fba_fee = 5.69
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 908 and weight <= 1135):
        fee_type = 11
        fba_fee = 6.1
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 1135 and weight <= 1362):
        fee_type = 12
        fba_fee = 6.39
    elif (longVal <= 43 and width <= 34 and high <= 19 and weight > 1362 and weight <= 9080):
        fee_type = 13
        fba_fee = 7.33
    elif (longVal <= 152.4 and (longVal + 2 * (width + high)) <= 330.2 and weight <= 31780):
        fee_type = 14
        fba_fee = 10.15
    elif (longVal <= 274.32 and (longVal + 2 * (width + high)) <= 419.1 and weight <= 68100):
        fee_type = 15
        fba_fee = 19.47
    elif (longVal <= 274.32 and (longVal + 2 * (width + high)) > 419.1 and weight <= 68100):
        fee_type = 16
        fba_fee = 90.81
    elif (longVal > 274.32 and (longVal + 2 * (width + high)) > 419.1 and weight > 68100):
        fee_type = 17
        fba_fee = 159.32
    return (fee_type, fba_fee)


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.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_parse_volume(asin_volume: str, site_name: str):
        """
        解析
        :param asin_volume:
        :return:
        """

        def safeIndex(list: list, index: int, default: object = None):
            if index <= len(list) - 1:
                return list[index]
            return default

        def to_float(obj, default: object = None):
            try:
                return float(obj)
            except:
                return default

        resultArr = []
        if asin_volume is not None:
            pattern = r"([0-9.]+)"
            resultArr = re.findall(pattern, asin_volume, flags=re.IGNORECASE)

        #  倒序
        resultArr.sort(reverse=True)

        longVal = to_float(safeIndex(resultArr, 0, None), None)
        width = to_float(safeIndex(resultArr, 1, None), None)
        height = to_float(safeIndex(resultArr, 2, None), None)

        if site_name == 'us':
            #  英寸
            rate = 2.54
        else:
            rate = 1

        return {
            "long": None if longVal is None else round(rate * longVal, 4),
            "width": None if width is None else round(rate * width, 4),
            "height": None if height is None else round(rate * height, 4),
        }

    @staticmethod
    def udf_calc_avg(
            val1_col=0,
            val2_col=0,
            val3_col=0,
            val1_count=0,
            val2_count=0,
            val3_count=0
    ):
        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 udf_sort_val():
        """
        长宽高重新排序
        :return:
        """

        def udf_sort_inner(val1, val2, val3):
            # 从大到小排序
            arr = [val1 or 0, val2 or 0, val3 or 0, ]
            arr.sort(reverse=True)
            return {
                "long": arr[0],
                "width": arr[1],
                "height": arr[2],
            }

        return F.udf(udf_sort_inner, MapType(StringType(), DecimalType()))

    @staticmethod
    def udf_calc_fba_reg(config_dict: dict):
        """
        根据配置表计算fba费用
        :param config_dict:
        :return:
        """

        def udf_calc_fba(one_categoy_id: str, price):
            if price is None:
                return None
            config_row = config_dict.get(str(one_categoy_id))
            if config_row is None:
                #  默认佣金比例是0.15
                return price * 0.15
            else:
                calc_type = config_row['calc_type']
                calc_json = json.loads(config_row['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

        # result_list.append()
        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):
        sql = f"""
    select dsam.search_term,
       osk.st_key       as search_term_id,
       dsam.asin,
       asin_volume,
       asin_weight,
       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 asin,
                asin_weight,
                asin_volume,
                asin_price
         from dim_asin_detail
             where date_type = '{CommonUtil.get_rel_date_type('dim_asin_detail', self.date_type)}'
           and date_info = '{self.date_info}'
           and site_name = '{self.site_name}'
     ) dad on dad.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("tmp_row", self.udf_parse_volume_reg(F.col("asin_volume"), F.lit(self.site_name)))
        df_all = df_all.withColumn("long", F.col("tmp_row").getField("long"))
        df_all = df_all.withColumn("width", F.col("tmp_row").getField("width"))
        df_all = df_all.withColumn("height", F.col("tmp_row").getField("height"))
        df_all.drop("tmp_row")

        st_agg_one = df_all.groupBy("search_term_id") \
            .agg(
            F.first("categoy_id").alias("categoy_id"),
            F.first("current_categoy_id").alias("current_categoy_id"),
            F.first("search_term").alias("search_term"),
            F.avg("asin_price").alias("price"),
            # 磅 => 克 并过滤无效数据 todo 其他us站点的
            F.expr("round(avg(asin_weight * 453.59237),4)").alias("weight")
        ).select(
            F.col("search_term"),
            F.col("search_term_id"),
            F.col("categoy_id"),
            F.col("current_categoy_id"),
            F.col("price"),
            F.col("weight")
        )

        long_df = df_all.withColumn("flag",
                                    F.when(F.expr("0 < long and long <= 50"), F.lit(1))
                                    .when(F.expr("50 < long and long <= 100"), F.lit(2))
                                    .when(F.expr("100 < long and long <= 500"), F.lit(3))
                                    .otherwise(None)
                                    ) \
            .groupBy(F.col("search_term_id")) \
            .pivot("flag", [1, 2, 3]) \
            .agg(F.avg("long").cast(DecimalType(10, 2)).alias("val"), F.count("flag").alias("row_count")) \
            .select(
            F.col("search_term_id"),
            self.udf_calc_avg_reg(
                F.col("1_val"), F.col("2_val"), F.col("3_val"),
                F.col("1_row_count"), F.col("2_row_count"), F.col("3_row_count")
            ).alias("long")
        ).cache()

        width_df = df_all.withColumn("flag",
                                     F.when(F.expr("0 < width and width <= 50"), F.lit(1))
                                     .when(F.expr("50 < width and width <= 100"), F.lit(2))
                                     .when(F.expr("100 < width and width <= 500"), F.lit(3))
                                     .otherwise(None)
                                     ) \
            .groupBy(F.col("search_term_id")) \
            .pivot("flag", [1, 2, 3]) \
            .agg(F.avg("width").cast(DecimalType(10, 2)).alias("val"), F.count("flag").alias("row_count")) \
            .select(
            F.col("search_term_id"),
            self.udf_calc_avg_reg(
                F.col("1_val"), F.col("2_val"), F.col("3_val"),
                F.col("1_row_count"), F.col("2_row_count"), F.col("3_row_count")
            ).alias("width")
        ).cache()

        height_df = df_all.withColumn("flag",
                                      F.when(F.expr("0 < height and height <= 50"), F.lit(1))
                                      .when(F.expr("50 < height and height <= 100"), F.lit(2))
                                      .when(F.expr("100 < height and height <= 500"), F.lit(3))
                                      .otherwise(None)
                                      ) \
            .groupBy(F.col("search_term_id")) \
            .pivot("flag", [1, 2, 3]) \
            .agg(F.avg("height").cast(DecimalType(10, 2)).alias("val"), F.count("flag").alias("row_count")) \
            .select(
            F.col("search_term_id"),
            self.udf_calc_avg_reg(
                F.col("1_val"), F.col("2_val"), F.col("3_val"),
                F.col("1_row_count"), F.col("2_row_count"), F.col("3_row_count")
            ).alias("height")
        ).cache()

        st_volume_info = st_agg_one \
            .join(long_df, "search_term_id") \
            .join(width_df, "search_term_id") \
            .join(height_df, "search_term_id") \
            .select(
            st_agg_one["search_term"],
            st_agg_one["search_term_id"],
            st_agg_one["categoy_id"],
            st_agg_one["current_categoy_id"],
            st_agg_one["price"],
            st_agg_one["weight"],
            long_df["long"],
            width_df["width"],
            height_df["height"].alias("high")
        ).fillna({
            "long": 0,
            "width": 0,
            "high": 0,
            "categoy_id": 0,
            "current_categoy_id": 0,
            "price": 0,
            "weight": 0
        })

        conn_info = DBUtil.get_connection_info("postgresql", "us")
        config_sql1 = f"""
        select categoy_id as one_categoy_id,
               categoy_name,
               referral_fee_formula,
               upfc.calc_type,
               upfc.config_json,
               adv::decimal(10, 3),
               (
                   select round(avg(return_ratio) / 100, 2)::decimal(10, 3)
                   from us_aba_profit_category_insights
               )          as return_ratio
        from us_profit_fba_config upfc
                 left join us_profit_adv upa on upa.category = upfc.categoy_name    
        """
        df_profit_join = SparkUtil.read_jdbc_query(
            session=self.spark,
            url=conn_info["url"],
            pwd=conn_info["pwd"],
            username=conn_info["username"],
            query=config_sql1
        ).cache()

        calc_profit_dict = {str(row['one_categoy_id']): row.asDict() for row in df_profit_join.collect()}

        config_sql2 = f"""
        select rel_category_id                         as rel_category_id,
               ubc.one_category_id                     as one_category_id,
               round(avg(cost), 2)::decimal(10, 3)     as cost_rate,
               round(avg(avg_cost), 2)::decimal(10, 3) as avg_cost_rate
    from (
             select upcn.name                                    as header_name,
                    (replace(cost, '%', '')::decimal) / 100      as cost,
                    (replace(avg_cost, '%', '') ::decimal) / 100 as avg_cost,
                    first_name,
                    last_name,
                    coalesce(tmp2.id, tmp1.id, upcn.first_id)    as rel_category_id
             from public.us_profit_cost_new upcn
                      left join (
                 select min(id)                   as id,
                        replace(en_name, ' ', '') as name
                 from us_bs_category
                 where nodes_num = 2
                 group by en_name
             ) tmp1 on upcn.first_name = tmp1.name
                      left join (
                 select min(id)                   as id,
                        replace(en_name, ' ', '') as name
                 from us_bs_category
                 where nodes_num > 2
                 group by en_name
             ) tmp2 on upcn.last_name = tmp2.name
         ) tmp
             inner join us_bs_category ubc on ubc.id = tmp.rel_category_id
    group by one_category_id, rel_category_id
        """
        df_cost_join = SparkUtil.read_jdbc_query(
            session=self.spark,
            url=conn_info["url"],
            pwd=conn_info["pwd"],
            username=conn_info["username"],
            query=config_sql2
        ).cache()

        # 平均成本比例
        def_cost_rate = df_cost_join.select(F.avg("cost_rate").cast(DecimalType(10, 3)).alias("cost_rate")).first()[0]
        # 平均广告费
        def_adv = df_profit_join.select(F.avg("adv").cast(DecimalType(10, 3)).alias("adv")).first()[0]
        # 平均退款率
        def_return_ratio = df_profit_join.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_save = st_volume_info \
            .join(df_profit_join, st_volume_info['categoy_id'].eqNullSafe(df_profit_join['one_categoy_id']), "left") \
            .join(df_cost_join, st_volume_info['current_categoy_id'].eqNullSafe(df_cost_join['rel_category_id']), "left")

        df_save = df_save.select(
            st_volume_info["search_term"],
            st_volume_info["search_term_id"],
            st_volume_info["categoy_id"],
            st_volume_info["current_categoy_id"],
            st_volume_info["price"],
            st_volume_info["long"],
            st_volume_info["width"],
            st_volume_info["high"],
            st_volume_info["weight"],
            df_profit_join["return_ratio"],
            df_profit_join["adv"],
            df_cost_join['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("categoy_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"),
            F.col("categoy_id"),
            st_volume_info["current_categoy_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,
        }
        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()