导出excel接口初版

app/api/export_excel.py

@@ -0,0 +1,99 @@
+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 ..schemas.export_excel import ExportExcelRequest
+from ..services.section_data import SectionDataService
+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()
+
+@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
+        }