asin跳转逻辑封装udf

Pyspark_job/yswg_utils/common_udf.py

@@ -916,3 +916,18 @@ def udf_detect_phrase_reg(lang_word_map):
             return {"lang": lang, "hint_word": hint_word}
         pass
     return F.udf(detect_phrase, MapType(StringType(), StringType()))
+
+
+def resolve_asin_jump(asin_col='asin', current_asin_col='current_asin'):
+    """asin跳转判断（分支拆开写，方便后续单独扩展特例）：
+    - current_asin 为空 → 不跳转，保留 asin
+    - current_asin 为合法10位字母数字ASIN码 → 跳转，取 current_asin
+    - current_asin 长度大于10（代表母体可卖）→ 不跳转，保留 asin
+    - 其余未覆盖情况 → 默认不跳转，保留 asin
+    """
+    current = F.col(current_asin_col)
+    asin = F.col(asin_col)
+    return F.when(current.isNull() | (current == ''), asin) \
+        .when(current.rlike(r'^[A-Za-z0-9]{10}$'), current) \
+        .when(F.length(current) > 10, asin) \
+        .otherwise(asin)