from fastapi import APIRouter, Depends, HTTPException, status from fastapi.responses import FileResponse, StreamingResponse from sqlalchemy.orm import Session from typing import Optional, Dict, Any from ..core.database import get_db from ..core.response_code import ResponseCode, ResponseMessage from ..core.exceptions import BusinessException, DataNotFoundException, AccountNotFoundException from ..schemas.export_excel import ExportExcelRequest, ExportSettlementRequest, ExportLevelDataRequest from ..services.section_data import SectionDataService from ..services.export_excel import ExportExcelService import logging import pandas as pd from io import BytesIO from datetime import datetime import os import tempfile router = APIRouter(prefix="/export", tags=["数据导出"]) logger = logging.getLogger(__name__) # 实例化服务类 section_service = SectionDataService() export_excel_service = ExportExcelService() @router.post("/section_data") def export_section_data( request: ExportExcelRequest, db: Session = Depends(get_db) ): """导出断面数据为Excel文件""" try: logger.info(f"导出断面数据，请求参数: section_id={request.section_id}, account_id={request.account_id}") result = section_service.search_sections_with_checkpoints( db, section_id=request.section_id, mileage=request.mileage, work_site=request.work_site, number=request.number, status=request.status, account_id=request.account_id, skip=0, limit=10000 # 设置一个较大的限制值 ) data_list = result.get('data', []) if not data_list: logger.warning("未找到符合条件的断面数据") return { "code": ResponseCode.SUCCESS, "message": "未找到符合条件的数据，无法导出", "data": None } # 转换为DataFrame df = pd.DataFrame(data_list) # 生成文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"断面数据_{timestamp}.xlsx" # 创建临时文件 temp_dir = tempfile.gettempdir() file_path = os.path.join(temp_dir, filename) # 保存Excel文件 with pd.ExcelWriter(file_path, engine='openpyxl') as writer: df.to_excel(writer, index=False, sheet_name='断面数据') # 获取工作表对象以便调整列宽 worksheet = writer.sheets['断面数据'] # 自动调整列宽 for column in worksheet.columns: max_length = 0 column_letter = column[0].column_letter for cell in column: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) # 最大宽度限制为50 worksheet.column_dimensions[column_letter].width = adjusted_width logger.info(f"成功导出{len(data_list)}条断面数据，保存到文件: {file_path}") # 返回文件下载响应 return FileResponse( path=file_path, filename=filename, media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) except Exception as e: logger.error(f"导出断面数据失败: {str(e)}", exc_info=True) return { "code": ResponseCode.EXPORT_FAILED, "message": f"{ResponseMessage.EXPORT_FAILED}: {str(e)}", "data": None } @router.post("/settlement_data") def export_settlement_data( request: ExportSettlementRequest, db: Session = Depends(get_db) ): """导出沉降数据Excel文件（包含断面、观测点、水准数据）""" try: logger.info(f"导出沉降数据，请求参数: project_name={request.project_name}") # 生成文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{request.project_name}_沉降数据_{timestamp}.xlsx" # 创建临时文件 temp_dir = tempfile.gettempdir() file_path = os.path.join(temp_dir, filename) # 调用服务层导出数据到文件 export_excel_service.export_settlement_data_to_file( db, project_name=request.project_name, file_path=file_path ) logger.info(f"成功生成沉降数据Excel文件: {file_path}") # 返回文件下载响应 return FileResponse( path=file_path, filename=filename, media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) except AccountNotFoundException as e: logger.warning(f"账号不存在: {str(e)}") raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "code": e.code, "message": e.message, "data": None } ) except DataNotFoundException as e: logger.warning(f"数据不存在: {str(e)}") raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "code": e.code, "message": e.message, "data": None } ) except BusinessException as e: logger.warning(f"业务异常: {str(e)}") raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "code": e.code, "message": e.message, "data": None } ) except Exception as e: logger.error(f"导出沉降数据失败: {str(e)}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={ "code": ResponseCode.EXPORT_FAILED, "message": f"{ResponseMessage.EXPORT_FAILED}: {str(e)}", "data": None } ) @router.post("/level_data") def export_level_data( request: ExportLevelDataRequest, db: Session = Depends(get_db) ): """导出水准数据Excel文件（以水准数据为主体，包含断面、观测点、沉降、原始数据）""" try: logger.info(f"导出水准数据，请求参数: project_name={request.project_name}") # 生成文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{request.project_name}_水准数据_{timestamp}.xlsx" # 创建临时文件 temp_dir = tempfile.gettempdir() file_path = os.path.join(temp_dir, filename) # 调用服务层导出数据到文件 export_excel_service.export_level_data_to_file( db, project_name=request.project_name, file_path=file_path ) logger.info(f"成功生成水准数据Excel文件: {file_path}") # 返回文件下载响应 return FileResponse( path=file_path, filename=filename, media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' ) except AccountNotFoundException as e: logger.warning(f"账号不存在: {str(e)}") raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "code": e.code, "message": e.message, "data": None } ) except DataNotFoundException as e: logger.warning(f"数据不存在: {str(e)}") raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "code": e.code, "message": e.message, "data": None } ) except BusinessException as e: logger.warning(f"业务异常: {str(e)}") raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "code": e.code, "message": e.message, "data": None } ) except Exception as e: logger.error(f"导出水准数据失败: {str(e)}", exc_info=True) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={ "code": ResponseCode.EXPORT_FAILED, "message": f"{ResponseMessage.EXPORT_FAILED}: {str(e)}", "data": None } )