139 lines
6.0 KiB
Python
139 lines
6.0 KiB
Python
from sqlalchemy.orm import Session
|
||
from typing import List, Optional, Dict, Any
|
||
from ..models.level_data import LevelData
|
||
from .base import BaseService
|
||
from ..models.settlement_data import SettlementData
|
||
|
||
class LevelDataService(BaseService[LevelData]):
|
||
def __init__(self):
|
||
super().__init__(LevelData)
|
||
|
||
def get_by_nyid(self, db: Session, nyid: str) -> List[LevelData]:
|
||
"""根据期数ID获取水准数据"""
|
||
return self.get_by_field(db, "NYID", nyid)
|
||
|
||
def get_by_linecode(self, db: Session, linecode: str) -> List[LevelData]:
|
||
"""根据水准线路编码获取水准数据"""
|
||
return self.get_by_field(db, "linecode", linecode)
|
||
|
||
def search_level_data(self, db: Session,
|
||
id: Optional[str] = None,
|
||
linecode: Optional[str] = None,
|
||
nyid: Optional[str] = None,
|
||
benchmarkids: Optional[str] = None) -> List[LevelData]:
|
||
"""根据多个条件搜索水准数据"""
|
||
conditions = {}
|
||
if linecode is not None:
|
||
conditions["linecode"] = linecode
|
||
if nyid is not None:
|
||
conditions["NYID"] = nyid
|
||
if benchmarkids is not None:
|
||
conditions["benchmarkids"] = benchmarkids
|
||
if id is not None:
|
||
conditions["id"] = id
|
||
|
||
return self.search_by_conditions(db, conditions)
|
||
|
||
|
||
def get_by_nyid_and_linecode(self, db: Session, nyid: str, linecode: str = None) -> Optional[LevelData]:
|
||
"""根据期数ID和线路编码获取水准数据"""
|
||
return db.query(LevelData).filter(
|
||
LevelData.NYID == nyid if nyid else True,
|
||
LevelData.linecode == linecode if linecode else True
|
||
).first()
|
||
|
||
def _check_settlement_exists(self, db: Session, nyid: str) -> bool:
|
||
"""检查期数id沉降数据是否存在"""
|
||
settlement = db.query(SettlementData).filter(SettlementData.NYID == nyid).first()
|
||
return settlement is not None
|
||
|
||
def batch_import_level_data(self, db: Session, data: List) -> Dict[str, Any]:
|
||
"""
|
||
批量导入水准数据，根据期数ID和线路编码判断是否重复，重复数据改为更新操作
|
||
支持事务回滚，失败时重试一次
|
||
"""
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
|
||
total_count = len(data)
|
||
success_count = 0
|
||
failed_count = 0
|
||
failed_items = []
|
||
|
||
for attempt in range(2): # 最多重试1次
|
||
try:
|
||
db.begin()
|
||
success_count = 0
|
||
failed_count = 0
|
||
failed_items = []
|
||
|
||
for item_data in data:
|
||
try:
|
||
|
||
# 判断期数id沉降数据是否存在
|
||
settlement = self._check_settlement_exists(db, item_data.get('NYID'))
|
||
if not settlement:
|
||
logger.error(f"Settlement {item_data.get('NYID')} not found")
|
||
raise Exception(f"Settlement {item_data.get('NYID')} not found")
|
||
|
||
level_data = self.get_by_nyid_and_linecode(
|
||
db,
|
||
# item_data.get('linecode'),
|
||
nyid=item_data.get('NYID')
|
||
)
|
||
if level_data:
|
||
# 更新操作
|
||
level_data.benchmarkids = item_data.get('benchmarkids')
|
||
level_data.wsphigh = item_data.get('wsphigh')
|
||
level_data.mtype = item_data.get('mtype')
|
||
level_data.createDate = item_data.get('createDate')
|
||
logger.info(f"Updated level data: {item_data.get('linecode')}-{item_data.get('NYID')}")
|
||
else:
|
||
# 新增操作
|
||
level_data = LevelData(
|
||
linecode=item_data.get('linecode'),
|
||
benchmarkids=item_data.get('benchmarkids'),
|
||
wsphigh=item_data.get('wsphigh'),
|
||
mtype=item_data.get('mtype'),
|
||
NYID=item_data.get('NYID'),
|
||
createDate=item_data.get('createDate')
|
||
)
|
||
db.add(level_data)
|
||
logger.info(f"Created level data: {item_data.get('linecode')}-{item_data.get('NYID')}")
|
||
|
||
success_count += 1
|
||
except Exception as e:
|
||
failed_count += 1
|
||
failed_items.append({
|
||
'data': item_data,
|
||
'error': str(e)
|
||
})
|
||
logger.error(f"Failed to process level data {item_data.get('linecode')}-{item_data.get('NYID')}: {str(e)}")
|
||
raise e
|
||
|
||
db.commit()
|
||
logger.info(f"Batch import level data completed. Success: {success_count}, Failed: {failed_count}")
|
||
break
|
||
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.warning(f"Batch import attempt {attempt + 1} failed: {str(e)}")
|
||
if attempt == 1: # 最后一次重试失败
|
||
logger.error("Batch import level data failed after retries")
|
||
return {
|
||
'success': False,
|
||
'message': f'批量导入失败: {str(e)}',
|
||
'total_count': total_count,
|
||
'success_count': 0,
|
||
'failed_count': total_count,
|
||
'failed_items': failed_items
|
||
}
|
||
|
||
return {
|
||
'success': failed_count == 0,
|
||
'message': '批量导入完成' if failed_count == 0 else f'部分导入失败',
|
||
'total_count': total_count,
|
||
'success_count': success_count,
|
||
'failed_count': failed_count,
|
||
'failed_items': failed_items
|
||
} |