From f2e4f5362eaf6cec6497ec9f404628e0e90dbe31 Mon Sep 17 00:00:00 2001 From: lhx Date: Mon, 29 Dec 2025 08:34:05 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=B2=89=E9=99=8D=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=AF=BC=E5=85=A5=E5=A2=9E=E5=8A=A0=E9=87=8D=E5=A4=8D=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E8=A6=86=E7=9B=96=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/comprehensive_data.py | 12 +++-- app/schemas/comprehensive_data.py | 1 + app/services/settlement_data.py | 85 ++++++++++++++++++++++++------- 3 files changed, 78 insertions(+), 20 deletions(-) diff --git a/app/api/comprehensive_data.py b/app/api/comprehensive_data.py index 568dbf2..020a2d8 100644 --- a/app/api/comprehensive_data.py +++ b/app/api/comprehensive_data.py @@ -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( diff --git a/app/schemas/comprehensive_data.py b/app/schemas/comprehensive_data.py index b3160c6..74901d4 100644 --- a/app/schemas/comprehensive_data.py +++ b/app/schemas/comprehensive_data.py @@ -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" # 允许额外字段 diff --git a/app/services/settlement_data.py b/app/services/settlement_data.py index 49ba10d..21b912b 100644 --- a/app/services/settlement_data.py +++ b/app/services/settlement_data.py @@ -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,17 +451,53 @@ class SettlementDataService(BaseService[SettlementData]): key = f"{point_id}_{nyid}" if key in existing_map: - # 数据已存在，跳过 - logger.info(f"Continue settlement data: {point_id}-{nyid}") - failed_count += 1 - failed_items.append({ - 'data': item_data, - 'error': '数据已存在，跳过插入操作' - }) + 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({ + 'data': item_data, + 'error': '数据已存在，跳过插入操作' + }) else: # 记录需要插入的数据 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 } From a6f809e47913004cd34d56cca15d5e814971a276 Mon Sep 17 00:00:00 2001 From: lhx Date: Mon, 29 Dec 2025 08:50:13 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E6=98=BE=E5=BC=8F=E5=9C=B0=E5=B0=86?= =?UTF-8?q?=E6=9B=B4=E6=96=B0=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/settlement_data.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/app/services/settlement_data.py b/app/services/settlement_data.py index 21b912b..b880a6c 100644 --- a/app/services/settlement_data.py +++ b/app/services/settlement_data.py @@ -497,6 +497,9 @@ class SettlementDataService(BaseService[SettlementData]): 'error': f'更新失败: {str(e)}' }) logger.error(f"Failed to update record: {str(e)}") + # 刷新更新到数据库 + db.flush() + logger.info(f"Updated {updated_count} records successfully") # ===== 执行批量插入 ===== if to_insert: