from sqlalchemy.orm import Session from typing import List, Optional, Dict, Any from ..models.original_data import OriginalData from .base import BaseService from ..models.settlement_data import SettlementData class OriginalDataService(BaseService[OriginalData]): def __init__(self): super().__init__(OriginalData) def get_by_nyid(self, db: Session, nyid: str) -> List[OriginalData]: """根据期数ID获取原始数据""" return self.get_by_field(db, "NYID", nyid) def get_by_bfpcode(self, db: Session, bfpcode: str) -> List[OriginalData]: """根据前(后)视点名称获取原始数据""" return self.get_by_field(db, "bfpcode", bfpcode) def search_original_data(self, db: Session, bfpcode: Optional[str] = None, bffb: Optional[str] = None, nyid: Optional[str] = None, bfpl: Optional[str] = None) -> List[OriginalData]: """根据多个条件搜索原始数据""" conditions = {} if bfpcode is not None: conditions["bfpcode"] = bfpcode if bffb is not None: conditions["bffb"] = bffb if nyid is not None: conditions["NYID"] = nyid if bfpl is not None: conditions["bfpl"] = bfpl return self.search_by_conditions(db, conditions) 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_original_data(self, db: Session, data: List) -> Dict[str, Any]: """ 批量导入原始数据,直接新增,无需检查重复 支持事务回滚,失败时重试一次 """ 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") # 直接新增操作 original_data = OriginalData( bfpcode=item_data.get('bfpcode'), mtime=item_data.get('mtime'), bffb=item_data.get('bffb'), bfpl=item_data.get('bfpl'), bfpvalue=item_data.get('bfpvalue'), times=item_data.get('times'), NYID=item_data.get('NYID'), sort=item_data.get('sort') ) db.add(original_data) logger.info(f"Created original data: {item_data.get('bfpcode')}-{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 original data {item_data.get('bfpcode')}-{item_data.get('NYID')}: {str(e)}") raise e db.commit() logger.info(f"Batch import original 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 original 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 }