报表合并历史记录功能

xuyonghao

2025-02-08 72a8a0a1ad6b79b8e9fb2facef121f9b5d584666

报表合并历史记录功能

 app/api/excel.py

@@ -1,13 +1,15 @@
from fastapi import APIRouter, File, UploadFile, Depends
import random
import string

from fastapi import APIRouter, File, UploadFile, Form, BackgroundTasks, Depends, Request
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 starlette.websockets import WebSocket

from app.api import get_current_user, get_current_user_websocket, Response
from app.models import UserModel, AgentType
from app.models.base_model import get_db
from app.models.user_model import UserModel
from app.service.session import SessionService
from app.utils.excelmerge.conformity import run_conformity
import shutil
import os
@@ -18,17 +20,17 @@
EXCEL_FILES_PATH = 'data/output'
SOURCE_FILES_PATH = 'data/source'

def allowed_file(filename):

def allowed_file(filename: str) -> bool:
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS


def create_dir_if_not_exists(path):
def create_dir_if_not_exists(path: str):
    if not os.path.exists(path):
        os.makedirs(path)


# 清理函数
def clear_directory(path):
def clear_directory(path: str) -> dict:
    for filename in os.listdir(path):
        file_path = os.path.join(path, filename)
        try:
@@ -41,99 +43,124 @@
    return {"message": "目录已清空"}


@router.post('/excel/upload')
async def upload_file(files: list[UploadFile] = File(...)):
    if not any(file.filename for file in files):
        return JSONResponse(content={"error": "没有文件部分"}, status_code=400)
def user_file_path(userid: str, path: str) -> str:
    return os.path.join(path, userid)

    create_dir_if_not_exists(SOURCE_FILES_PATH)
    create_dir_if_not_exists(EXCEL_FILES_PATH)
    clear_directory(SOURCE_FILES_PATH)
    clear_directory(EXCEL_FILES_PATH)

def generate_db_id(prefix: str = "me") -> str:
    random_part = ''.join(random.choices(string.ascii_letters + string.digits, k=13))
    return prefix + random_part


def db_create_session(db: Session, user_id: str):
    db_id = generate_db_id()
    session = SessionService(db).create_session(
        db_id,
        "合并Excel",
        "basic_excel_merge",
        AgentType.BASIC,
        int(user_id)
    )
    return session


@router.post('/excel/upload', response_model=Response)
async def upload_file(files: list[UploadFile] = File(...), current_user: UserModel = Depends(get_current_user)):
    user_id = str(current_user.id)
    if not any(file.filename for file in files):
        return Response(code=400, msg="没有文件部分", data={})
    if not user_id:
        return Response(code=400, msg="缺少参数user_id", data={})
    user_source = user_file_path(user_id, SOURCE_FILES_PATH)
    user_excel = EXCEL_FILES_PATH

    create_dir_if_not_exists(user_source)
    create_dir_if_not_exists(user_excel)
    clear_directory(user_source)

    save_path_list = []
    for file in files:
        if file.filename == '':
            return JSONResponse(content={"error": "没有选择文件"}, status_code=400)
        if file and allowed_file(file.filename):
            save_path = os.path.join(SOURCE_FILES_PATH, file.filename)
            save_path = os.path.join(user_source, file.filename)
            with open(save_path, 'wb') as buffer:
                shutil.copyfileobj(file.file, buffer)
            save_path_list.append(save_path)
        else:
            return JSONResponse(content={"error": "不允许的文件类型"}, status_code=400)
    return JSONResponse(content={"code": 200, "msg": "", "data": {}}, status_code=200)
            return Response(code=400, msg="不允许的文件类型", data={})
    return Response(code=200, msg="上传成功", data={})


# ws://localhost:9201/api/document/ws/excel
@router.websocket("/ws/excel")
async def ws_excel(websocket: WebSocket):
async def ws_excel(websocket: WebSocket,
                   current_user: UserModel = Depends(get_current_user_websocket),
                   db: Session = Depends(get_db)):
    await websocket.accept()
    user_id = str(current_user.id)

    create_dir_if_not_exists(SOURCE_FILES_PATH)
    create_dir_if_not_exists(EXCEL_FILES_PATH)
    user_source = user_file_path(user_id, SOURCE_FILES_PATH)
    user_excel = EXCEL_FILES_PATH
    create_dir_if_not_exists(user_source)
    create_dir_if_not_exists(user_excel)

    while True:
        data = await websocket.receive_text()
        try:
            if data == "\"合并Excel\"":
                output_file_path = run_conformity()
                output_file_path = run_conformity()
                files = os.listdir(EXCEL_FILES_PATH)
                if files:
                    first_file = files[0]
                    file_name = os.path.basename(first_file)
                    file_url = f"./api/document/download/{first_file}"
                merge_file = run_conformity(user_source, user_excel)
                if merge_file is not None:

                    await websocket.send_json({
                        "message": "文档合并成功！",
                        "type": "stream",
                        "files": [{
                            "file_name": file_name,
                            "file_url": file_url
                        }]
                        "files": [
                            {
                                "file_name": "Excel",
                                "file_url": f"./api/document/download/{merge_file}.xlsx?file_type=excel",
                            }
                        ]
                    })
                    await websocket.send_json({
                        "message": "文档合并成功！",
                        "message": "合并成功",
                        "type": "close",
                    })
                    # 创建会话记录
                    session = db_create_session(db, user_id)
                    # 更新会话记录
                    if session:
                        session_id = session.id
                        new_message = {
                            "role": "assistant",
                            "content": {
                                "message": "\u5408\u5e76\u6210\u529f",
                                "type": "message",
                                "file_name": "Excel",
                                "file_url": f"/api/document/download/{merge_file}.xlsx?file_type=excel"
                            }
                        }
                        session_service = SessionService(db)
                        session_service.update_session(session_id, message=new_message)
                else:
                    await websocket.send_json({"error": "合并操作未生成文件", "type": "stream", "files": []})
            elif data == "\"查询合并进度\"":
                files = os.listdir(EXCEL_FILES_PATH)
                if not files:
                    await websocket.send_json({"step_message": "正在合并中", "type": "stream", "files": []})
                else:
                    await websocket.send_json({"step_message": "文档合并成功！", "type": "stream", "files": []})
            elif data == "\"获取文件\"":
                files = os.listdir(EXCEL_FILES_PATH)
                if not files:
                    await websocket.send_json({"error": "目录下没有生成的文件", "type": "stream", "files": []})
                else:
                    first_file = files[0]
                    file_name = os.path.basename(first_file)
                    file_url = f"./api/document/download/{first_file}"
                    await websocket.send_json({
                        "step_message": "文档合并成功！",
                        "type": "stream",
                        "files": [{
                            "file_name": file_name,
                            "file_url": file_url
                        }]
                    })
                    await websocket.send_json({"error": "合并失败", "type": "stream", "files": []})
                    await websocket.close()
            else:
                print(f"Received data: {data}")
                await websocket.send_json({"error": "未知指令", "data": str(data)})
                await websocket.close()
        except Exception as e:
            await websocket.send_json({"error": str(e)})
            await websocket.close()


@router.get("/download/{filename}")
async def download_file(filename: str):
    try:
        return FileResponse(os.path.join(EXCEL_FILES_PATH, filename), filename=filename,
                            media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail="文件不存在")
    except Exception as e:
        raise HTTPException(status_code=500, detail="服务器错误")
@router.get("/download/{file_full_name}")
async def download_file(file_full_name: str):
    file_name = os.path.basename(file_full_name)
    user_excel = EXCEL_FILES_PATH
    file_path = os.path.join(user_excel, file_full_name)

    if not os.path.exists(file_path):
        return JSONResponse(content={"error": "文件不存在"}, status_code=404)
    return FileResponse(
        path=file_path,
        filename="Excel.xlsx",
        media_type='application/octet-stream',
    )

 app/utils/excelmerge/conformity.py

@@ -1,73 +1,82 @@
from openpyxl import load_workbook
from openpyxl.utils import get_column_letter
from datetime import datetime
import os
import random
import shutil
import string

from datetime import datetime
from openpyxl import load_workbook


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')
def clear_blank_rows(sheet):
    last_row = sheet.max_row
    for row in range(last_row, 1, -1):
        if all(cell.value is None or cell.value == '' for cell in sheet[row]):
            sheet.delete_rows(row)

    # 获取工作表
    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('~$')]
def copy_data(source_sheet, target_sheet, start_row):
    for row in range(start_row, source_sheet.max_row + 1):
        a_cell_value = source_sheet.cell(row=row, column=1).value
        if isinstance(a_cell_value, (int, float)) and any(
                source_sheet.cell(row=row, column=col).value for col in range(4, source_sheet.max_column + 1)):
            target_sheet.append(
                [source_sheet.cell(row=row, column=col).value for col in range(1, source_sheet.max_column + 1)])

    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['典型经验交流']
def run_conformity(file_path, print_path):
    try:
        # 加载模板文件
        template_path = os.path.join('app', 'utils', 'excelmerge', '国网上海电力整合模版.xlsx')
        template_excel = load_workbook(template_path)
        template_sheets = {sheet.title: sheet for sheet in template_excel}
        source_files = [f for f in os.listdir(file_path) if f.endswith('.xlsx') and not f.startswith('~$')]

        # 清除无效空白行
        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 file in source_files:
            source_path = os.path.join(file_path, file)
            source_excel = load_workbook(source_path)

        # 复制数据
        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_sheets = {sheet.title: sheet for sheet in source_excel}

        source_excel.close()
            for name in template_sheets:
                if name in source_sheets:
                    clear_blank_rows(source_sheets[name])

    # 在目标工作表中添加序号
    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 name, start_row in [('技术监督工作统计表', 4), ('技术监督告（预）警单统计表', 3),
                                    ('投产前技术监督报告统计表', 3), ('技术监督细则完善建议', 4),
                                    ('典型经验交流', 3)]:
                if name in source_sheets and name in template_sheets:
                    copy_data(source_sheets[name], template_sheets[name], start_row)

    # 保存文件
    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()
            source_excel.close()

    return timestamp
        for name, start_row in [('技术监督工作统计表', 4), ('技术监督告（预）警单统计表', 3),
                                ('投产前技术监督报告统计表', 3), ('技术监督细则完善建议', 4),
                                ('典型经验交流', 3)]:
            if name in template_sheets:
                last_row = template_sheets[name].max_row
                for i in range(start_row, last_row + 1):
                    template_sheets[name].cell(row=i, column=1).value = i - start_row + 1

        timestamp = datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
        random_string = ''.join(random.choice(string.ascii_letters + string.digits) for i in range(5))
        file_name = f'{random_string}_{timestamp}'
        output_path = os.path.join(print_path, f'{file_name}.xlsx')
        template_excel.save(output_path)
        template_excel.close()

        try:
            for filename in os.listdir(file_path):
                file_path_full = os.path.join(file_path, filename)
                if os.path.isfile(file_path_full) or os.path.islink(file_path_full):
                    os.unlink(file_path_full)
                elif os.path.isdir(file_path_full):
                    shutil.rmtree(file_path_full)
            os.rmdir(file_path)
        except Exception as e:
            print(f"删除文件时发生错误: {e}")

        return file_name
    except Exception as e:
        print(f"读取数据发生错误: {e}")
        return None