admin/railway_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

沉降数据导入增加重复数据覆盖处理

Browse Source
This commit is contained in:
lhx
2025-12-29 08:34:05 +08:00
parent e09ebb52e6
commit f2e4f5362e
3 changed files with 78 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/api/comprehensive_data.py
View File

@@ -99,11 +99,17 @@ def batch_import_checkpoints(request: BatchCheckpointDataImportRequest, db: Sess
@router.post("/batch_import_settlement_data", response_model=DataImportResponse)
def batch_import_settlement_data(request: BatchSettlementDataImportRequest, db: Session = Depends(get_db)):
"""批量导入沉降数据"""
"""批量导入沉降数据
duplicate_action: 重复数据处理方式
- skip: 跳过重复数据（默认）
- overwrite: 覆盖重复数据（根据观测点ID和期数ID判断重复）
"""
try:
logger.info(f"Starting batch import settlement data, count: {len(request.data)}")
logger.info(f"Starting batch import settlement data, count: {len(request.data)}, duplicate_action: {request.duplicate_action}")
data_list = request.data
result = settlement_service.batch_import_settlement_data(db, data_list)
duplicate_action = request.duplicate_action or "skip"
result = settlement_service.batch_import_settlement_data(db, data_list, duplicate_action=duplicate_action)
logger.info(f"Batch import settlement data completed: {result['message']}")
return DataImportResponse(

1
app/schemas/comprehensive_data.py
View File

@@ -207,6 +207,7 @@ class BatchCheckpointDataImportRequest(BaseModel):
class BatchSettlementDataImportRequest(BaseModel):
data: List[Dict[str, Any]]
duplicate_action: Optional[str] = "skip" # 重复数据处理方式：skip=跳过（默认），overwrite=覆盖
class Config:
extra = "allow" # 允许额外字段

73
app/services/settlement_data.py
View File

@@ -356,11 +356,13 @@ class SettlementDataService(BaseService[SettlementData]):
checkpoint = db.query(Checkpoint).filter(Checkpoint.point_id == point_id).first()
return checkpoint is not None
def batch_import_settlement_data(self, db: Session, data: List) -> Dict[str, Any]:
def batch_import_settlement_data(self, db: Session, data: List, duplicate_action: str = "skip") -> Dict[str, Any]:
"""
批量导入沉降数据 - 性能优化版
使用批量查询和批量操作，大幅提升导入速度
1.根据观测点ID和期数ID判断是否重复，修复记录，跳过插入操作
1.根据观测点ID和期数ID判断是否重复
- duplicate_action="skip": 跳过重复数据（默认）
- duplicate_action="overwrite": 覆盖重复数据（更新现有记录）
2.判断观测点数据是否存在，不存在则记录，跳过插入操作
支持事务回滚，失败时重试一次
"""
@@ -371,6 +373,7 @@ class SettlementDataService(BaseService[SettlementData]):
success_count = 0
failed_count = 0
failed_items = []
updated_count = 0 # 新增：记录更新数量
if total_count == 0:
return {
@@ -388,6 +391,7 @@ class SettlementDataService(BaseService[SettlementData]):
success_count = 0
failed_count = 0
failed_items = []
updated_count = 0
# ===== 性能优化1：批量查询观测点数据（IN查询） =====
# 统一转换为字符串处理（数据库point_id字段是VARCHAR类型）
@@ -435,8 +439,9 @@ class SettlementDataService(BaseService[SettlementData]):
}
logger.info(f"Found {len(existing_data)} existing settlement records")
# ===== 性能优化3：批量处理插入和跳过 =====
# ===== 性能优化3：批量处理插入、更新和跳过 =====
to_insert = []
to_update = [] # 新增：需要更新的数据
for item_data in valid_items:
point_id = str(item_data.get('point_id')) # 统一转换为字符串
@@ -446,7 +451,11 @@ class SettlementDataService(BaseService[SettlementData]):
key = f"{point_id}_{nyid}"
if key in existing_map:
# 数据已存在，跳过
if duplicate_action == "overwrite":
# 覆盖模式：记录需要更新的数据
to_update.append((existing_map[key], item_data))
else:
# 跳过模式：数据已存在，跳过
logger.info(f"Continue settlement data: {point_id}-{nyid}")
failed_count += 1
failed_items.append({
@@ -457,6 +466,38 @@ class SettlementDataService(BaseService[SettlementData]):
# 记录需要插入的数据
to_insert.append(item_data)
# ===== 执行批量更新（覆盖模式） =====
if to_update:
logger.info(f"Updating {len(to_update)} existing records (overwrite mode)")
for existing_record, item_data in to_update:
try:
# 更新现有记录的字段
existing_record.CVALUE = item_data.get('CVALUE')
existing_record.MAVALUE = item_data.get('MAVALUE')
existing_record.MTIME_W = item_data.get('MTIME_W')
existing_record.PRELOADH = item_data.get('PRELOADH')
existing_record.PSTATE = item_data.get('PSTATE')
existing_record.REMARK = item_data.get('REMARK')
existing_record.WORKINFO = item_data.get('WORKINFO')
existing_record.createdate = item_data.get('createdate')
existing_record.day = item_data.get('day')
existing_record.day_jg = item_data.get('day_jg')
existing_record.isgzjdxz = item_data.get('isgzjdxz')
existing_record.mavalue_bc = item_data.get('mavalue_bc')
existing_record.mavalue_lj = item_data.get('mavalue_lj')
existing_record.sjName = item_data.get('sjName')
existing_record.useflag = item_data.get('useflag')
existing_record.workinfoname = item_data.get('workinfoname')
existing_record.upd_remark = item_data.get('upd_remark')
updated_count += 1
except Exception as e:
failed_count += 1
failed_items.append({
'data': item_data,
'error': f'更新失败: {str(e)}'
})
logger.error(f"Failed to update record: {str(e)}")
# ===== 执行批量插入 =====
if to_insert:
logger.info(f"Inserting {len(to_insert)} new records")
@@ -504,22 +545,22 @@ class SettlementDataService(BaseService[SettlementData]):
logger.error(f"Failed to insert batch: {str(e)}")
raise e
# 如果有插入失败记录（不是跳过记录），不提交事务
# 跳过记录不应该影响事务，只插入失败的记录才需要回滚
insert_failed_items = [item for item in failed_items if '插入失败' in item.get('error', '')]
if insert_failed_items:
# 如果有插入/更新失败记录（不是跳过记录），不提交事务
operation_failed_items = [item for item in failed_items if '插入失败' in item.get('error', '') or '更新失败' in item.get('error', '')]
if operation_failed_items:
db.rollback()
return {
'success': False,
'message': f'批量导入失败: {len(insert_failed_items)}条记录插入失败',
'message': f'批量导入失败: {len(operation_failed_items)}条记录操作失败',
'total_count': total_count,
'success_count': success_count,
'failed_count': failed_count,
'updated_count': updated_count,
'failed_items': failed_items
}
db.commit()
logger.info(f"Batch import settlement data completed. Success: {success_count}, Failed: {failed_count}")
logger.info(f"Batch import settlement data completed. Success: {success_count}, Updated: {updated_count}, Failed: {failed_count}")
break
except Exception as e:
@@ -533,14 +574,24 @@ class SettlementDataService(BaseService[SettlementData]):
'total_count': total_count,
'success_count': 0,
'failed_count': total_count,
'updated_count': 0,
'failed_items': failed_items
}
# 构建成功消息
if duplicate_action == "overwrite" and updated_count > 0:
message = f'批量导入完成，新增{success_count}条，更新{updated_count}条'
elif failed_count == 0:
message = '批量导入完成'
else:
message = '部分导入失败'
return {
'success': True,
'message': '批量导入完成' if failed_count == 0 else f'部分导入失败',
'message': message,
'total_count': total_count,
'success_count': success_count,
'updated_count': updated_count,
'failed_count': failed_count,
'failed_items': failed_items
}