common_udf.py

import re
from datetime import datetime
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, MapType, StringType

from yswg_utils.udf_util import UdfUtil
import json


def udf_title_number_parse_reg():
    def udf_title_number_parse(title):
        val = udf_get_package_quantity(title)
        if val is not None:
            return [{
                "match": None,
                "label": None,
                "value": udf_get_package_quantity(title),
            }]
        return None

    return F.udf(udf_title_number_parse, ArrayType(MapType(StringType(), StringType())))


def get_Fba_Fee(longVal: float,
                width: float,
                high: float,
                weight: float,
                ):
    """
    根据长宽高计算fba类型，长宽高
    :param longVal:长=> cm
    :param width: 宽=> cm
    :param high:  高=> cm
    :param weight: 重量单位为g
    :return:
    """
    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)


def udf_get_package_quantity_with_flag(title):
    """
    获取打包数量
    :param title:
    :return:
    """
    if title != '':
        title = str(title).lower()
        title = title.replace(' ', ' ')
        eligible_list = []
        unit_list = []
        parse_list = []
        thousand_bit_symmbol_count = 0
        eligible_value_map = {"one": 1, "two": 2, "three": 3, "four": 4, "five": 5,
                              "six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10}
        key_list = ['set', 'pack', 'pair', 'box', 'quantity']
        if title not in ['null', 'none']:
            patterns = [
                r'\b(?<!\d\.\d)(?<!\d\sx\s)((?:\d{1,3})(?:,\d{3})*|\d+)(?!\.\d)(?!%)[-_\s]*(bulk|total|pc|pcs|piece|pieces|set|pack|packs|pairs|pk|pair|count|ct|counts|sets|sheets|sheet|wrap|wraps|roll|rolls|box|boxes|quantity)(?![a-zA-Z])',
                r'\b((?:set|pack|pair|pairs|box|case|carton|quantity) of) ((?:\d{1,3})(?:,\d{3})*|\d+)(?!\.\d)(?!\sx\s\d)(?!%)\b',
                r'\b(total|count|quantity)\s*[-_\s]*\s*((?:\d{1,3})(?:,\d{3})*|\d+)(?!\.)(?!\sx\s\d)\b(?!%)',
                r'\b(one|two|three|four|five|six|seven|eight|nine|ten)(?: +)(bulk|total|pc|pcs|piece|pieces|set|pack|packs|pairs|pk|pair|count|ct|counts|sets|sheets|sheet|wrap|wraps|roll|rolls|box|boxes|quantity)(?![a-zA-Z])',
                r'\b((?:set|pack|pair|pairs|box|case|carton|quantity) of) (one|two|three|four|five|six|seven|eight|nine|ten)\b',
                r'\b(total|count|quantity)(?: +)(one|two|three|four|five|six|seven|eight|nine|ten)\b'
            ]
            for i in range(len(patterns)):
                pattern = patterns[i]
                result_list = re.findall(pattern, title)
                if len(result_list) > 0:
                    for result in result_list:
                        if i in [0, 3]:
                            eligible_element = result[0]
                            unit_element = result[1]
                        else:
                            eligible_element = result[1]
                            unit_element = result[0]
                        eligible_list.append(eligible_element)
                        unit_list.append(unit_element)
            if eligible_list and unit_list:
                for key in key_list:
                    if key in unit_list and f'{key} of' in unit_list:
                        index = unit_list.index(key)
                        unit_list.pop(index)
                        eligible_list.pop(index)
                for eligible_element in eligible_list:
                    if eligible_element in eligible_value_map.keys():
                        eligible_element_value = eligible_value_map[eligible_element]
                        parse_list.append(int(eligible_element_value))
                    else:
                        if str(eligible_element).count(',') > 0:
                            thousand_bit_symmbol_count = thousand_bit_symmbol_count + 1
                            eligible_element = str(eligible_element).replace(',', '')
                        if (not str(eligible_element).startswith('0')) and (int(eligible_element) < 10000) and (
                                int(eligible_element) >= 0):
                            parse_list.append(int(eligible_element))
        if len(parse_list) == 1:
            return parse_list[0], 0
        elif len(parse_list) > 1:
            if thousand_bit_symmbol_count >= 2 and len(parse_list) >= 3:
                return min(parse_list), 1
            else:
                return max(parse_list), 1
        else:
            return None, None


def udf_get_package_quantity(title):
    """
    获取打包数量
    :param title:
    :return:
    """
    if title != '':
        title = str(title).lower()
        title = title.replace(' ', ' ')
        eligible_list = []
        unit_list = []
        parse_list = []
        thousand_bit_symmbol_count = 0
        eligible_value_map = {"one": 1, "two": 2, "three": 3, "four": 4, "five": 5,
                              "six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10}
        key_list = ['set', 'pack', 'pair', 'box', 'quantity']
        if title not in ['null', 'none']:
            patterns = [
                r'\b(?<!\d\.\d)(?<!\d\sx\s)((?:\d{1,3})(?:,\d{3})*|\d+)(?!\.\d)(?!%)[-_\s]*(bulk|total|pc|pcs|piece|pieces|set|pack|packs|pairs|pk|pair|count|ct|counts|sets|sheets|sheet|wrap|wraps|roll|rolls|box|boxes|quantity)(?![a-zA-Z])',
                r'\b((?:set|pack|pair|pairs|box|case|carton|quantity) of) ((?:\d{1,3})(?:,\d{3})*|\d+)(?!\.\d)(?!\sx\s\d)(?!%)\b',
                r'\b(total|count|quantity)\s*[-_\s]*\s*((?:\d{1,3})(?:,\d{3})*|\d+)(?!\.)(?!\sx\s\d)\b(?!%)',
                r'\b(one|two|three|four|five|six|seven|eight|nine|ten)(?: +)(bulk|total|pc|pcs|piece|pieces|set|pack|packs|pairs|pk|pair|count|ct|counts|sets|sheets|sheet|wrap|wraps|roll|rolls|box|boxes|quantity)\b',
                r'\b((?:set|pack|pair|pairs|box|case|carton|quantity) of) (one|two|three|four|five|six|seven|eight|nine|ten)\b',
                r'\b(total|count|quantity)(?: +)(one|two|three|four|five|six|seven|eight|nine|ten)\b'
            ]
            for i in range(len(patterns)):
                pattern = patterns[i]
                result_list = re.findall(pattern, title)
                if len(result_list) > 0:
                    for result in result_list:
                        if i in [0, 3]:
                            eligible_element = result[0]
                            unit_element = result[1]
                        else:
                            eligible_element = result[1]
                            unit_element = result[0]
                        eligible_list.append(eligible_element)
                        unit_list.append(unit_element)
            if eligible_list and unit_list:
                for key in key_list:
                    if key in unit_list and f'{key} of' in unit_list:
                        index = unit_list.index(key)
                        unit_list.pop(index)
                        eligible_list.pop(index)
                for eligible_element in eligible_list:
                    if eligible_element in eligible_value_map.keys():
                        eligible_element_value = eligible_value_map[eligible_element]
                        parse_list.append(int(eligible_element_value))
                    else:
                        if str(eligible_element).count(',') > 0:
                            thousand_bit_symmbol_count = thousand_bit_symmbol_count + 1
                            eligible_element = str(eligible_element).replace(',', '')
                        if (not str(eligible_element).startswith('0')) and (int(eligible_element) < 10000) and (
                                int(eligible_element) >= 0):
                            parse_list.append(int(eligible_element))
        if len(parse_list) == 1:
            return parse_list[0]
        elif len(parse_list) > 1:
            if thousand_bit_symmbol_count >= 2 and len(parse_list) >= 3:
                return min(parse_list)
            else:
                return max(parse_list)
        else:
            return None


# 公用函数-处理String类型空值返回NoneType
def udf_handle_string_null_value(value):
    # 转小写并去除头尾空值
    if value is not None:
        handle_value = str(value).strip().lower()
        if handle_value in ['null', 'none', '', '-1']:
            return None
        else:
            return value
    return None


def parse_best_sellers_href(href: str):
    """
    根据asin想去 best_sellers 解析获取一级分类/当前分类
    :param href:
    :return:
    """
    arr = href.split("/")

    last_val = UdfUtil.safeIndex(arr, len(arr) - 1, None)

    if "ref=" in last_val:
        category_id = UdfUtil.safeIndex(arr, len(arr) - 2, None)
        category_first_id = UdfUtil.safeIndex(arr, len(arr) - 3, None)
    else:
        category_id = UdfUtil.safeIndex(arr, len(arr) - 1, None)
        category_first_id = UdfUtil.safeIndex(arr, len(arr) - 2, None)

    return (category_id, category_first_id)
    pass


def parse_bsr_url(nodes_num: int, url: str):
    """
    统一解析链接获取 bsr 分类id等数据
    :param nodes_num:
    :param url:
    :return:
    """
    arr = url.split("/")
    if not "ref=" in url:
        ref_suffix = None
        category_id = UdfUtil.safeIndex(arr, len(arr) - 1, None)
        category_first_id = UdfUtil.safeIndex(arr, len(arr) - 2, None)
    else:
        ref_suffix = UdfUtil.safeIndex(arr, len(arr) - 1, None)
        category_first_id = UdfUtil.safeIndex(arr, len(arr) - 3, None)
        category_id = UdfUtil.safeIndex(arr, len(arr) - 2, None)
    if nodes_num == 1:
        level = 1
    elif url.endswith("_0"):
        level = 2
    elif url.endswith(f"{category_first_id}_1"):
        level = 3
    else:
        level = 4

    # 获取 parent id
    if level == 1:
        # 根节点
        category_id = "0"
        category_first_id = None
        category_parent_id = None
    elif level == 2:
        # 一级节点
        category_id = category_id
        category_first_id = category_id
        category_parent_id = "0"
    elif level == 3:
        # 一级节点下的次级节点
        category_id = category_id
        category_first_id = category_first_id
        category_parent_id = category_first_id
    elif level == 4:
        category_id = category_id
        category_first_id = category_first_id
        if ref_suffix is not None:
            category_parent_id = ref_suffix[ref_suffix.rfind("_") + 1:]
        else:
            category_parent_id = None
    else:
        category_id = None
        category_first_id = None
        category_parent_id = None

    return {
        "category_id": category_id,
        "category_first_id": category_first_id,
        "category_parent_id": category_parent_id
    }
    pass


def parse_weight_str(weight_str: str, site_name: str):
    """
    解析重量字符串获取重量和单位，逗号分隔
    :param weight_str:
    :param site_name:
    :return:
    """
    val = None
    weight_type = 'pounds' if site_name == 'us' else 'grams'
    if weight_str is not None:
        if 'pounds' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}pounds", weight_str)
            val = round(float(match.group(1)), 3) if site_name == 'us' and match else round(
                float(match.group(1)) * 1000 * 0.454, 3) if match else None
        elif 'ounces' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}ounces", weight_str)
            val = round(float(match.group(1)) / 16, 3) if site_name == 'us' and match else round(
                float(match.group(1)) / 16 * 1000 * 0.454, 3) if match else None
        elif any(substring in weight_str for substring in ['kilogram', ' kg']):
            weight_str = weight_str.replace(' kg', ' kilogram')
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}kilogram", weight_str)
            val = round(float(match.group(1)) / 0.454, 3) if site_name == 'us' and match else round(
                float(match.group(1)) * 1000, 3) if match else None
        elif any(substring in weight_str for substring in ['milligrams']):
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}milligrams", weight_str)
            val = round(float(match.group(1)) / 1000 / 1000 / 0.454, 3) if site_name == 'us' and match else round(
                float(match.group(1)) / 1000, 3) if match else None
        elif ' gram' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,} gram", weight_str)
            val = round(float(match.group(1)) / 1000 / 0.454, 3) if site_name == 'us' and match else round(
                float(match.group(1)), 3) if match else None
        elif ' g' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,} g", weight_str)
            val = round(float(match.group(1)) / 1000 / 0.454, 3) if site_name == 'us' and match else round(
                float(match.group(1)), 3) if match else None

    return val, weight_type


def udf_new_asin_flag(launch_time, cal_day):
    """
    计算asin是否新品标记公共udf函数
    :param launch_time: asin上架时间
    :param cal_day: 计算日期（周取周最后一天，月取月最后一天，需调用工具类get_calDay_by_dateInfo获取）
    :return: 是否新品标记，1为新品；0为非新品
    """
    if launch_time is None or cal_day is None:
        return None

    date_format = "%Y-%m-%d"
    try:
        # 将日期字符串转换为 datetime 对象
        datetime1 = datetime.strptime(launch_time, date_format)
        datetime2 = datetime.strptime(cal_day, date_format)
        # 计算日期的偏移量
        offset = (datetime2 - datetime1).days
        # 判断偏移量是否小于180天，是则确定为新品
        if offset <= 180:
            return 1
        else:
            return 0
    except ValueError:
        # 日期字符串格式不正确
        return None


def category_craw_flag(category_first_id, asin: str = None):
    """
    用于判断asin或者分类是否爬取
    :param category_first_id:
    :param asin:
    """
    if asin is not None and not asin.startswith("B0"):
        return False
    arr = [
        "audible",
        "books",
        "digital-text",
        "dmusic",
        "mobile-apps",
        "movies-tv",
        "music",
        "software",
        "videogames"
    ]
    if category_first_id is None or category_first_id in arr:
        return False
    return True


def sort_volume(val1, val2, val3):
    """
    排序长宽高
    """

    def custom_sort(item):
        if item is None:
            return 0
        return item

    arr = [val1, val2, val3]
    arr.sort(key=custom_sort, reverse=True)

    l = UdfUtil.safeIndex(arr, 0, None)
    w = UdfUtil.safeIndex(arr, 1, None)
    h = UdfUtil.safeIndex(arr, 2, None)
    return l, w, h


def parse_asin_volume_str(volume_str, sortFlag=False):
    """
    解析 volume_str
    :param volume_str: 体积长宽高字符串
    :param sortFlag: 是否按照大小定义长宽高排序
    :return: l, w, h, type 返回长宽高单位（均为原始数据）
    """
    if volume_str is None:
        return None, None, None, None

    types = re.findall(r"inches|inch|cm|centímetros|centimetres|milímetros|millimeter|mm|metros", volume_str)
    #  多个单位的截取第一个单位
    if len(types) >= 2:
        volume_str = volume_str[0:volume_str.find(types[0])]

    matches = re.findall(r"(\d+(\.\d+)?)", volume_str)
    values = [float(val[0]) for val in matches]
    if sortFlag:
        values.sort(reverse=True)

    l = UdfUtil.safeIndex(values, 0, None)
    w = UdfUtil.safeIndex(values, 1, None)
    h = UdfUtil.safeIndex(values, 2, None)
    type = UdfUtil.safeIndex(types, 0, None)
    if type in ['inches', 'inch']:
        type = "inches"
    elif type in ['cm', 'centímetros', 'centimetres']:
        type = "cm"
    elif type in ['milímetros', 'millimeter', 'mm']:
        type = "mm"
    elif type in ['metros']:
        type = "m"
    else:
        type = "inches"
        sortVal = re.findall(r"l|d|w|h", volume_str)
        if len(sortVal) > 0:
            tmpMap = {
                str(key): UdfUtil.safeIndex(values, i, None) for i, key in enumerate(sortVal)
            }
            l = tmpMap.get("l") or tmpMap.get("d")
            w = tmpMap.get("w")
            h = tmpMap.get("h")
    return l, w, h, type


def udf_rank_and_category(best_sellers_rank):
    """
    解析bs分类名称和排名
    """
    pattern = r"#([\d,]+) in ([\w&' ]+)"
    matches = re.findall(pattern, best_sellers_rank)

    bs_rank_str = ",".join([rank.replace(",", "") for rank, category in matches])
    bs_category_str = ",".join([category.strip().replace(",", " ") for rank, category in matches])
    return bs_rank_str, bs_category_str


def udf_ele_mattch(match_text: str, ele_list_str: str):
    """
    字符串多包含多个元素精准匹配
    :param match_text: 待匹配的字符串
    :param ele_list_str: 需要匹配的匹配词list(此处可将list直接str(list)传入)
    :return: 返回字符串中匹配到的多个匹配词的字符串。采用”,“拼接，可根据","拆分;如都无匹配结果则为None
    """
    pattern = re.compile(r'(?<!\+|\*|\-|\%|\.)\b({})\b'.format('|'.join([re.escape(x) for x in ele_list_str])), flags=re.IGNORECASE)
    ele_list = re.findall(pattern, match_text)
    if ele_list:
        return ','.join(set(ele_list))
    else:
        return None


# 插件-体积标准提取
def udf_extract_volume_format(volume_str: str):
    # 解析类型
    # pattern = r'\b\w+\b'
    volume_str = str(volume_str).lower()
    pattern = r'[a-z]+'
    matches = re.findall(pattern, volume_str)

    # 使用集合存储匹配的单词
    type_set = set()
    for word in matches:
        if word in ['inches', 'inch']:
            type_set.add('inches')
        elif word in ['cm', 'centímetros', 'centimetres']:
            type_set.add('cm')
        elif word in ['milímetros', 'millimeter', 'mm']:
            type_set.add('mm')
        elif word in ['metros']:
            type_set.add('m')

    # 根据集合的长度返回结果
    if len(type_set) == 1:
        asin_volume_type = list(type_set)[0]
    elif len(type_set) >= 2:
        asin_volume_type = ','.join(type_set)
    else:
        asin_volume_type = ''

    # 解析长宽高
    # length, width, height = None, None, None
    if asin_volume_type == 'cm,inches':
        num_inches = volume_str.find('inch')
        num_cm = volume_str.find('cm')
        volume_str = volume_str[:num_inches] if num_cm > num_inches else volume_str[num_cm:num_inches]
    dimensions = re.findall(r"(\d+(\.\d+)?)", volume_str)
    dimensions = [float(dim[0]) for dim in dimensions]

    if len(dimensions) == 1:
        length = dimensions[0]
        result = f"{length}"
    elif len(dimensions) == 2:
        # if asin_volume_type == '':
        #     if "l" in volume_str and "w" in volume_str:
        #         length, width = dimensions
        #     elif "w" in volume_str and "h" in volume_str:
        #         width, height = dimensions
        #     elif "l" in volume_str and "h" in volume_str:
        #         length, height = dimensions
        #     elif "d" in volume_str and "w" in volume_str:
        #         length, width = dimensions
        #     elif "d" in volume_str and "h" in volume_str:
        #         length, height = dimensions
        # else:
        #     length, width = dimensions
        length, width = dimensions
        result = f"{length}*{width}"
    elif len(dimensions) == 3:
        length, width, height = dimensions
        result = f"{length}*{width}*{height}"
    elif len(dimensions) >= 4:
        length, width, height = dimensions[:3]
        result = f"{length}*{width}*{height}"
    else:
        result = ""

    if asin_volume_type == "inches":
        # 单位转换成cm
        return "*".join([str(round(float(dim) * 2.54, 2)) for dim in result.split("*")]) + "cm"
    else:
        return f"{result}{asin_volume_type}"


# 大数据 -- 返回长+宽+高+类型
def udf_extract_volume_dimensions(volume_str: str):
    # 解析类型
    # pattern = r'\b\w+\b'
    volume_str = str(volume_str).lower()
    pattern = r'[a-z]+'
    matches = re.findall(pattern, volume_str)

    # 使用集合存储匹配的单词
    type_set = set()
    for word in matches:
        if word in ['inches', 'inch']:
            type_set.add('inches')
        elif word in ['cm', 'centímetros', 'centimetres']:
            type_set.add('cm')
        elif word in ['milímetros', 'millimeter', 'mm']:
            type_set.add('mm')
        elif word in ['metros']:
            type_set.add('m')

    # 根据集合的长度返回结果
    if len(type_set) == 1:
        asin_volume_type = list(type_set)[0]
    elif len(type_set) >= 2:
        asin_volume_type = ','.join(type_set)
    else:
        asin_volume_type = ''

    # 解析长宽高
    length, width, height = None, None, None
    if asin_volume_type == 'cm,inches':
        num_inches = volume_str.find('inch')
        num_cm = volume_str.find('cm')
        volume_str = volume_str[:num_inches] if num_cm > num_inches else volume_str[num_cm:num_inches]
    dimensions = re.findall(r"(\d+(\.\d+)?)", volume_str)
    dimensions = [float(dim[0]) for dim in dimensions]

    if len(dimensions) == 1:
        length = dimensions[0]
    elif len(dimensions) == 2:
        if asin_volume_type == '':
            if "l" in volume_str and "w" in volume_str:
                length, width = dimensions
            elif "w" in volume_str and "h" in volume_str:
                width, height = dimensions
            elif "l" in volume_str and "h" in volume_str:
                length, height = dimensions
            elif "d" in volume_str and "w" in volume_str:
                length, width = dimensions
            elif "d" in volume_str and "h" in volume_str:
                length, height = dimensions
            else:
                length, width = dimensions
        else:
            length, width = dimensions
        asin_volume_type = "inches"
    elif len(dimensions) >= 3:
        length, width, height = dimensions[:3]
        asin_volume_type = "inches"

    # 降序排序
    example_list = [length, width, height]

    # 使用 sorted 函数进行排序
    # key=lambda x: (x is not None, x) 确保 None 值被视为最小值并排在最后
    # reverse=True 以实现降序排序
    sorted_list = sorted(example_list, key=lambda x: (x is not None, x), reverse=True)
    length, width, height = sorted_list
    return (length, width, height, asin_volume_type)


# 插件-重量标准提取
def udf_extract_weight_format(weight_str: str):
    """
    解析重量字符串获取重量和单位，逗号分隔
    :param weight_str:
    :param site_name:
    :return:
    """
    val = None
    # weight_type = 'pounds' if site_name == 'us' else 'grams'
    weight_type = 'g'
    if weight_str is not None:
        if 'pounds' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}pounds", weight_str)
            val = round(float(match.group(1)) * 1000 * 0.454, 3) if match else None
        elif 'ounces' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}ounces", weight_str)
            val = round(float(match.group(1)) / 16 * 1000 * 0.454, 3) if match else None
        elif any(substring in weight_str for substring in ['kilogram', ' kg']):
            weight_str = weight_str.replace(' kg', ' kilogram')
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}kilogram", weight_str)
            val = round(float(match.group(1)) * 1000, 3) if match else None
        elif any(substring in weight_str for substring in ['milligrams']):
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,}milligrams", weight_str)
            val = round(float(match.group(1)) / 1000, 3) if match else None
        elif ' gram' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,} gram", weight_str)
            val = round(float(match.group(1)), 3) if match else None
        elif ' g' in weight_str:
            match = re.search(r"(\d+\.{0,}\d{0,})\D{0,} g", weight_str)
            val = round(float(match.group(1)), 3) if match else None
    if val:
        return f"{round(val, 2)}{weight_type}"
    else:
        return f"{val}"


# 分类提取-返回: 一级/当前分类id+一级/当前分类排名
# 参考dim_asin_bs_info.py使用
def udf_parse_bs_category(asin_bs_sellers_rank_lower, last_herf, all_best_sellers_href, cate_current_pattern,
                          cate_1_pattern):
    """
    asin_bs_sellers_rank_lower: 底部分类字符串
    last_herf： 最后一级分类链接
    all_best_sellers_href： 所有分类链接
    cate_current_pattern： 当前分类排名匹配规则
    cate_1_pattern： 一级分类排名匹配规则
    """

    # if (site_name == 'us' and date_type in ['month', 'month_week'] and date_info >= '2023-11') or (site_name != 'us' and date_type in ['week'] and date_info >= '2023-41'):
    #     href_list = all_best_sellers_href.split("&&&&")

    # 1. 判断用哪个字段来解析分类
    if str(all_best_sellers_href).lower() not in ['', 'none', 'null']:
        bs_href = all_best_sellers_href
    elif str(last_herf).lower() not in ['', 'none', 'null']:
        bs_href = last_herf
    else:
        bs_href = ''
    href_list = bs_href.replace("?tf=1", "").split("&&&&")

    # 新增climate-pledge分类优化--若最后一级是climate-pledge的分类，则向前取
    rank_flag = None
    while True:
        if '/climate-pledge' in href_list[-1] and len(href_list) >= 2:
            href_list.pop()
            rank_flag = True
        else:
            break

    # 2. 解析一级和当前 分类 + 排名
    # 2.1 提取分类
    if href_list:
        if len(href_list) == 1:
            cate_list = re.findall('bestsellers/(.*)/ref', href_list[0])
            if cate_list:
                if "/" in cate_list[0]:
                    cate_1_id, cate_current_id = cate_list[0].split("/")[0], cate_list[0].split("/")[-1]
                else:
                    cate_1_id, cate_current_id = cate_list[0].split("/")[0], None
            else:
                cate_1_id, cate_current_id = None, None
        else:
            cate_1_id = re.findall('bestsellers/(.*)/ref', href_list[0])[0] if re.findall('bestsellers/(.*)/ref',
                                                                                          href_list[0]) else None
            cate_current_id = re.findall('bestsellers/(.*)/ref', href_list[-1])[0] if re.findall('bestsellers/(.*)/ref',
                                                                                                 href_list[
                                                                                                     -1]) else None
            if "/" in cate_1_id:
                cate_1_id = cate_1_id.split("/")[0]
            if "/" in cate_current_id:
                cate_current_id = cate_current_id.split("/")[-1]
    else:
        cate_1_id, cate_current_id = None, None

    # 2.2 提取排名
    if asin_bs_sellers_rank_lower is not None:
        asin_bs_sellers_rank_lower2 = asin_bs_sellers_rank_lower.replace(".", "").replace(",", "").replace(" 100 ", "")
    else:
        asin_bs_sellers_rank_lower2 = ''
    rank_list = re.findall(cate_current_pattern, asin_bs_sellers_rank_lower2)  # 匹配排名
    rank_list = [int(rank) for rank in rank_list]  # 转换成int类型
    # print("rank_list:", rank_list)
    if rank_flag:
        if len(rank_list) > len(href_list):
            rank_list = rank_list[:len(href_list)]

    if rank_list:
        if len(rank_list) == 1:
            if cate_1_pattern in asin_bs_sellers_rank_lower:
                cate_1_rank, cate_current_rank = rank_list[0], None
            else:
                cate_1_rank, cate_current_rank = None, rank_list[0]
        else:
            if cate_1_pattern in asin_bs_sellers_rank_lower:
                cate_1_rank, cate_current_rank = rank_list[0], rank_list[-1]
            else:
                cate_1_rank, cate_current_rank = None, rank_list[0]
    else:
        cate_1_rank, cate_current_rank = None, None

    return cate_1_id, cate_current_id, cate_1_rank, cate_current_rank


# 将asin转换成1-10亿数值--从而可以划分指定分区表
def udf_asin_to_number(asin):
    """
    Convert a 10-character ASIN string to a unique number.
    This function assumes that ASIN consists of uppercase letters and digits.
    """

    def char_to_number(char):
        if char.isdigit():
            return int(char)
        else:
            return ord(char) - 55  # 'A' -> 10, 'B' -> 11, ..., 'Z' -> 35

    if len(asin) != 10:
        raise ValueError("ASIN must be 10 characters long")

    base = 36
    asin_number = 0
    for i, char in enumerate(reversed(asin)):
        asin_number += char_to_number(char) * (base ** i)

    # The final number is taken modulo 1 billion to fit the range 1-10 billion
    return asin_number % 1000000000


# 判断buy_box_seller_type类型
def udf_parse_seller_json(seller_json):
    """
    :param ship_from: 爬虫爬取asin详情页面上的字段信息
    :param sold_by: 爬虫爬取asin详情页面上的字段信息
    :param fulfilled_by：爬虫爬取asin详情页面上的字段信息
    :return: buy_box_seller_type: 1.amazon,2.fba,3.fbm,4.默认值（无类型）
    :return: 类型、店铺名称、店铺id
    """
    if not seller_json:
        return 0, None, None
    else:
        seller_info_parsed = json.loads(seller_json)
        ship_from = seller_info_parsed["ship_from"]
        sold_by = seller_info_parsed["sold_by"]
        fulfilled_by = seller_info_parsed["fulfilled_by"]
        seller_id = seller_info_parsed["seller_id"]
        if (ship_from and ship_from.lower().strip().startswith("amazon")) or (
                fulfilled_by and 'amazon' in fulfilled_by.lower()):
            if sold_by and not sold_by.lower().strip().startswith("amazon"):
                return 2, sold_by, seller_id  # FBA
            elif sold_by and sold_by.lower().strip().startswith("amazon"):
                return 1, sold_by, seller_id  # Amazon
        elif (ship_from or fulfilled_by) and sold_by:
            return 3, sold_by, seller_id  # FBM
        return 4, sold_by, seller_id  # Other


def udf_parse_amazon_orders(asin_amazon_orders_str):
    """
    :param asin_amazon_orders_str: 示例: '50+ bought in past month'
    解析asin详情页面的月销字段, 适配usukde3个站点
    """
    pattern = "(\d+[k]{0,})\+"
    results_list = re.findall(pattern, str(asin_amazon_orders_str).lower())
    if len(results_list) == 1:
        result = int(results_list[0].replace("k", "000").replace(" ", ""))
    else:
        result = None
    return result


# 解析ABA词的语种
def udf_detect_phrase_reg(lang_word_map):
    def detect_phrase(phrase: str):
        # + 号替换为空格用于分词
        phrase = re.sub(r'(\+)', ' ', phrase).strip()
        # 分词
        from nltk.tokenize import word_tokenize
        word_list = list(filter(lambda x: len(x) >= 2, word_tokenize(phrase, "english")))
        tmp_map = {
            "en": {"frequency": 0, "word": []},
            "fr": {"frequency": 0, "word": []},
            "es": {"frequency": 0, "word": []},
            "de": {"frequency": 0, "word": []},
        }
        for word in word_list:
            lang_rank_map: dict = lang_word_map.get(word)
            if lang_rank_map is not None:
                for lang in lang_rank_map.keys():
                    frequency = lang_rank_map[lang]
                    tmp_map[lang]["frequency"] = tmp_map[lang]["frequency"] + frequency
                    tmp_map[lang]["word"].append(word)
            pass
        # 先根据word名称个数倒序后根据分数
        lang, hint_word_map = sorted(tmp_map.items(), key=lambda it: (len(it[1]['word']), it[1]['frequency']), reverse=True)[0]
        if hint_word_map['frequency'] == 0:
            return {"lang": None, "hint_word": None}
        else:
            # 如果en的频率大于0，优先设为en
            if tmp_map['en']['frequency'] > 0:
                lang = 'en'
                hint_word_map = tmp_map['en']
            hint_word_list = hint_word_map['word']
            hint_word = " ".join(hint_word_list)
            if len(hint_word) <= 2:
                return {"lang": None, "hint_word": None}
            return {"lang": lang, "hint_word": hint_word}
        pass
    return F.udf(detect_phrase, MapType(StringType(), StringType()))