非标业绩提报增加字段
This commit is contained in:
@@ -112,7 +112,7 @@ CONCURRENT_WORKERS = 8
|
||||
USE_BATCH_CREATE = True # True=批量创建（快），False=逐条创建（慢）
|
||||
|
||||
# 批量创建大小（每批次创建的记录数）
|
||||
BATCH_CREATE_SIZE = 100
|
||||
BATCH_CREATE_SIZE = 90
|
||||
|
||||
# ====================================================
|
||||
|
||||
@@ -607,7 +607,8 @@ class UpdateAllNGVDataDaily:
|
||||
update_data_list.append({
|
||||
'org_code': org_code,
|
||||
'data_id': str(data_id),
|
||||
'data_dict': data_dict.copy()
|
||||
'data_dict': data_dict.copy(),
|
||||
'row_data': row # 保存原始数据用于输出
|
||||
})
|
||||
else:
|
||||
# 创建操作：必须包含门店编码字段
|
||||
@@ -617,7 +618,8 @@ class UpdateAllNGVDataDaily:
|
||||
|
||||
create_data_list.append({
|
||||
'org_code': idx,
|
||||
'data_dict': data_dict
|
||||
'data_dict': data_dict,
|
||||
'row_data': row # 保存原始数据用于输出
|
||||
})
|
||||
|
||||
logger.info(f" - 需要更新: {len(update_data_list)} 条")
|
||||
@@ -630,6 +632,8 @@ class UpdateAllNGVDataDaily:
|
||||
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
|
||||
@@ -638,6 +642,8 @@ class UpdateAllNGVDataDaily:
|
||||
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} 条")
|
||||
|
||||
@@ -1052,6 +1058,118 @@ class UpdateAllNGVDataDaily:
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user