#!/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 classification_id_get_classification_info(db,id): query = db.query(TpPatternClassification) query = query.filter_by(classification_id = id,del_flag = '0') return query.first() def template_id_get_classification_list(db,id): query = db.query(TpPatternClassification) query = query.filter_by(template_id = id,del_flag = '0') query = query.order_by(TpPatternClassification.order_num.asc()) return query.all() def template_id_get_template_info(db,id): query = db.query(TpPatternTemplate) query = query.filter_by(template_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: if 'order_num' in body: order_num=body['order_num'] else: info = db.query(func.max(TpPatternClassification.order_num).label('max_value')).filter_by(template_id=body['template_id']).first() if info.max_value: order_num=info.max_value+1 else: order_num=1 if 'visible'in body: visible=body['visible'] else: visible='1' new_classification = TpPatternClassification( classification_id=new_guid(), template_id=body['template_id'], name=body['name'], value=body['value'], order_num = order_num, visible=visible, image=body['image'], icon=body['icon'], size=body['size'], create_by = user_id ) db.add(new_classification) 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_classification( id :str , user_id=Depends(valid_access_token), body=Depends(remove_xss_json), db: Session = Depends(get_db) ): try: update_classification = classification_id_get_classification_info(db,id) if not update_classification: return JSONResponse(status_code=404,content={"code":404,"msg":"classification not found"}) update_classification.template_id = body['template_id'] update_classification.name = body['name'] update_classification.value=body['value'] update_classification.order_num=body['order_num'] update_classification.visible=body['visible'] update_classification.image=body['image'] update_classification.icon=body['icon'] update_classification.size=body['size'] update_classification.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 = classification_id_get_classification_info(db,id) if not info: return JSONResponse(status_code=404,content={"code":404,"msg":"classification not found"}) template_info = template_id_get_template_info(db,info.template_id) data = {"classification_id": info.classification_id, "template_id": info.template_id, "template_name":template_info.name, "name": info.name, "value": info.value, "order_num": info.order_num, "visible": info.visible, "image": info.image, "icon": info.icon, "size": info.size, "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(None, gt=0, description='每页条目数量'), db: Session = Depends(get_db) ): try: query = db.query(TpPatternClassification) query = query.filter_by(del_flag='0') if name: query = query.filter(TpPatternClassification.name.like(f'%{name}%')) total_items = query.count() # 排序 if pageSize is None: pageSize=total_items query = query.order_by(TpPatternClassification.order_num.asc()) # 执行分页查询 lists = query.offset((page - 1) * pageSize).limit(pageSize).all() data = [ ] for info in lists: template_info = template_id_get_template_info(db, info.template_id) data.append({"classification_id": info.classification_id, "template_id": info.template_id, "template_name":template_info.name, "name": info.name, "value": info.value, "order_num": info.order_num, "visible": info.visible, "image": info.image, "icon": info.icon, "size": info.size, "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 = classification_id_get_classification_info(db, id) if not info: return JSONResponse(status_code=404, content={"code": 404, "msg": "classification 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)}") @router.put('/changeVisible') async def change_classification_visible( db: Session = Depends(get_db), body=Depends(remove_xss_json), user_id=Depends(valid_access_token) ): try: classification_id = body['classification_id'] visible = body['visible'] info = classification_id_get_classification_info(db, classification_id) if not info: return JSONResponse(status_code=404, content={"code": 404, "msg": "classification not found"}) info.visible= visible info.update_by=user_id db.commit() return { "code": 200, "msg": "操作成功" } except Exception as e: # 处理异常 traceback.print_exc() raise HTTPException(status_code=500, detail=str(e)) @router.get("/templateTree") async def get_pattern_list( visible: str = Query(None, description='是否显示'), # page: int = Query(1, gt=0, description='页码'), # pageSize: int = Query(None, gt=0, description='每页条目数量'), db: Session = Depends(get_db) ): try: query = db.query(TpPatternTemplate) query = query.filter_by(del_flag='0') if visible is not None: query = query.filter_by(visible=visible) # if name: # query = query.filter(TpPatternTemplate.name.like(f'%{name}%')) # total_items = query.count() # # 排序 # if pageSize is None: # pageSize=total_items query = query.order_by(TpPatternTemplate.order_num.asc()) # 执行分页查询 lists = query.all() # .offset((page - 1) * pageSize).limit(pageSize) data = [] for info in lists: # print(1111) query1 = db.query(TpPatternClassification) query1 = query1.filter_by(template_id=info.template_id, del_flag='0') if visible is not None: query1 = query1.filter_by(visible=visible) query1 = query1.order_by(TpPatternClassification.order_num.asc()) classification_list = query1.all() # =template_id_get_classification_list(db,info.template_id) template_info = {"template_id": info.template_id, "name": info.name, "value": info.value, "order_num": info.order_num, "visible": info.visible, "create_time": info.create_time} if classification_list: template_info['children']=[{"classification_id": classification.classification_id, "name": classification.name, "value": classification.value, "order_num": classification.order_num, "visible": classification.visible, "image": classification.image, "icon": classification.icon, "size": classification.size, "create_time": classification.create_time} for classification in classification_list] data.append(template_info) 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)}")