续约待办派发

test/续约待办派发.py

@@ -30,15 +30,92 @@ common_module = CommonModule()
 output_dir = "output"  # 设置输出目录
 os.makedirs(output_dir, exist_ok=True)
 
+
 class RenewalToDo:
     def __init__(self):
         self.json_list = None
         self.data_NGV = None
         self.staff_id_list = None
         self.NGV_data_list = None
-
+        self.field_map = {
+            "关联数据": "_widget_1764820541663",
+            "公司名称": "_widget_1764820541616",
+            "门店名称": "_widget_1764820541617",
+            "门店编码": "_widget_1764820541661",
+            "加盟商": "_widget_1764820541618",
+            "过期日": "_widget_1764820541672",
+            "Saas版本": "_widget_1764820541623",
+            "上次购买价格": "_widget_1764820541624",
+            "联系人": "_widget_1764820541621",
+            "联系手机号": "_widget_1764820541622",
+            "专属运营顾问": "_widget_1764820541625",
+            "区域客服": "_widget_1764820541715",
+            "运营专家": "_widget_1764820541678",
+            "120天是否跟进": "_widget_1764820541628",
+            "120天处理人": "_widget_1764820541634",
+            "120天跟进时间": "_widget_1765352838631",
+            "60天是否跟进": "_widget_1764820541630",
+            "60天处理人": "_widget_1764820541635",
+            "60天跟进时间": "_widget_1765352838632",
+            "30天是否跟进": "_widget_1764820541632",
+            "30天处理人": "_widget_1764820541636",
+            "30天跟进时间": "_widget_1765352838633",
+            "是否联系上客户": "_widget_1764820541638",
+            "客户现阶段问题分类": "_widget_1764820541641",
+            "未联系上原因字段": "_widget_1765330820509",
+            "联系情况及问题说明": "_widget_1764820541653",
+            "潜在商机": "_widget_1764820541657",
+            "商机详情": "_widget_1764820541659",
+            "门店续约意愿": "_widget_1764820541654",
+            "不续约原因": "_widget_1764820541700",
+            "产品原因": "_widget_1764820541707",
+            "服务问题": "_widget_1764820541709",
+            "门店原因": "_widget_1764820541711",
+            "价格原因": "_widget_1764820541713",
+            "不续约具体情况说明": "_widget_1764820541702",
+            "回访完成方式": "_widget_1764820541697",
+            "周期性增购": "_widget_1764820541717",
+            "周期性增购.商品名称": "_widget_1764820541717._widget_1764820541719",
+            "周期性增购.分母金额": "_widget_1764820541717._widget_1764820541720",
+            "周期性增购.应续约日": "_widget_1764820541717._widget_1764820541721",
+            "周期性增购.上次购买数量": "_widget_1764820541717._widget_1764820541722",
+            "周期性增购.不续约原因": "_widget_1764820541717._widget_1764820541723",
+            "周期性增购.是否愿意续约": "_widget_1764820541717._widget_1764820541724",
+            "周期性增购.续约后订单编码": "_widget_1764820541717._widget_1764820541725",
+            "订单编码": "_widget_1764820541674",
+            "订单支付日期": "_widget_1764820541679",
+            "本次-实付金额（元）": "_widget_1764820541676",
+            "业务类型（续约、升级）": "_widget_1764820541680",
+            "连锁门店待办同步处理": "_widget_1764820541681",
+            "选择需要同步的门店名称": "_widget_1765330820391",
+            "过期日后90天日期": "_widget_1764820541865",
+            "当前所处节点": "_widget_1765352838609",
+            "流程状态": "_widget_1765352838610",
+            "提交人": "creator",
+            "提交时间": "createTime",
+            "更新时间": "updateTime"
+        }
+        self.cn_field_map = {
+            "related_data": "关联数据",
+            "group_name": "公司名称",
+            "org_name": "门店名称",
+            "org_code": "门店编码",
+            "franchisee": "加盟商",  # 新加字段
+            "expiry_time": "过期日",
+            "saas_edition_fmt": "Saas版本",
+            "last_purchase_price": "上次购买价格",  # 新加字段
+            "contacts": "联系人",
+            "contact_mobile": "联系手机号",
+            "service_impl_principal": "专属运营顾问",
+            "regional_customer_service": "区域客服",  # 新加字段
+            "technician": "运营专家",
+        }
 
     def load_all_data(self):
+        """
+        从各类来源加载数据上加载数据
+        :return:
+        """
         # 获取NGV数据
         payload = {"api_key": "675b900991ad2491c69389ca", "entry_id": "675bb02bd2d53c2034c665e4"}
         self.NGV_data_list = api_instance.entry_data_list(payload).get("data")
@@ -54,20 +131,218 @@ class RenewalToDo:
         payload = {"api_key": "675b900991ad2491c69389ca", "entry_id": "676512ac3e54dc3159460c0a"}
         json_dict = api_instance.entry_data_list(payload)
         if json_dict and "data" in json_dict:
-            self.json_list = json_dict.get("data")
+            self.staff_id_list = json_dict.get("data")
         else:
             print("加载省市区人员关系表失败")
-            self.json_list = []
+            self.staff_id_list = []
 
         # 数据库获取续约回访数据
         self.data_NGV = common_module.get_renewal_details()
 
+    @staticmethod
+    def replace_names_with_staff_ids(df, name_columns, staff_id_list):
+        """
+        将 DataFrame 中多个姓名列替换为对应的员工ID。
+
+        :param df: pandas.DataFrame
+        :param name_columns: list[str]，需要替换的姓名列名列表，例如 ["col1", "col2"]
+        :return: 修改后的 DataFrame（原列被替换）
+        """
+        # 1. 构建姓名 -> 员工ID 的映射字典（只做一次）
+        name_to_id = {}
+        for item in staff_id_list or []:
+            name = item.get("_widget_1734942794144")
+            staff_id = item.get("_widget_1734942794145")
+            if name and staff_id:
+                name_to_id[str(name).strip()] = str(staff_id)
+
+        # 2. 对每个指定的列进行替换
+        df = df.copy()  # 避免修改原始数据
+        for col in name_columns:
+            if col not in df.columns:
+                continue  # 跳过不存在的列
+            # 替换：姓名 → ID，找不到的保留原值（可改为 fillna(None)）
+            df[col] = (
+                df[col]
+                .astype(str)
+                .str.strip()
+                .map(name_to_id)
+                .fillna(df[col])
+            )
+        return df
+
+    @staticmethod
+    def row_to_dict(row, field_mapping):
+        """将一行数据转换为指定格式的字典"""
+        result = {}
+        for col_name, widget_id in field_mapping.items():
+            if col_name in row:
+                value = row[col_name]
+                # 处理Timestamp类型
+                if pd.isna(value):
+                    clean_value = None
+                elif isinstance(value, pd.Timestamp):
+                    clean_value = value.strftime('%Y-%m-%dT%H:%M:%SZ')
+                else:
+                    clean_value = value
+                result[widget_id] = {"value": clean_value}
+        return result
+
+    @staticmethod
+    def en_row_to_cn_row(en_row, en_to_cn_map):
+        """
+        将英文字段的行数据转换为中文字段的行数据
+
+        :param en_row: dict 或 pandas.Series，key 为英文字段
+        :param en_to_cn_map: dict, 英文字段名 -> 中文字段名
+        :return: dict，key 为中文字段名
+        """
+        cn_row = {}
+        for en_key, value in en_row.items():
+            if en_key in en_to_cn_map:
+                cn_key = en_to_cn_map[en_key]
+                cn_row[cn_key] = value
+            # 可选：忽略无法映射的字段，或记录警告
+        return cn_row
+
+    @staticmethod
+    def get_customer_service_by_location(province_name, city_name, area_name, staff_id_list):
+        """
+        直接遍历 self.staff_id_list，根据省市区匹配续约回访客服。
+
+        :return: 客服用户名（str），未找到则返回提示信息
+        """
+        if not all([province_name, city_name, area_name]):
+            return "数据缺失: 省市区不完整"
+
+        for item in staff_id_list or []:
+            try:
+                prov = item.get('_widget_1734677164861', '').strip()
+                city = item.get('_widget_1734677164862', '').strip()
+                area = item.get('_widget_1734677164863', '').strip()
+
+                if (prov == province_name.strip() and
+                        city == city_name.strip() and
+                        area == area_name.strip()):
+                    # 提取客服用户名
+                    staff_info = item.get('_widget_1734677164869', {})  # 续约回访客服
+                    username = staff_info.get('username')
+                    return username if username else "数据缺失: 客服用户名为空"
+            except Exception:
+                continue  # 跳过格式异常的记录
+
+        return "数据缺失: 未找到对应的续约回访客服"
+
     def process_data(self):
         """
-
+        数据处理加工
         :return:
         """
+        data_NGV = self.data_NGV
+        # data_NGV.to_csv(os.path.join(output_dir, "续约回访待办派发.csv"))
 
+        # 日期字段替换为UTC时间
+        time_columns = ['expiry_time']
+
+        data_NGV[time_columns] = data_NGV[time_columns].apply(
+            lambda col: pd.to_datetime(col, errors='coerce')
+            .dt.tz_localize('Asia/Shanghai')  # 假设原时间是北京时间
+            .dt.tz_convert('UTC')  # 转为 UTC
+            .dt.strftime('%Y-%m-%d %H:%M:%S')  # 格式化为字符串（无时区标记）
+        )
+
+        # 成员字段替换
+        staff_name_cols = [
+            "service_impl_principal",
+            "technician",
+        ]
+        data_NGV = self.replace_names_with_staff_ids(data_NGV, staff_name_cols, self.staff_id_list)
+
+        return data_NGV
+
+    def dispatch_task(self, data_NGV):
+        """
+        一次性构建所有记录并派发（依赖 api_instance 内部的分批与容错）
+        """
+        records = []
+        no_customer_service_data = []
+        for _, row in data_NGV.iterrows():
+            NGV_data_id = None
+            # 派发逻辑
+            # step1:优先从 data_NGV 获取省市区信息
+            province_name = row.get('province_name')
+            city_name = row.get('city_name')
+            area_name = row.get('area_name') if 'area_name' in row else row.get('district_name')
+
+            # 检查省市区是否完整（省市区是一体的，任意一个缺失就需要从NGV获取）
+            use_ngv_location = False
+            if (not province_name or province_name in ['', 'None', 'NA'] or
+                    not city_name or city_name in ['', 'None', 'NA'] or
+                    not area_name or area_name in ['', 'None', 'NA']):
+                use_ngv_location = True
+                logger.info(f"门店 {row['org_code']} 的省市区信息不完整，将从NGV_data_list获取")
+
+            # step2:获取关联数据
+            for NGV_Data in self.NGV_data_list:
+                # NGV_Data = NGV_Data.get("data")
+                if row["org_code"] == NGV_Data.get("_widget_1734062123071"):  # 门店编码
+                    NGV_data_id = NGV_Data.get("_id") # 数据id
+
+                    # 如果需要从 NGV_data_list 获取省市区信息
+                    if use_ngv_location:
+                        province_name = NGV_Data.get("_widget_1734062123090")
+                        city_name = NGV_Data.get("_widget_1734062123092")
+                        area_name = NGV_Data.get("_widget_1734062123094")
+                        logger.info(
+                            f"【从NGV获取省市区】门店 {row['org_code']}: {province_name}, {city_name}, {area_name}")
+
+            # step3:根据省市区填充续约待办客服
+            # 检查省市区是否都有值，如果有任何一个为空，则客服为空
+            if (not province_name or province_name in ['', 'None', 'NA'] or
+                    not city_name or city_name in ['', 'None', 'NA'] or
+                    not area_name or area_name in ['', 'None', 'NA']):
+                customer_service = None
+                logger.warning(
+                    f"【省市区信息缺失】门店 {row['org_code']} 省市区信息不完整，续约回访客服设置为空")
+                logger.warning(f"省: {province_name}, 市: {city_name}, 区: {area_name}")
+            else:
+                customer_service = self.get_customer_service_by_location(province_name, city_name, area_name,
+                                                                         self.staff_id_list)
+                logger.info(f"【派发客服】门店 {row['org_code']} 派发给客服: {customer_service}")
+
+            # step4:非NGV字段添加到row中
+            if NGV_data_id:  # 关联数据
+                row["related_data"] = NGV_data_id
+            else:
+                row["related_data"] = None
+                logger.warning(f"未找到关联数据，请检查门店编码: {row['org_code']}")
+            if customer_service:  # 区域客服
+                row["regional_customer_service"] = customer_service
+            else:
+                # 找不到省市区续约待办客服
+                no_customer_service_data.append(row)
+                row["regional_customer_service"] = None
+                logger.warning(f"未找到区域客服，请检查门店编码: {row['org_code']}")
+
+            # 列名替换为中文
+            cn_row = self.en_row_to_cn_row(row, self.cn_field_map)
+            # 简道云字段替换
+            widget_dict = self.row_to_dict(cn_row, self.field_map)
+
+            records.append(widget_dict)
+
+        if not records:
+            logger.info("无数据需要派发")
+            return
+
+        payload = {
+            "api_key": "675b900991ad2491c69389ca",
+            "entry_id": "6931063d64187eaf6b927557",
+            "data_list": records
+        }
+
+        api_instance.entry_data_batch_create(payload)
+        logger.info(f"已提交 {len(records)} 条数据进行派发")
 
     def main(self):
         task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@@ -77,7 +352,9 @@ class RenewalToDo:
             self.load_all_data()
             logger.info("加载数据完成")
             # step2:数据处理
-            self.process_data()
+            data_NGV = self.process_data()
+            # step3:数据派发
+            self.dispatch_task(data_NGV)
 
             common_module.send_task_status(task_start_time, "续约回访待办")
         except Exception as e: