增加设备分析

zhaoqingang

2025-02-25 383c21560fcb7012cd2e5b15d559e46d038c87b0

 app/api/excel.py

@@ -1,105 +1,178 @@
from fastapi import APIRouter, File, UploadFile
import random
import string

from fastapi import APIRouter, File, UploadFile, Form, BackgroundTasks, Depends, Request, WebSocket
from fastapi.responses import JSONResponse, FileResponse
from fastapi.exceptions import HTTPException
from werkzeug.utils import secure_filename
from sqlalchemy.orm import Session
# 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.service.session import SessionService
from app.utils.excelmerge.conformity import run_conformity
from pathlib import Path
import subprocess
import shutil
import os


router = APIRouter()

ALLOWED_EXTENSIONS = {'xlsx'}
EXCEL_FILES_PATH = 'data/output'
SOURCE_FILES_PATH = 'data/source'
output_path_value = None


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)


@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)

    create_dir_if_not_exists(SOURCE_FILES_PATH)

    # 清空SOURCE_FILES_PATH目录
    for filename in os.listdir(SOURCE_FILES_PATH):
        file_path = os.path.join(SOURCE_FILES_PATH, filename)
def clear_directory(path: str) -> dict:
    for filename in os.listdir(path):
        file_path = os.path.join(path, filename)
        try:
            if os.path.isfile(file_path) or os.path.islink(file_path):
                os.unlink(file_path)
            elif os.path.isdir(file_path):
                shutil.rmtree(file_path)
        except Exception as e:
            return JSONResponse(content={"error": "文件处理出错"}, status_code=500)
            return {"error": "清空出错"}
    return {"message": "目录已清空"}


def user_file_path(userid: str, path: str) -> str:
    return os.path.join(path, userid)


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, message:str, upload_filenames: list):
    db_id = generate_db_id()
    session = SessionService(db).create_session(
        db_id,
        message,
        "basic_excel_merge",
        AgentType.BASIC,
        int(user_id),
        {"role": "user", "content": message, "upload_filenames": upload_filenames}
    )
    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):
            filename = secure_filename(file.filename)
            save_path = os.path.join(SOURCE_FILES_PATH, 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={"message": "文件上传成功", "paths": save_path_list}, status_code=201)
            return Response(code=400, msg="不允许的文件类型", data={})
    return Response(code=200, msg="上传成功", data={})


@router.post('/excel/conformity')
async def run_conformity_api():
    global output_path_value  # 声明全局变量
    try:
        create_dir_if_not_exists(EXCEL_FILES_PATH)
# ws://localhost:9201/api/document/ws/excel
@router.websocket("/ws/excel")
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)

        # 清空EXCEL_FILES_PATH目录
        for filename in os.listdir(EXCEL_FILES_PATH):
            file_path = os.path.join(EXCEL_FILES_PATH, filename)
            try:
                if os.path.isfile(file_path) or os.path.islink(file_path):
                    os.unlink(file_path)
                elif os.path.isdir(file_path):
                    shutil.rmtree(file_path)
            except Exception as e:
                return JSONResponse(content={"error": "文件处理出错"}, status_code=500)
    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()git
        receive_message = await websocket.receive_json()
        try:
            if receive_message.get("message") == "合并Excel":
                upload_filenames = receive_message.get('upload_filenames', [])
                merge_file = run_conformity(user_source, user_excel)
                if merge_file is not None:

        # 运行方法
        output_path = run_conformity()
        output_path_value = output_path
        return JSONResponse(content={"message": "conformity.py 运行成功", "output_path": str(output_path)}, status_code=200)
    except Exception as e:
        return JSONResponse(content={"error": str(e)}, status_code=500)
                    await websocket.send_json({
                        "type": "stream",
                        "files": [
                            {
                                "file_name": "Excel",
                                "file_url": f"./api/document/download/{merge_file}.xlsx?file_type=excel",
                            }
                        ]
                    })
                    await websocket.send_json({
                        "message": "合并成功",
                        "type": "close",
                    })
                    # 创建会话记录
                    session = db_create_session(db, user_id, receive_message.get("message"), upload_filenames)
                    # 更新会话记录
                    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": []})
                    await websocket.close()
            else:
                print(f"Received data: {receive_message.get('message')}")
                await websocket.send_json({"error": "未知指令", "data": str(receive_message.get('message'))})
                await websocket.close()
        except Exception as e:
            await websocket.send_json({"error": str(e)})
            await websocket.close()


@router.get('/excel/file/status')
async def get_file_status():
    try:
        return JSONResponse(content={"output_path": str(output_path_value)}, status_code=200)
    except Exception as e:
        return JSONResponse(content={"error": str(e)}, status_code=500)
@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',
    )
    # def delete_file():
    #     try:
    #         os.unlink(file_path)
    #     except OSError as e:
    #         print(f"Deleting file error")

@router.get('/excel/download_excel')
async def download_excel():
    try:
        files = os.listdir(EXCEL_FILES_PATH)
        first_file = files[0]
        return FileResponse(os.path.join(EXCEL_FILES_PATH, first_file), filename=first_file,
                            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="服务器错误")
    # 待下载完成后删除生成的文件
    # background_tasks.add_task(delete_file)
    # return FileResponse(path=file_path, filename=file_name,
    #                    media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")