"""观测点数据服务""" from typing import List, Dict, Tuple from sqlalchemy import text from sqlalchemy.orm import Session from app.core.logging_config import get_logger from app.models.checkpoint import Checkpoint from app.models.section_data import SectionData from app.schemas.checkpoint import CheckpointCreate, CheckpointQuery from app.schemas.common import BatchImportResponse from .table_manager import TableManager logger = get_logger(__name__) class CheckpointService: """观测点数据服务""" @staticmethod def batch_import(db: Session, account_id: int, data: List[CheckpointCreate]) -> BatchImportResponse: """批量导入观测点数据""" table_name = Checkpoint.get_table_name(account_id) if not TableManager.ensure_table_exists(db, "checkpoint", account_id): return BatchImportResponse( success=False, total=len(data), inserted=0, skipped=0, message="创建表失败" ) # 获取已存在的point_id point_ids = [item.point_id for item in data if item.point_id] existing_ids = set() if point_ids: placeholders = ",".join([f":id_{i}" for i in range(len(point_ids))]) params = {f"id_{i}": pid for i, pid in enumerate(point_ids)} result = db.execute( text(f"SELECT point_id FROM {table_name} WHERE point_id IN ({placeholders})"), params ) existing_ids = {row[0] for row in result.fetchall()} to_insert = [] skipped_ids = [] for item in data: if item.point_id in existing_ids: skipped_ids.append(item.point_id) else: to_insert.append(item) existing_ids.add(item.point_id) if to_insert: try: values = [] params = {} for i, item in enumerate(to_insert): values.append(f"(:name_{i}, :burial_date_{i}, :objstate_{i}, :monitoring_type_{i}, " f":period_number_{i}, :first_time_{i}, :manufacturer_{i}, " f":point_code_{i}, :point_id_{i}, :section_id_{i})") params[f"name_{i}"] = item.name params[f"burial_date_{i}"] = item.burial_date params[f"objstate_{i}"] = item.objstate params[f"monitoring_type_{i}"] = item.monitoring_type params[f"period_number_{i}"] = item.period_number params[f"first_time_{i}"] = item.first_time params[f"manufacturer_{i}"] = item.manufacturer params[f"point_code_{i}"] = item.point_code params[f"point_id_{i}"] = item.point_id params[f"section_id_{i}"] = item.section_id sql = f"""INSERT INTO {table_name} (name, burial_date, objstate, monitoring_type, period_number, first_time, manufacturer, point_code, point_id, section_id) VALUES {','.join(values)}""" db.execute(text(sql), params) db.commit() logger.info(f"观测点数据导入成功: account_id={account_id}, 插入={len(to_insert)}, 跳过={len(skipped_ids)}") except Exception as e: db.rollback() logger.error(f"观测点数据导入失败: {e}") return BatchImportResponse( success=False, total=len(data), inserted=0, skipped=len(skipped_ids), skipped_ids=skipped_ids, message=f"插入失败: {str(e)}" ) return BatchImportResponse( success=True, total=len(data), inserted=len(to_insert), skipped=len(skipped_ids), skipped_ids=skipped_ids, message="导入成功" ) @staticmethod def query(db: Session, params: CheckpointQuery) -> Tuple[List[Dict], int]: """查询观测点数据""" table_name = Checkpoint.get_table_name(params.account_id) if not TableManager.ensure_table_exists(db, "checkpoint", params.account_id): return [], 0 conditions = [] query_params = {} if params.section_id: conditions.append("section_id = :section_id") query_params["section_id"] = params.section_id if params.point_id: conditions.append("point_id = :point_id") query_params["point_id"] = params.point_id if params.name: conditions.append("name LIKE :name") query_params["name"] = f"%{params.name}%" where_clause = " AND ".join(conditions) if conditions else "1=1" count_sql = f"SELECT COUNT(*) FROM {table_name} WHERE {where_clause}" total = db.execute(text(count_sql), query_params).scalar() offset = (params.page - 1) * params.page_size query_params["limit"] = params.page_size query_params["offset"] = offset data_sql = f"SELECT * FROM {table_name} WHERE {where_clause} LIMIT :limit OFFSET :offset" result = db.execute(text(data_sql), query_params) items = [dict(row._mapping) for row in result.fetchall()] return items, total @staticmethod def query_by_department(db: Session, account_id: int, department_id: str, page: int = 1, page_size: int = 20) -> Tuple[List[Dict], int]: """根据department_id查询观测点数据（包含断面信息）""" checkpoint_table = Checkpoint.get_table_name(account_id) section_table = SectionData.get_table_name(account_id) if not TableManager.ensure_table_exists(db, "checkpoint", account_id): return [], 0 if not TableManager.ensure_table_exists(db, "section_data", account_id): return [], 0 # 先查询该department下的所有section_id section_sql = f"SELECT section_id, mileage, rock_mass_classification FROM {section_table} WHERE department_id = :department_id" section_result = db.execute(text(section_sql), {"department_id": department_id}) section_map = {row[0]: {"mileage": row[1], "rock_mass_classification": row[2]} for row in section_result.fetchall()} if not section_map: return [], 0 section_ids = list(section_map.keys()) placeholders = ",".join([f":sid_{i}" for i in range(len(section_ids))]) params = {f"sid_{i}": sid for i, sid in enumerate(section_ids)} # 查询总数 count_sql = f"SELECT COUNT(*) FROM {checkpoint_table} WHERE section_id IN ({placeholders})" total = db.execute(text(count_sql), params).scalar() # 分页查询 offset = (page - 1) * page_size params["limit"] = page_size params["offset"] = offset data_sql = f"SELECT * FROM {checkpoint_table} WHERE section_id IN ({placeholders}) LIMIT :limit OFFSET :offset" result = db.execute(text(data_sql), params) items = [] for row in result.fetchall(): item = dict(row._mapping) section_info = section_map.get(item.get("section_id"), {}) item["section_mileage"] = section_info.get("mileage") item["rock_mass_classification"] = section_info.get("rock_mass_classification") items.append(item) return items, total @staticmethod def get_by_point_ids(db: Session, account_id: int, point_ids: List[str]) -> Dict[str, Dict]: """根据point_id批量获取观测点数据""" if not point_ids: return {} table_name = Checkpoint.get_table_name(account_id) if not TableManager.ensure_table_exists(db, "checkpoint", account_id): return {} placeholders = ",".join([f":id_{i}" for i in range(len(point_ids))]) params = {f"id_{i}": pid for i, pid in enumerate(point_ids)} sql = f"SELECT point_id, name, section_id FROM {table_name} WHERE point_id IN ({placeholders})" result = db.execute(text(sql), params) return {row[0]: {"name": row[1], "section_id": row[2]} for row in result.fetchall()}