1184 lines
50 KiB
Python
1184 lines
50 KiB
Python
# -*- coding: utf-8 -*-
|
||
|
||
import pandas as pd
|
||
import datetime
|
||
import os
|
||
import sys
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
import time
|
||
import numpy as np
|
||
|
||
# 添加父目录到Python路径，以便导入项目模块
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from config import Config
|
||
from api import API
|
||
from back_ground_module import CommonModule
|
||
from log_config import configure_task_logger, configure_error_task_logger
|
||
from tqdm import tqdm
|
||
|
||
# 获取已经配置好的日志记录器
|
||
logger = configure_task_logger()
|
||
error_task_logger = configure_error_task_logger()
|
||
|
||
# 初始化实例
|
||
api_instance = API()
|
||
common_module = CommonModule()
|
||
|
||
# 输出目录配置
|
||
output_dir = "output"
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# ==================== 全局配置开关 ====================
|
||
|
||
# 【1. 本地缓存配置】
|
||
# 开发调试开关：设置为True时从本地缓存读取数据，False时从API获取
|
||
USE_LOCAL_CACHE = False
|
||
|
||
# 【2. 删除状态重置配置】
|
||
# 首次运行开关：设置为True时会批量标记所有数据为"未删除"（仅首次运行需要）
|
||
RESET_ALL_DELETED_STATUS = False # 首次运行设为True，之后设为False
|
||
|
||
# 【3. 并发更新配置】
|
||
# 是否使用并发更新（多线程同时更新，速度快）
|
||
USE_CONCURRENT_UPDATE = True # True=并发更新（快），False=串行更新（慢）
|
||
|
||
# 并发线程数（同时执行的更新任务数）
|
||
# 建议值：5-20，过大可能被API限流，过小影响速度
|
||
# 如果API限流严重，可以降低到3-5
|
||
CONCURRENT_WORKERS = 4
|
||
|
||
# 【4. 批量创建配置】
|
||
# 是否使用批量创建（批量创建速度快）
|
||
USE_BATCH_CREATE = True # True=批量创建（快），False=逐条创建（慢）
|
||
|
||
# 批量创建大小（每批次创建的记录数）
|
||
BATCH_CREATE_SIZE = 90
|
||
|
||
# ====================================================
|
||
|
||
# 本地缓存文件路径
|
||
CACHE_DIR = os.path.join(output_dir, "cache")
|
||
os.makedirs(CACHE_DIR, exist_ok=True)
|
||
|
||
CACHE_FILES = {
|
||
'jdy_ngv_data': os.path.join(CACHE_DIR, 'jdy_ngv_data.csv'),
|
||
'staff_data': os.path.join(CACHE_DIR, 'staff_data.csv'),
|
||
'ngv_data_today': os.path.join(CACHE_DIR, 'ngv_data_today.csv'),
|
||
'ngv_data_yesterday': os.path.join(CACHE_DIR, 'ngv_data_yesterday.csv'),
|
||
}
|
||
|
||
|
||
class UpdateAllNGVDataDaily:
|
||
"""NGV数据每日更新"""
|
||
|
||
# 门店编码字段的Widget ID
|
||
ORG_CODE_WIDGET_ID = '_widget_1734062123071'
|
||
|
||
def __init__(self):
|
||
"""初始化"""
|
||
self.field_mapping = {}
|
||
self._init_field_mapping()
|
||
# org_code到所有_id的映射（用于更新所有重复记录）
|
||
self.org_code_to_all_ids = {}
|
||
|
||
def main(self):
|
||
"""主流程"""
|
||
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
try:
|
||
logger.info("=" * 60)
|
||
logger.info(f"开始执行NGV数据更新任务：{task_start_time}")
|
||
logger.info("=" * 60)
|
||
|
||
# 步骤1: 加载基础数据
|
||
jdy_ngv_data, staff_id_map = self._load_base_data()
|
||
|
||
# 步骤2: 获取并处理NGV源数据
|
||
ngv_data_today, ngv_data_yesterday = self._load_ngv_source_data(task_start_time)
|
||
|
||
# 步骤3: 处理已删除的门店
|
||
self._handle_deleted_stores(jdy_ngv_data, ngv_data_today)
|
||
|
||
# 步骤4: 对比数据变化
|
||
changed_data = self._compare_data_changes(ngv_data_today, ngv_data_yesterday, jdy_ngv_data)
|
||
|
||
# 步骤5: 数据转换
|
||
prepared_data = self._prepare_data_for_sync(changed_data, staff_id_map)
|
||
|
||
# 步骤6: 同步到简道云
|
||
self._sync_to_jiandaoyun(prepared_data)
|
||
|
||
logger.info("=" * 60)
|
||
logger.info("NGV更新数据任务已完成")
|
||
common_module.send_task_status(task_start_time, "NGV更新数据")
|
||
logger.info("=" * 60)
|
||
|
||
except Exception as e:
|
||
error_task_logger.error(f"NGV更新数据执行时发生异常: {e}", exc_info=True)
|
||
common_module.send_task_error(task_start_time, "NGV更新数据", str(e))
|
||
raise
|
||
|
||
def _compose_key_values(self, org_name, group_name, org_code, id_own_group, id_own_org):
|
||
"""将五个字段组合为稳定索引键，空值用空字符串，占位并去除首尾空格。"""
|
||
def nv(x):
|
||
return '' if pd.isna(x) else str(x).strip()
|
||
parts = [nv(org_name), nv(group_name), nv(org_code), nv(id_own_group), nv(id_own_org)]
|
||
return '||'.join(parts)
|
||
|
||
def _compose_key_df_ngv(self, df):
|
||
"""为NGV数据增加composite_key列（基于字段名）。"""
|
||
cols = ['org_name', 'group_name', 'org_code', 'id_own_group', 'id_own_org']
|
||
for c in cols:
|
||
if c not in df.columns:
|
||
df[c] = ''
|
||
df['composite_key'] = [
|
||
self._compose_key_values(r['org_name'], r['group_name'], r['org_code'], r['id_own_group'], r['id_own_org'])
|
||
for _, r in df.iterrows()
|
||
]
|
||
return df
|
||
|
||
def _compose_key_df_jdy(self, df):
|
||
"""为简道云数据增加composite_key列（基于widget列名）。"""
|
||
# 对应字段widget id
|
||
col_map = {
|
||
'_widget_1734062123070': 'org_name',
|
||
'_widget_1734062123068': 'group_name',
|
||
'_widget_1734062123071': 'org_code',
|
||
'_widget_1734062123067': 'id_own_group',
|
||
'_widget_1734062123069': 'id_own_org',
|
||
}
|
||
tmp = df.copy()
|
||
for wid in col_map.keys():
|
||
if wid not in tmp.columns:
|
||
tmp[wid] = ''
|
||
tmp_renamed = tmp.rename(columns=col_map)
|
||
tmp_renamed['composite_key'] = [
|
||
self._compose_key_values(r.get('org_name', ''), r.get('group_name', ''), r.get('org_code', ''), r.get('id_own_group', ''), r.get('id_own_org', ''))
|
||
for _, r in tmp_renamed.iterrows()
|
||
]
|
||
return tmp_renamed
|
||
|
||
def _load_base_data(self):
|
||
"""
|
||
步骤1: 加载基础数据
|
||
返回: (简道云NGV数据, 员工ID映射字典)
|
||
"""
|
||
logger.info("步骤1: 开始加载基础数据...")
|
||
|
||
# 检查是否使用本地缓存
|
||
if USE_LOCAL_CACHE:
|
||
logger.info(" [缓存模式] 从本地缓存读取数据...")
|
||
jdy_ngv_data = self._load_from_cache('jdy_ngv_data')
|
||
staff_df = self._load_from_cache('staff_data')
|
||
|
||
# 从缓存的staff_data重建映射字典
|
||
staff_id_map = {}
|
||
if not staff_df.empty and '_widget_1734942794144' in staff_df.columns and '_widget_1734942794145' in staff_df.columns:
|
||
staff_id_map = dict(zip(
|
||
staff_df['_widget_1734942794144'].astype(str),
|
||
staff_df['_widget_1734942794145']
|
||
))
|
||
|
||
logger.info(f" - [缓存] 简道云NGV数据: {len(jdy_ngv_data)} 条")
|
||
logger.info(f" - [缓存] 员工ID映射: {len(staff_id_map)} 个员工")
|
||
return jdy_ngv_data, staff_id_map
|
||
|
||
# 从API获取数据
|
||
logger.info(" [API模式] 从简道云API获取数据...")
|
||
|
||
# 获取简道云现有NGV数据
|
||
payload = {
|
||
"api_key": "675b900991ad2491c69389ca",
|
||
"entry_id": "675bb02bd2d53c2034c665e4"
|
||
}
|
||
ngv_data_list = api_instance.entry_data_list(payload).get("data", [])
|
||
jdy_ngv_data = pd.DataFrame(ngv_data_list)
|
||
logger.info(f" - 已获取简道云NGV数据: {len(jdy_ngv_data)} 条")
|
||
|
||
# 保存到本地缓存
|
||
self._save_to_cache(jdy_ngv_data, 'jdy_ngv_data')
|
||
|
||
# 获取员工信息并构建映射字典
|
||
payload = {
|
||
"api_key": "6694d3c4fcb69ca9a111a6c4",
|
||
"entry_id": "6769204a1902c9341340a1bc"
|
||
}
|
||
staff_data = api_instance.entry_data_list(payload)
|
||
staff_list = staff_data.get("data", [])
|
||
staff_df = pd.DataFrame(staff_list)
|
||
|
||
# 保存到本地缓存
|
||
self._save_to_cache(staff_df, 'staff_data')
|
||
|
||
# 构建员工姓名到ID的映射字典（优化查询性能）
|
||
staff_id_map = {
|
||
str(staff['_widget_1734942794144']): staff['_widget_1734942794145']
|
||
for staff in staff_list
|
||
if '_widget_1734942794144' in staff and '_widget_1734942794145' in staff
|
||
}
|
||
logger.info(f" - 已构建员工ID映射字典: {len(staff_id_map)} 个员工")
|
||
|
||
return jdy_ngv_data, staff_id_map
|
||
|
||
def _load_ngv_source_data(self, task_start_time):
|
||
"""
|
||
步骤2: 获取并处理NGV源数据
|
||
返回: (昨天的数据, 前天的数据)
|
||
"""
|
||
logger.info("步骤2: 开始获取NGV源数据...")
|
||
|
||
# 检查是否使用本地缓存
|
||
if USE_LOCAL_CACHE:
|
||
logger.info(" [缓存模式] 从本地缓存读取NGV源数据...")
|
||
ngv_data_1 = self._load_from_cache('ngv_data_today')
|
||
ngv_data_2 = self._load_from_cache('ngv_data_yesterday')
|
||
|
||
logger.info(f" - [缓存] 昨天数据: {len(ngv_data_1)} 条")
|
||
logger.info(f" - [缓存] 前天数据: {len(ngv_data_2)} 条")
|
||
return ngv_data_1, ngv_data_2
|
||
|
||
# 从API获取数据
|
||
logger.info(" [API模式] 从数据库获取NGV源数据...")
|
||
|
||
# 获取最近两天的NGV数据
|
||
ngv_data_1 = common_module.get_ngv_details(days_back=1)
|
||
ngv_data_2 = common_module.get_ngv_details(days_back=2)
|
||
|
||
import time
|
||
nowtime = time.time()
|
||
|
||
# 存储每天获取到的数据
|
||
ngv_data_1.to_csv(f"ngv_data_today.csv", index=False)
|
||
ngv_data_2.to_csv(f"ngv_data_yesterday.csv", index=False)
|
||
|
||
# 只保留 org_type 为 "一般" 的记录
|
||
ngv_data_1 = ngv_data_1[ngv_data_1['org_type'] == '一般']
|
||
ngv_data_2 = ngv_data_2[ngv_data_2['org_type'] == '一般']
|
||
|
||
# 保存到本地缓存
|
||
self._save_to_cache(ngv_data_1, 'ngv_data_today')
|
||
self._save_to_cache(ngv_data_2, 'ngv_data_yesterday')
|
||
|
||
logger.info(f" - 昨天数据: {len(ngv_data_1)} 条")
|
||
logger.info(f" - 前天数据: {len(ngv_data_2)} 条")
|
||
|
||
return ngv_data_1, ngv_data_2
|
||
|
||
def _handle_deleted_stores(self, jdy_ngv_data, ngv_current_data):
|
||
"""
|
||
步骤3: 处理已删除的门店
|
||
逻辑:
|
||
1. 【首次运行】批量标记所有数据为"未删除"（通过RESET_ALL_DELETED_STATUS开关控制）
|
||
2. 批量标记不存在的门店为"已删除"
|
||
|
||
重要：这个方法会修改简道云数据！
|
||
"""
|
||
logger.info("步骤3: 开始处理已删除的门店...")
|
||
|
||
# 准备简道云数据
|
||
temp_jdy_data = jdy_ngv_data.copy()
|
||
temp_jdy_data.reset_index(drop=True, inplace=True)
|
||
|
||
if self.ORG_CODE_WIDGET_ID not in temp_jdy_data.columns:
|
||
error_task_logger.error("列 '门店编码' 不存在")
|
||
return
|
||
|
||
# 添加org_code列
|
||
temp_jdy_data['org_code'] = temp_jdy_data[self.ORG_CODE_WIDGET_ID]
|
||
|
||
# 检查是否有空的org_code
|
||
null_org_codes = temp_jdy_data['org_code'].isna().sum()
|
||
if null_org_codes > 0:
|
||
logger.warning(f" ⚠ 警告: 简道云数据中有 {null_org_codes} 条记录的门店编码为空，将被忽略")
|
||
temp_jdy_data = temp_jdy_data[temp_jdy_data['org_code'].notna()]
|
||
|
||
# 检查并记录重复的org_code【数据质量问题】
|
||
duplicate_org_codes = temp_jdy_data[temp_jdy_data['org_code'].duplicated(keep=False)]
|
||
if len(duplicate_org_codes) > 0:
|
||
unique_duplicate_org_codes = duplicate_org_codes['org_code'].nunique()
|
||
logger.warning(
|
||
f" ⚠ 数据质量问题：发现 {unique_duplicate_org_codes} 个org_code重复，共 {len(duplicate_org_codes)} 条记录")
|
||
logger.warning(f" ⚠ 建议：请在简道云中清理重复数据，每个org_code只保留一条记录")
|
||
|
||
# 保存重复记录到文件供分析
|
||
if output_dir:
|
||
duplicate_org_codes[['org_code', '_id']].to_csv(
|
||
os.path.join(output_dir, '重复org_code记录.csv'),
|
||
index=False,
|
||
encoding='utf-8-sig'
|
||
)
|
||
logger.warning(f" - 重复详情已保存到: 重复org_code记录.csv")
|
||
|
||
# 【重要】标记删除时不去重，需要标记所有重复记录
|
||
logger.info(f" - 简道云原始数据共 {len(temp_jdy_data)} 条记录（包含重复）")
|
||
logger.info(f" - NGV源数据: {len(ngv_current_data)} 条")
|
||
|
||
# 检查_id列是否存在
|
||
if '_id' not in temp_jdy_data.columns:
|
||
error_task_logger.error(" ✗ 错误: 简道云数据中没有_id列，无法标记删除状态")
|
||
return
|
||
|
||
# 3.1 【首次运行】批量标记所有数据为"未删除"
|
||
if RESET_ALL_DELETED_STATUS:
|
||
logger.info(" ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
|
||
logger.info(" 【首次运行模式】批量重置所有数据删除状态")
|
||
logger.info(" ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
|
||
|
||
all_data_ids = temp_jdy_data['_id'].dropna().tolist()
|
||
if all_data_ids:
|
||
logger.info(f" - 准备批量标记所有数据为'未删除': {len(all_data_ids)} 条")
|
||
|
||
batch_data = {
|
||
'api_key': Config.SaaS_Tasks_APP_ID,
|
||
'entry_id': Config.NGV_TASKS_ENTRY_ID,
|
||
"data_ids": all_data_ids,
|
||
"data": {"_widget_1754285499851": {"value": "未删除"}}
|
||
}
|
||
api_instance.entry_data_banch_update(data=batch_data, max_retries=20)
|
||
logger.info(f" ✓ 已批量标记所有数据为'未删除': {len(all_data_ids)} 条")
|
||
logger.info(f" 💡 提示：首次运行完成后，请将 RESET_ALL_DELETED_STATUS 设置为 False")
|
||
else:
|
||
logger.warning(" ⚠ 警告: 没有有效的_id可以标记")
|
||
|
||
# 3.2 找出需要标记为"已删除"的门店
|
||
ngv_org_codes = set(ngv_current_data['org_code'].dropna().unique())
|
||
jdy_org_codes_unique = set(temp_jdy_data['org_code'].dropna().unique())
|
||
|
||
# 找出在简道云存在但NGV中不存在的门店（唯一复合索引）
|
||
missing_org_codes = jdy_org_codes_unique - ngv_org_codes
|
||
|
||
if len(missing_org_codes) == 0:
|
||
logger.info(" - 无需处理的已删除门店（简道云和NGV数据完全一致）")
|
||
return
|
||
|
||
logger.info(f" - 发现 {len(missing_org_codes)} 个门店在简道云存在但NGV中不存在")
|
||
|
||
# 【关键】筛选出所有匹配的记录（包括重复的）
|
||
# 这样可以把同一个org_code的所有记录都标记为已删除
|
||
only_in_jdy = temp_jdy_data[temp_jdy_data['org_code'].isin(missing_org_codes)].copy()
|
||
|
||
logger.info(f" - 需要标记删除的org_code数: {len(missing_org_codes)}")
|
||
logger.info(f" - 需要标记删除的记录数: {len(only_in_jdy)} 条（包含重复记录）")
|
||
|
||
# 3.3 批量标记为"已删除"
|
||
delete_data_ids = only_in_jdy['_id'].dropna().tolist()
|
||
if not delete_data_ids:
|
||
logger.warning(" ⚠ 警告: 没有有效的_id可以标记为已删除")
|
||
return
|
||
|
||
logger.info(f" - 准备批量标记为'已删除': {len(delete_data_ids)} 条")
|
||
|
||
batch_data = {
|
||
'api_key': Config.SaaS_Tasks_APP_ID,
|
||
'entry_id': Config.NGV_TASKS_ENTRY_ID,
|
||
"data_ids": delete_data_ids,
|
||
"data": {"_widget_1754285499851": {"value": "已删除"}}
|
||
}
|
||
api_instance.entry_data_banch_update(data=batch_data, max_retries=20)
|
||
logger.info(f" ✓ 已批量标记为'已删除': {len(delete_data_ids)} 条")
|
||
|
||
def _compare_data_changes(self, ngv_today, ngv_yesterday, jdy_ngv_data):
|
||
"""
|
||
步骤4: 对比数据变化，找出需要更新的记录
|
||
返回: 带有_id的变化数据DataFrame
|
||
"""
|
||
logger.info("步骤4: 开始对比数据变化...")
|
||
|
||
# 移除不需要对比的列
|
||
columns_to_remove = {'date_id', 'date_fmt', 'pt', 'etl_time','id_own_org'}
|
||
|
||
# 过滤列
|
||
df1_filtered = ngv_today[[col for col in ngv_today.columns if col not in columns_to_remove]]
|
||
df2_filtered = ngv_yesterday[[col for col in ngv_yesterday.columns if col not in columns_to_remove]]
|
||
|
||
# 设置索引
|
||
df1_indexed = df1_filtered.set_index('org_code')
|
||
df2_indexed = df2_filtered.set_index('org_code')
|
||
|
||
# 数据清洗：统一处理空值
|
||
df1_indexed = df1_indexed.astype(str).replace(['nan', 'None'], '').fillna('')
|
||
df2_indexed = df2_indexed.astype(str).replace(['nan', 'None'], '').fillna('')
|
||
|
||
# 找到共同的索引和列
|
||
common_index = df1_indexed.index.intersection(df2_indexed.index)
|
||
df1_common = df1_indexed.reindex(common_index).fillna('')
|
||
df2_common = df2_indexed.reindex(common_index).fillna('')
|
||
|
||
common_columns = df1_common.columns.intersection(df2_common.columns)
|
||
df1_common = df1_common[common_columns]
|
||
df2_common = df2_common[common_columns]
|
||
|
||
# 对比数据
|
||
matches = (df1_common == df2_common).all(axis=1)
|
||
df1_common['match_status'] = matches.map({True: '一致', False: '不一致'})
|
||
|
||
# 统计一致和不一致的数量
|
||
consistent_count = (df1_common['match_status'] == '一致').sum()
|
||
inconsistent_count = (df1_common['match_status'] == '不一致').sum()
|
||
|
||
logger.info(f" 📊 共同门店数据对比结果：")
|
||
logger.info(f" - 共同门店总数: {len(df1_common)}")
|
||
logger.info(f" - 数据一致: {consistent_count} 条")
|
||
logger.info(f" - 数据不一致(需要更新): {inconsistent_count} 条")
|
||
|
||
# 只保留不一致的数据
|
||
changed_data = df1_common[df1_common['match_status'] == '不一致'].copy()
|
||
|
||
# 关联简道云的_id（基于org_code）
|
||
temp_jdy = jdy_ngv_data.copy()
|
||
temp_jdy.reset_index(drop=True, inplace=True)
|
||
|
||
# 【新增】保存org_code到所有_id的映射（用于更新所有重复记录）
|
||
org_code_to_all_ids = {}
|
||
|
||
if self.ORG_CODE_WIDGET_ID in temp_jdy.columns:
|
||
temp_jdy.rename(columns={self.ORG_CODE_WIDGET_ID: 'org_code'}, inplace=True)
|
||
|
||
# 构建org_code到所有_id的映射
|
||
for _, row in temp_jdy.iterrows():
|
||
org_code = row['org_code']
|
||
data_id = row['_id']
|
||
if pd.notna(org_code) and pd.notna(data_id):
|
||
if org_code not in org_code_to_all_ids:
|
||
org_code_to_all_ids[org_code] = []
|
||
org_code_to_all_ids[org_code].append(data_id)
|
||
|
||
# 统计重复情况
|
||
multi_id_count = sum(1 for ids in org_code_to_all_ids.values() if len(ids) > 1)
|
||
if multi_id_count > 0:
|
||
logger.info(f" - 发现 {multi_id_count} 个org_code有多个_id，更新时将同步所有记录")
|
||
|
||
# 去重用于join（避免数据爆炸）
|
||
if '_createTime' in temp_jdy.columns:
|
||
temp_jdy = temp_jdy.sort_values('_createTime', ascending=False)
|
||
temp_jdy = temp_jdy.drop_duplicates(subset=['org_code'], keep='first')
|
||
temp_jdy.set_index('org_code', inplace=True)
|
||
changed_data = changed_data.join(temp_jdy['_id'], how='left')
|
||
|
||
# 保存映射关系，供同步时使用
|
||
self.org_code_to_all_ids = org_code_to_all_ids
|
||
|
||
# 【查缺补漏】查找NGV昨天有但简道云没有的门店，需要补充新建
|
||
logger.info(f"")
|
||
logger.info(f" 📊 查缺补漏分析（NGV有但简道云没有）：")
|
||
|
||
# 获取NGV昨天的所有org_code（唯一值）
|
||
ngv_org_codes = set(df1_indexed.index.unique())
|
||
# 获取简道云的所有org_code（已去重的）
|
||
jdy_org_codes = set(temp_jdy.index.unique()) if len(temp_jdy) > 0 else set()
|
||
|
||
# 找出NGV有但简道云没有的门店（需要补充新建）
|
||
missing_in_jdy = ngv_org_codes - jdy_org_codes
|
||
|
||
logger.info(f" - NGV昨天的门店总数: {len(ngv_org_codes)}")
|
||
logger.info(f" - 简道云的门店总数: {len(jdy_org_codes)}")
|
||
logger.info(f" - NGV有但简道云没有(需补充): {len(missing_in_jdy)} 条")
|
||
|
||
if len(missing_in_jdy) > 0:
|
||
# 获取这些缺失门店的完整数据
|
||
missing_stores = df1_indexed.loc[list(missing_in_jdy)].copy()
|
||
missing_stores['match_status'] = '查缺补漏-需新建'
|
||
missing_stores['_id'] = None # 这些门店在简道云中没有_id
|
||
|
||
# 【关键】检查是否与已有的changed_data重复
|
||
if len(changed_data) > 0:
|
||
# 找出在changed_data中已存在的org_code
|
||
existing_org_codes = set(changed_data.index)
|
||
overlap_org_codes = missing_in_jdy & existing_org_codes
|
||
if len(overlap_org_codes) > 0:
|
||
logger.warning(f" ⚠ 警告：发现 {len(overlap_org_codes)} 个org_code同时在'不一致'和'查缺补漏'中")
|
||
logger.warning(f" 这些记录可能是数据问题，已从查缺补漏中排除")
|
||
# 从missing_stores中移除重复的
|
||
missing_stores = missing_stores[~missing_stores.index.isin(overlap_org_codes)]
|
||
missing_in_jdy = missing_in_jdy - overlap_org_codes
|
||
|
||
if len(missing_stores) > 0:
|
||
logger.info(f" - 实际需要补充新建: {len(missing_stores)} 条")
|
||
# 将缺失门店加入到需要处理的数据中
|
||
changed_data = pd.concat([changed_data, missing_stores])
|
||
else:
|
||
logger.info(f" ✓ 排除重复后无需补充")
|
||
else:
|
||
logger.info(f" ✓ 无需补充，简道云数据完整")
|
||
|
||
logger.info(f"")
|
||
logger.info(f" 📊 汇总统计：")
|
||
logger.info(f" - 需要处理的数据总数: {len(changed_data)} 条")
|
||
logger.info(f" ├─ 数据不一致(需更新): {inconsistent_count} 条")
|
||
logger.info(f" └─ 查缺补漏(需新建): {len(missing_in_jdy)} 条")
|
||
|
||
return changed_data
|
||
|
||
def _prepare_data_for_sync(self, data_df, staff_id_map):
|
||
"""
|
||
步骤5: 数据转换（日期转UTC、人员字段转换）
|
||
"""
|
||
logger.info("步骤5: 开始数据转换...")
|
||
|
||
if len(data_df) == 0:
|
||
logger.info(" - 无需转换的数据")
|
||
return data_df
|
||
|
||
prepared_df = data_df.copy()
|
||
|
||
# 5.1 日期字段转换为UTC格式
|
||
time_columns = ['saas_create_time', 'expiry_time', 'install_create_time',
|
||
'last_end_date', 'renew_date']
|
||
|
||
for col in time_columns:
|
||
if col not in prepared_df.columns:
|
||
continue
|
||
|
||
# 转换为datetime类型
|
||
temp_series = pd.to_datetime(prepared_df[col], errors='coerce', utc=False)
|
||
|
||
# 转换为UTC时区格式
|
||
prepared_df[col + '_date'] = (
|
||
temp_series
|
||
.dt.tz_localize('Asia/Shanghai', ambiguous='infer', nonexistent='NaT')
|
||
.dt.tz_convert('UTC')
|
||
.dt.strftime('%Y-%m-%dT%H:%M:%SZ')
|
||
)
|
||
|
||
logger.info(" - 日期字段已转换为UTC格式")
|
||
|
||
# 5.2 人员字段转换为员工ID（使用字典映射，优化性能）
|
||
staff_columns = ['area_manager', 'service_impl_principal',
|
||
'service_salesmen', 'technician']
|
||
|
||
for col in staff_columns:
|
||
if col not in prepared_df.columns:
|
||
continue
|
||
|
||
# 使用向量化操作替代三重循环
|
||
prepared_df[col + '_staff_id'] = prepared_df[col].astype(str).map(
|
||
lambda x: staff_id_map.get(x, None)
|
||
)
|
||
|
||
logger.info(" - 人员字段已转换为员工ID")
|
||
|
||
# 5.3G转化率保留3位小数
|
||
prepared_df['g_month_percentage'] = (pd.to_numeric(prepared_df['g_month_percentage'], errors='coerce')
|
||
.round(3)
|
||
.apply(lambda x: f"{x:.3f}" if pd.notna(x) else ''))
|
||
logger.info(" - G转化率已保留3位小数")
|
||
|
||
return prepared_df
|
||
|
||
def _sync_to_jiandaoyun(self, data_df):
|
||
"""
|
||
步骤6: 同步数据到简道云
|
||
支持批量更新和批量创建，大幅提升速度
|
||
"""
|
||
logger.info("步骤6: 开始同步数据到简道云...")
|
||
|
||
if len(data_df) == 0:
|
||
logger.info(" - 无需同步的数据")
|
||
return
|
||
|
||
# 分离更新和创建数据
|
||
update_data_list = []
|
||
create_data_list = []
|
||
|
||
logger.info(f" - 准备同步数据...")
|
||
for idx, row in data_df.iterrows():
|
||
# 构建数据字典
|
||
data_dict = self._build_data_dict(row)
|
||
|
||
# 判断是更新还是创建
|
||
has_id = '_id' in data_df.columns and pd.notna(row.get('_id'))
|
||
|
||
if has_id:
|
||
# 更新操作：移除门店编码字段（不允许修改）
|
||
if self.ORG_CODE_WIDGET_ID in data_dict:
|
||
del data_dict[self.ORG_CODE_WIDGET_ID]
|
||
|
||
# 【关键】获取该org_code的所有_id（包括重复记录）
|
||
org_code = idx
|
||
all_data_ids = self.org_code_to_all_ids.get(org_code, [str(row['_id'])])
|
||
|
||
# 为每个_id都准备更新数据
|
||
for data_id in all_data_ids:
|
||
update_data_list.append({
|
||
'org_code': org_code,
|
||
'data_id': str(data_id),
|
||
'data_dict': data_dict.copy(),
|
||
'row_data': row # 保存原始数据用于输出
|
||
})
|
||
else:
|
||
|
||
# 新增：仅创建内容非全空的记录
|
||
# 检查除了门店编码外，其他字段是否全为空
|
||
all_empty = True
|
||
for k, v in data_dict.items():
|
||
if k != self.ORG_CODE_WIDGET_ID and v.get('value') not in (None, '', float('nan')):
|
||
all_empty = False
|
||
break
|
||
if all_empty:
|
||
logger.info(f" - 跳过内容全空的新建记录 org_code: {idx}")
|
||
continue
|
||
|
||
create_data_list.append({
|
||
'org_code': idx,
|
||
'data_dict': data_dict,
|
||
'row_data': row # 保存原始数据用于输出
|
||
})
|
||
|
||
logger.info(f" - 需要更新: {len(update_data_list)} 条")
|
||
logger.info(f" - 需要创建: {len(create_data_list)} 条")
|
||
|
||
# 执行更新
|
||
update_count = 0
|
||
if len(update_data_list) > 0:
|
||
if USE_CONCURRENT_UPDATE:
|
||
update_count = self._concurrent_update(update_data_list)
|
||
else:
|
||
update_count = self._single_update(update_data_list)
|
||
# 输出更新统计
|
||
self._save_update_stats(update_data_list)
|
||
|
||
# 执行创建
|
||
create_count = 0
|
||
|
||
create_df = pd.DataFrame(create_data_list)
|
||
create_df.to_csv(f"create_data.csv", index=False)
|
||
# if len(create_data_list) > 0:
|
||
# if USE_BATCH_CREATE:
|
||
# create_count = self._batch_create(create_data_list)
|
||
# else:
|
||
# create_count = self._single_create(create_data_list)
|
||
# # 输出新增数据
|
||
# self._save_create_data(create_data_list)
|
||
|
||
logger.info(f" ✓ 同步完成: 更新 {update_count} 条, 创建 {create_count} 条")
|
||
|
||
def _concurrent_update(self, update_data_list):
|
||
"""
|
||
并发更新（多线程，速度快）
|
||
|
||
使用线程池并发执行更新操作，大幅提升速度。
|
||
预期速度提升：3-10倍（取决于CONCURRENT_WORKERS设置）
|
||
"""
|
||
logger.info(f" - 使用并发更新模式 (并发数: {CONCURRENT_WORKERS})")
|
||
|
||
total_count = len(update_data_list)
|
||
success_count = 0
|
||
failed_count = 0
|
||
start_time = time.time()
|
||
|
||
# 定义单条更新函数
|
||
def update_single_record(item):
|
||
try:
|
||
update_data = {
|
||
'api_key': Config.SaaS_Tasks_APP_ID,
|
||
'entry_id': Config.NGV_TASKS_ENTRY_ID,
|
||
'data_id': item['data_id'],
|
||
'data': item['data_dict']
|
||
}
|
||
api_instance.entry_data_update(data=update_data, max_retries=20)
|
||
return {'success': True, 'org_code': item['org_code']}
|
||
except Exception as e:
|
||
error_task_logger.error(f"更新失败 (org_code={item['org_code']}): {e}")
|
||
return {'success': False, 'org_code': item['org_code'], 'error': str(e)}
|
||
|
||
# 使用线程池并发执行
|
||
with ThreadPoolExecutor(max_workers=CONCURRENT_WORKERS) as executor:
|
||
# 提交所有任务
|
||
future_to_item = {executor.submit(update_single_record, item): item for item in update_data_list}
|
||
|
||
# 使用tqdm显示进度
|
||
with tqdm(total=total_count, desc="并发更新") as pbar:
|
||
for future in as_completed(future_to_item):
|
||
result = future.result()
|
||
if result['success']:
|
||
success_count += 1
|
||
else:
|
||
failed_count += 1
|
||
pbar.update(1)
|
||
|
||
elapsed_time = time.time() - start_time
|
||
speed = total_count / elapsed_time if elapsed_time > 0 else 0
|
||
|
||
logger.info(f" - 更新完成: 成功 {success_count} 条, 失败 {failed_count} 条")
|
||
logger.info(f" - 耗时: {elapsed_time:.1f}秒, 速度: {speed:.1f}条/秒")
|
||
|
||
return success_count
|
||
|
||
def _single_update(self, update_data_list):
|
||
"""逐条更新（串行模式，速度慢但易定位问题）"""
|
||
logger.info(f" - 使用串行更新模式（逐条）")
|
||
|
||
success_count = 0
|
||
failed_count = 0
|
||
total_count = len(update_data_list)
|
||
start_time = time.time()
|
||
|
||
for item in tqdm(update_data_list, desc="串行更新"):
|
||
try:
|
||
update_data = {
|
||
'api_key': Config.SaaS_Tasks_APP_ID,
|
||
'entry_id': Config.NGV_TASKS_ENTRY_ID,
|
||
'data_id': item['data_id'],
|
||
'data': item['data_dict']
|
||
}
|
||
api_instance.entry_data_update(data=update_data, max_retries=20)
|
||
success_count += 1
|
||
except Exception as e:
|
||
failed_count += 1
|
||
error_task_logger.error(f"更新失败 (org_code={item['org_code']}): {e}")
|
||
|
||
elapsed_time = time.time() - start_time
|
||
speed = total_count / elapsed_time if elapsed_time > 0 else 0
|
||
|
||
logger.info(f" - 更新完成: 成功 {success_count} 条, 失败 {failed_count} 条")
|
||
logger.info(f" - 耗时: {elapsed_time:.1f}秒, 速度: {speed:.1f}条/秒")
|
||
|
||
return success_count
|
||
|
||
def _batch_create(self, create_data_list):
|
||
"""批量创建（速度快）"""
|
||
logger.info(f" - 使用批量创建模式 (批次大小: {BATCH_CREATE_SIZE})")
|
||
|
||
total_count = len(create_data_list)
|
||
success_count = 0
|
||
|
||
# 分批处理
|
||
for i in range(0, total_count, BATCH_CREATE_SIZE):
|
||
batch = create_data_list[i:i + BATCH_CREATE_SIZE]
|
||
|
||
try:
|
||
# 批量创建
|
||
batch_create_data = {
|
||
'api_key': Config.SaaS_Tasks_APP_ID,
|
||
'entry_id': Config.NGV_TASKS_ENTRY_ID,
|
||
'data': [item['data_dict'] for item in batch]
|
||
}
|
||
result = api_instance.data_batch_create(data=batch_create_data, max_retries=20)
|
||
success_count += len(batch)
|
||
|
||
if (i + BATCH_CREATE_SIZE) % (BATCH_CREATE_SIZE * 5) == 0 or (i + BATCH_CREATE_SIZE) >= total_count:
|
||
logger.info(f" - 已创建 {min(i + BATCH_CREATE_SIZE, total_count)}/{total_count} 条")
|
||
except Exception as e:
|
||
error_task_logger.error(f"批量创建失败 (批次 {i // BATCH_CREATE_SIZE + 1}): {e}")
|
||
# 批量失败时，尝试逐条创建这一批
|
||
logger.warning(f" - 批量创建失败，尝试逐条创建该批次...")
|
||
for item in batch:
|
||
try:
|
||
single_create_data = {
|
||
'api_key': Config.SaaS_Tasks_APP_ID,
|
||
'entry_id': Config.NGV_TASKS_ENTRY_ID,
|
||
'data': item['data_dict']
|
||
}
|
||
api_instance.data_batch_create(data=single_create_data, max_retries=20)
|
||
success_count += 1
|
||
except Exception as e2:
|
||
error_task_logger.error(f"逐条创建失败 (org_code={item['org_code']}): {e2}")
|
||
|
||
return success_count
|
||
|
||
def _single_create(self, create_data_list):
|
||
"""逐条创建（速度慢但易定位问题）"""
|
||
logger.info(f" - 使用逐条创建模式")
|
||
|
||
success_count = 0
|
||
|
||
for item in tqdm(create_data_list, desc="逐条创建"):
|
||
try:
|
||
create_data = {
|
||
'api_key': Config.SaaS_Tasks_APP_ID,
|
||
'entry_id': Config.NGV_TASKS_ENTRY_ID,
|
||
'data': item['data_dict'],
|
||
'is_start_trigger': 'true',
|
||
|
||
}
|
||
api_instance.data_batch_create(data=create_data, max_retries=20)
|
||
success_count += 1
|
||
except Exception as e:
|
||
error_task_logger.error(f"创建失败 (org_code={item['org_code']}): {e}")
|
||
|
||
return success_count
|
||
|
||
def _build_data_dict(self, row):
|
||
"""
|
||
根据field_mapping构建简道云数据字典
|
||
"""
|
||
data_dict = {}
|
||
|
||
for col_name, widget_id in self.field_mapping.items():
|
||
if col_name in row.index:
|
||
value = row[col_name]
|
||
clean_value = None if pd.isna(value) else value
|
||
data_dict[widget_id] = {"value": clean_value}
|
||
|
||
return data_dict
|
||
|
||
def _init_field_mapping(self):
|
||
"""初始化字段映射关系"""
|
||
self.field_mapping = {
|
||
# 基础信息
|
||
'date_id': '_widget_1734062123065',
|
||
'date_fmt': '_widget_1734062123066',
|
||
'id_own_group': '_widget_1734062123067',
|
||
'group_name': '_widget_1734062123068',
|
||
'id_own_org': '_widget_1734062123069',
|
||
'org_name': '_widget_1734062123070',
|
||
'org_code': '_widget_1734062123071',
|
||
'group_grade': '_widget_1734062123072',
|
||
'org_type': '_widget_1734062123073',
|
||
'org_status': '_widget_1734062123074',
|
||
|
||
# SaaS相关
|
||
'saas_version': '_widget_1734062123075',
|
||
'is_wechat': '_widget_1734062123076',
|
||
'is_mini_app': '_widget_1734062123077',
|
||
'is_wx_shop': '_widget_1734062123078',
|
||
'is_camera_service': '_widget_1734062123079',
|
||
'is_maintenance_service': '_widget_1734062123080',
|
||
'saas_create_time': '_widget_1734062123081',
|
||
'expiry_time': '_widget_1734062123177', # 改过期日
|
||
'saas_use_days': '_widget_1734062123083',
|
||
'saas_use_year': '_widget_1734062123084',
|
||
'is_main_org': '_widget_1734062123085',
|
||
|
||
# 证照信息
|
||
'license_code': '_widget_1734062123086',
|
||
'license_name': '_widget_1734062123087',
|
||
'org_crm_id': '_widget_1734062123088',
|
||
|
||
# 地理信息
|
||
'province_id': '_widget_1734062123089',
|
||
'province_name': '_widget_1734062123090',
|
||
'city_id': '_widget_1734062123091',
|
||
'city_name': '_widget_1734062123092',
|
||
'area_id': '_widget_1734062123093',
|
||
'area_name': '_widget_1734062123094',
|
||
'region_name': '_widget_1734062123095',
|
||
'region_short_name': '_widget_1734062123096',
|
||
'branch_name': '_widget_1734062123097',
|
||
|
||
# 车主邦相关
|
||
'carzone_store_id': '_widget_1734062123098',
|
||
'carzone_store_name': '_widget_1734062123099',
|
||
'customer_carzone_id': '_widget_1734062123100',
|
||
|
||
# 人员信息
|
||
'salesmen': '_widget_1734062123101',
|
||
'area_manager': '_widget_1734062123102',
|
||
'service_salesmen': '_widget_1734062123103',
|
||
'impl_principal': '_widget_1734062123104',
|
||
'service_impl_principal': '_widget_1734062123105',
|
||
|
||
# 用户统计
|
||
'active_user_count': '_widget_1734062123106',
|
||
'active_user_type': '_widget_1734062123107',
|
||
'limit_user_count': '_widget_1734062123108',
|
||
'limit_user_type': '_widget_1734062123109',
|
||
|
||
# NGV标记
|
||
'is_n': '_widget_1734062123110',
|
||
'is_g': '_widget_1734062123111',
|
||
'is_v': '_widget_1734062123112',
|
||
'is_visited': '_widget_1734062123113',
|
||
'is_active': '_widget_1734062123114',
|
||
'active_status_fmt': '_widget_1734062123115',
|
||
|
||
# 单据统计
|
||
'bill_count_last_30_day': '_widget_1734062123116',
|
||
'bill_day_count_last_30_day': '_widget_1734062123117',
|
||
'bill_day_count_this_month': '_widget_1734062123118',
|
||
'bill_count_last_7_day': '_widget_1734062123119',
|
||
'bill_day_count_last_7_day': '_widget_1734062123120',
|
||
'pv_count': '_widget_1734062123121',
|
||
'uv_count': '_widget_1734062123122',
|
||
|
||
# 每日单据数（1-31天）
|
||
'bill_count_1d': '_widget_1734062123123',
|
||
'bill_count_2d': '_widget_1734062123124',
|
||
'bill_count_3d': '_widget_1734062123125',
|
||
'bill_count_4d': '_widget_1734062123126',
|
||
'bill_count_5d': '_widget_1734062123127',
|
||
'bill_count_6d': '_widget_1734062123128',
|
||
'bill_count_7d': '_widget_1734062123129',
|
||
'bill_count_8d': '_widget_1734062123130',
|
||
'bill_count_9d': '_widget_1734062123131',
|
||
'bill_count_10d': '_widget_1734062123132',
|
||
'bill_count_11d': '_widget_1734062123133',
|
||
'bill_count_12d': '_widget_1734062123134',
|
||
'bill_count_13d': '_widget_1734062123135',
|
||
'bill_count_14d': '_widget_1734062123136',
|
||
'bill_count_15d': '_widget_1734062123137',
|
||
'bill_count_16d': '_widget_1734062123138',
|
||
'bill_count_17d': '_widget_1734062123139',
|
||
'bill_count_18d': '_widget_1734062123140',
|
||
'bill_count_19d': '_widget_1734062123141',
|
||
'bill_count_20d': '_widget_1734062123142',
|
||
'bill_count_21d': '_widget_1734062123143',
|
||
'bill_count_22d': '_widget_1734062123144',
|
||
'bill_count_23d': '_widget_1734062123145',
|
||
'bill_count_24d': '_widget_1734062123146',
|
||
'bill_count_25d': '_widget_1734062123147',
|
||
'bill_count_26d': '_widget_1734062123148',
|
||
'bill_count_27d': '_widget_1734062123149',
|
||
'bill_count_28d': '_widget_1734062123150',
|
||
'bill_count_29d': '_widget_1734062123151',
|
||
'bill_count_30d': '_widget_1734062123152',
|
||
'bill_count_31d': '_widget_1734062123153',
|
||
|
||
# ETL时间
|
||
'etl_time': '_widget_1734062123154',
|
||
|
||
# 业务类型统计
|
||
'maintain_bill_count_last_30_day': '_widget_1734062123155',
|
||
'washing_bill_count_last_30_day': '_widget_1734062123156',
|
||
'maintain_bill_day_count_last_30_day': '_widget_1734062123157',
|
||
'washing_bill_day_count_last_30_day': '_widget_1734062123158',
|
||
'retail_bill_count_last_30_day': '_widget_1734062123159',
|
||
'retail_bill_day_count_last_30_day': '_widget_1734062123160',
|
||
'purchase_bill_count_last_30_day': '_widget_1734062123161',
|
||
'purchase_bill_day_count_last_30_day': '_widget_1734062123162',
|
||
'card_bill_count_last_30_day': '_widget_1734062123163',
|
||
'card_bill_day_count_last_30_day': '_widget_1734062123164',
|
||
'gd_sales_bill_count_last_30_day': '_widget_1734062123165',
|
||
'gd_sales_bill_day_count_last_30_day': '_widget_1734062123166',
|
||
|
||
# G标记相关
|
||
'g_change_flag': '_widget_1734062123167',
|
||
'saas_package': '_widget_1734062123168',
|
||
'manage_model': '_widget_1734062123169',
|
||
|
||
# 联系信息
|
||
'contacts': '_widget_1734062123170',
|
||
'contact_number': '_widget_1734062123171',
|
||
'contact_mobile': '_widget_1734062123172',
|
||
|
||
# G月度统计
|
||
'g_month_count': '_widget_1734062123173',
|
||
'g_month_percentage': '_widget_1734062123174',
|
||
|
||
# 安装服务
|
||
'is_install_service': '_widget_1734062123175',
|
||
'install_create_time': '_widget_1734062123176',
|
||
'last_end_date': '_widget_1734062123082',
|
||
'renew_date': '_widget_1734062123178',
|
||
|
||
# 连锁信息
|
||
'is_chain_owner': '_widget_1734062123179',
|
||
'group_org_count': '_widget_1734062123180',
|
||
|
||
# 预警信息
|
||
'recent_bill_warning_days': '_widget_1734062123181',
|
||
'g_change_flag_d': '_widget_1734062123182',
|
||
'g_lost_warning_days': '_widget_1734062123183',
|
||
|
||
# SaaS版本
|
||
'saas_edition_fmt': '_widget_1734062123184',
|
||
|
||
# G标记月度（1-6月）
|
||
'g_flag_1m': '_widget_1734062123185',
|
||
'g_flag_2m': '_widget_1734062123186',
|
||
'g_flag_3m': '_widget_1734062123187',
|
||
'g_flag_4m': '_widget_1734062123188',
|
||
'g_flag_5m': '_widget_1734062123189',
|
||
'g_flag_6m': '_widget_1734062123190',
|
||
'g_flag_day_count': '_widget_1734062123191',
|
||
|
||
# 其他标记
|
||
'add_org_flag': '_widget_1734062123192',
|
||
'pt': '_widget_1734062123193',
|
||
|
||
# 门店属性
|
||
'org_size': '_widget_1734062123194',
|
||
'qualification_type_fmt': '_widget_1734062123195',
|
||
'business_scope_fmt': '_widget_1734062123196',
|
||
'store_type_fmt': '_widget_1734062123197',
|
||
'area': '_widget_1734062123198',
|
||
'station_number': '_widget_1734062123199',
|
||
'header_type_fmt': '_widget_1734062123200',
|
||
'org_stage': '_widget_1734062123201',
|
||
|
||
# 本月统计
|
||
'g_count_this_month': '_widget_1734062123202',
|
||
'saas_customer_type': '_widget_1734062123203',
|
||
'technician': '_widget_1734062123204',
|
||
'tmall_maintain_service_status_desc': '_widget_1734062123205',
|
||
|
||
# 日期字段（UTC格式）
|
||
'date_fmt_date': '_widget_1749000071375',
|
||
'saas_create_time_date': '_widget_1749000071377',
|
||
'expiry_time_date': '_widget_1749000071389', # 过期日
|
||
'install_create_time_date': '_widget_1749000071384',
|
||
'last_end_date_date': '_widget_1749000071382',
|
||
'renew_date_date': '_widget_1749000071391',
|
||
|
||
# 人员ID字段
|
||
'area_manager_staff_id': '_widget_1748496855779',
|
||
'service_salesmen_staff_id': '_widget_1748496855778',
|
||
'service_impl_principal_staff_id': '_widget_1748496855780',
|
||
'technician_staff_id': '_widget_1751877712235',
|
||
}
|
||
|
||
def _save_to_cache(self, df, cache_key):
|
||
"""
|
||
保存DataFrame到本地缓存
|
||
|
||
参数:
|
||
df: 要保存的DataFrame
|
||
cache_key: 缓存键名（对应CACHE_FILES中的键）
|
||
"""
|
||
if cache_key not in CACHE_FILES:
|
||
logger.warning(f" ⚠ 未知的缓存键: {cache_key}")
|
||
return
|
||
|
||
cache_file = CACHE_FILES[cache_key]
|
||
try:
|
||
df.to_csv(cache_file, index=False, encoding='utf-8-sig')
|
||
logger.info(f" ✓ 已保存到缓存: {cache_file} ({len(df)} 条记录)")
|
||
except Exception as e:
|
||
logger.warning(f" ⚠ 保存缓存失败 ({cache_key}): {e}")
|
||
|
||
def _load_from_cache(self, cache_key):
|
||
"""
|
||
从本地缓存加载DataFrame
|
||
|
||
参数:
|
||
cache_key: 缓存键名（对应CACHE_FILES中的键）
|
||
|
||
返回:
|
||
DataFrame或空DataFrame（如果缓存不存在）
|
||
"""
|
||
if cache_key not in CACHE_FILES:
|
||
logger.warning(f" ⚠ 未知的缓存键: {cache_key}")
|
||
return pd.DataFrame()
|
||
|
||
cache_file = CACHE_FILES[cache_key]
|
||
|
||
if not os.path.exists(cache_file):
|
||
logger.warning(f" ⚠ 缓存文件不存在: {cache_file}")
|
||
logger.warning(f" 请先关闭USE_LOCAL_CACHE运行一次以生成缓存文件")
|
||
return pd.DataFrame()
|
||
|
||
try:
|
||
df = pd.read_csv(cache_file, encoding='utf-8-sig')
|
||
logger.info(f" ✓ 已从缓存加载: {cache_file} ({len(df)} 条记录)")
|
||
return df
|
||
except Exception as e:
|
||
logger.error(f" ✗ 加载缓存失败 ({cache_key}): {e}")
|
||
return pd.DataFrame()
|
||
|
||
def _save_create_data(self, create_data_list):
|
||
"""
|
||
保存新增数据到文件
|
||
|
||
参数:
|
||
create_data_list: 新增数据列表
|
||
"""
|
||
if len(create_data_list) == 0:
|
||
return
|
||
|
||
try:
|
||
# 生成时间戳
|
||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
|
||
# 提取数据到DataFrame
|
||
create_records = []
|
||
for item in create_data_list:
|
||
row_data = item['row_data']
|
||
record = {
|
||
'org_code': item['org_code'],
|
||
'org_name': row_data.get('org_name', ''),
|
||
'group_name': row_data.get('group_name', ''),
|
||
'org_type': row_data.get('org_type', ''),
|
||
'province_name': row_data.get('province_name', ''),
|
||
'city_name': row_data.get('city_name', ''),
|
||
'saas_version': row_data.get('saas_version', ''),
|
||
'active_status_fmt': row_data.get('active_status_fmt', ''),
|
||
}
|
||
create_records.append(record)
|
||
|
||
create_df = pd.DataFrame(create_records)
|
||
|
||
# 使用相对路径保存（支持跨平台）
|
||
file_path = os.path.join(output_dir, f'新增门店_{timestamp}.csv')
|
||
create_df.to_csv(file_path, index=False, encoding='utf-8-sig')
|
||
|
||
logger.info(f" ✓ 新增数据已保存: {file_path} ({len(create_df)} 条)")
|
||
except Exception as e:
|
||
error_task_logger.error(f"保存新增数据失败: {e}", exc_info=True)
|
||
|
||
def _save_update_stats(self, update_data_list):
|
||
"""
|
||
保存更新统计信息到文件
|
||
|
||
参数:
|
||
update_data_list: 更新数据列表
|
||
"""
|
||
if len(update_data_list) == 0:
|
||
return
|
||
|
||
try:
|
||
# 生成时间戳
|
||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
|
||
# 统计每个org_code的更新记录数（去重）
|
||
org_code_counts = {}
|
||
org_code_info = {}
|
||
|
||
for item in update_data_list:
|
||
org_code = item['org_code']
|
||
if org_code not in org_code_counts:
|
||
org_code_counts[org_code] = 0
|
||
row_data = item['row_data']
|
||
org_code_info[org_code] = {
|
||
'org_name': row_data.get('org_name', ''),
|
||
'group_name': row_data.get('group_name', ''),
|
||
'org_type': row_data.get('org_type', ''),
|
||
'province_name': row_data.get('province_name', ''),
|
||
'city_name': row_data.get('city_name', ''),
|
||
'saas_version': row_data.get('saas_version', ''),
|
||
'active_status_fmt': row_data.get('active_status_fmt', ''),
|
||
}
|
||
org_code_counts[org_code] += 1
|
||
|
||
# 构建统计DataFrame
|
||
update_stats = []
|
||
for org_code, count in org_code_counts.items():
|
||
info = org_code_info[org_code]
|
||
stat = {
|
||
'org_code': org_code,
|
||
'org_name': info['org_name'],
|
||
'group_name': info['group_name'],
|
||
'org_type': info['org_type'],
|
||
'province_name': info['province_name'],
|
||
'city_name': info['city_name'],
|
||
'saas_version': info['saas_version'],
|
||
'active_status_fmt': info['active_status_fmt'],
|
||
'update_count': count,
|
||
'note': '同一org_code有多个记录' if count > 1 else ''
|
||
}
|
||
update_stats.append(stat)
|
||
|
||
update_df = pd.DataFrame(update_stats)
|
||
update_df = update_df.sort_values('update_count', ascending=False)
|
||
|
||
# 使用相对路径保存（支持跨平台）
|
||
file_path = os.path.join(output_dir, f'更新统计_{timestamp}.csv')
|
||
update_df.to_csv(file_path, index=False, encoding='utf-8-sig')
|
||
|
||
# 统计汇总
|
||
total_org_codes = len(org_code_counts)
|
||
total_records = len(update_data_list)
|
||
duplicate_org_codes = sum(1 for count in org_code_counts.values() if count > 1)
|
||
|
||
logger.info(f" ✓ 更新统计已保存: {file_path}")
|
||
logger.info(f" - 更新的org_code数: {total_org_codes}")
|
||
logger.info(f" - 更新的记录总数: {total_records}")
|
||
if duplicate_org_codes > 0:
|
||
logger.info(f" - 包含多条记录的org_code: {duplicate_org_codes}")
|
||
except Exception as e:
|
||
error_task_logger.error(f"保存更新统计失败: {e}", exc_info=True)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
updater = UpdateAllNGVDataDaily()
|
||
updater.main()
|