修改excel脚本

app/api/excel.py

@@ -1,13 +1,7 @@
-from fastapi import APIRouter, File, UploadFile, Depends
+from fastapi import APIRouter, File, UploadFile
 from fastapi.responses import JSONResponse, FileResponse
 from fastapi.exceptions import HTTPException
-from sqlalchemy.orm import Session
-from starlette.websockets import WebSocket, WebSocketDisconnect
-from werkzeug.utils import secure_filename
-from app.api import get_current_user_websocket
-from app.models.agent_model import AgentModel, AgentType
-from app.models.base_model import get_db
-from app.models.user_model import UserModel
+from starlette.websockets import WebSocket
 from app.utils.excelmerge.conformity import run_conformity
 import shutil
 import os
@@ -18,6 +12,7 @@ ALLOWED_EXTENSIONS = {'xlsx'}
 EXCEL_FILES_PATH = 'data/output'
 SOURCE_FILES_PATH = 'data/source'
 
+
 def allowed_file(filename):
     return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
 
@@ -77,19 +72,18 @@ async def ws_excel(websocket: WebSocket):
         data = await websocket.receive_text()
         try:
             if data == "\"合并Excel\"":
-                output_file_path = run_conformity()
-                output_file_path = run_conformity()
+                run_excel = run_conformity()
                 files = os.listdir(EXCEL_FILES_PATH)
-                if files:
+                if run_excel:
                     first_file = files[0]
                     file_name = os.path.basename(first_file)
-                    file_url = f"./api/document/download/{first_file}"
+                    download_url = f"./api/document/download/{first_file}"
                     await websocket.send_json({
                         "message": "文档合并成功！",
                         "type": "stream",
                         "files": [{
                             "file_name": file_name,
-                            "file_url": file_url
+                            "file_url": download_url
                         }]
                     })
                     await websocket.send_json({
@@ -97,7 +91,7 @@ async def ws_excel(websocket: WebSocket):
                         "type": "close",
                     })
                 else:
-                    await websocket.send_json({"error": "合并操作未生成文件", "type": "stream", "files": []})
+                    await websocket.send_json({"error": "合并失败", "type": "stream", "files": []})
             elif data == "\"查询合并进度\"":
                 files = os.listdir(EXCEL_FILES_PATH)
                 if not files:
@@ -136,4 +130,4 @@ async def download_file(filename: str):
     except FileNotFoundError:
         raise HTTPException(status_code=404, detail="文件不存在")
     except Exception as e:
-        raise HTTPException(status_code=500, detail="服务器错误")
+        raise HTTPException(status_code=500, detail="服务器错误")

app/utils/excelmerge/conformity.py

@@ -1,73 +1,77 @@
 from openpyxl import load_workbook
-from openpyxl.utils import get_column_letter
 from datetime import datetime
 import os
 
 
 def run_conformity():
-    # 加载模板文件
-    template_path = os.path.join('app', 'utils', 'excelmerge', '国网上海电力整合模版.xlsx')
-    template_excel = load_workbook(template_path)
-    EXCEL_FILES_PATH = os.path.join('data', 'output')
+    try:
+        # 加载模板文件
+        template_path = os.path.join('app', 'utils', 'excelmerge', '国网上海电力整合模版.xlsx')
+        template_excel = load_workbook(template_path)
+        EXCEL_FILES_PATH = os.path.join('data', 'output')
 
-    # 获取工作表
-    sheet1 = template_excel['技术监督工作统计表']
-    sheet2 = template_excel['技术监督告（预）警单统计表']
-    sheet3 = template_excel['投产前技术监督报告统计表']
-    sheet4 = template_excel['技术监督细则完善建议']
-    sheet5 = template_excel['典型经验交流']
+        # 获取工作表
+        sheet1 = template_excel['技术监督工作统计表']
+        sheet2 = template_excel['技术监督告（预）警单统计表']
+        sheet3 = template_excel['投产前技术监督报告统计表']
+        sheet4 = template_excel['技术监督细则完善建议']
+        sheet5 = template_excel['典型经验交流']
 
-    # 获取源文件路径
-    source_folder = os.path.join('data', 'source')
-    source_files = [f for f in os.listdir(source_folder) if f.endswith('.xlsx') and not f.startswith('~$')]
+        # 获取源文件路径
+        source_folder = os.path.join('data', 'source')
+        source_files = [f for f in os.listdir(source_folder) if f.endswith('.xlsx') and not f.startswith('~$')]
 
-    for file in source_files:
-        source_path = os.path.join(source_folder, file)
-        source_excel = load_workbook(source_path)
+        for file in source_files:
+            source_path = os.path.join(source_folder, file)
+            source_excel = load_workbook(source_path)
 
-        # 获取源工作表
-        ssheet1 = source_excel['技术监督工作统计表']
-        ssheet2 = source_excel['技术监督告（预）警单统计表']
-        ssheet3 = source_excel['投产前技术监督报告统计表']
-        ssheet4 = source_excel['技术监督细则完善建议']
-        ssheet5 = source_excel['典型经验交流']
+            # 获取源工作表
+            ssheet1 = source_excel['技术监督工作统计表']
+            ssheet2 = source_excel['技术监督告（预）警单统计表']
+            ssheet3 = source_excel['投产前技术监督报告统计表']
+            ssheet4 = source_excel['技术监督细则完善建议']
+            ssheet5 = source_excel['典型经验交流']
 
-        # 清除无效空白行
-        for ssheet in [ssheet1, ssheet2, ssheet3, ssheet4, ssheet5]:
-            last_row = ssheet.max_row
-            if last_row > 1:
-                for row in range(last_row, 1, -1):
-                    if all(ssheet.cell(row=row, column=col).value is None or ssheet.cell(row=row,
-                                                                                         column=col).value == ''
-                           for col in range(1, ssheet.max_column + 1)):
-                        ssheet.delete_rows(row)
+            # 清除无效空白行
+            for ssheet in [ssheet1, ssheet2, ssheet3, ssheet4, ssheet5]:
+                last_row = ssheet.max_row
+                if last_row > 1:
+                    for row in range(last_row, 1, -1):
+                        if all(ssheet.cell(row=row, column=col).value is None or ssheet.cell(row=row,
+                                                                                             column=col).value == ''
+                               for col in range(1, ssheet.max_column + 1)):
+                            ssheet.delete_rows(row)
 
-        # 复制数据
-        for (ssheet, tsheet, start_point) in [
-            (ssheet1, sheet1, 4),
-            (ssheet2, sheet2, 3),
-            (ssheet3, sheet3, 3),
-            (ssheet4, sheet4, 4),
-            (ssheet5, sheet5, 3),
-        ]:
-            for row in range(start_point, ssheet.max_row + 1):
-                a_cell_value = ssheet.cell(row=row, column=1).value
-                if isinstance(a_cell_value, (int, float)) and any(
-                        ssheet.cell(row=row, column=col).value for col in range(2, ssheet.max_column + 1)):
-                    tsheet.append([ssheet.cell(row=row, column=col).value for col in range(1, ssheet.max_column + 1)])
+            # 复制数据
+            for (ssheet, tsheet, start_point) in [
+                (ssheet1, sheet1, 4),
+                (ssheet2, sheet2, 3),
+                (ssheet3, sheet3, 3),
+                (ssheet4, sheet4, 4),
+                (ssheet5, sheet5, 3),
+            ]:
+                for row in range(start_point, ssheet.max_row + 1):
+                    a_cell_value = ssheet.cell(row=row, column=1).value
+                    if isinstance(a_cell_value, (int, float)) and any(
+                            ssheet.cell(row=row, column=col).value for col in range(2, ssheet.max_column + 1)):
+                        tsheet.append(
+                            [ssheet.cell(row=row, column=col).value for col in range(1, ssheet.max_column + 1)])
 
-        source_excel.close()
+            source_excel.close()
 
-    # 在目标工作表中添加序号
-    for tsheet, start_point in [(sheet1, 4), (sheet2, 3), (sheet3, 3), (sheet4, 4), (sheet5, 3)]:
-        last_row = tsheet.max_row
-        for i in range(start_point, last_row + 1):
-            tsheet.cell(row=i, column=1).value = i - start_point + 1
+        # 在目标工作表中添加序号
+        for tsheet, start_point in [(sheet1, 4), (sheet2, 3), (sheet3, 3), (sheet4, 4), (sheet5, 3)]:
+            last_row = tsheet.max_row
+            for i in range(start_point, last_row + 1):
+                tsheet.cell(row=i, column=1).value = i - start_point + 1
 
-    # 保存文件
-    timestamp = datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
-    output_path = os.path.join(EXCEL_FILES_PATH, f'{timestamp}.xlsx')
-    template_excel.save(output_path)
-    template_excel.close()
+        # 保存文件
+        timestamp = datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
+        output_path = os.path.join(EXCEL_FILES_PATH, f'{timestamp}.xlsx')
+        template_excel.save(output_path)
+        template_excel.close()
 
-    return timestamp
+        return True
+    except Exception as e:
+        print(f"读取数据发生错误: {e}")
+        return False