#!/usr/bin/env python3 # -*- coding: utf-8 -*- from fastapi import APIRouter, Request, Depends, Query, HTTPException, status,WebSocket,WebSocketDisconnect from common.security import valid_access_token,valid_websocket_token from fastapi.responses import JSONResponse from sqlalchemy.orm import Session from sqlalchemy.sql import func from common.auth_user import * from sqlalchemy import text from pydantic import BaseModel from common.BigDataCenterAPI import * from database import get_db from typing import List from models import * from utils import * from utils.spatial import * from utils.ry_system_util import * from common.websocketManager import * import json import traceback router = APIRouter() def warehouse_room_id_get_warehouse_room_info(db,id): query = db.query(ResourceProvisionWarehouseRoomInfo) query = query.filter_by(id = id,del_flag = '0') return query.first() def warehouse_id_get_warehouse_info(db,id): query = db.query(ResourceProvisionWarehouseInfo) query = query.filter_by(warehouse_id = id,del_flag = '0') return query.first() @router.post("/create") async def create_pattern( user_id=Depends(valid_access_token), body = Depends(remove_xss_json), db: Session = Depends(get_db) ): try: new_type = ResourceProvisionWarehouseRoomInfo( id = new_guid(), room_name=body['room_name'], warehouse=body['warehouse_id'], room_area=body['room_area'], available_area=body['available_area'], height=body['height'], room_volume=body['room_volume'], available_volume=body['available_volume'], storage_type=body['storage_type'], level=body['level'], storage_type_code=body['storage_type_code'], last_inventory_time=body['last_inventory_time'], create_by = user_id ) db.add(new_type) db.commit() return {"code": 200, "msg": "创建成功", "data": None} except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") @router.put("/update/{id}") async def update_pattern( id :str , user_id=Depends(valid_access_token), body=Depends(remove_xss_json), db: Session = Depends(get_db) ): try: info = warehouse_room_id_get_warehouse_room_info(db,id) if not info: return JSONResponse(status_code=404,content={"code":404,"msg":"warehouse room not found"}) info.room_name = body['room_name'] info.warehouse = body['warehouse_id'] info.room_area = body['room_area'] info.available_area = body['available_area'] info.height = body['height'] info.room_volume = body['room_volume'] info.available_volume = body['available_volume'] info.storage_type = body['storage_type'] info.level = body['level'] info.storage_type_code = body['storage_type_code'] info.last_inventory_time = body['last_inventory_time'] info.update_by = user_id db.commit() return {"code": 200, "msg": "更新成功"} except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") @router.get("/info/{id}") async def get_pattern_info( id: str, db: Session = Depends(get_db) ): try: info = warehouse_room_id_get_warehouse_room_info(db,id) if not info: return JSONResponse(status_code=404,content={"code":404,"msg":"warehouse room not found"}) warehouse_info = warehouse_id_get_warehouse_info(db,info.warehouse) if not warehouse_info: warehouse_name=None else: warehouse_name = warehouse_info.warehouse_name data = { "id":info.id, "warehouse_id": info.warehouse, "warehouse_name": warehouse_name, "room_name": info.room_name, "room_area": info.room_area, "available_area": info.available_area, "height": info.height, "room_volume": info.room_volume, "available_volume": info.available_volume, "storage_type": info.storage_type, "level": info.level, "storage_type_code": info.storage_type_code, "last_inventory_time": info.last_inventory_time, "create_time":info.create_time} return {"code": 200, "msg": "获取成功", "data": data} except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") @router.get("/list") async def get_pattern_list( # name: str = Query(None, description='名称'), page: int = Query(1, gt=0, description='页码'), pageSize: int = Query(10, gt=0, description='每页条目数量'), db: Session = Depends(get_db) ): try: query = db.query(ResourceProvisionWarehouseRoomInfo) query = query.filter_by(del_flag='0') # if name: # query = query.filter(ResourceProvisionWarehouseInfo.material_category_name.like(f'%{name}%')) total_items = query.count() # 排序 query = query.order_by(ResourceProvisionWarehouseRoomInfo.create_time.desc()) # 执行分页查询 lists = query.offset((page - 1) * pageSize).limit(pageSize).all() data = [] for info in lists: warehouse_info = warehouse_id_get_warehouse_info(db, info.warehouse) if not warehouse_info: warehouse_name = None else: warehouse_name = warehouse_info.warehouse_name data.append({ "id":info.id, "warehouse_id": info.warehouse, "warehouse_name": warehouse_name, "room_name": info.room_name, "room_area": info.room_area, "available_area": info.available_area, "height": info.height, "room_volume": info.room_volume, "available_volume": info.available_volume, "storage_type": info.storage_type, "level": info.level, "storage_type_code": info.storage_type_code, "last_inventory_time": info.last_inventory_time, "create_time":info.create_time} ) return {"code": 200, "msg": "查询成功", "data": data, "total": total_items, "page": page, "pageSize": pageSize, "totalPages": (total_items + pageSize - 1) // pageSize } except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") @router.delete("/delete/{id}") async def delete_pattern( id: str, db: Session = Depends(get_db) ): try: # 检查图案是否存在 info = warehouse_room_id_get_warehouse_room_info(db, id) if not info: return JSONResponse(status_code=404, content={"code": 404, "msg": "warehouse room not found"}) info.del_flag='2' db.commit() return {"code": 200, "msg": "删除成功"} except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")