新增dim_fb_detail店铺详情表

Pyspark_job/dim/dim_fb_detail.py

+"""
+   @Author      : CT
+   @Description : 店铺详情维表——把 dwt_fb_base_report.py 里对 ods_seller_account_feedback
+                  原始字段(seller_address/seller_rating/feedback_histogram/metadata_json)的
+                  解析计算抽离到 dim 层，解析逻辑保持跟 dwt_fb_base_report.py 一致
+                  后续 dwt 只需直接读本表，专注做跨 ASIN/搜索词等维度的聚合计算，不用重复解析
+                  另外把 dim_seller_address.py 里"解析 business_addr 匹配中国省市"这一步也接入
+                  本脚本，生成 seller_province/seller_city 两个字段；这部分逻辑单独封装成内部类
+                  DimFbDetail.SellerAddressParser，跟主流程（read_data/save_data）解耦，不混在一起
+                  写 Doris dim.dim_seller_address 的逻辑保留在 save_data 方法中（与写 Hive
+                  dim_fb_detail 表一起完成，不再单独拆脚本）
+                  匹配思路（只针对 fb_country_name = 'CN' 的行，与 dim_seller_address.py 完全一致）：
+                    1. 邮编精确匹配优先：business_addr 按双空格切分，倒数第2个 token 视为邮编，
+                       去匹配 dim_china_zipcode 聚合出的"邮编→省市"表。只有一个邮编在
+                       dim_china_zipcode 里对应唯一一组(province,city)才算命中；对应0组
+                       （查不到/邮编本身有问题）或者多组（同一邮编跨了不同省市）统一都算
+                       "邮编匹配不到"，不再取出现次数最多的省市兜底
+                    2. 邮编匹配不到的，退化用文本匹配兜底：扫描地址里除国家和邮编外的所有
+                       token，逐个跟 dim_china_zipcode 去重出来的 省份/城市/区县 字典做包含
+                       关系匹配（中文名 + 拼音，拼音现算，不额外维护拼音字段）
+   @Dependency  : pypinyin（只在 driver 端生成拼音字典再广播，executor 不需要装这个包）
+   @SourceTable :
+                  ①ods_seller_account_feedback
+                  ②ods_seller_account_syn
+                  ③selection.accounts（MySQL）
+                  ④dim_china_zipcode
+   @SinkTable   :
+                  ①dim_fb_detail（Hive）
+                  ②dim.dim_seller_address（Doris）
+"""
+import os
+import re
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
+
+from pypinyin import lazy_pinyin
+from pyspark.sql.window import Window
+from pyspark.sql import functions as F
+from pyspark.sql.types import StringType, IntegerType, DoubleType
+from utils.spark_util import SparkUtil
+from utils.hdfs_utils import HdfsUtils
+from utils.common_util import CommonUtil
+from utils.db_util import DBUtil
+from utils.DorisHelper import DorisHelper
+
+DORIS_DB = "dim"
+DORIS_TABLE = "dim_seller_address"
+DORIS_COLUMNS = "seller_id, account_name, business_addr, fb_country_name, seller_province, seller_city, fb_crawl_date, site_name"
+
+# ============================================================================
+# 以下常量表只给 DimFbDetail.SellerAddressParser 内部匹配逻辑使用，属于纯静态数据，
+# 放在模块级别（而不是塞进类属性）是为了避免 Spark UDF 闭包意外捕获类实例（见该内部类的说明）
+# ============================================================================
+
+# 省/市/区县三级分开维护后缀表：同一个"市"字在不同层级都会出现（直辖市是省级、
+# 地级市是市级），必须按当前查的是哪一级字典选用对应的后缀表，不能混用一张表。
+# 顺序要把更长/更具体的后缀排在前面（比如"自治县"要排在"县"前面），不然会被短
+# 后缀先命中截断错。列表依据对 dim_china_zipcode 全量 distinct 值的实际核查确定。
+PROVINCE_SUFFIXES = ["自治区", "特别行政区", "省", "市"]
+PROVINCE_PINYIN_SUFFIXES = ["zizhiqu", "tebiexingzhengqu", "sheng", "shi", "province"]
+
+CITY_SUFFIXES = ["自治州", "地区", "特别行政区", "盟", "市"]
+CITY_PINYIN_SUFFIXES = ["zizhizhou", "diqu", "tebiexingzhengqu", "meng", "shi", "city"]
+
+# 区县原来不剥后缀（怕剥完只剩1个字，比如"单县"->"单"太容易误匹配），现在改成剥，
+# 靠 strip_suffix 里"剥完至少剩2字"的安全网挡住这种情况，"单县"这种还是不剥
+DISTRICT_SUFFIXES = ["自治县", "自治旗", "林区", "矿区", "新城", "市", "区", "县", "旗"]
+DISTRICT_PINYIN_SUFFIXES = [
+    "zizhixian", "zizhiqi", "xincheng", "shi", "qu", "xian", "qi", "district", "county",
+]
+
+# city 字段这几个值是无效占位符，不是真实城市名，构建 city 字典前先过滤掉
+INVALID_CITY_VALUES = ["县", "省直辖县级行政区划", "自治区直辖县级行政区划"]
+
+# 新疆几个自治区直辖县级市（图木舒克/石河子/阿拉尔）在源数据里 city 字段是"占位前缀+
+# 真实市名"拼在一起的脏文本（比如"自治区直辖县图木舒克市"），剥掉前缀还原真实市名，
+# 不然这三个市永远匹配不到
+INVALID_CITY_PREFIX = "自治区直辖县"
+
+# pypinyin 对多音字地名默认读音有误，人工核实过的案例收在这份覆盖表里；城市级 key 剥后缀，区县级 key 带后缀
+PINYIN_OVERRIDE = {
+    # ===== 城市级（剥后缀后查表）=====
+    "昌都": "changdu",  # 昌都市（西藏），"都"在此读 dū，pypinyin 默认读成 dōu
+    "长治": "changzhi",  # 长治市（山西），"长"在此读 cháng，pypinyin 默认读成 zhǎng
+    "朝阳": "chaoyang",  # 朝阳市（辽宁），"朝"在此读 cháo，pypinyin 默认读成 zhāo
+    "阿勒泰": "aletai",  # 阿勒泰地区（新疆），"勒"在此读 lè，pypinyin 默认读成 lēi
+    "锡林郭勒": "xilinguole",  # 锡林郭勒盟（内蒙古），"勒"读 lè，同上
+    "克孜勒苏": "kezilesu",  # 克孜勒苏柯尔克孜自治州（新疆），"勒"读 lè，同上
+
+    # ===== 区县级（原始值直接查，部分源数据本身就不带后缀）=====
+    # "都"在此读 dū，pypinyin 对不常见的地名默认读成 dōu
+    "三都": "sandu", "都兰": "dulan", "都安": "duan", "都匀": "duyun",
+    "丰都县": "fengduxian", "于都县": "yuduxian", "商都县": "shangduxian", "宁都县": "ningduxian",
+    "尧都区": "yaoduqu", "新都区": "xinduqu", "曾都区": "cengduqu", "望都县": "wangduxian",
+    "武都区": "wuduqu", "殷都区": "yinduqu", "盐都区": "yanduqu", "秦都区": "qinduqu",
+    "花都区": "huaduqu", "莲都区": "lianduqu", "都昌县": "duchangxian", "魏都区": "weiduqu",
+    "昆都仑区": "kundulunqu",
+
+    # "长"在此读 cháng，pypinyin 对不常见的地名默认读成 zhǎng
+    "长岛": "changdao", "长白": "changbai", "长阳": "changyang", "长顺": "changshun", "南长": "nanchang",
+    "长丰县": "changfengxian", "长宁区": "changningqu", "长宁县": "changningxian", "长岭县": "changlingxian",
+    "长武县": "changwuxian", "长泰区": "changtaiqu", "长洲区": "changzhouqu", "长海县": "changhaixian",
+    "长清区": "changqingqu", "天长市": "tianchangshi", "子长市": "zichangshi",
+    "长葛市": "changgeshi", "长垣市": "changyuanshi",
+
+    # "朝"在此读 cháo（"朝阳区"默认已经读对，不用加）
+    "朝阳县": "chaoyangxian",
+
+    # "什"在此读 shí，pypinyin 对不常见的地名默认读成 shén
+    "阿图什": "atushi", "乌什县": "wushixian", "克什克腾旗": "keshiketengqi",
+    "塔什库尔干塔吉克自治县": "tashikuergantajikezizhixian", "什邡市": "shifangshi",
+
+    # 少数民族地名音译，"勒"本该读 lè，pypinyin 默认读成 lēi
+    "库尔勒": "kuerle", "策勒县": "celexian", "阿勒泰市": "aletaishi", "霍林郭勒市": "huolinguoleshi",
+    "尼勒克": "nileke",
+
+    # 其他单独核实过的多音字地名
+    "泊头市": "botoushi",  # "泊"在此读 bó（停泊），pypinyin 默认读成 pō（湖泊）
+    "单县": "shanxian",  # "单"在此读 shàn，pypinyin 默认读成 dān
+    "民乐县": "minlexian",  # 民乐县（甘肃），pypinyin 词组词典把它当成"音乐"词组读成了 mínyuè
+    "蚌山区": "bengshanqu",  # 蚌山区（安徽蚌埠），"蚌"跟"蚌埠"一样读 bèng，pypinyin 默认读成 bàng
+    "荥经县": "yingjingxian",  # 荥经县（四川），"荥"在此读 yíng；跟"荥阳"(xíng)是两个不同读音的地名
+
+    # 以下几个源表 dim_china_zipcode 里暂时查不到，先保留人工核实过的读音
+    "信都": "xindu", "襄都": "xiangdu", "郫都": "pidu",  # 邢台信都区/邢台襄都区/成都郫都区
+    "长白朝鲜族": "changbaichaoxianzu", "长阳土家族": "changyangtujiazu",  # xx自治县
+    "康巴什": "kangbashi",  # 鄂尔多斯康巴什区
+    "龙圩": "longxu",  # 梧州龙圩区，"圩"在此读 xū（南方圩镇），pypinyin 默认读成 wéi（圩堤）
+}
+
+# dim_china_zipcode 里没有港澳台，手动补充：这三个地区不分省市，省份和城市都是自己本身
+SPECIAL_REGIONS = [
+    ("香港", ["xianggang", "hongkong"]),
+    ("澳门", ["aomen", "macau", "macao"]),
+    ("台湾", ["taiwan"]),
+]
+
+# 广西/宁夏/新疆这三个自治区官方全名带了民族限定词（壮族/回族/维吾尔），单纯剥掉"自治区"
+# 后缀会剩"广西壮族"这种还带着民族限定词的名字，跟地址里几乎都直接写的简称"广西"对不上，
+# 所以这三个单独覆盖成简称（内蒙古自治区、西藏自治区本身没有民族限定词，不受影响）
+PROVINCE_SHORT_NAME_OVERRIDE = {
+    "广西壮族自治区": "广西",
+    "宁夏回族自治区": "宁夏",
+    "新疆维吾尔自治区": "新疆",
+}
+
+
+# ============================================================================
+# 以下几个纯函数必须留在模块级别、不能挪进 SellerAddressParser 类里当方法（哪怕是
+# @staticmethod）！
+# 原因：build_zip_map/parse 会把它们包进闭包函数里注册成 Spark UDF。如果写成类方法，
+# 函数体里一旦出现 "DimFbDetail.SellerAddressParser.xxx(...)" 这种引用外层类名的写法，
+# Spark/cloudpickle 序列化这个 UDF 时会连带把整个 DimFbDetail 类"按值"打包（包括
+# SellerAddressParser 里所有其他方法，比如 to_pinyin），而 to_pinyin 引用了 pypinyin
+# 模块——这会导致 executor 反序列化 UDF 时尝试 import pypinyin，报
+# "ModuleNotFoundError: No module named 'pypinyin'"（executor 机器本来就不需要装这个
+# 包，这个问题已经在生产环境实际复现过）。用 pyspark 自带的 cloudpickle 手动验证过：
+# 一旦某个方法体里出现对外层类名的引用，整个类（含不相关的兄弟方法）都会被拉进闭包；
+# 改成模块级别的普通函数、互相直接按函数名调用（而不是通过类名/self）就不会有这个问题，
+# 这也是 dim_seller_address.py 原始实现的写法，这里保持一致，不要改回类方法。
+# ============================================================================
+
+def strip_suffix(name, suffixes):
+    if not name:
+        return name
+    for suf in suffixes:
+        if name.endswith(suf) and len(name) - len(suf) >= 2:
+            return name[: -len(suf)]
+    return name
+
+
+def clean_city_name(raw_city):
+    if raw_city and raw_city.startswith(INVALID_CITY_PREFIX) and len(raw_city) > len(INVALID_CITY_PREFIX):
+        return raw_city[len(INVALID_CITY_PREFIX):]
+    return raw_city
+
+
+def to_pinyin(word):
+    if not word:
+        return None
+    return PINYIN_OVERRIDE.get(word) or "".join(lazy_pinyin(word))
+
+
+def lookup_backfill_city(backfill_map, province, district):
+    if not province or not district:
+        return None
+    # 回填字典的 key 同时来自 city 侧（剥 CITY_SUFFIXES）和 district 侧（剥
+    # DISTRICT_SUFFIXES），两者唯一的公共后缀是"市"，用 DISTRICT_SUFFIXES 剥这里
+    # 传入的查询值即可覆盖实际会出现的场景（district 字段本身的后缀只可能是市/区/县/旗）
+    return backfill_map.get((province, strip_suffix(district, DISTRICT_SUFFIXES)))
+
+
+def match_text_city(addr, province_lookup, city_lookup, district_lookup):
+    """
+    在 business_addr 里（去掉国家、邮编两个 token 后）扫描剩余 token，只认"干净命中"：
+    命中后面剩下的部分要么是空的、要么开头是一个行政后缀（比如"广东省"命中"广东"剩
+    "省"、"南阳市邓州市"命中"南阳"剩"市邓州市"、"jiangsushengsuzhoushi"命中
+    "suzhou"剩"shi"）——前缀不做任何校验，因为"广东深圳市"这种省市连写、
+    "新郑市龙湖镇"这种区县+乡镇连写都是合法地址，不能要求命中的内容必须是token开头。
+    校验后缀时按当前查的是省/市/区哪一级字典，选用对应层级的后缀表（同一个"市"字
+    在省级/市级都合法，必须区分开，不能混用一张表）。
+    "黑龙江北路"命中"黑龙江"剩"北路"（不以任何省级后缀开头）不算命中——不然满大街
+    带省市名的路名/街道名都会被误当成地址真实归属。
+    先定省份缩小范围，再依次尝试命中城市/区县，返回 "province|-|city" 或 None
+    """
+    if not addr:
+        return None
+    tokens = [t.strip() for t in addr.split("  ") if t.strip()]
+    if len(tokens) < 2:
+        return None
+    tokens = tokens[:-1]  # 去掉倒数第1个token(国家)
+    # 倒数第2个token长得像邮编（纯数字）才一并去掉；有些地址本身就没填邮编，
+    # 这种情况倒数第2个token其实是真实地名（比如"...顺德区  佛山市  CN"），
+    # 误当邮编砍掉会把仅有的城市信息也搜没了，所以不像邮编就保留它参与匹配
+    if tokens and re.fullmatch(r"\d{4,10}", tokens[-1]):
+        tokens = tokens[:-1]
+    if not tokens:
+        return None
+    # 去空格再比较拼音：不然"Guang Dong"这种按音节分开写的拼音永远匹配不上"guangdong"
+    tokens_low = [re.sub(r"\s+", "", t.lower()) for t in tokens]
+
+    def is_clean(base, text, suffixes):
+        if not base:
+            return False
+        idx = text.find(base)
+        if idx == -1:
+            return False
+        leftover = text[idx + len(base):]
+        return leftover == "" or any(leftover.startswith(suf) for suf in suffixes)
+
+    def is_hit(base_cn, base_py, cn_suffixes, py_suffixes):
+        for tok, tok_low in zip(tokens, tokens_low):
+            if is_clean(base_cn, tok, cn_suffixes):
+                return True
+            if is_clean(base_py, tok_low, py_suffixes):
+                return True
+        return False
+
+    def unique_match(lookup, key_of, restrict_province, cn_suffixes, py_suffixes):
+        # lookup 里每行最后两列固定是 (base_cn, base_py)，前面几列由 key_of 决定取哪些做候选key
+        matched = set()
+        for row in lookup:
+            if restrict_province and row[0] != restrict_province:
+                continue
+            if is_hit(row[-2], row[-1], cn_suffixes, py_suffixes):
+                matched.add(key_of(row))
+        return next(iter(matched)) if len(matched) == 1 else None
+
+    matched_province = unique_match(
+        province_lookup, key_of=lambda row: row[0], restrict_province=None,
+        cn_suffixes=PROVINCE_SUFFIXES, py_suffixes=PROVINCE_PINYIN_SUFFIXES,
+    )
+
+    picked = unique_match(
+        city_lookup, key_of=lambda row: (row[0], row[1]), restrict_province=matched_province,
+        cn_suffixes=CITY_SUFFIXES, py_suffixes=CITY_PINYIN_SUFFIXES,
+    )
+    if picked:
+        return f"{picked[0]}|-|{picked[1]}"
+
+    picked = unique_match(
+        district_lookup, key_of=lambda row: (row[0], row[1]), restrict_province=matched_province,
+        cn_suffixes=DISTRICT_SUFFIXES, py_suffixes=DISTRICT_PINYIN_SUFFIXES,
+    )
+    if picked:
+        return f"{picked[0]}|-|{picked[1]}"
+
+    return None
+
+
+class DimFbDetail(object):
+
+    class SellerAddressParser(object):
+        """
+        把 business_addr 解析出中国省市（seller_province/seller_city）的逻辑单独封装成一个类，
+        跟 DimFbDetail 主流程（read_data/save_data）解耦，避免混在一起。迁移自 dim_seller_address.py，
+        匹配逻辑原样保留：
+          1. 邮编精确匹配优先（对应0组/多组都算匹配不到，不做兜底）
+          2. 邮编匹配不到的，退化用文本匹配（先定位省份缩小范围，再依次尝试命中城市/区县）
+        注意：strip_suffix/clean_city_name/to_pinyin/lookup_backfill_city/match_text_city 这几个
+        纯函数特意留在模块级别、没有挪进这个类里，原因见模块级别处的详细注释——简单说就是这几个
+        函数最终会被塞进 Spark UDF 闭包，一旦写成类方法且互相通过类名调用，会把整个类"按值"序列化，
+        连带把只有 to_pinyin 需要的 pypinyin 依赖也一起带上，导致 executor 报 ModuleNotFoundError。
+        """
+
+        def __init__(self, spark):
+            self.spark = spark
+            self.province_lookup = None
+            self.city_lookup = None
+            self.district_lookup = None
+            self.city_backfill_map = None
+            self.province_lookup_bc = None
+            self.city_lookup_bc = None
+            self.district_lookup_bc = None
+            self.city_backfill_bc = None
+            self.df_zip_map = None
+
+        def build_lookups(self):
+            print("构建省份/城市/区县文本匹配字典 + city缺失回填字典（driver端生成拼音，广播给executor）")
+            provinces = [
+                r["province"] for r in
+                self.spark.sql("select distinct province from dim_china_zipcode where province is not null").collect()
+            ]
+            def province_base_name(p):
+                return PROVINCE_SHORT_NAME_OVERRIDE.get(p) or strip_suffix(p, PROVINCE_SUFFIXES)
+
+            self.province_lookup = []
+            for p in provinces:
+                base = province_base_name(p)
+                self.province_lookup.append((p, base, to_pinyin(base)))
+
+            invalid_city_clause = ", ".join(f"'{v}'" for v in INVALID_CITY_VALUES)
+            city_rows = self.spark.sql(
+                "select distinct province, city from dim_china_zipcode "
+                "where province is not null and city is not null "
+                f"and city not in ({invalid_city_clause})"
+            ).collect()
+            self.city_lookup = []
+            for r in city_rows:
+                city_name = clean_city_name(r["city"])
+                base = strip_suffix(city_name, CITY_SUFFIXES)
+                self.city_lookup.append((r["province"], city_name, base, to_pinyin(base)))
+
+            # 同一个省份+剥后缀名字，如果同时存在"XX地区"和"XX市"两个原始城市名（比如"铜仁地区"/
+            # "铜仁市"，2011年撤地设市遗留的新旧两个名字都留在源数据里），文本匹配时两个候选长度
+            # 一样、谁都选不出来，所以只保留现在还在用的"市"，把过时的"地区"去掉
+            by_stripped = {}
+            for row in self.city_lookup:
+                by_stripped.setdefault((row[0], row[2]), []).append(row)
+            deduped_city_lookup = []
+            for (province, stripped), rows in by_stripped.items():
+                if len(rows) > 1:
+                    shi_rows = [r for r in rows if r[1].endswith("市")]
+                    rows = shi_rows or rows
+                deduped_city_lookup.extend(rows)
+            self.city_lookup = deduped_city_lookup
+
+            # 港澳台不分省市，省份和城市都是自己本身
+            for name, pinyins in SPECIAL_REGIONS:
+                for py in pinyins:
+                    self.province_lookup.append((name, name, py))
+                    self.city_lookup.append((name, name, name, py))
+
+            # length(district)>=2：源数据里有极少数区县字段是单字垃圾值（比如上海某几条记录
+            # district 字段直接是"区"这一个字），单字后面又刚好是行政后缀本身，会跟全国任何
+            # 以"区"结尾的 token 撞车，真实的区县名没有单字的，直接过滤掉
+            district_rows = self.spark.sql(
+                "select distinct province, city, district from dim_china_zipcode "
+                "where province is not null and city is not null and district is not null and district != '' "
+                "and length(district) >= 2"
+            ).collect()
+            self.district_lookup = []
+            for r in district_rows:
+                base = strip_suffix(r["district"], DISTRICT_SUFFIXES)
+                self.district_lookup.append((r["province"], r["city"], base, to_pinyin(base)))
+
+            # 同一个(省份,区县)如果同时归属不止一个城市（比如"公主岭市"曾归四平市、2014年后
+            # 改由长春市代管；"江口县"曾归"铜仁地区"、现归"铜仁市"），能靠"只有一个以市结尾"
+            # 消歧的就消歧（这种是地区改市/地区代管变更，取现在这个）；两个都以市结尾的说明是
+            # 更复杂的历史沿革（比如"襄樊市"改名"襄阳市"）或者单纯两个市恰好有同名区县，没法
+            # 安全判断，这条区县整个丢弃，不瞎猜
+            by_province_district = {}
+            for row in self.district_lookup:
+                by_province_district.setdefault((row[0], row[2]), []).append(row)
+            deduped_district_lookup = []
+            for (province, district), rows in by_province_district.items():
+                if len({r[1] for r in rows}) > 1:
+                    shi_rows = [r for r in rows if r[1].endswith("市")]
+                    if len(shi_rows) != 1:
+                        continue
+                    rows = shi_rows
+                deduped_district_lookup.extend(rows)
+            self.district_lookup = deduped_district_lookup
+
+            # 上面"地区/市"去重把过时的城市名（比如"铜仁地区"）从 city_lookup 删掉了；区县
+            # 字典里如果还有挂在这些过时城市名下面的区县（比如撤地设市前，"铜仁地区"下面
+            # 有个同名的县级市"铜仁市"，这条归属数据现在已经陈旧、不可信），先清掉，避免
+            # 拿这种陈旧归属去跟当前有效的城市名做下面的跨级去重判断
+            valid_cities_by_province = {}
+            for row in self.city_lookup:
+                valid_cities_by_province.setdefault(row[0], set()).add(row[1])
+            self.district_lookup = [
+                row for row in self.district_lookup
+                if row[1] in valid_cities_by_province.get(row[0], set())
+            ]
+
+            # "省直辖县级市"这种（比如新郑市实际由郑州市代管）在源数据里经常同时出现在两个
+            # 层级：city_lookup 里是独立一条"新郑市"，district_lookup 里又挂在"郑州市"下面
+            # 当区县，两边都留着会导致城市这一级判不出唯一结果（"新郑市"/"郑州市"两个候选
+            # 打平）。以 district 层级的归属为准：city_lookup 里的城市名如果在同一个省份的
+            # district_lookup 里也作为"挂在别的城市下面的区县"出现，就不再当独立城市，从
+            # city_lookup 剔除，只留 district 那条，靠区县反查城市的机制处理。
+            # 必须限定"别的城市"：很多地级市自己就辖有同名的县（长沙市/长沙县、衡阳市/衡阳
+            # 县这种极常见的市县同名），这种归属城市就是自己本身，不能算冲突去掉自己
+            # district_lookup 的 row[2] 现在是剥过 DISTRICT_SUFFIXES 的区县名，city_lookup
+            # 的 row[1] 还是原始城市名，两边唯一的公共后缀是"市"，按 CITY_SUFFIXES 剥一下
+            # city 名字再比较，才能对上（"新郑市"剥完是"新郑"，跟区县那边剥完的"新郑"一致）
+            district_names_by_province = {}
+            for row in self.district_lookup:
+                district_names_by_province.setdefault(row[0], set()).add((row[2], row[1]))
+            self.city_lookup = [
+                row for row in self.city_lookup
+                if not any(
+                    district_name == strip_suffix(row[1], CITY_SUFFIXES) and governing_city != row[1]
+                    for district_name, governing_city in district_names_by_province.get(row[0], set())
+                )
+            ]
+
+            # 新版邮编源数据里有一批行 city 是空的，实际城市名被错放进了 district 字段（比如
+            # 省=江苏省 市="" 区=苏州），或者 district 是真区县、只是这一行恰好没填 city（比如
+            # 省=广东省 市="" 区=宝安）。用"city不为空"的正常数据反过来建一个
+            # (省份, 剥后缀的名字) -> 城市 的回填字典：
+            #   1) 名字剥完后缀等于某个城市剥后缀的名字，回填这个城市（"苏州"->"苏州市"）
+            #   2) 名字是某个区县、且这个区县在正常数据里归属唯一，回填归属的城市（"宝安"->"深圳市"）
+            # 同一个 (省份,名字) 两种来源指向不同城市的，说明有歧义，不回填，city 继续留空不参与匹配
+            backfill, conflict = {}, set()
+
+            def add_candidate(province, name, city, suffixes):
+                key = (province, strip_suffix(name, suffixes))
+                if key in backfill and backfill[key] != city:
+                    conflict.add(key)
+                else:
+                    backfill[key] = city
+
+            for r in city_rows:
+                clean_city = clean_city_name(r["city"])
+                add_candidate(r["province"], clean_city, clean_city, CITY_SUFFIXES)
+            for r in district_rows:
+                add_candidate(r["province"], r["district"], r["city"], DISTRICT_SUFFIXES)
+            self.city_backfill_map = {k: v for k, v in backfill.items() if k not in conflict}
+
+            # city 缺失但能靠回填字典唯一推出城市的行，额外补进区县字典，让文本匹配也能兜住
+            # （地址里正好写了这种不带后缀的区县名，比如"宝安"而不是"宝安区"）
+            backfill_district_rows = self.spark.sql(
+                "select distinct province, district from dim_china_zipcode "
+                "where province is not null and city is null and district is not null and district != '' "
+                "and length(district) >= 2"
+            ).collect()
+            for r in backfill_district_rows:
+                city = lookup_backfill_city(self.city_backfill_map, r["province"], r["district"])
+                if city:
+                    self.district_lookup.append((r["province"], city, r["district"], to_pinyin(r["district"])))
+
+            print(
+                f"字典规模：province={len(self.province_lookup)}，city={len(self.city_lookup)}，"
+                f"district={len(self.district_lookup)}，city回填={len(self.city_backfill_map)}"
+            )
+
+            # 显式广播，避免字典跟着 UDF 闭包在每个 task 里重复序列化
+            self.province_lookup_bc = self.spark.sparkContext.broadcast(self.province_lookup)
+            self.city_lookup_bc = self.spark.sparkContext.broadcast(self.city_lookup)
+            self.district_lookup_bc = self.spark.sparkContext.broadcast(self.district_lookup)
+            self.city_backfill_bc = self.spark.sparkContext.broadcast(self.city_backfill_map)
+
+        def build_zip_map(self):
+            print("读取 dim_china_zipcode 全量原始行，用回填字典补全 city 缺失的行，再聚合出邮编->省市归属"
+                  "（同邮编对应不止一组省市的，视为匹配不到）")
+            df_raw = self.spark.sql(
+                "select province, city, district, zipcode from dim_china_zipcode "
+                "where province is not null and zipcode is not null"
+            )
+
+            city_backfill_bc = self.city_backfill_bc
+            # clean_city_name/lookup_backfill_city 是模块级别的普通函数（不是 self 的方法），
+            # 直接调用即可——UDF 闭包只会带上这两个函数本身和用到的模块常量，不会把 self 也序列化进去
+            def backfill_city(province, city, district):
+                if city:
+                    return clean_city_name(city)
+                return lookup_backfill_city(city_backfill_bc.value, province, district)
+
+            u_backfill_city = F.udf(backfill_city, StringType())
+
+            df_filled = df_raw.withColumn(
+                "city_filled", u_backfill_city(F.col("province"), F.col("city"), F.col("district"))
+            ).filter(F.col("city_filled").isNotNull())
+
+            df_distinct = df_filled.select(
+                F.col("province"), F.col("city_filled").alias("city"), F.col("zipcode")
+            ).distinct()
+
+            combo_cnt = F.count(F.lit(1)).over(Window.partitionBy("zipcode"))
+            self.df_zip_map = df_distinct.withColumn("combo_cnt", combo_cnt).filter(F.col("combo_cnt") == 1).select(
+                F.col("zipcode"),
+                F.col("province").alias("map_province"),
+                F.col("city").alias("map_city"),
+            )
+
+        def parse(self, df):
+            """接收带 business_addr/fb_country_name 列的 df，返回新增 seller_province/seller_city 两列的 df"""
+            print("解析 business_addr 匹配省市（仅 fb_country_name = 'CN' 的行）")
+            addr_split = F.split(F.col("business_addr"), "  ")
+            # 邮编 token 里偶尔会混入零宽字符等不可见字符（复制粘贴带过来的），精确比较会
+            # 匹配不上字典里干净的邮编，这里只保留数字部分再去 join，不影响正常邮编
+            addr_zip_clean = F.regexp_replace(F.element_at(addr_split, -2), r"[^0-9]", "")
+            df = df.withColumn(
+                "addr_zip",
+                F.when(F.col("fb_country_name") == "CN", addr_zip_clean).otherwise(F.lit(None)),
+            )
+
+            df = df.join(self.df_zip_map, df["addr_zip"] == self.df_zip_map["zipcode"], how="left")
+
+            province_lookup_bc, city_lookup_bc, district_lookup_bc = (
+                self.province_lookup_bc, self.city_lookup_bc, self.district_lookup_bc,
+            )
+
+            # 短路判断必须写在函数体内部：F.when 不会让 Spark 跳过 UDF 调用本身（实测过）
+            # match_text_city 是模块级别的普通函数，直接调用，不经过 self
+            def udf_match_text(addr, country, zip_matched):
+                if zip_matched or country != "CN":
+                    return None
+                return match_text_city(addr, province_lookup_bc.value, city_lookup_bc.value, district_lookup_bc.value)
+
+            u_match_text = F.udf(udf_match_text, StringType())
+
+            df = df.withColumn(
+                "text_match",
+                u_match_text(F.col("business_addr"), F.col("fb_country_name"), F.col("map_province").isNotNull()),
+            )
+            text_split = F.split(F.col("text_match"), r"\|-\|")
+
+            return df.withColumn(
+                "seller_province", F.coalesce(F.col("map_province"), text_split.getItem(0))
+            ).withColumn(
+                "seller_city", F.coalesce(F.col("map_city"), text_split.getItem(1))
+            ).drop("addr_zip", "zipcode", "map_province", "map_city", "text_match")
+
+    def __init__(self, site_name, date_type, date_info):
+        self.site_name = site_name
+        self.date_type = date_type
+        self.date_info = date_info
+        self.hive_tb = "dim_fb_detail"
+        self.partition_dict = {
+            "site_name": site_name,
+            "date_type": date_type,
+            "date_info": date_info,
+        }
+        # 落表路径校验
+        self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)
+
+        # 创建spark_session对象相关
+        app_name = f"{self.hive_tb}:{self.site_name}_{self.date_type}_{self.date_info}"
+        self.spark = SparkUtil.get_spark_session(app_name)
+        self.partitions_num = CommonUtil.reset_partitions(self.site_name, 20)
+
+        # 初始化全局变量df
+        self.df_fb_feedback = self.spark.sql("select 1+1;")
+        self.df_seller_account = self.spark.sql("select 1+1;")
+        self.df_self_seller_id = self.spark.sql("select 1+1;")
+
+        # 中国省市匹配（seller_province/seller_city）单独封装成内部类，跟主流程解耦
+        self.addr_parser = DimFbDetail.SellerAddressParser(self.spark)
+
+        # 初始化UDF函数
+        self.u_get_business_val = F.udf(self.get_business_val, StringType())  # 解析 seller_address 字段
+
+    # 解析seller_address字段
+    @staticmethod
+    def get_business_val(seller_address, key):
+        if not seller_address:
+            return None
+
+        parts = [p.strip() for p in seller_address.split("|-|")]
+        for i, p in enumerate(parts):
+            if p.startswith(key):
+                # Business Address: 拼接后续所有内容，因为地址可能有多段（街道、城市、邮编），用 |-| 分隔
+                if key in ("Business Address", "Geschäftsadresse"):
+                    return "  ".join(parts[i + 1:]).strip()
+                # 其他key: 只取下一个
+                elif i + 1 < len(parts):
+                    return parts[i + 1].strip()
+        return None
+
+    def read_data(self):
+        # ods_seller_account_feedback 店铺详情原始表
+        print("获取 ods_seller_account_feedback")
+        sql = f"""
+            select
+                seller_id,
+                num as fb_web_asin_num,
+                count_30_day as count_30_day_num,
+                count_1_year as count_1_year_num,
+                count_lifetime as count_lifetime_num,
+                country_name as fb_country_name,
+                seller_address,
+                seller_rating,
+                feedback_histogram,
+                metadata_json,
+                created_at
+            from ods_seller_account_feedback
+            where site_name = '{self.site_name}'
+              and date_type = '{self.date_type}'
+              and date_info = '{self.date_info}'
+              and length(seller_id) > 2
+        """
+        self.df_fb_feedback = self.spark.sql(sqlQuery=sql)
+        print(sql)
+
+        # 去重
+        w_fb = Window.partitionBy('seller_id').orderBy(F.col('created_at').desc())
+        self.df_fb_feedback = self.df_fb_feedback \
+            .withColumn('rank', F.row_number().over(w_fb)) \
+            .filter('rank = 1') \
+            .withColumn('fb_crawl_date', F.date_format(F.col('created_at'), 'yyyy-MM-dd HH:mm:ss')) \
+            .drop('rank', 'created_at')
+
+        # 解析seller_rating: 格式为"30天|-|90天|-|1年|-|历史"，切分后转double，-1改为0
+        # -1.0 代表该时间段无评分数据，统一替换成 0.0，避免负值干扰后续计算
+        _rating_split = F.split(F.col("seller_rating"), r"\|-\|")
+        self.df_fb_feedback = self.df_fb_feedback \
+            .withColumn("rating_30_day_num",
+                F.when(_rating_split.getItem(0).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rating_split.getItem(0).cast(DoubleType()))) \
+            .withColumn("rating_90_day_num",
+                F.when(_rating_split.getItem(1).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rating_split.getItem(1).cast(DoubleType()))) \
+            .withColumn("rating_1_year_num",
+                F.when(_rating_split.getItem(2).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rating_split.getItem(2).cast(DoubleType()))) \
+            .withColumn("rating_lifetime_num",
+                F.when(_rating_split.getItem(3).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rating_split.getItem(3).cast(DoubleType()))) \
+            .withColumn("fb_star_5_pct",
+                F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
+                 .otherwise(F.coalesce(
+                     F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star5").cast(DoubleType()) / 100, 4),
+                     F.lit(-1.0)
+                 ))) \
+            .withColumn("fb_star_4_pct",
+                F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
+                 .otherwise(F.coalesce(
+                     F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star4").cast(DoubleType()) / 100, 4),
+                     F.lit(-1.0)
+                 ))) \
+            .withColumn("fb_star_3_pct",
+                F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
+                 .otherwise(F.coalesce(
+                     F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star3").cast(DoubleType()) / 100, 4),
+                     F.lit(-1.0)
+                 ))) \
+            .withColumn("fb_star_2_pct",
+                F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
+                 .otherwise(F.coalesce(
+                     F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star2").cast(DoubleType()) / 100, 4),
+                     F.lit(-1.0)
+                 ))) \
+            .withColumn("fb_star_1_pct",
+                F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
+                 .otherwise(F.coalesce(
+                     F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star1").cast(DoubleType()) / 100, 4),
+                     F.lit(-1.0)
+                 ))) \
+            .withColumn("fb_web_asin_num",
+                # metadata_json.totalResultCount 更准确（实时爬取的真实总数），有值优先用；无值回退原字段 num
+                F.when(
+                    F.get_json_object(F.col("metadata_json"), "$.totalResultCount").isNotNull(),
+                    F.get_json_object(F.col("metadata_json"), "$.totalResultCount").cast(IntegerType())
+                ).otherwise(F.col("fb_web_asin_num")))
+
+        # 解析卖家公司数据（seller_address），不同语言区别处理
+        if self.site_name in ("us", "uk"):
+            self.df_fb_feedback = self.df_fb_feedback.withColumn(
+                "business_name", self.u_get_business_val(F.col("seller_address"), F.lit("Business Name"))
+            ).withColumn(
+                "business_addr", self.u_get_business_val(F.col("seller_address"), F.lit("Business Address"))
+            )
+        elif self.site_name == "de":
+            self.df_fb_feedback = self.df_fb_feedback.withColumn(
+                "business_name", self.u_get_business_val(F.col("seller_address"), F.lit("Geschäftsname"))
+            ).withColumn(
+                "business_addr", self.u_get_business_val(F.col("seller_address"), F.lit("Geschäftsadresse"))
+            )
+        else:
+            self.df_fb_feedback = self.df_fb_feedback.withColumn(
+                "business_name", F.lit(None).cast(StringType())
+            ).withColumn(
+                "business_addr", F.lit(None).cast(StringType())
+            )
+
+        self.df_fb_feedback = self.df_fb_feedback.drop(
+            'seller_address', 'seller_rating', 'feedback_histogram', 'metadata_json'
+        )
+
+        # 获取ods_seller_account_syn提取account_name
+        print("获取 ods_seller_account_syn")
+        sql = f"""
+            select seller_id, account_name, lower(account_name) as account_name_lower, url as fb_url from ods_seller_account_syn where site_name='{self.site_name}'
+        """
+        self.df_seller_account = self.spark.sql(sqlQuery=sql)
+        print(sql)
+
+        # 获取mysql：selection.accounts，用于排除公司内部店铺
+        print("获取 selection.accounts")
+        sql = """
+            select seller_id, 1 as is_self_account from (select distinct seller_id from selection.accounts) t1
+        """
+        conn_info = DBUtil.get_connection_info("mysql", "us")
+        self.df_self_seller_id = SparkUtil.read_jdbc_query(
+            session=self.spark,
+            url=conn_info["url"],
+            pwd=conn_info["pwd"],
+            username=conn_info["username"],
+            query=sql
+        )
+
+        # 构建中国省市匹配所需的字典/邮编映射（内部类自己的读数据步骤）
+        self.addr_parser.build_lookups()
+        self.addr_parser.build_zip_map()
+
+    def handle_seller_address(self):
+        self.df_fb_feedback = self.addr_parser.parse(self.df_fb_feedback)
+
+    def save_data(self):
+        df_joined = self.df_fb_feedback.join(
+            self.df_seller_account, on='seller_id', how='inner'
+        ).join(
+            self.df_self_seller_id, on='seller_id', how='left'
+        ).na.fill({"is_self_account": 0})
+
+        df_save = df_joined.select(
+            F.col('seller_id'),
+            F.col('account_name'),
+            F.col('account_name_lower'),
+            F.col('fb_url'),
+            F.col('fb_country_name'),
+            F.col('fb_web_asin_num'),
+            F.col('is_self_account'),
+            F.col('count_30_day_num'),
+            F.col('count_1_year_num'),
+            F.col('count_lifetime_num'),
+            F.col('business_name'),
+            F.col('business_addr'),
+            F.col('rating_30_day_num'),
+            F.col('rating_90_day_num'),
+            F.col('rating_1_year_num'),
+            F.col('rating_lifetime_num'),
+            F.col('fb_star_5_pct'),
+            F.col('fb_star_4_pct'),
+            F.col('fb_star_3_pct'),
+            F.col('fb_star_2_pct'),
+            F.col('fb_star_1_pct'),
+            F.col('seller_province'),
+            F.col('seller_city'),
+            F.col('fb_crawl_date'),
+            F.lit(self.site_name).alias('site_name'),
+            F.lit(self.date_type).alias('date_type'),
+            F.lit(self.date_info).alias('date_info'),
+        ).cache()  # 下面 Hive 写入和 Doris 写入两个 action 共用同一份 df，cache 避免重复算一遍join+匹配
+
+        # ---------------- 写入 Hive：dim_fb_detail ----------------
+        print(f"清除hdfs目录中:{self.hdfs_path}")
+        HdfsUtils.delete_file_in_folder(self.hdfs_path)
+        df_hive_save = df_save.repartition(self.partitions_num)
+        partition_by = ["site_name", "date_type", "date_info"]
+        print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}")
+        df_hive_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=partition_by)
+
+        # ---------------- 写入 Doris：dim.dim_seller_address ----------------
+        # fb_crawl_date 源字段是 date_format 出来的字符串，Doris 表定义的是 DATETIME，
+        # 写入前要显式转成 timestamp，不然 Doris 连接器会报 "Cannot safely cast 'fb_crawl_date': string to timestamp"
+        df_doris_save = df_save.select(
+            F.col('seller_id'),
+            F.col('account_name'),
+            F.col('business_addr'),
+            F.col('fb_country_name'),
+            F.col('seller_province'),
+            F.col('seller_city'),
+            F.to_timestamp(F.col('fb_crawl_date'), 'yyyy-MM-dd HH:mm:ss').alias('fb_crawl_date'),
+            F.lit(self.site_name).alias('site_name'),
+        )
+        total = df_doris_save.count()
+        print(f"写入 Doris {DORIS_DB}.{DORIS_TABLE}，共 {total} 条")
+        DorisHelper.spark_export_with_columns(
+            df_save=df_doris_save,
+            db_name=DORIS_DB,
+            table_name=DORIS_TABLE,
+            table_columns=DORIS_COLUMNS,
+            use_type="selection",
+        )
+
+        df_save.unpersist()
+        print("success")
+
+    def run(self):
+        self.read_data()
+        self.handle_seller_address()
+        self.save_data()
+
+
+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)  # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
+    handle_obj = DimFbDetail(site_name=site_name, date_type=date_type, date_info=date_info)
+    handle_obj.run()