实现门控制接口

ruoyi-fastapi-backend/module_admin/controller/door_controller.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from fastapi import APIRouter, Depends, Form, Request
+from fastapi import APIRouter, Depends, Form, Request, Body
 from pydantic_validation_decorator import ValidateFields
 from sqlalchemy.ext.asyncio import AsyncSession
 from config.enums import BusinessType
@@ -9,16 +9,35 @@ from module_admin.aspect.interface_auth import CheckUserInterfaceAuth
 from module_admin.entity.vo.user_vo import CurrentUserModel
 from module_admin.service.login_service import LoginService
 from module_admin.service.door_service import DoorService
+from module_admin.service.haikang_service import HaiKangService
 from module_admin.entity.vo.door_vo import DeleteDoorModel, DoorModel, DoorPageQueryModel
 from utils.common_util import bytes2file_response
 from utils.log_util import logger
 from utils.page_util import PageResponseModel
 from utils.response_util import ResponseUtil
 
+import time
+
+
 
 doorController = APIRouter(prefix='/system/door', dependencies=[Depends(LoginService.get_current_user)])
 
+# 机器人控制开门
+@doorController.post('/control_door')
+async def control_door(request: Request, door_index_code: str = Body(), control_type: int = Body()):
+    print("door_index_code: ", door_index_code)
+    print("control_type: ", control_type)
+    
+    start_time = time.time()
+    result = await HaiKangService.door_do_control_service(door_index_code, control_type)
+    print("end_time: ", time.time() - start_time)
+    if result[0]:
+        logger.info('门控制成功')
+        return ResponseUtil.success(data=result[1])
+    logger.error('门控制失败')
+    return ResponseUtil.error(msg=result[1])
 
+# 获取门禁设备列表
 @doorController.get(
     '/list', response_model=PageResponseModel
     # , dependencies=[Depends(CheckUserInterfaceAuth('system:door:list'))]
@@ -88,8 +107,6 @@ async def delete_system_door(request: Request, ids: str, query_db: AsyncSession
 
     return ResponseUtil.success(msg=delete_door_result.message)
 
-# 视频流地址
-
 # 获取设备状态
 @doorController.get('/door_status',)
 async def get_door_status(request: Request, query_db: AsyncSession = Depends(get_db)):

ruoyi-fastapi-backend/module_admin/service/haikang_service.py

@@ -27,10 +27,16 @@ class HaiKangService:
     # 门禁控制
     @classmethod
     async def door_do_control_service(cls, door_index_codes, control_type):
-        result = await HaikangUtil.door_do_control(door_index_codes, control_type)
-        print(result)
-        with open("door_do_control_service.json", "w", encoding="utf-8") as f:
-            f.write(json.dumps(result[1]))
+        try:
+            result = await cls.door_do_control(door_index_codes, control_type)
+            return result
+        except Exception as e:
+            return [False,"请求失败"]
+        
+    
+        # print(result)
+        # with open("door_do_control_service.json", "w", encoding="utf-8") as f:
+        #     f.write(json.dumps(result[1]))
 
     # 查询门禁点事件
     @classmethod

ruoyi-fastapi-backend/utils/haikang_util.py

@@ -97,7 +97,7 @@ class HaikangUtil:
 	
         async with httpx.AsyncClient(verify=False) as client:
             response = await client.request(
-                method, url, headers=headers, content=body, timeout=60
+                method, url, headers=headers, content=body, timeout=5 # timeout 单位秒
             )
             if response.status_code in (301, 302, 303, 307, 308):
                 location = response.headers.get("Location")
@@ -688,17 +688,17 @@ class HaikangUtil:
         return headers
 
     # 发送请求
-    @classmethod
-    async def send_request(cls, method: str, url: str, headers: dict, body: str | None):
-        async with httpx.AsyncClient() as client:
-            response = await client.request(
-                method, url, headers=headers, content=body, timeout=60
-            )
-            response.raise_for_status()
-            # if response.status_code >= 400:
-            #     logger.error(f"发送请求失败: {url} , {response.status_code} {response.text}")
-            #     raise HTTPException(status_code=response.status_code, detail=response.text)
-            return response.json()
+    # @classmethod
+    # async def send_request(cls, method: str, url: str, headers: dict, body: str | None):
+    #     async with httpx.AsyncClient() as client:
+    #         response = await client.request(
+    #             method, url, headers=headers, content=body, timeout=5
+    #         )
+    #         response.raise_for_status()
+    #         # if response.status_code >= 400:
+    #         #     logger.error(f"发送请求失败: {url} , {response.status_code} {response.text}")
+    #         #     raise HTTPException(status_code=response.status_code, detail=response.text)
+    #         return response.json()
 
     # 获取access_token
     @classmethod