feat(lzwcai-demp-tool-server-dify-to-mcp): 初始化 Dify 集成工具模块

新增 Dify 到 MCP 的集成工具,支持通过 Dify API 将模型部署到 MCP 平台并进行推理。
该模块包含完整的服务器实现、依赖配置和命令行启动脚本。

主要功能:
- 支持 Workflow 和 Completion 模式的调用
- 自动翻译工具名称为驼峰命名格式
- 提供文件上传与任务停止接口
- 兼容流式与非流式响应处理
This commit is contained in:
2025-12-16 17:52:04 +08:00
commit ec7e7fd7dc
179 changed files with 18443 additions and 0 deletions

View File

@@ -0,0 +1,140 @@
# IoT 设备 MCP 工具说明
## 📋 工具清单
本服务提供 **4 个核心工具**,用于智能设备的查询、定位和控制。
---
## 🔧 工具详情
### 1. **iot_get_devices_by_location** - 根据位置获取设备
**功能**:查询指定位置/房间的所有智能设备
**使用场景**
- "办公室有哪些设备"
- "会议室有什么设备"
- "客厅设备列表"
**参数**
- `location` (必填):位置/房间名称
**返回**该位置的设备清单设备ID、名称、类型、状态、控制命令等
---
### 2. **iot_get_all_spaces_and_devices** - 获取所有空间位置信息
**功能**:获取系统中所有可用空间位置的列表(仅空间名称,不包含设备详情)
**使用场景**
- "显示所有空间"
- "有哪些位置"
- "空间列表"
- "一共有多少个房间"
**参数**:无需参数
**返回**
- 空间总数
- 所有空间名称的列表
**注意**:此工具只返回空间清单,如需查看某个空间的设备,请使用 `iot_get_devices_by_location` 工具
---
### 3. **iot_device_precise_controller** - IoT设备精确控制
**功能**通过设备ID精确控制特定设备
**使用场景**
- 控制特定的灯光、空调、门禁等
- 需配合查询工具获取设备信息后使用
**参数**
- `entityId` (必填)设备唯一ID
- `command` (必填):操作命令(如 turn_on、turn_off、set_temperature
- `params` (必填):操作参数(根据命令类型提供,如温度值、亮度等)
- `userId` (可选)用户ID
**返回**:设备操作结果(成功/失败、设备反馈)
---
### 4. **smart_space_device_locator_matcher** - 智能空间设备定位
**功能**:查询用户当前所属的空间/位置
**使用场景**
- "我现在在哪"
- "当前位置是什么"
- "确认一下位置"
**参数**
- `userId` (必填)用户ID
**返回**:用户所属的空间名称
---
## 💡 典型使用流程
### 方式一:查看所有空间
```
1. 调用 iot_get_all_spaces_and_devices 获取所有空间列表
2. 选择感兴趣的空间
3. 调用 iot_get_devices_by_location 查看该空间的设备
```
### 方式二:查看特定位置的设备
```
1. 调用 iot_get_devices_by_location 指定位置
2. 查看该位置的设备清单和状态
```
### 方式三:控制设备(两步操作)
```
1. 调用 iot_get_devices_by_location 获取设备列表
2. 从结果中提取 entityId 和 command
3. 调用 iot_device_precise_controller 执行控制
```
### 方式四:定位用户
```
1. 调用 smart_space_device_locator_matcher
2. 获取用户当前所属空间
3. 基于位置查询或控制设备
```
---
## 📝 注意事项
1. **企业ID配置**:服务启动时需要配置 `ENTERPRISE_ID` 环境变量,系统会自动初始化向量库
2. **日志记录**:所有操作都会记录到日志文件 `lzwcai_mcp_iot.log`
3. **传输方式**:使用 stdio标准输入输出方式运行
4. **控制工具配合**:精确控制工具必须配合查询工具使用,不能单独随意填写参数
## 🔄 重要更新v0.3.2
**配置变更:**
- ❌ 废弃:`employeeId` 环境变量
- ✅ 新增:`ENTERPRISE_ID` 环境变量(必需)
**初始化流程优化:**
- 移除了通过员工ID查询企业ID的步骤
- 现在直接使用企业ID进行初始化
- 提高了服务启动效率
---
## 🎯 核心特性
- ✅ 支持位置筛选查询设备
- ✅ 支持获取所有可用空间列表
- ✅ 支持精确的设备ID控制
- ✅ 支持用户空间定位
- ✅ 自动格式化设备列表输出
- ✅ 完整的错误处理和日志记录

22
lzwcai_mcp_iot/PKG-INFO Normal file
View File

@@ -0,0 +1,22 @@
Metadata-Version: 2.4
Name: lzwcai-mcp-smartIot
Version: 0.2.21
Summary: IoT设备控制服务器使用FastMCP框架提供设备操作功能
Author-email: LZWCAI开发团队 <dev@lzwcai.com>
License: 专有软件
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: fastmcp>=0.1.0
Requires-Dist: requests
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: black>=23.1.0; extra == "dev"
Requires-Dist: isort>=5.12.0; extra == "dev"
Requires-Dist: flake8>=6.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"

200
lzwcai_mcp_iot/README.md Normal file
View File

@@ -0,0 +1,200 @@
# lzwcai-mcp-iot
[![Version](https://img.shields.io/badge/version-0.3.1-blue.svg)](https://pypi.org/project/lzwcai-mcp-iot/)
[![Python](https://img.shields.io/badge/python-3.8%2B-brightgreen.svg)](https://www.python.org/)
[![License](https://img.shields.io/badge/license-Proprietary-red.svg)]()
> IoT设备控制服务器使用 FastMCP 框架提供智能设备的查询、定位和控制功能
## ✨ 特性
- ✅ 支持位置筛选查询设备
- ✅ 支持获取所有可用空间列表
- ✅ 支持精确的设备ID控制
- ✅ 支持用户空间定位
- ✅ 自动格式化设备列表输出
- ✅ 完整的错误处理和日志记录
## 📦 安装
```bash
pip install lzwcai-mcp-iot
```
或从源码安装:
```bash
git clone <repository-url>
cd lzwcai_mcp_iot
pip install -e .
```
## 🚀 快速开始
### 启动服务
```bash
lzwcai-mcp-iot
```
### 配置要求
服务启动时需要配置 `employeeId`系统会自动初始化企业ID。
## 🔧 核心工具
本服务提供 **4 个核心工具**,用于智能设备的查询、定位和控制。
### 1. iot_get_devices_by_location - 根据位置获取设备
**功能**:查询指定位置/房间的所有智能设备
**使用场景**
- "办公室有哪些设备"
- "会议室有什么设备"
- "客厅设备列表"
**参数**
- `location` (必填):位置/房间名称
**返回**该位置的设备清单设备ID、名称、类型、状态、控制命令等
---
### 2. iot_get_all_spaces_and_devices - 获取所有空间位置信息
**功能**:获取系统中所有可用空间位置的列表(仅空间名称,不包含设备详情)
**使用场景**
- "显示所有空间"
- "有哪些位置"
- "空间列表"
- "一共有多少个房间"
**参数**:无需参数
**返回**
- 空间总数
- 所有空间名称的列表
**注意**:此工具只返回空间清单,如需查看某个空间的设备,请使用 `iot_get_devices_by_location` 工具
---
### 3. iot_device_precise_controller - IoT设备精确控制
**功能**通过设备ID精确控制特定设备
**使用场景**
- 控制特定的灯光、空调、门禁等
- 需配合查询工具获取设备信息后使用
**参数**
- `entityId` (必填)设备唯一ID
- `command` (必填):操作命令(如 turn_on、turn_off、set_temperature
- `params` (必填):操作参数(根据命令类型提供,如温度值、亮度等)
- `userId` (可选)用户ID
**返回**:设备操作结果(成功/失败、设备反馈)
---
### 4. smart_space_device_locator_matcher - 智能空间设备定位
**功能**:查询用户当前所属的空间/位置
**使用场景**
- "我现在在哪"
- "当前位置是什么"
- "确认一下位置"
**参数**
- `userId` (必填)用户ID
**返回**:用户所属的空间名称
---
## 💡 典型使用流程
### 方式一:查看所有空间
```
1. 调用 iot_get_all_spaces_and_devices 获取所有空间列表
2. 选择感兴趣的空间
3. 调用 iot_get_devices_by_location 查看该空间的设备
```
### 方式二:查看特定位置的设备
```
1. 调用 iot_get_devices_by_location 指定位置
2. 查看该位置的设备清单和状态
```
### 方式三:控制设备(两步操作)
```
1. 调用 iot_get_devices_by_location 获取设备列表
2. 从结果中提取 entityId 和 command
3. 调用 iot_device_precise_controller 执行控制
```
### 方式四:定位用户
```
1. 调用 smart_space_device_locator_matcher
2. 获取用户当前所属空间
3. 基于位置查询或控制设备
```
## 📝 注意事项
1. **企业ID初始化**:服务启动时需要配置 `employeeId`系统会自动初始化企业ID
2. **日志记录**:所有操作都会记录到日志文件 `lzwcai_mcp_iot.log`
3. **传输方式**:使用 stdio标准输入输出方式运行
4. **控制工具配合**:精确控制工具必须配合查询工具使用,不能单独随意填写参数
## 🛠️ 开发
### 安装开发依赖
```bash
pip install -e ".[dev]"
```
### 代码格式化
```bash
# 使用 black 格式化
black lzwcai_mcp_iot/
# 使用 isort 排序导入
isort lzwcai_mcp_iot/
```
### 代码检查
```bash
# 使用 flake8
flake8 lzwcai_mcp_iot/
# 使用 mypy
mypy lzwcai_mcp_iot/
```
### 运行测试
```bash
pytest
```
## 📄 许可证
专有软件 - 版权所有 © LZWCAI开发团队
## 📧 联系方式
- 开发团队LZWCAI开发团队
- 邮箱dev@lzwcai.com
## 📚 更多文档
详细的工具使用说明请参考 [IoT设备工具说明.md](IoT设备工具说明.md)

View File

@@ -0,0 +1,223 @@
Metadata-Version: 2.4
Name: lzwcai-mcp-iot
Version: 0.3.3
Summary: IoT设备控制服务器使用FastMCP框架提供设备操作功能
Author-email: LZWCAI开发团队 <dev@lzwcai.com>
License-Expression: LicenseRef-Proprietary
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: fastmcp>=0.1.0
Requires-Dist: requests
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: black>=23.1.0; extra == "dev"
Requires-Dist: isort>=5.12.0; extra == "dev"
Requires-Dist: flake8>=6.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
# lzwcai-mcp-iot
[![Version](https://img.shields.io/badge/version-0.3.1-blue.svg)](https://pypi.org/project/lzwcai-mcp-iot/)
[![Python](https://img.shields.io/badge/python-3.8%2B-brightgreen.svg)](https://www.python.org/)
[![License](https://img.shields.io/badge/license-Proprietary-red.svg)]()
> IoT设备控制服务器使用 FastMCP 框架提供智能设备的查询、定位和控制功能
## ✨ 特性
- ✅ 支持位置筛选查询设备
- ✅ 支持获取所有可用空间列表
- ✅ 支持精确的设备ID控制
- ✅ 支持用户空间定位
- ✅ 自动格式化设备列表输出
- ✅ 完整的错误处理和日志记录
## 📦 安装
```bash
pip install lzwcai-mcp-iot
```
或从源码安装:
```bash
git clone <repository-url>
cd lzwcai_mcp_iot
pip install -e .
```
## 🚀 快速开始
### 启动服务
```bash
lzwcai-mcp-iot
```
### 配置要求
服务启动时需要配置 `employeeId`系统会自动初始化企业ID。
## 🔧 核心工具
本服务提供 **4 个核心工具**,用于智能设备的查询、定位和控制。
### 1. iot_get_devices_by_location - 根据位置获取设备
**功能**:查询指定位置/房间的所有智能设备
**使用场景**
- "办公室有哪些设备"
- "会议室有什么设备"
- "客厅设备列表"
**参数**
- `location` (必填):位置/房间名称
**返回**该位置的设备清单设备ID、名称、类型、状态、控制命令等
---
### 2. iot_get_all_spaces_and_devices - 获取所有空间位置信息
**功能**:获取系统中所有可用空间位置的列表(仅空间名称,不包含设备详情)
**使用场景**
- "显示所有空间"
- "有哪些位置"
- "空间列表"
- "一共有多少个房间"
**参数**:无需参数
**返回**
- 空间总数
- 所有空间名称的列表
**注意**:此工具只返回空间清单,如需查看某个空间的设备,请使用 `iot_get_devices_by_location` 工具
---
### 3. iot_device_precise_controller - IoT设备精确控制
**功能**通过设备ID精确控制特定设备
**使用场景**
- 控制特定的灯光、空调、门禁等
- 需配合查询工具获取设备信息后使用
**参数**
- `entityId` (必填)设备唯一ID
- `command` (必填):操作命令(如 turn_on、turn_off、set_temperature
- `params` (必填):操作参数(根据命令类型提供,如温度值、亮度等)
- `userId` (可选)用户ID
**返回**:设备操作结果(成功/失败、设备反馈)
---
### 4. smart_space_device_locator_matcher - 智能空间设备定位
**功能**:查询用户当前所属的空间/位置
**使用场景**
- "我现在在哪"
- "当前位置是什么"
- "确认一下位置"
**参数**
- `userId` (必填)用户ID
**返回**:用户所属的空间名称
---
## 💡 典型使用流程
### 方式一:查看所有空间
```
1. 调用 iot_get_all_spaces_and_devices 获取所有空间列表
2. 选择感兴趣的空间
3. 调用 iot_get_devices_by_location 查看该空间的设备
```
### 方式二:查看特定位置的设备
```
1. 调用 iot_get_devices_by_location 指定位置
2. 查看该位置的设备清单和状态
```
### 方式三:控制设备(两步操作)
```
1. 调用 iot_get_devices_by_location 获取设备列表
2. 从结果中提取 entityId 和 command
3. 调用 iot_device_precise_controller 执行控制
```
### 方式四:定位用户
```
1. 调用 smart_space_device_locator_matcher
2. 获取用户当前所属空间
3. 基于位置查询或控制设备
```
## 📝 注意事项
1. **企业ID初始化**:服务启动时需要配置 `employeeId`系统会自动初始化企业ID
2. **日志记录**:所有操作都会记录到日志文件 `lzwcai_mcp_iot.log`
3. **传输方式**:使用 stdio标准输入输出方式运行
4. **控制工具配合**:精确控制工具必须配合查询工具使用,不能单独随意填写参数
## 🛠️ 开发
### 安装开发依赖
```bash
pip install -e ".[dev]"
```
### 代码格式化
```bash
# 使用 black 格式化
black lzwcai_mcp_iot/
# 使用 isort 排序导入
isort lzwcai_mcp_iot/
```
### 代码检查
```bash
# 使用 flake8
flake8 lzwcai_mcp_iot/
# 使用 mypy
mypy lzwcai_mcp_iot/
```
### 运行测试
```bash
pytest
```
## 📄 许可证
专有软件 - 版权所有 © LZWCAI开发团队
## 📧 联系方式
- 开发团队LZWCAI开发团队
- 邮箱dev@lzwcai.com
## 📚 更多文档
详细的工具使用说明请参考 [IoT设备工具说明.md](IoT设备工具说明.md)

View File

@@ -0,0 +1,19 @@
README.md
pyproject.toml
setup.cfg
lzwcai_mcp_iot/__init__.py
lzwcai_mcp_iot/config.py
lzwcai_mcp_iot/iot_device_tool.py
lzwcai_mcp_iot.egg-info/PKG-INFO
lzwcai_mcp_iot.egg-info/SOURCES.txt
lzwcai_mcp_iot.egg-info/dependency_links.txt
lzwcai_mcp_iot.egg-info/entry_points.txt
lzwcai_mcp_iot.egg-info/requires.txt
lzwcai_mcp_iot.egg-info/top_level.txt
lzwcai_mcp_iot/src/__init__.py
lzwcai_mcp_iot/src/device_operations.py
lzwcai_mcp_iot/src/device_results_pretreatment.py
lzwcai_mcp_iot/src/init_mcp.py
lzwcai_mcp_iot/src/iot_device_dicts_prompt.py
lzwcai_mcp_iot/src/logger_config.py
lzwcai_mcp_iot/src/vector_service.py

View File

@@ -0,0 +1 @@

View File

@@ -0,0 +1,2 @@
[console_scripts]
lzwcai-mcp-iot = lzwcai_mcp_iot.iot_device_tool:main

View File

@@ -0,0 +1,9 @@
fastmcp>=0.1.0
requests
[dev]
pytest>=7.0.0
black>=23.1.0
isort>=5.12.0
flake8>=6.0.0
mypy>=1.0.0

View File

@@ -0,0 +1 @@
lzwcai_mcp_iot

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,44 @@
{
"code": 200,
"data": {
"devices": [
{
"location_key": "32",
"location_desc": "研发办公区;研发部",
"entityId": "climate.qjiang_cn_741479129_wb20",
"device_desc": "空调;制冷设备",
"operations": [
{
"command": "turn_on",
"operation_desc": "打开空调;开空调;打开;开",
"operation_params": []
},
{
"command": "turn_off",
"operation_desc": "关空调;关闭空调;关闭;关;闭空调;空调",
"operation_params": []
},
{
"command": "set_temperature",
"operation_desc": "设置温度;调节温度;",
"operation_params": [
{
"description": "温度",
"key": "temperature",
"value": "7-35°C"
},
{
"description": "运行模式",
"key": "hvac_mode",
"value": "heat/cool/auto/dry/fan_only/off"
}
]
}
]
}
],
"count": 1,
"location_filter": "爱一拍展厅"
},
"msg": "成功获取公司 1952978233106669569 在位置 '爱一拍展厅' 的 2 个设备"
}

View File

@@ -0,0 +1,99 @@
"""
配置文件
该模块包含了项目的所有配置常量和设置。
"""
import os
from typing import Dict, Any
from .src.logger_config import get_logger
# 延迟初始化日志器,避免在导入时立即执行
logger = None
def _ensure_logger():
"""确保日志器已初始化"""
global logger
if logger is None:
logger = get_logger(__name__)
return logger
# 生产环境
DEFAULT_DEVICE_API_BASE_URL = "http://lzwcai-demp-corp-manager:8086"
DEFAULT_VECTOR_API_BASE_URL = "http://lzwcai-demp-tool-server:5002"
# 本地环境
# DEFAULT_DEVICE_API_BASE_URL = "http://192.168.2.236:8088"
# DEFAULT_VECTOR_API_BASE_URL = "http://192.168.2.236:5002"
# 默认企业ID
# DEFAULT_ENTERPRISE_ID = "1952978233106669569"
DEFAULT_ENTERPRISE_ID = ""
# 默认员工ID
# DEFAULT_EMPLOYEE_ID = "1955949384389005313"
DEFAULT_EMPLOYEE_ID = ""
# 请求配置
REQUEST_TIMEOUT = 30 # 请求超时时间(秒)
MAX_RETRIES = 3 # 最大重试次数
# 向量服务配置
DEFAULT_VECTOR_STORE_NAME = "设备库"
DEFAULT_VECTOR_STORE_DESCRIPTION = "向量库"
DEFAULT_ENCODER_TYPE = "word2vec"
DEFAULT_TOP_K = 1
DEFAULT_AUTO_CREATE = True
# 日志配置
LOG_LEVEL = "INFO"
LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
# 环境变量配置
def get_config() -> Dict[str, Any]:
"""
获取配置信息,支持环境变量覆盖
返回:
Dict[str, Any]: 配置字典
"""
logger = _ensure_logger()
logger.debug("开始加载配置...")
config = {
"device_api_base_url": os.getenv(
"DEVICE_API_BASE_URL", DEFAULT_DEVICE_API_BASE_URL
),
"vector_api_base_url": os.getenv(
"VECTOR_API_BASE_URL", DEFAULT_VECTOR_API_BASE_URL
),
"enterprise_id": os.getenv("enterpriseId", DEFAULT_ENTERPRISE_ID),
"employeeId": os.getenv("employeeId", DEFAULT_EMPLOYEE_ID),
"request_timeout": int(os.getenv("REQUEST_TIMEOUT", REQUEST_TIMEOUT)),
"max_retries": int(os.getenv("MAX_RETRIES", MAX_RETRIES)),
"vector_store_name": os.getenv("VECTOR_STORE_NAME", DEFAULT_VECTOR_STORE_NAME),
"vector_store_description": os.getenv(
"VECTOR_STORE_DESCRIPTION", DEFAULT_VECTOR_STORE_DESCRIPTION
),
"encoder_type": os.getenv("ENCODER_TYPE", DEFAULT_ENCODER_TYPE),
"default_top_k": int(os.getenv("DEFAULT_TOP_K", DEFAULT_TOP_K)),
"default_auto_create": os.getenv(
"DEFAULT_AUTO_CREATE", str(DEFAULT_AUTO_CREATE)
).lower()
== "true",
"log_level": os.getenv("LOG_LEVEL", LOG_LEVEL),
"log_format": os.getenv("LOG_FORMAT", LOG_FORMAT),
}
logger.info("配置加载完成")
logger.debug(f"设备API地址: {config['device_api_base_url']}")
logger.debug(f"向量API地址: {config['vector_api_base_url']}")
logger.debug(f"员工ID: {config['employeeId']}")
logger.debug(f"请求超时: {config['request_timeout']}")
return config
# 全局配置实例
CONFIG = get_config()

View File

@@ -0,0 +1,410 @@
"""
IoT设备服务器主模块
该模块实现了一个简单的IoT设备控制服务器使用FastMCP框架提供设备操作功能。
"""
import os
import json
from mcp.server.fastmcp import FastMCP
from .src.device_results_pretreatment import (
format_devices_list,
)
from .src.device_operations import DeviceOperator
from .src.vector_service import VectorService
from .src.logger_config import get_logger
from .config import CONFIG
from .src.iot_device_dicts_prompt import (
iot_get_devices_by_location_prompt,
iot_get_all_spaces_and_devices_prompt,
iot_device_precise_controller_prompt,
smart_space_device_locator_matcher_prompt,
)
from .src.init_mcp import init_mcp_server
# 延迟初始化日志器,避免在导入时立即执行
logger = None
def _ensure_logger():
"""确保日志器已初始化"""
global logger
if logger is None:
logger = get_logger(__name__)
return logger
# 创建FastMCP实例
mcp = FastMCP("iot_device_server")
# 创建设备操作实例
device_op = DeviceOperator(api_base_url=CONFIG["device_api_base_url"])
# 创建向量服务实例
vector_service = VectorService(base_url=CONFIG["vector_api_base_url"])
@mcp.tool(description=iot_get_devices_by_location_prompt())
async def iot_get_devices_by_location(
location: str,
) -> str:
logger = _ensure_logger()
try:
# 输入参数验证
if not location:
error_msg = "location参数是必需的"
logger.error(error_msg)
return json.dumps(
{"code": 400, "msg": error_msg, "data": None}, ensure_ascii=False
)
logger.info(
f"开始处理IoT设备查询请求 - 位置: {location}"
)
# 从环境变量获取企业ID已在启动时初始化
enterprise_id = os.environ.get("enterpriseId")
if not enterprise_id:
error_msg = f"企业ID未初始化请检查员工ID配置"
logger.error(error_msg)
return json.dumps(
{"code": 404, "msg": error_msg, "data": None}, ensure_ascii=False
)
logger.info(f"获取到企业ID: {enterprise_id}")
# 查询设备列表
query_result = vector_service.query_devices_by_location(
keyId=enterprise_id,
location=location
)
logger.info(f"查询设备列表结果: {query_result}")
# 提取设备列表
devices = query_result.get("devices", [])
device_count = query_result.get("count", 0)
if device_count == 0 or not devices:
return json.dumps({
"code": 404,
"msg": f"在位置「{location}」未找到任何设备",
"data": None
}, ensure_ascii=False)
# 使用格式化函数将设备列表转换为易读的文本
formatted_text = format_devices_list(devices)
logger.info(f"设备列表已格式化,设备数量: {device_count}")
# 返回包含格式化文本和原始数据的结果
result = {
"code": 200,
"msg": formatted_text, # 直接将格式化文本放在msg字段
"data": {
"devices": devices, # 原始设备数据,供后续操作使用
"count": device_count,
"location_filter": query_result.get("location_filter", location)
}
}
return json.dumps(result, ensure_ascii=False)
except Exception as e:
error_msg = f"IoT设备查询过程中发生异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return json.dumps(
{"code": 500, "msg": error_msg, "data": None}, ensure_ascii=False
)
@mcp.tool(description=iot_get_all_spaces_and_devices_prompt())
async def iot_get_all_spaces_and_devices() -> str:
logger = _ensure_logger()
try:
logger.info("开始处理获取所有空间位置信息请求")
# 从环境变量获取企业ID已在启动时初始化
enterprise_id = os.environ.get("enterpriseId")
if not enterprise_id:
error_msg = f"企业ID未初始化请检查员工ID配置"
logger.error(error_msg)
return json.dumps(
{"code": 404, "msg": error_msg, "data": None}, ensure_ascii=False
)
logger.info(f"获取到企业ID: {enterprise_id}")
# 查询所有设备(不指定位置,获取所有)
query_result = vector_service.query_devices_by_location(
keyId=enterprise_id,
location="" # 空字符串表示获取所有
)
logger.info(f"查询所有设备结果: {query_result}")
# 提取设备列表
devices = query_result.get("devices", [])
device_count = query_result.get("count", 0)
if device_count == 0 or not devices:
return json.dumps({
"code": 404,
"msg": "系统中未找到任何设备和空间",
"data": None
}, ensure_ascii=False)
# 提取所有唯一的空间名称
spaces_set = set()
for device in devices:
# 提取空间信息
location_desc = device.get("location_desc", "")
if location_desc and location_desc != "未知空间":
spaces_set.add(location_desc)
# 转换为列表并排序
spaces_list = sorted(list(spaces_set))
if not spaces_list:
return json.dumps({
"code": 404,
"msg": "系统中未找到任何有效空间",
"data": None
}, ensure_ascii=False)
# 构建格式化文本
formatted_lines = []
formatted_lines.append("=" * 60)
formatted_lines.append(f"所有空间位置信息")
formatted_lines.append("=" * 60)
formatted_lines.append(f"空间总数: {len(spaces_list)}")
formatted_lines.append("")
# 列出所有空间
for space_idx, space_name in enumerate(spaces_list, 1):
formatted_lines.append(f"{space_idx}. {space_name}")
formatted_lines.append("")
formatted_lines.append("=" * 60)
formatted_text = "\n".join(formatted_lines)
logger.info(f"已获取所有空间位置信息,空间数: {len(spaces_list)}")
# 返回包含格式化文本和空间列表的结果
result = {
"code": 200,
"msg": formatted_text,
"data": {
"spaces": spaces_list, # 空间名称列表
"space_count": len(spaces_list)
}
}
return json.dumps(result, ensure_ascii=False)
except Exception as e:
error_msg = f"获取所有空间位置信息过程中发生异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return json.dumps(
{"code": 500, "msg": error_msg, "data": None}, ensure_ascii=False
)
@mcp.tool(description=iot_device_precise_controller_prompt())
async def iot_device_precise_controller(
entityId: str,
command: str,
params: dict = None,
userId: str = None,
) -> str:
logger = _ensure_logger()
logger.info("=" * 60)
logger.info("调用iot_device_precise_controller工具")
logger.info(f"参数 - entityId: {entityId}, command: {command}, params: {params}, userId: {userId}")
logger.info("=" * 60)
try:
# 输入参数验证
logger.debug("开始参数验证...")
if not all([entityId, command]):
error_msg = "所有参数都是必需的: entityId,command"
logger.error(f"参数验证失败: {error_msg}")
return json.dumps(
{"code": 400, "msg": error_msg, "data": None}, ensure_ascii=False
)
logger.debug("参数验证通过")
# 从环境变量获取企业ID已在启动时初始化
enterprise_id = os.environ.get("enterpriseId")
if not enterprise_id:
error_msg = f"企业ID未初始化请检查员工ID配置"
logger.error(error_msg)
return json.dumps(
{"code": 404, "msg": error_msg, "data": None}, ensure_ascii=False
)
logger.info(f"获取到企业ID: {enterprise_id}")
map_result = {
"enterpriseId": enterprise_id,
"entityId": entityId,
"command": command,
"params": params if params is not None else {},
}
# 如果提供了userId则添加到map_result中
if userId:
map_result["userId"] = userId
# 操作设备
result = device_op.operate_device(map_result)
logger.info(f"设备操作结果: {result}")
# 将结果字典转换为JSON字符串
if isinstance(result, dict):
result = json.dumps(result, ensure_ascii=False)
logger.info("iot_device_precise_controller工具执行完成")
logger.info("=" * 60)
return result
except Exception as e:
error_msg = f"IoT设备操作过程中发生异常: {str(e)}"
logger.error(error_msg, exc_info=True)
logger.error("iot_device_precise_controller工具执行失败")
logger.error("=" * 60)
return json.dumps(
{"code": 500, "msg": error_msg, "data": None}, ensure_ascii=False
)
@mcp.tool(description=smart_space_device_locator_matcher_prompt())
async def smart_space_device_locator_matcher(
userId: str,
) -> str:
logger = _ensure_logger()
try:
if not userId:
error_msg = "所有参数都是必需的: userId"
logger.error(error_msg)
return json.dumps(
{"code": 400, "msg": error_msg, "data": None}, ensure_ascii=False
)
result = device_op.get_user_spaces_by_user_id(userId)
if isinstance(result, dict):
# 检查返回的data是否为None如果是则提供友好的描述
if result.get("data") is None:
result["msg"] = "我目前还没发现您在哪里 请您告诉我你所属位置;"
return json.dumps(result, ensure_ascii=False)
return result
except Exception as e:
error_msg = f"定位空间过程中发生异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return json.dumps(
{"code": 500, "msg": error_msg, "data": None}, ensure_ascii=False
)
def testFn() -> str:
logger = _ensure_logger()
try:
# 读取test.json文件
test_file_path = r"E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcp_iot\test.json"
with open(test_file_path, "r", encoding="utf-8") as f:
test_data = json.load(f)
# 获取results数据
results = test_data.get("results", [])
logger.info(f"从test.json读取到的results数据: {results}")
# 参数预处理
map_result = device_op.preprocess_results(results)
logger.info(f"参数预处理结果: {map_result}")
# 保存map_result到本地JSON文件
output_file_path = r"E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcp_iot\map_result.json"
with open(output_file_path, "w", encoding="utf-8") as f:
json.dump(map_result, f, ensure_ascii=False, indent=2)
logger.info(f"map_result已保存到: {output_file_path}")
return json.dumps(map_result, ensure_ascii=False)
except FileNotFoundError:
error_msg = "test.json文件未找到"
logger.error(error_msg)
return json.dumps(
{"code": 404, "msg": error_msg, "data": None}, ensure_ascii=False
)
except json.JSONDecodeError as e:
error_msg = f"test.json文件格式错误: {str(e)}"
logger.error(error_msg)
return json.dumps(
{"code": 400, "msg": error_msg, "data": None}, ensure_ascii=False
)
except Exception as e:
error_msg = f"测试过程中发生异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return json.dumps(
{"code": 500, "msg": error_msg, "data": None}, ensure_ascii=False
)
def main():
"""主函数启动MCP服务器"""
logger = _ensure_logger()
try:
logger.info("=" * 80)
logger.info("IoT设备MCP服务器启动流程开始")
logger.info(f"配置信息: {CONFIG}")
logger.info("=" * 80)
# 初始化组件
logger.info("正在初始化核心组件...")
logger.info(f"设备操作器API地址: {CONFIG['device_api_base_url']}")
logger.info(f"向量服务API地址: {CONFIG['vector_api_base_url']}")
logger.info("核心组件初始化完成")
# 获取企业ID并初始化MCP服务器
enterprise_id = CONFIG.get("enterprise_id")
if enterprise_id:
logger.info(f"检测到企业ID配置: {enterprise_id}")
logger.info("开始初始化MCP服务器...")
init_result = init_mcp_server(device_op, vector_service, enterprise_id)
logger.info(f"MCP服务器初始化结果: {init_result}")
if init_result.get("code") != 200:
logger.warning(f"MCP服务器初始化失败但服务器仍将启动: {init_result}")
else:
logger.info("MCP服务器初始化成功")
# 保存企业ID到环境变量
returned_enterprise_id = init_result.get("data", {}).get("enterprise_id")
if returned_enterprise_id:
device_op.set_enterprise_id_to_env(returned_enterprise_id)
logger.info(f"企业ID已保存到环境变量: {returned_enterprise_id}")
else:
# 如果返回结果中没有enterprise_id使用配置中的
device_op.set_enterprise_id_to_env(enterprise_id)
logger.info(f"企业ID已保存到环境变量: {enterprise_id}")
else:
logger.warning("未配置企业ID跳过预初始化")
logger.info("提示您可以在配置文件或环境变量中设置ENTERPRISE_ID来启用自动初始化功能")
# 注册的工具列表日志
logger.info("已注册的MCP工具:")
logger.info("1. iot_get_devices_by_location - 根据位置获取设备列表")
logger.info("2. iot_get_all_spaces_and_devices - 获取所有空间位置信息")
logger.info("3. iot_device_precise_controller - IoT设备精确控制")
logger.info("4. smart_space_device_locator_matcher - 智能空间设备定位")
logger.info("=" * 80)
logger.info("MCP服务器即将启动等待客户端连接...")
logger.info("传输方式: stdio (标准输入输出)")
logger.info("注意: 控制台日志已禁用,所有日志将写入文件")
logger.info("=" * 80)
# 使用标准输入输出作为传输方式运行服务器
mcp.run(transport="stdio")
except Exception as e:
logger.error("=" * 80)
logger.error(f"服务器启动失败: {str(e)}")
logger.error("错误详情:", exc_info=True)
logger.error("=" * 80)
raise
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,751 @@
"""
设备操作模块
该模块提供了设备操作的接口,用于控制不同位置的各种设备。
"""
import os
import requests
import json
from typing import Dict, Any, Optional, List
from ..config import CONFIG
from .device_results_pretreatment import process_device_data, process_device_results
from .logger_config import get_logger
# 延迟初始化日志器,避免在导入时立即执行
logger = None
def _ensure_logger():
"""确保日志器已初始化"""
global logger
if logger is None:
logger = get_logger(__name__)
return logger
enterprise_id = os.environ.get("enterpriseId", CONFIG["enterprise_id"])
class DeviceOperator:
"""设备操作类,提供设备控制功能"""
def __init__(self, api_base_url: str = None):
"""
初始化设备操作器
参数:
api_base_url: API服务的基础URL
"""
logger = _ensure_logger()
self.api_base_url = api_base_url or CONFIG["device_api_base_url"]
self.session = requests.Session()
# 设置默认超时
self.session.timeout = CONFIG["request_timeout"]
logger.info(f"DeviceOperator初始化完成API基础URL: {self.api_base_url}")
def set_enterprise_id_to_env(self, enterprise_id_value: str) -> None:
"""
将企业ID存入环境变量
参数:
enterprise_id_value: 企业ID值
"""
logger = _ensure_logger()
if not enterprise_id_value or not isinstance(enterprise_id_value, str):
logger.warning(f"无效的企业ID值: {enterprise_id_value}")
return
os.environ["enterpriseId"] = enterprise_id_value
global enterprise_id
enterprise_id = enterprise_id_value
logger.info(f"企业ID已更新: {enterprise_id_value}")
def get_enterprise_id_by_user(self, user_id: str) -> Optional[str]:
"""
根据用户ID获取企业ID并存入环境变量
参数:
user_id: 用户ID
返回:
Optional[str]: 企业ID获取失败时返回None
"""
logger = _ensure_logger()
if not user_id or not isinstance(user_id, str):
logger.error(f"无效的用户ID: {user_id}")
return None
# 调用实际API获取企业ID
url = f"{self.api_base_url}/system/enterprise/user/{user_id}"
headers = {
"Accept": "*/*",
}
try:
logger.info(f"正在获取用户 {user_id} 的企业ID...")
response = self.session.get(
url=url, headers=headers, timeout=CONFIG["request_timeout"]
)
if response.status_code == 200:
data = response.json()
if (
data.get("code") == 200
and data.get("data")
and data["data"].get("id")
):
enterprise_id_value = str(data["data"]["id"])
# 将企业ID存入环境变量
self.set_enterprise_id_to_env(enterprise_id_value)
logger.info(f"成功获取企业ID: {enterprise_id_value}")
return enterprise_id_value
else:
logger.warning(f"API返回数据格式异常: {data}")
else:
logger.error(
f"API请求失败状态码: {response.status_code}, 响应: {response.text}"
)
# 如果API调用失败或数据格式不符预期返回默认值
logger.info(f"使用默认企业ID: {enterprise_id}")
return enterprise_id
except requests.exceptions.Timeout:
logger.error(f"获取企业ID请求超时用户ID: {user_id}")
return enterprise_id
except requests.exceptions.RequestException as e:
logger.error(f"获取企业ID网络请求异常: {str(e)}")
return enterprise_id
except Exception as e:
logger.error(f"获取企业ID时发生未知异常: {str(e)}", exc_info=True)
return enterprise_id
# 设备结果分数判断
def check_device_results_score(
self, best_match_data: Optional[Dict[str, Any]]
) -> Dict[str, Any]:
"""
检查设备匹配的准确性
根据你的代码分析,评分系统说明:
- 综合分数(combined_score)或score是最终评判标准
- 0.85-0.88: 准确匹配,可信度高 (随机阈值)
- 0.7-0.84: 一般匹配,需要确认
- 0.5-0.69: 较差匹配,不建议使用
- <0.5: 很差匹配,不推荐
Args:
best_match_data: best_match数据包含device和score信息
Returns:
Dict: {
"checkResult": True/False, # 是否满足分数要求(随机阈值0.85-0.88)
"data": data # 满足要求返回对应项不满足返回整个data
}
"""
logger.debug("开始检查设备匹配准确性...")
import random
# 提取分数 - 优先使用combined_score其次使用score
score = best_match_data.get("combined_score", best_match_data.get("score", 0.0))
logger.debug(f"提取的匹配分数: {score}")
# 生成随机阈值 (0.85-0.88之间)
random_threshold = round(random.uniform(0.6, 0.65), 2)
logger.debug(f"生成的随机阈值: {random_threshold}")
# 判断是否满足准确性要求 (随机阈值-1.0)
is_accurate = random_threshold <= score <= 1.0
logger.info(f"设备匹配准确性检查: 分数={score}, 阈值={random_threshold}, 通过={is_accurate}")
# 构造返回结果
result = {"checkResult": is_accurate, "data": best_match_data}
return result
def preprocess_results(self, raw_data: List[List[Any]]) -> List[Dict[str, Any]]:
"""
预处理数据,将原始数据转换为简化的数组对象格式
如果 entityId 相同,则合并到一起
Args:
raw_data: 原始数据,格式为 [[entity, score], ...]
Returns:
处理后的数据数组每个对象包含合并后的信息不包含id和score_details字段
"""
logger.debug(f"开始预处理设备结果数据,原始数据条数: {len(raw_data) if raw_data else 0}")
try:
result = process_device_results(raw_data)
logger.info(f"设备结果数据预处理完成,输出条数: {len(result) if result else 0}")
return result
except Exception as e:
logger.error(f"预处理设备结果数据时发生异常: {str(e)}", exc_info=True)
return []
def preprocess_parameters(
self,
data: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
"""
预处理设备操作参数使用process_device_data方法处理数据
参数:
data: 需要预处理的数据包含best_match信息
格式如:{
"best_match": {
"device": {
"device": {
"key": "switch.zimi_cn_1144138206_dhkg01_on_p_2_1",
"description": "灵泽办公区过道吊灯 开关 按键"
},
"operation": {
"key": "turn_on",
"description": "开关"
}
}
}
}
返回:
Dict[str, Any]: 预处理后的设备操作参数
格式如:{
"enterpriseId": "1932095424144715777",
"entityId": "switch.zimi_cn_1144138206_dhkg01_on_p_2_1",
"command": "turn_on"
}
"""
# 默认结果结构
default_result = {"enterpriseId": enterprise_id, "entityId": "", "command": ""}
try:
# 使用process_device_data方法处理数据
if data and isinstance(data, dict):
processed_result = process_device_data(data, enterprise_id)
if processed_result:
logger.info(f"成功处理设备数据: {processed_result}")
return processed_result
else:
logger.warning("process_device_data返回None使用默认结果")
else:
logger.warning("输入数据为空或格式异常")
except Exception as e:
logger.error(f"预处理参数时发生异常: {str(e)}", exc_info=True)
return default_result
def operate_device(self, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
操作设备的方法
参数:
data: 设备操作数据包含entityId、command等信息
返回:
Dict[str, Any]: 包含操作结果的字典
"""
# 初始化操作参数
if data is None:
data = {}
# 检查并设置 enterpriseId
if not data.get("enterpriseId"):
data["enterpriseId"] = enterprise_id
logger.info(f"data中未包含enterpriseId使用默认值: {enterprise_id}")
# 验证必要参数
if not data.get("entityId") or not data.get("command"):
error_msg = "缺少必要的设备操作参数: entityId 和 command"
logger.error(error_msg)
return {"code": 400, "msg": error_msg, "data": None}
logger.info(
f"开始操作设备 - entityId: {data.get('entityId')}, command: {data.get('command')}"
)
# # 保存map_result到本地JSON文件
# output_file_path = r"E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcp_iot\output\data.json"
# with open(output_file_path, "w", encoding="utf-8") as f:
# json.dump(data, f, ensure_ascii=False, indent=2)
# logger.info(f"map_result已保存到: {output_file_path}")
# 调用API接口
result = self.call_device_api(data)
# result = {"code": 200, "msg": "操作成功", "data": None}
return result
def batch_operate_devices(self, devices_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
批量操作设备的方法
参数:
devices_data: 设备操作数据数组每个元素包含entityId、command等信息
数据结构与operate_device方法的data参数相同
返回:
Dict[str, Any]: 批量操作结果,包含成功/失败统计和详细结果
{
"code": 200, # 整体状态码
"msg": "批量操作完成",
"data": {
"total": 总数量,
"success": 成功数量,
"failed": 失败数量,
"results": [
{
"index": 索引,
"deviceInfo": 设备信息,
"result": 操作结果,
"success": True/False
}
]
}
}
"""
if not devices_data or not isinstance(devices_data, list):
error_msg = "无效的设备数据devices_data必须是一个非空数组"
logger.error(error_msg)
return {"code": 400, "msg": error_msg, "data": None}
total_count = len(devices_data)
success_count = 0
failed_count = 0
results = []
logger.info(f"开始批量操作设备,总数量: {total_count}")
for index, device_data in enumerate(devices_data):
device_info = { **device_data,"index": index}
try:
logger.info(f"正在操作第 {index + 1}/{total_count} 个设备: {device_info['entityId']}")
# 验证设备数据格式
if not isinstance(device_data, dict):
raise ValueError(f"设备数据格式错误,必须是字典类型")
# 调用单个设备操作方法
operation_result = self.operate_device(device_data)
# 判断操作是否成功通常code=200表示成功
is_success = operation_result.get("code") == 200
if is_success:
success_count += 1
logger.info(f"设备操作成功: {device_info['entityId']}")
else:
failed_count += 1
logger.warning(f"设备操作失败: {device_info['entityId']}, 原因: {operation_result.get('msg', '未知错误')}")
# 记录结果
results.append({
"index": index,
"deviceInfo": device_info,
"result": operation_result,
"success": is_success
})
except Exception as e:
failed_count += 1
error_msg = f"设备操作异常: {str(e)}"
logger.error(f"{index + 1} 个设备操作异常: {error_msg}", exc_info=True)
# 记录异常结果
results.append({
"index": index,
"deviceInfo": device_info,
"result": {"code": 500, "msg": error_msg, "data": None},
"success": False
})
# 构建返回结果
summary_msg = f"批量操作完成,总数: {total_count}, 成功: {success_count}, 失败: {failed_count}"
logger.info(summary_msg)
# 整体状态码判断如果全部成功则返回200部分成功返回206全部失败返回500
if success_count == total_count:
overall_code = 200
overall_msg = "批量操作全部成功"
elif success_count > 0:
overall_code = 206 # 部分成功
overall_msg = f"批量操作部分成功,成功: {success_count}, 失败: {failed_count}"
else:
overall_code = 500
overall_msg = "批量操作全部失败"
return {
"code": overall_code,
"msg": overall_msg,
"data": {
"total": total_count,
"success": success_count,
"failed": failed_count,
"success_rate": round(success_count / total_count * 100, 2) if total_count > 0 else 0,
"results": results
}
}
def get_digital_employee_by_id(self, employee_id: str) -> Dict[str, Any]:
"""
根据员工ID获取数字员工信息
参数:
employee_id: 员工ID
返回:
Dict[str, Any]: 包含数字员工信息的字典
"""
if not employee_id or not isinstance(employee_id, str):
error_msg = f"无效的员工ID: {employee_id}"
logger.error(error_msg)
return {"code": 400, "msg": error_msg, "data": None}
# API配置
url = f"{self.api_base_url}/system/mcpServer/getByEmployeeId/{employee_id}"
headers = {}
try:
logger.info(f"正在获取员工ID为 {employee_id} 的数字员工信息...")
response = self.session.get(
url=url, headers=headers, timeout=CONFIG["request_timeout"]
)
# 处理响应
if response.status_code == 200:
data = response.json()
logger.info(f"成功获取数字员工信息: {data}")
return data
else:
error_msg = f"API请求失败状态码: {response.status_code}, 响应: {response.text}"
logger.error(error_msg)
return {
"code": response.status_code,
"msg": error_msg,
"data": None,
}
except requests.exceptions.Timeout:
error_msg = f"获取数字员工信息请求超时员工ID: {employee_id}"
logger.error(error_msg)
return {"code": 408, "msg": error_msg, "data": None}
except requests.exceptions.RequestException as e:
error_msg = f"获取数字员工信息网络请求异常: {str(e)}"
logger.error(error_msg)
return {"code": 500, "msg": error_msg, "data": None}
except Exception as e:
error_msg = f"获取数字员工信息时发生未知异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {"code": 500, "msg": error_msg, "data": None}
def get_user_spaces_by_user_id(self, user_id: str) -> Dict[str, Any]:
"""
根据用户ID获取该用户关联的空间信息
参数:
user_id: 用户ID
返回:
Dict[str, Any]: 接口响应结果,结构为 {"code": number, "msg": string, "data": any}
"""
if not user_id or not isinstance(user_id, str):
error_msg = f"无效的用户ID: {user_id}"
logger.error(error_msg)
return {"code": 400, "msg": error_msg, "data": None}
url = f"{self.api_base_url}/system/mcpServer/space/user/{user_id}"
headers = {
"Accept": "*/*",
}
try:
logger.info(f"正在获取用户ID为 {user_id} 的空间信息...")
response = self.session.get(
url=url, headers=headers, timeout=CONFIG["request_timeout"]
)
if response.status_code == 200:
try:
data = response.json()
except ValueError:
data = {}
if isinstance(data, dict):
# 统一返回格式,确保存在 code/msg/data 字段
if "code" not in data:
data["code"] = 200
if "msg" not in data:
data["msg"] = "success"
# 接口可能没有返回 data 时,显式置为 None
if "data" not in data or data["data"] in (None, [], {}):
data["data"] = None
logger.info(f"成功获取用户空间信息: {data}")
return data
else:
wrapped = {"code": 200, "msg": "success", "data": None}
logger.info(f"成功获取用户空间信息(非字典响应已包装): {wrapped}")
return wrapped
else:
error_msg = f"API请求失败状态码: {response.status_code}, 响应: {response.text}"
logger.error(error_msg)
return {
"code": response.status_code,
"msg": error_msg,
"data": None,
}
except requests.exceptions.Timeout:
error_msg = f"获取用户空间信息请求超时用户ID: {user_id}"
logger.error(error_msg)
return {"code": 408, "msg": error_msg, "data": None}
except requests.exceptions.RequestException as e:
error_msg = f"获取用户空间信息网络请求异常: {str(e)}"
logger.error(error_msg)
return {"code": 500, "msg": error_msg, "data": None}
except Exception as e:
error_msg = f"获取用户空间信息时发生未知异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {"code": 500, "msg": error_msg, "data": None}
def call_device_api(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
调用设备操作API
参数:
data: 设备操作数据包含entityId、command等信息
返回:
Dict[str, Any]: API调用结果
"""
# API配置
url = f"{self.api_base_url}/space_iot/operation/call"
headers = {
"Content-Type": "application/json",
}
# 构建请求数据
request_data = {**data}
try:
logger.info(f"调用设备操作API: {url}")
logger.debug(f"请求数据: {request_data}")
# 发送POST请求
response = self.session.post(
url=url,
headers=headers,
data=json.dumps(request_data),
timeout=CONFIG["request_timeout"],
)
# 处理响应
if response.status_code == 200:
result = response.json()
logger.info(f"设备操作API调用成功: {result}")
return result
else:
error_msg = f"API请求失败状态码: {response.status_code}, 响应: {response.text}"
logger.error(error_msg)
return {
"code": response.status_code,
"msg": error_msg,
"data": None,
}
except requests.exceptions.Timeout:
error_msg = "设备操作API请求超时"
logger.error(error_msg)
return {"code": 408, "msg": error_msg, "data": None}
except requests.exceptions.RequestException as e:
error_msg = f"设备操作API网络请求异常: {str(e)}"
logger.error(error_msg)
return {"code": 500, "msg": error_msg, "data": None}
except Exception as e:
error_msg = f"设备操作API调用异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {"code": 500, "msg": error_msg, "data": None}
def get_operation_analysis_result(self, tool_data: Dict[str, Any], best_match: Optional[Dict[str, Any]]) -> str:
"""
分析操作结果,根据 tool_data 和 best_match 返回合适的操作命令
参数:
tool_data: 工具数据,包含 location、device、operation、keyId
best_match: 最佳匹配结果,包含设备信息和评分
返回:
str: 操作命令字符串
"""
logger.debug("开始分析操作结果...")
logger.debug(f"输入参数 - tool_data: {tool_data}, best_match: {best_match}")
def contains_chinese(text: str) -> bool:
"""检查字符串是否包含中文字符"""
if not text:
return False
for char in text:
if '\u4e00' <= char <= '\u9fff':
return True
return False
try:
# 获取 tool_data 中的 operation
tool_operation = tool_data.get("operation", "") if tool_data else ""
logger.debug(f"tool_data 中的 operation: {tool_operation}")
# 判断 tool_data 的 operation 是否是纯英文(不包含中文)
if tool_operation and not contains_chinese(tool_operation):
logger.info(f"tool_data operation 为纯英文,直接使用: {tool_operation}")
return tool_operation
# 如果 tool_data 的 operation 包含中文或为空,检查 best_match
logger.debug("tool_data operation 包含中文或为空,尝试使用 best_match")
if best_match and isinstance(best_match, dict):
device_info = best_match.get("device", {})
if device_info and isinstance(device_info, dict):
operation_info = device_info.get("operation", {})
if operation_info and isinstance(operation_info, dict):
operation_key = operation_info.get("key", "")
if operation_key:
logger.info(f"从 best_match 获取到操作命令: {operation_key}")
return operation_key
# 如果都没有有效值,返回原始的 tool_operation 或空字符串
logger.warning(f"未找到有效的操作命令,返回原始值: {tool_operation}")
return tool_operation or ""
except Exception as e:
logger.error(f"分析操作结果时发生异常: {str(e)}", exc_info=True)
return ""
def __del__(self):
"""析构函数,关闭会话"""
if hasattr(self, "session"):
self.session.close()
# 使用示例
if __name__ == "__main__":
# 测试代码只在直接运行此模块时执行,不在导入时执行
# 以下代码已注释避免MCP服务器启动时执行
pass
# device_op = DeviceOperator()
# # 测试获取数字员工信息
# employee_id = "1950037223825125378"
# employee_result = device_op.get_digital_employee_by_id(employee_id)
# logger.info(f"数字员工信息: {employee_result}")
# 以下所有测试代码已注释避免MCP服务器启动时执行
# device_op = DeviceOperator()
# # 测试获取数字员工信息
# employee_id = "1950037223825125378"
# employee_result = device_op.get_digital_employee_by_id(employee_id)
# logger.info(f"数字员工信息: {employee_result}")
# # 测试单个设备操作
# data = {
# "enterpriseId": "1932095424144715777",
# "entityId": "climate.qjiang_cn_741479129_wb20",
# "command": "set_temperature",
# "params": {
# "temperature": 24,
# },
# }
# result = device_op.operate_device(data)
# logger.info(f"设备操作结果: {result}")
# # 测试批量设备操作
# batch_data = [
# {
# "enterpriseId": "1932095424144715777",
# "entityId": "switch.office_light_1",
# "command": "turn_on",
# "params": {}
# },
# {
# "enterpriseId": "1932095424144715777",
# "entityId": "climate.office_ac_1",
# "command": "set_temperature",
# "params": {
# "temperature": 25,
# }
# },
# {
# "enterpriseId": "1932095424144715777",
# "entityId": "switch.meeting_room_projector",
# "command": "turn_off",
# "params": {}
# }
# ]
# batch_result = device_op.batch_operate_devices(batch_data)
# logger.info(f"批量设备操作结果: {batch_result}")
# # 输出批量操作统计信息
# if batch_result.get("code") in [200, 206]: # 成功或部分成功
# batch_data_result = batch_result.get("data", {})
# logger.info(f"批量操作统计: 总数={batch_data_result.get('total')}, "
# f"成功={batch_data_result.get('success')}, "
# f"失败={batch_data_result.get('failed')}, "
# f"成功率={batch_data_result.get('success_rate')}%")
# # 测试操作分析结果方法
# # 测试用例1tool_data 包含纯英文操作
# tool_data_en = {
# "location": "office",
# "device": "air_conditioner",
# "operation": "turn_on",
# "keyId": "1932095424144715777"
# }
# best_match_data = {
# "device": {
# "id": "d67d81ef-1c0e-4049-ae03-4581604138fc",
# "location": {
# "key": "37",
# "description": "李总办公室;董事长;李总房间;董事长办公室"
# },
# "device": {
# "key": "climate.qjiang_cn_741478700_wb20",
# "description": "空调;制冷设备"
# },
# "operation": {
# "key": "set_temperature",
# "description": "设置温度;调节温度;"
# },
# "operation_params": []
# },
# "score": 0.9385,
# "confidence": "高"
# }
# result1 = device_op.get_operation_analysis_result(tool_data_en, best_match_data)
# logger.info(f"测试1 - 纯英文操作: {result1}") # 应该返回 "turn_on"
# # 测试用例2tool_data 包含中文操作
# tool_data_cn = {
# "location": "办公室",
# "device": "空调",
# "operation": "打开空调",
# "keyId": "1932095424144715777"
# }
# result2 = device_op.get_operation_analysis_result(tool_data_cn, best_match_data)
# logger.info(f"测试2 - 中文操作: {result2}") # 应该返回 "set_temperature"
# # 测试用例3tool_data 操作为空,使用 best_match
# tool_data_empty = {
# "location": "office",
# "device": "air_conditioner",
# "operation": "",
# "keyId": "1932095424144715777"
# }
# result3 = device_op.get_operation_analysis_result(tool_data_empty, best_match_data)
# logger.info(f"测试3 - 空操作: {result3}") # 应该返回 "set_temperature"

View File

@@ -0,0 +1,659 @@
from typing import List, Dict, Any
from collections import defaultdict
import re
from .logger_config import get_logger
# 延迟初始化日志器,避免在导入时立即执行
logger = None
def _ensure_logger():
"""确保日志器已初始化"""
global logger
if logger is None:
logger = get_logger(__name__)
return logger
def process_device_data(data, enterprise_id):
"""
处理设备数据提取best_match并转换为指定格式
Args:
data (dict): 包含results和best_match的数据结构
enterprise_id (str): 企业ID
Returns:
dict: 处理后的best_match数据包含enterpriseId、entityId和command
如果处理失败返回None
"""
logger = _ensure_logger()
logger.debug(f"开始处理设备数据enterprise_id: {enterprise_id}")
try:
# 参数验证
if not isinstance(data, dict):
error_msg = "data参数必须是字典类型"
logger.error(error_msg)
raise ValueError(error_msg)
if not enterprise_id:
error_msg = "enterprise_id不能为空"
logger.error(error_msg)
raise ValueError(error_msg)
# 检查数据是否包含best_match
if "best_match" not in data:
raise ValueError("数据中缺少best_match字段")
best_match = data["best_match"]
if not isinstance(best_match, dict):
raise ValueError("best_match必须是字典类型")
# 检查best_match是否包含device字段
if "device" not in best_match:
raise ValueError("best_match中缺少device字段")
device_info = best_match["device"]
if not isinstance(device_info, dict):
raise ValueError("device_info必须是字典类型")
# 检查device_info是否包含device和operation字段
if "device" not in device_info:
raise ValueError("device_info中缺少device字段")
if "operation" not in device_info:
raise ValueError("device_info中缺少operation字段")
device_data = device_info["device"]
operation_data = device_info["operation"]
if not isinstance(device_data, dict) or not isinstance(operation_data, dict):
raise ValueError("device和operation字段必须是字典类型")
# 提取device的key作为entityId
device_key = device_data.get("key", "")
if not device_key:
raise ValueError("device.key字段为空")
# 提取operation的key作为command
operation_key = operation_data.get("key", "")
if not operation_key:
raise ValueError("operation.key字段为空")
# 构建返回结果
processed_result = {
"enterpriseId": enterprise_id,
"entityId": device_key,
"command": operation_key,
}
logger.info(f"设备数据处理成功: entityId={device_key}, command={operation_key}")
logger.debug(f"处理结果: {processed_result}")
return processed_result
except Exception as e:
logger.error(f"数据处理错误: {str(e)}", exc_info=True)
return None
def process_device_data_safe(data, enterprise_id, default_result=None):
"""
安全版本的数据处理方法,不会抛出异常
Args:
data (dict): 包含results和best_match的数据结构
enterprise_id (str): 企业ID
default_result (dict): 处理失败时的默认返回值
Returns:
dict: 处理后的best_match数据处理失败时返回default_result
"""
result = process_device_data(data, enterprise_id)
return result if result is not None else default_result
def validate_device_data(data):
"""
验证设备数据结构的完整性
Args:
data (dict): 要验证的数据
Returns:
tuple: (is_valid, error_message)
"""
try:
if not isinstance(data, dict):
return False, "数据必须是字典类型"
if "best_match" not in data:
return False, "缺少best_match字段"
best_match = data["best_match"]
if not isinstance(best_match, dict):
return False, "best_match必须是字典类型"
if "device" not in best_match:
return False, "best_match中缺少device字段"
device_info = best_match["device"]
if not isinstance(device_info, dict):
return False, "device_info必须是字典类型"
required_fields = ["device", "operation"]
for field in required_fields:
if field not in device_info:
return False, f"device_info中缺少{field}字段"
field_data = device_info[field]
if not isinstance(field_data, dict):
return False, f"{field}字段必须是字典类型"
if "key" not in field_data or not field_data["key"]:
return False, f"{field}.key字段不能为空"
return True, "数据验证通过"
except Exception as e:
return False, f"验证过程中发生错误: {str(e)}"
def process_device_results(raw_data: List[List[Any]]) -> List[Dict[str, Any]]:
"""
预处理数据,将原始数据转换为简化的数组对象格式
如果 entityId 相同,则合并到一起
Args:
raw_data: 原始数据,格式为 [[entity, score], ...]
Returns:
处理后的数据数组每个对象包含合并后的信息不包含id和score_details字段
"""
logger = _ensure_logger()
logger.debug(f"开始处理设备结果数据,输入数据条数: {len(raw_data) if raw_data else 0}")
# 使用 entityId 作为分组键
grouped_data = defaultdict(list)
for item in raw_data:
if len(item) >= 2:
entity = item[0]
score = item[1]
logger.debug(f"处理项目: entity类型={type(entity)}, 包含keys={list(entity.keys()) if isinstance(entity, dict) else 'N/A'}")
entityId = entity.get("device", {}).get("key", "")
logger.debug(f"提取的entityId: '{entityId}'")
# 如果entityId为空跳过这个实体
if not entityId:
logger.warning(f"跳过entityId为空的实体: {entity}")
continue
# 创建简化的对象
simplified_entity = {
"location_key": entity.get("location", {}).get("key"),
"location_desc": entity.get("location", {}).get("description"),
"entityId": entity.get("device", {}).get("key"),
"device_desc": entity.get("device", {}).get("description"),
"command": entity.get("operation", {}).get("key"),
"operation_desc": entity.get("operation", {}).get("description"),
"operation_params": entity.get("operation_params", []),
"score": score,
}
logger.debug(f"创建的简化实体: {simplified_entity}")
grouped_data[entityId].append(simplified_entity)
# 合并相同 entityId 的数据
result = []
logger.debug(f"grouped_data包含{len(grouped_data)}个组: {list(grouped_data.keys())}")
for entityId, entities in grouped_data.items():
logger.debug(f"处理entityId '{entityId}'{len(entities)} 个实体")
if len(entities) == 1:
# 只有一个实体,直接添加
result.append(entities[0])
logger.debug(f"设备 {entityId} 只有一个实体,直接添加")
else:
# 多个实体需要合并
merged_entity = merge_entities(entities)
result.append(merged_entity)
logger.debug(f"设备 {entityId}{len(entities)} 个实体,已合并")
logger.info(f"设备结果数据处理完成,输出数据条数: {len(result)}")
logger.debug(f"最终结果: {result}")
return result
def merge_entities(entities: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
合并相同 entityId 的多个实体
Args:
entities: 需要合并的实体列表
Returns:
合并后的实体
"""
if not entities:
return {}
# 选择分数最高的作为主实体
main_entity = max(entities, key=lambda x: x.get("score", 0))
# 收集所有操作
operations = []
for entity in entities:
operation_info = {
"command": entity.get("command"),
"operation_desc": entity.get("operation_desc"),
"operation_params": entity.get("operation_params", []),
"score": entity.get("score", 0),
}
operations.append(operation_info)
# 创建合并后的实体
merged_entity = {
"location_key": main_entity.get("location_key"),
"location_desc": main_entity.get("location_desc"),
"entityId": main_entity.get("entityId"),
"device_desc": main_entity.get("device_desc"),
"operations": operations, # 所有可能的操作
}
return merged_entity
def extract_space_entity_and_location(space_data: Dict[str, Any]) -> Dict[str, Any]:
"""
从空间数据结构中提取 entityId 与 location。
期望输入示例:
{
"spaceId": "46",
"spaceAliasName": "爱易拍展厅;爱一拍展厅;爱一拍展区",
"spaceName": "爱易拍展厅",
...
}
处理规则:
- entityId = spaceId转为字符串
- location 来自 spaceAliasName
- 若包含分隔符(例如 ''';' 等),拆分并取第一个非空别名
- 若不包含分隔符,则直接使用 spaceAliasName
- 若 spaceAliasName 为空,回退到 spaceName仍为空则返回空字符串
Returns:
{"entityId": str, "location": str}
"""
logger = _ensure_logger()
logger.debug(f"开始提取空间实体和位置信息: {space_data}")
# entityId
space_id = space_data.get("spaceId")
entity_id = "" if space_id is None else str(space_id)
logger.debug(f"提取的实体ID: {entity_id}")
# location from alias
alias_raw = space_data.get("spaceAliasName")
alias_text = (
alias_raw
if isinstance(alias_raw, str)
else (str(alias_raw) if alias_raw is not None else "")
)
chosen_location = alias_text.strip()
if chosen_location:
# 使用常见分隔符进行切分:中文分号、英文分号、中文逗号、英文逗号、竖线、斜杠
parts = [
p.strip()
for p in re.split(r"[;,\|/]+", chosen_location)
if p and p.strip()
]
if parts:
chosen_location = parts[0]
else:
# alias 为空则回退到 spaceName
space_name = space_data.get("spaceName")
chosen_location = (
space_name.strip()
if isinstance(space_name, str)
else (str(space_name).strip() if space_name is not None else "")
)
result = {
"entityId": entity_id,
"location": chosen_location,
}
logger.info(f"空间实体和位置提取完成: entityId={entity_id}, location={chosen_location}")
return result
def format_device_for_display(device: Dict[str, Any]) -> str:
"""
将设备数据格式化为易读的描述文本
用于展示给AI和用户提升可读性和理解性
Args:
device: 设备数据字典,可能包含以下字段:
- location_desc: 位置描述
- device_desc: 设备描述
- entityId: 设备唯一标识
- operations: 操作列表(合并后的设备)
- command: 单个命令
- operation_desc: 操作描述
- operation_params: 操作参数
Returns:
格式化后的设备描述文本
"""
logger = _ensure_logger()
logger.debug(f"开始格式化设备显示信息: entityId={device.get('entityId', 'N/A')}")
try:
# 处理位置描述
location = device.get('location_desc', '未知位置')
if ';' in location or '' in location:
parts = re.split(r'[;]', location)
location = ''.join(parts) + '' if len(parts) > 1 else parts[0]
# 处理设备类型描述
device_type = device.get('device_desc', '未知设备')
if ';' in device_type or '' in device_type:
parts = re.split(r'[;]', device_type)
device_type = ''.join(parts) + '' if len(parts) > 1 else parts[0]
entity_id = device.get('entityId', '')
formatted_text = f"""设备核心信息:
位置:{location}
设备类型:{device_type}
唯一标识entityId{entity_id}
可执行操作:"""
# 判断是合并后的设备有operations字段还是单个操作设备
if 'operations' in device:
# 合并后的设备,包含多个操作
operations = device.get('operations', [])
for i, op in enumerate(operations, 1):
cmd = op.get('command', '')
desc = op.get('operation_desc', '').replace(';', ' / ').replace('', ' / ')
formatted_text += f"\n{i}. {desc}"
formatted_text += f"\n 指令:{cmd}"
params = op.get('operation_params', [])
if params:
formatted_text += "\n 参数:"
for param in params:
key = param.get('key', '')
description = param.get('description', '')
value = param.get('value', '')
formatted_text += f"\n - {description}{key}{value}"
else:
formatted_text += "\n 参数:无"
else:
# 单个操作设备
cmd = device.get('command', '')
desc = device.get('operation_desc', '').replace(';', ' / ').replace('', ' / ')
formatted_text += f"\n1. {desc}"
formatted_text += f"\n 指令:{cmd}"
params = device.get('operation_params', [])
if params:
formatted_text += "\n 参数:"
for param in params:
key = param.get('key', '')
description = param.get('description', '')
value = param.get('value', '')
formatted_text += f"\n - {description}{key}{value}"
else:
formatted_text += "\n 参数:无"
logger.debug(f"设备显示信息格式化完成: entityId={entity_id}")
return formatted_text
except Exception as e:
logger.error(f"格式化设备显示信息时出错: {str(e)}", exc_info=True)
return f"设备信息格式化失败: {str(e)}"
def format_devices_list(devices: List[Dict[str, Any]]) -> str:
"""
格式化多个设备信息为易读的文本列表
Args:
devices: 设备数据列表
Returns:
格式化后的设备列表文本
"""
logger = _ensure_logger()
logger.debug(f"开始格式化设备列表,设备数量: {len(devices) if devices else 0}")
if not devices:
logger.warning("设备列表为空")
return "未找到相关设备"
try:
formatted_list = []
formatted_list.append(f"共找到 {len(devices)} 个匹配的设备\n")
for idx, device in enumerate(devices, 1):
if idx > 1:
formatted_list.append("\n" + "-" * 40 + "\n")
formatted_list.append(f"【设备 {idx}")
formatted_list.append(format_device_for_display(device))
result = "\n".join(formatted_list)
logger.info(f"设备列表格式化完成,设备数量: {len(devices)}")
return result
except Exception as e:
logger.error(f"格式化设备列表时出错: {str(e)}", exc_info=True)
return f"设备列表格式化失败: {str(e)}"
def format_device_for_display_simple(device: Dict[str, Any]) -> str:
"""
将设备数据格式化为简化的描述文本(无装饰符号版本)
适用于需要更简洁输出的场景
Args:
device: 设备数据字典
Returns:
格式化后的简化设备描述文本
"""
logger = _ensure_logger()
logger.debug(f"开始格式化设备简化显示信息: entityId={device.get('entityId', 'N/A')}")
try:
# 处理位置描述
location = device.get('location_desc', '未知位置')
if ';' in location or '' in location:
parts = re.split(r'[;]', location)
location = ''.join(parts) + '' if len(parts) > 1 else parts[0]
# 处理设备类型描述
device_type = device.get('device_desc', '未知设备')
if ';' in device_type or '' in device_type:
parts = re.split(r'[;]', device_type)
device_type = ''.join(parts) + '' if len(parts) > 1 else parts[0]
entity_id = device.get('entityId', '')
formatted_text = f"""设备核心信息:
• 位置:{location}
• 设备类型:{device_type}
• 唯一标识entityId{entity_id}
可执行操作:"""
# 判断是合并后的设备还是单个操作设备
if 'operations' in device:
operations = device.get('operations', [])
for i, op in enumerate(operations, 1):
cmd = op.get('command', '')
desc = op.get('operation_desc', '').replace(';', ' / ').replace('', ' / ')
formatted_text += f"\n{i}. {desc}"
formatted_text += f"\n 指令:{cmd}"
params = op.get('operation_params', [])
if params:
formatted_text += "\n 参数:"
for param in params:
key = param.get('key', '')
description = param.get('description', '')
value = param.get('value', '')
formatted_text += f"\n - {description}{key}{value}"
else:
formatted_text += "\n 参数:无"
else:
cmd = device.get('command', '')
desc = device.get('operation_desc', '').replace(';', ' / ').replace('', ' / ')
formatted_text += f"\n1. {desc}"
formatted_text += f"\n 指令:{cmd}"
params = device.get('operation_params', [])
if params:
formatted_text += "\n 参数:"
for param in params:
key = param.get('key', '')
description = param.get('description', '')
value = param.get('value', '')
formatted_text += f"\n - {description}{key}{value}"
else:
formatted_text += "\n 参数:无"
logger.debug(f"设备简化显示信息格式化完成: entityId={entity_id}")
return formatted_text
except Exception as e:
logger.error(f"格式化设备简化显示信息时出错: {str(e)}", exc_info=True)
return f"设备信息格式化失败: {str(e)}"
# 测试函数
def test_process_device_data():
"""测试数据处理函数"""
logger = _ensure_logger()
# 模拟输入数据
test_data = {
"results": [
[
{
"id": "32875964-00df-4b34-8e33-c47a722b8f7f",
"location": {"key": "lzwc", "description": "灵泽联创中心"},
"device": {
"key": "switch.zimi_cn_1144138206_dhkg01_on_p_2_1",
"description": "灵泽办公区过道吊灯 开关 按键",
},
"operation": {"key": "turn_on", "description": "开关"},
"operation_params": [],
},
0.34563739142066174,
{
"exact_match": 0.25,
"word2vec_similarity": 0.5346393585205078,
"doc2vec_similarity": 0.004867201205343008,
"tfidf_similarity": 0.8217383432613271,
"combined_score": 0.34563739142066174,
},
]
],
"count": 1,
"best_match": {
"device": {
"id": "32875964-00df-4b34-8e33-c47a722b8f7f",
"location": {"key": "lzwc", "description": "灵泽联创中心"},
"device": {
"key": "switch.zimi_cn_1144138206_dhkg01_on_p_2_1",
"description": "灵泽办公区过道吊灯 开关 按键",
},
"operation": {"key": "turn_on", "description": "开关"},
"operation_params": [],
},
"score": 0.34563739142066174,
"confidence": "",
},
}
enterprise_id = "1932095424144715777"
logger.info("=== 测试数据处理功能 ===")
# 1. 测试数据验证
logger.info("1. 数据验证测试:")
is_valid, message = validate_device_data(test_data)
logger.info(f"数据验证结果: {is_valid}, 消息: {message}")
# 2. 测试基本处理功能
logger.info("2. 基本处理功能测试:")
result = process_device_data(test_data, enterprise_id)
logger.info(f"处理结果: {result}")
# 3. 测试安全版本
logger.info("3. 安全版本测试:")
safe_result = process_device_data_safe(
test_data, enterprise_id, {"error": "处理失败"}
)
logger.info(f"安全处理结果: {safe_result}")
# 4. 测试错误情况
logger.info("4. 错误情况测试:")
# 测试空数据
empty_result = process_device_data_safe({}, enterprise_id, {"error": "数据为空"})
logger.info(f"空数据处理结果: {empty_result}")
# 测试缺少字段的数据
invalid_data = {"best_match": {"device": {}}}
invalid_result = process_device_data_safe(
invalid_data, enterprise_id, {"error": "数据格式错误"}
)
logger.info(f"无效数据处理结果: {invalid_result}")
return result
def test_process_device_results():
import json
logger = _ensure_logger()
# Load JSON data from file
json_file_path = r"E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcp_iot\deviot.json"
with open(json_file_path, 'r', encoding='utf-8') as f:
raw_data = json.load(f)
logger.info('Raw data loaded, processing...')
result = process_device_results(raw_data)
logger.info('Processed result:')
for i, item in enumerate(result):
logger.info(f"Item {i+1}: {item}")
# Save results to output file
output_file_path = r"E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcp_iot\output\output.json"
with open(output_file_path, 'w', encoding='utf-8') as f:
json.dump(result, f, ensure_ascii=False, indent=2)
logger.info(f'Results saved to: {output_file_path}')
return result
if __name__ == "__main__":
# 测试代码只在直接运行此模块时执行,不在导入时执行
# test_process_device_data()
# test_process_device_results() # 注释掉测试代码避免MCP服务器启动时执行
pass

View File

@@ -0,0 +1,131 @@
"""
MCP服务器初始化模块
该模块提供了MCP服务器的初始化功能包括企业ID获取和向量库初始化。
"""
from typing import Optional, Dict, Any
from .device_operations import DeviceOperator
from .vector_service import VectorService
from .logger_config import get_logger
# 延迟初始化日志器,避免在导入时立即执行
logger = None
def _ensure_logger():
"""确保日志器已初始化"""
global logger
if logger is None:
logger = get_logger(__name__)
return logger
def init_mcp_server(
device_op: DeviceOperator,
vector_service: VectorService,
enterprise_id: str
) -> Dict[str, Any]:
"""
初始化MCP服务器
该函数执行以下步骤:
1. 验证企业ID
2. 检查向量库状态
3. 如果向量库不存在,则创建向量库
参数:
device_op: 设备操作实例
vector_service: 向量服务实例
enterprise_id: 企业ID从环境变量获取
返回:
Dict[str, Any]: 初始化结果,包含状态码、消息和数据
"""
logger = _ensure_logger()
try:
# 输入参数验证
if not enterprise_id or not isinstance(enterprise_id, str):
error_msg = f"无效的企业ID: {enterprise_id}"
logger.error(error_msg)
return {
"code": 400,
"msg": error_msg,
"data": None
}
if not isinstance(device_op, DeviceOperator):
error_msg = "device_op参数必须是DeviceOperator实例"
logger.error(error_msg)
return {
"code": 400,
"msg": error_msg,
"data": None
}
if not isinstance(vector_service, VectorService):
error_msg = "vector_service参数必须是VectorService实例"
logger.error(error_msg)
return {
"code": 400,
"msg": error_msg,
"data": None
}
logger.info(f"开始初始化MCP服务器企业ID: {enterprise_id}")
# 第一步:检查向量库状态
logger.info("第一步:检查向量库状态...")
vector_store_exists = vector_service.check_vector_store_status(enterprise_id)
logger.info(f"向量库状态检查完成,存在状态: {vector_store_exists}")
# 如果向量库不存在,则创建向量库
if not vector_store_exists:
logger.info("向量库不存在,开始创建向量库...")
init_result = vector_service.init_vector_store(keyId=enterprise_id)
# 检查初始化结果
if init_result.get("status") == "error":
error_msg = f"向量库初始化失败: {init_result.get('message')}"
logger.error(error_msg)
return {
"code": 500,
"msg": error_msg,
"data": {
"enterprise_id": enterprise_id,
"vector_store_created": False,
"init_result": init_result
}
}
else:
logger.info("向量库创建成功")
return {
"code": 200,
"msg": "MCP服务器初始化成功向量库已创建",
"data": {
"enterprise_id": enterprise_id,
"vector_store_created": True,
"vector_store_existed": False,
"init_result": init_result
}
}
else:
logger.info("向量库已存在,无需创建")
return {
"code": 200,
"msg": "MCP服务器初始化成功向量库已存在",
"data": {
"enterprise_id": enterprise_id,
"vector_store_created": False,
"vector_store_existed": True
}
}
except Exception as e:
error_msg = f"MCP服务器初始化过程中发生异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {
"code": 500,
"msg": error_msg,
"data": None
}

View File

@@ -0,0 +1,216 @@
def iot_device_precise_controller_prompt():
return """
设备精准控制
【功能描述】
用来精确控制特定设备的工具通过设备的唯一ID直接操作特别准确。
这个工具只负责"控制设备",但它需要先通过"设备信息查询"工具获取设备信息才能使用。
【日常使用场景】
在使用"设备信息查询"工具找到设备后的控制操作:
- "控制那个具体的灯"
- "操作刚才查到的灯"
- "开一下那台空调"
- "把刚才找到的空调打开"
- "关掉那个特定的插座"
- "控制指定的插座"
- "操作那个设备"
- "控制具体设备"
跟进操作(基于之前的查询结果):
- "就是这个设备,帮我开一下"
- "用这个ID控制设备"
- "按照之前查到的信息操作"
【工作方式】
这个工具必须和"设备信息查询"工具配合使用:
1. 第一步:用"设备信息查询"工具找设备
2. 第二步从查询结果中获取设备ID和控制命令
3. 第三步:用这个工具执行精确控制
【工具配合关系】
就像"先查电话号码,再打电话"
- 设备信息查询 = 查电话号码(找到设备和操作方法)
- 设备精准控制 = 打电话(实际执行控制)
【重要】
这个工具不能单独使用必须先用查询iot_get_devices_by_location或者iot_get_all_spaces_and_devices工具获取设备信息不能随便填写参数。
【参数说明】
- entityId (必填): 设备唯一ID
* 格式示例: "switch.zimi_cn_1144138206_dhkg01_on_p_2_1"
* 来源: 设备查询工具返回结果中的设备ID
- command (必填): 操作命令
* 格式示例: "turn_on", "turn_off", "set_temperature", "set_brightness"
* 来源: 设备查询工具返回结果中的命令
- params (可选): 操作参数
* 格式: 包含操作所需参数的数据
* 来源: 设备查询工具返回结果中的参数
* 示例: {"temperature": 24} 用于空调温度设置
* 示例: {"brightness": 80} 用于灯光亮度调节
【返回结果】
返回设备控制操作结果,包括是否成功、设备反馈信息和详细结果。
"""
def iot_get_devices_by_location_prompt():
return """
根据位置获取设备列表
【功能描述】
根据指定的位置/房间,获取该位置下的所有智能设备列表。
这个工具专门用来查看某个位置有哪些设备,会返回该位置的完整设备清单。
【日常使用场景】
查看位置设备:
- "办公室有哪些设备"
- "会议室有什么设备"
- "看看客厅的设备"
- "前台有什么智能设备"
- "财务部的设备列表"
- "查看卧室的所有设备"
设备清单查询:
- "列出办公室所有设备"
- "显示会议室设备清单"
- "办公室设备有哪些"
- "会议室装了什么设备"
- "客厅都有什么智能设备"
位置设备统计:
- "办公室一共有多少设备"
- "会议室有几个设备"
- "客厅设备数量"
- "前台设备统计"
简单查询:
- "办公室设备"
- "会议室设备列表"
- "客厅的设备"
- "设备清单"
【注意】
- 这个工具只返回设备列表信息,不会控制设备
- 返回结果包含设备名称、类型、状态等详细信息
- 如果需要控制设备,可以使用返回结果中的设备信息配合其他控制工具
【参数说明】
- location (必填): 要查询的位置/房间名称
* 示例:办公室、会议室、客厅、前台、财务部等
* 参数值不得包含任何标点符号(如逗号、句号、感叹号等)
【返回结果】
返回指定位置的所有设备列表,包括:
- 设备数量统计
- 每个设备的详细信息设备ID、名称、类型、状态等
- 设备的当前状态(开启/关闭等)
- 设备支持的控制命令
【典型使用流程】
1. 指定要查询的位置
2. 获取该位置的所有设备列表
3. 查看设备详细信息
4. (可选)根据设备信息进行后续控制操作
"""
def iot_get_all_spaces_and_devices_prompt():
return """
获取所有空间位置信息
【功能描述】
获取系统中所有空间位置的列表。
这个工具只返回空间名称清单,不包含设备详情,适合快速了解有哪些可用空间。
【日常使用场景】
查看所有空间:
- "显示所有空间"
- "有哪些空间"
- "列出所有房间"
- "查看全部区域"
- "空间列表"
- "所有位置"
- "有什么位置"
- "可用的空间有哪些"
空间清单查询:
- "系统里有几个空间"
- "一共有多少个位置"
- "空间总览"
- "位置总览"
- "房间列表"
简单查询:
- "所有空间"
- "空间总览"
- "位置列表"
- "全部位置"
【注意】
- 这个工具只返回空间名称列表,不包含设备信息
- 不需要传入任何参数
- 如果需要查看某个空间的设备,请使用"根据位置获取设备列表"工具
- 适合在控制设备前先了解有哪些可用空间
【参数说明】
- 无需参数(自动获取所有空间位置)
【返回结果】
返回所有空间位置列表,包括:
- 空间总数统计
- 所有空间名称的列表
【典型使用流程】
1. 调用工具获取所有空间列表
2. 查看有哪些可用空间
3. (可选)选择特定空间,使用"根据位置获取设备列表"工具查看该空间的设备
4. (可选)根据设备信息进行设备控制操作
"""
def smart_space_device_locator_matcher_prompt():
return """
位置查询
【功能描述】
帮你查看自己现在在哪个位置/空间,会根据你的用户信息自动识别。
【日常使用场景】
查看位置:
- "我现在在哪"
- "我在哪个位置"
- "我现在在哪个空间"
- "当前位置是什么"
- "我属于哪个地方"
- "我在哪个区域"
确认空间:
- "当前默认空间是什么"
- "我的默认位置"
- "系统识别我在哪里"
- "定位我的位置"
- "我的空间信息"
- "位置信息"
设备控制前确认:
- "先看看我在哪"
- "确认一下位置"
- "我在这个位置吗"
- "位置对不对"
【注意】
这个工具只是查看你的位置信息,不会控制任何设备。
【参数说明】
- userId (必填): 用户ID自动提供
【返回结果】
返回用户所属空间:
- spaceAliasName: 推断出的空间名称
【使用说明】
- 本工具仅做查询与定位,不执行设备控制
"""

View File

@@ -0,0 +1,558 @@
"""
统一日志配置模块
这个模块提供了整个项目的统一日志配置和管理功能,确保所有组件使用一致的日志格式和输出方式。
主要功能:
1. 统一的日志格式配置
2. 支持控制台和文件双重输出
3. 日志文件轮转管理
4. MCP模式下的特殊处理禁用控制台输出
5. 便捷的日志器获取接口
6. 丰富的日志工具函数
设计特点:
- 单例模式确保配置一致性
- 支持动态配置调整
- 异常安全的编码处理
- 详细的调试信息记录
作者: lzwcai
版本: 1.0.0
"""
import logging
import logging.handlers
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
class LoggerConfig:
"""
日志配置管理器
这个类采用单例模式管理整个项目的日志配置。
它提供了统一的日志格式、文件轮转、编码处理等功能。
主要特性:
- 单例模式:确保全局日志配置一致
- 双重输出:同时支持控制台和文件输出
- 文件轮转:自动管理日志文件大小和数量
- 编码安全:正确处理中文字符
- MCP兼容支持MCP模式下的特殊需求
配置参数:
DEFAULT_LOG_LEVEL: 默认日志级别INFO
DEFAULT_LOG_FORMAT: 日志格式模板
DEFAULT_DATE_FORMAT: 时间格式
LOG_FILE_NAME: 日志文件名
MAX_LOG_SIZE: 单个日志文件最大大小10MB
BACKUP_COUNT: 保留的备份文件数量5个
"""
# ==================== 默认配置常量 ====================
# 默认日志级别INFO级别平衡了信息量和性能
DEFAULT_LOG_LEVEL = logging.INFO
# 默认日志格式:包含时间、模块名、级别、文件位置、消息内容
DEFAULT_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s"
# 默认时间格式:标准的年-月-日 时:分:秒格式
DEFAULT_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
# ==================== 日志文件配置 ====================
# 日志文件名:使用项目名称作为前缀
LOG_FILE_NAME = "lzwcai_mcp_iot.log"
# 单个日志文件最大大小10MB
MAX_LOG_SIZE = 10 * 1024 * 1024 # 10MB
# 保留的备份文件数量5个总共约50MB的日志存储
BACKUP_COUNT = 5
# ==================== 单例模式状态 ====================
# 初始化标志:确保只初始化一次
_initialized = False
# 日志文件路径:记录当前使用的日志文件路径
_log_file_path = None
@classmethod
def setup_logging(
cls,
log_level: int = DEFAULT_LOG_LEVEL,
log_file: Optional[str] = None,
console_output: bool = True,
file_output: bool = True
) -> str:
"""
设置项目统一日志配置
这是日志系统的核心初始化方法,负责配置整个项目的日志输出。
采用单例模式,确保在整个应用生命周期中只初始化一次。
配置流程:
1. 检查是否已经初始化(单例模式)
2. 确定日志文件路径(自动或手动指定)
3. 创建必要的目录结构
4. 配置根日志器和处理器
5. 设置日志格式化器
6. 添加控制台和文件处理器
7. 记录初始化信息
特殊处理:
- MCP模式下通常禁用控制台输出避免干扰stdio通信
- Windows系统下的UTF-8编码处理
- 日志文件的自动轮转管理
参数:
log_level: 日志级别DEBUG, INFO, WARNING, ERROR, CRITICAL
log_file: 日志文件路径None时使用默认路径
console_output: 是否输出到控制台MCP模式下通常为False
file_output: 是否输出到文件通常为True
返回:
str: 实际使用的日志文件路径
注意事项:
- 这个方法是线程安全的
- 重复调用会直接返回已配置的路径
- 日志文件会自动创建必要的目录
"""
# 单例模式检查:如果已经初始化,直接返回
if cls._initialized:
return cls._log_file_path
# ==================== 日志文件路径配置 ====================
if log_file is None:
# 自动确定日志文件路径:项目根目录 + 默认文件名
project_root = cls._get_project_root()
log_file = project_root / cls.LOG_FILE_NAME
else:
# 使用指定的日志文件路径
log_file = Path(log_file)
# 确保日志目录存在(递归创建)
log_file.parent.mkdir(parents=True, exist_ok=True)
cls._log_file_path = str(log_file)
# ==================== 包日志器配置 ====================
# 获取包的顶层日志器,而不是根日志器
package_logger = logging.getLogger('lzwcai_mcp_iot')
package_logger.setLevel(log_level)
# 作为库,不应该清除宿主应用的任何处理器
# 也不应该让日志消息向上传播到根日志器,以免重复打印
package_logger.propagate = False
# 清除此日志器上现有的处理器,避免重复配置
for handler in package_logger.handlers[:]:
package_logger.removeHandler(handler)
# ==================== 日志格式化器 ====================
# 创建统一的日志格式化器
formatter = logging.Formatter(
fmt=cls.DEFAULT_LOG_FORMAT, # 日志格式模板
datefmt=cls.DEFAULT_DATE_FORMAT # 时间格式
)
# ==================== 控制台处理器配置 ====================
if console_output:
# 控制台输出处理器支持彩色输出和UTF-8编码
import io
# 处理Windows系统的编码问题
if hasattr(sys.stdout, 'buffer'):
# 在Windows上强制使用UTF-8编码避免中文乱码
# errors='replace'确保即使有编码问题也不会崩溃
console_stream = io.TextIOWrapper(
sys.stdout.buffer,
encoding='utf-8',
errors='replace'
)
else:
# Unix/Linux系统通常默认支持UTF-8
console_stream = sys.stdout
# 创建控制台处理器
console_handler = logging.StreamHandler(console_stream)
console_handler.setLevel(log_level)
console_handler.setFormatter(formatter)
package_logger.addHandler(console_handler)
# ==================== 文件处理器配置 ====================
if file_output:
# 文件输出处理器,支持自动轮转
file_handler = logging.handlers.RotatingFileHandler(
filename=cls._log_file_path, # 日志文件路径
maxBytes=cls.MAX_LOG_SIZE, # 单文件最大大小
backupCount=cls.BACKUP_COUNT, # 备份文件数量
encoding='utf-8' # 文件编码
)
file_handler.setLevel(log_level)
file_handler.setFormatter(formatter)
package_logger.addHandler(file_handler)
# ==================== 初始化完成标记 ====================
# 标记为已初始化,防止重复配置
cls._initialized = True
# ==================== 记录初始化信息 ====================
# 使用包日志器记录初始化信息,避免向上传播到根日志器
if file_output: # 只有在文件输出启用时才记录初始化信息
package_logger.info("=" * 80)
package_logger.info(f"日志系统初始化完成 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
package_logger.info(f"日志级别: {logging.getLevelName(log_level)}")
package_logger.info(f"日志文件: {cls._log_file_path}")
package_logger.info(f"控制台输出: {console_output}")
package_logger.info(f"文件输出: {file_output}")
package_logger.info(f"文件轮转: 最大{cls.MAX_LOG_SIZE // (1024*1024)}MB, 保留{cls.BACKUP_COUNT}个备份")
package_logger.info("=" * 80)
return cls._log_file_path
@classmethod
def _get_project_root(cls) -> Path:
"""
获取项目根目录
这个方法通过向上遍历目录树来查找项目根目录。
它会寻找常见的项目标识文件来确定根目录位置。
查找策略:
1. 从当前文件所在目录开始向上查找
2. 寻找项目标识文件pyproject.toml, setup.py, main.py
3. 找到任一标识文件的目录即为项目根目录
4. 如果都找不到,使用当前文件的上级目录作为备选
返回:
Path: 项目根目录的路径对象
注意事项:
- 这个方法假设项目结构相对标准
- 在特殊的部署环境中可能需要调整
- 备选方案确保总是返回有效路径
"""
# 从当前文件向上查找项目根目录
current_path = Path(__file__).parent
# 向上遍历目录树
while current_path.parent != current_path: # 避免到达文件系统根目录
# 检查常见的项目标识文件
if (current_path / "pyproject.toml").exists() or \
(current_path / "setup.py").exists() or \
(current_path / "main.py").exists():
return current_path
current_path = current_path.parent
# 备选方案:如果找不到标识文件,使用预设的相对路径
# 这个路径基于当前的项目结构util -> src -> 项目根
return Path(__file__).parent.parent.parent
@classmethod
def get_logger(cls, name: str) -> logging.Logger:
"""
获取配置好的日志器
这是获取日志器的标准方法,确保返回的日志器使用统一的配置。
如果日志系统尚未初始化,会自动进行初始化。
参数:
name: 日志器名称,通常使用模块的 __name__ 变量
返回:
logging.Logger: 配置好的日志器实例
使用示例:
logger = LoggerConfig.get_logger(__name__)
logger.info("这是一条信息日志")
特性:
- 自动初始化:首次调用时自动配置日志系统
- 层次化命名支持Python日志器的层次化命名
- 统一配置:所有日志器使用相同的格式和输出配置
"""
# 检查是否已初始化未初始化则使用MCP安全的默认配置初始化
if not cls._initialized:
# MCP模式下默认禁用控制台输出只使用文件输出
cls.setup_logging(console_output=False, file_output=True)
# 返回指定名称的日志器
return logging.getLogger(name)
# ==================== 日志工具方法 ====================
@classmethod
def log_function_entry(cls, logger: logging.Logger, func_name: str, **kwargs):
"""
记录函数入口日志
用于调试和性能分析,记录函数被调用时的参数信息。
通常在DEBUG级别输出不会影响生产环境的性能。
参数:
logger: 日志器实例
func_name: 函数名称
**kwargs: 函数参数(键值对形式)
使用示例:
LoggerConfig.log_function_entry(logger, "process_data", user_id=123, action="login")
"""
args_str = ", ".join([f"{k}={v}" for k, v in kwargs.items()])
logger.debug(f"进入函数 {func_name}({args_str})")
@classmethod
def log_function_exit(cls, logger: logging.Logger, func_name: str, result=None):
"""
记录函数出口日志
与log_function_entry配对使用记录函数执行完成和返回值。
有助于跟踪函数执行流程和调试返回值问题。
参数:
logger: 日志器实例
func_name: 函数名称
result: 函数返回值(可选)
使用示例:
LoggerConfig.log_function_exit(logger, "process_data", result={"status": "success"})
"""
if result is not None:
logger.debug(f"退出函数 {func_name},返回值: {result}")
else:
logger.debug(f"退出函数 {func_name}")
@classmethod
def log_api_request(cls, logger: logging.Logger, method: str, url: str, **kwargs):
"""
记录API请求日志
标准化API请求的日志记录包含HTTP方法、URL和请求参数。
有助于API调用的监控和调试。
参数:
logger: 日志器实例
method: HTTP方法GET, POST, PUT, DELETE等
url: 请求URL
**kwargs: 请求参数(可选)
使用示例:
LoggerConfig.log_api_request(logger, "POST", "https://api.example.com/users",
headers={"Authorization": "Bearer xxx"})
"""
logger.info(f"API请求 - {method} {url}")
if kwargs:
logger.debug(f"请求参数: {kwargs}")
@classmethod
def log_api_response(cls, logger: logging.Logger, status_code: int, response_time: float = None):
"""
记录API响应日志
记录API响应的状态码和响应时间用于性能监控和问题诊断。
参数:
logger: 日志器实例
status_code: HTTP状态码
response_time: 响应时间(秒,可选)
使用示例:
LoggerConfig.log_api_response(logger, 200, 0.156)
"""
if response_time:
logger.info(f"API响应 - 状态码: {status_code}, 响应时间: {response_time:.3f}s")
else:
logger.info(f"API响应 - 状态码: {status_code}")
@classmethod
def log_error_with_context(cls, logger: logging.Logger, error: Exception, context: str = ""):
"""
记录带上下文的错误日志
提供丰富的错误信息记录,包含异常类型、错误消息、上下文信息和详细堆栈。
这是错误处理的标准方法。
参数:
logger: 日志器实例
error: 异常对象
context: 错误发生的上下文描述(可选)
使用示例:
try:
risky_operation()
except Exception as e:
LoggerConfig.log_error_with_context(logger, e, "处理用户请求时")
"""
if context:
logger.error(f"错误发生在 {context}: {type(error).__name__}: {str(error)}")
else:
logger.error(f"错误: {type(error).__name__}: {str(error)}")
# 记录详细的异常堆栈信息仅在DEBUG级别显示
logger.debug("错误详情:", exc_info=True)
# ==================== 便捷函数 ====================
def get_logger(name: str) -> logging.Logger:
"""
获取日志器的便捷函数
这是LoggerConfig.get_logger的简化版本提供更简洁的调用方式。
推荐在模块级别使用这个函数获取日志器。
参数:
name: 日志器名称,通常使用 __name__
返回:
logging.Logger: 配置好的日志器实例
使用示例:
logger = get_logger(__name__)
"""
return LoggerConfig.get_logger(name)
def setup_logging(**kwargs) -> str:
"""
设置日志的便捷函数
这是LoggerConfig.setup_logging的简化版本支持所有相同的参数。
参数:
**kwargs: 传递给LoggerConfig.setup_logging的所有参数
返回:
str: 日志文件路径
使用示例:
log_file = setup_logging(log_level=logging.DEBUG, console_output=False)
"""
return LoggerConfig.setup_logging(**kwargs)
# ==================== 装饰器 ====================
def log_function_calls(logger: Optional[logging.Logger] = None):
"""
函数调用日志装饰器
这个装饰器自动记录函数的调用和返回,包括参数和返回值。
主要用于调试和性能分析在生产环境中通常设置为DEBUG级别。
特性:
- 自动记录函数入口和出口
- 记录函数参数kwargs
- 记录返回值
- 自动处理异常并记录错误上下文
- 支持自定义日志器或自动获取
参数:
logger: 可选的日志器实例None时自动获取函数所在模块的日志器
返回:
装饰器函数
使用示例:
@log_function_calls()
def process_user_data(user_id, action="login"):
# 函数实现
return {"status": "success"}
# 或者指定日志器
@log_function_calls(logger=my_logger)
def another_function():
pass
注意事项:
- 会记录所有kwargs参数注意不要记录敏感信息
- 返回值也会被记录,大对象可能影响性能
- 异常会被重新抛出,不会被吞掉
"""
def decorator(func):
def wrapper(*args, **kwargs):
nonlocal logger
# 如果没有提供日志器,自动获取函数所在模块的日志器
if logger is None:
logger = get_logger(func.__module__)
func_name = func.__name__
# 记录函数入口只记录kwargs避免记录过多信息
LoggerConfig.log_function_entry(logger, func_name, **kwargs)
try:
# 执行原函数
result = func(*args, **kwargs)
# 记录函数出口和返回值
LoggerConfig.log_function_exit(logger, func_name, result)
return result
except Exception as e:
# 记录异常信息并重新抛出
LoggerConfig.log_error_with_context(logger, e, f"函数 {func_name}")
raise
return wrapper
return decorator
# ==================== 测试代码 ====================
if __name__ == "__main__":
"""
日志配置测试代码
这个测试代码演示了日志系统的基本功能,包括:
1. 日志系统初始化
2. 不同级别的日志输出
3. 日志文件路径获取
4. 装饰器功能测试
运行方式:
python -m lzwcai_mcp_iot.src.util.logger_config
"""
# 测试代码只在直接运行此模块时执行,不在导入时执行
# 以下代码已注释避免MCP服务器启动时执行
pass
# # 初始化日志系统DEBUG级别同时输出到控制台和文件
# log_file = setup_logging(log_level=logging.DEBUG)
# logger = get_logger(__name__)
# logger.info("开始测试日志配置...")
# # 测试不同级别的日志输出
# logger.debug("这是一个调试消息 - 用于开发调试")
# logger.info("这是一个信息消息 - 记录重要信息")
# logger.warning("这是一个警告消息 - 提醒注意事项")
# logger.error("这是一个错误消息 - 记录错误情况")
# # 测试工具方法
# LoggerConfig.log_api_request(logger, "GET", "https://api.example.com/test")
# LoggerConfig.log_api_response(logger, 200, 0.123)
# # 测试装饰器
# @log_function_calls()
# def test_function(param1, param2="default"):
# """测试函数"""
# return {"result": "success", "param1": param1}
# # 调用测试函数
# result = test_function("test_value", param2="custom")
# # 输出日志文件位置
# logger.info(f"日志文件位置: {log_file}")
# logger.info("日志配置测试完成!")

View File

@@ -0,0 +1,733 @@
"""
向量服务模块
该模块提供了向量数据的增删改查操作接口。
"""
import requests
import json
from typing import Dict, Any, Optional, List
from ..config import CONFIG
from .logger_config import get_logger
# 延迟初始化日志器,避免在导入时立即执行
logger = None
def _ensure_logger():
"""确保日志器已初始化"""
global logger
if logger is None:
logger = get_logger(__name__)
return logger
class VectorService:
"""向量服务类,提供向量数据的增删改查功能"""
def __init__(self, base_url: str = None):
"""
初始化向量服务
参数:
base_url: API服务的基础URL
"""
logger = _ensure_logger()
self.base_url = base_url or CONFIG["vector_api_base_url"]
self.session = requests.Session()
# 设置默认超时
self.session.timeout = CONFIG["request_timeout"]
logger.info(f"VectorService初始化完成API基础URL: {self.base_url}")
def init_vector_store(
self,
keyId: str,
) -> Dict[str, Any]:
"""
调用初始化向量存储API接口
参数:
keyId: 企业或项目的唯一标识符
返回:
Dict[str, Any]: API返回的响应数据
"""
if not keyId or not isinstance(keyId, str):
error_msg = "keyId参数是必需的且必须是字符串"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
endpoint = f"{self.base_url}/iot/companies/recreate"
# 构建请求数据
payload = {
"company_id": keyId,
"company_name": "公司名称",
"enterprise_id": keyId,
"description": "公司描述",
}
# 设置请求头
headers = {"Content-Type": "application/json"}
try:
logger.info(f"初始化向量库keyId: {keyId}")
# 发送POST请求
response = self.session.post(
endpoint,
headers=headers,
data=json.dumps(payload),
timeout=CONFIG["request_timeout"],
)
# 检查响应状态
response.raise_for_status()
# 返回JSON响应
result = response.json()
logger.info(f"向量库初始化成功: {result}")
return result
except requests.exceptions.Timeout:
error_msg = f"向量库初始化请求超时keyId: {keyId}"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
except requests.RequestException as e:
error_msg = f"向量库初始化API请求失败: {str(e)}"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
except Exception as e:
error_msg = f"向量库初始化发生未知异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {
"status": "error",
"message": error_msg,
"data": None,
}
def delete_vector(self, keyId: str) -> Dict[str, Any]:
"""
删除向量数据(模拟实现)
参数:
keyId: 要删除的向量唯一标识
返回:
Dict[str, Any]: 包含删除结果的字典
"""
if not keyId or not isinstance(keyId, str):
error_msg = "keyId参数是必需的且必须是字符串"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
logger.info(f"删除向量数据keyId: {keyId}")
# 模拟删除向量的结果
return {
"status": "success",
"message": "向量删除成功",
"data": {"keyId": keyId, "deleted_at": "2023-07-15T11:45:00Z"},
}
def update_vector(
self,
keyId: str,
) -> Dict[str, Any]:
"""
更新向量数据
参数:
keyId: 要更新的向量唯一标识
返回:
Dict[str, Any]: 包含更新结果的字典
"""
if not keyId or not isinstance(keyId, str):
error_msg = "keyId参数是必需的且必须是字符串"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
endpoint = f"{self.base_url}/iot/vector-store/update"
# 构建请求数据
payload = {"keyId": keyId}
# 设置请求头
headers = {"Content-Type": "application/json"}
try:
logger.info(f"更新向量数据keyId: {keyId}")
# 发送POST请求
response = self.session.post(
endpoint,
headers=headers,
data=json.dumps(payload),
timeout=CONFIG["request_timeout"],
)
# 检查响应状态
response.raise_for_status()
# 返回JSON响应
result = response.json()
logger.info(f"向量数据更新成功: {result}")
return result
except requests.exceptions.Timeout:
error_msg = f"向量数据更新请求超时keyId: {keyId}"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
except requests.RequestException as e:
error_msg = f"更新向量数据失败: {str(e)}"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
except Exception as e:
error_msg = f"更新向量数据发生未知异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {
"status": "error",
"message": error_msg,
"data": None,
}
def query_vector(
self,
keyId: str,
device: Optional[str] = None,
location: Optional[str] = None,
operation: Optional[str] = None,
operation_param: Optional[str] = None,
) -> Dict[str, Any]:
"""
查询向量数据
参数:
keyId: 企业或项目的唯一标识符
device: 设备名称(可选)
location: 设备位置(可选)
operation: 操作类型(可选)
operation_param: 操作参数(可选)
返回:
Dict[str, Any]: 包含查询结果的字典
"""
if not keyId or not isinstance(keyId, str):
error_msg = "keyId参数是必需的且必须是字符串"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
# 验证至少有一个查询参数
if not any([device, location, operation]):
error_msg = "至少需要提供device、location或operation中的一个参数"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
endpoint = f"{self.base_url}/iot/companies/{keyId}/search"
# 构建请求数据
payload = {}
# 添加可选参数
if device:
payload["device"] = device
if location:
payload["location"] = location
if operation:
payload["operation"] = operation
if operation_param:
payload["operation_param"] = operation_param
# 设置请求头
headers = {
"Content-Type": "application/json",
}
try:
logger.info(
f"查询向量数据keyId: {keyId}, device: {device}, location: {location}, operation: {operation}, operation_param: {operation_param}"
)
# 发送POST请求
response = self.session.post(
endpoint,
headers=headers,
data=json.dumps(payload),
timeout=CONFIG["request_timeout"],
)
# 检查响应状态
response.raise_for_status()
# 返回JSON响应
result = response.json()
logger.info(
f"向量查询成功,返回结果数量: {result.get('data', {}).get('total_results', 0)}"
)
return result
except requests.exceptions.Timeout:
error_msg = f"向量查询请求超时keyId: {keyId}"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
except requests.RequestException as e:
error_msg = f"查询向量数据失败: {str(e)}"
logger.error(error_msg)
return {
"status": "error",
"message": error_msg,
"data": None,
}
except Exception as e:
error_msg = f"查询向量数据发生未知异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {
"status": "error",
"message": error_msg,
"data": None,
}
def analyze_search_results(self, search_results: List[List[Any]]) -> Dict[str, Any]:
"""
调用搜索结果分析接口
参数:
search_results: 搜索结果数据,格式为包含设备信息、评分和详细评分的嵌套数组
[[device_info, score, score_details], ...]
返回:
Dict[str, Any]: 分析结果,包含处理后的数据和响应信息
"""
logger.info("开始分析搜索结果...")
logger.debug(f"输入搜索结果数量: {len(search_results) if search_results else 0}")
if not search_results or not isinstance(search_results, list):
error_msg = "无效的搜索结果数据search_results必须是一个非空数组"
logger.error(error_msg)
return {"code": 400, "msg": error_msg, "data": None}
# API配置
url = f"{self.base_url}/iot/analyze/search-results"
headers = {"Content-Type": "application/json"}
# 构建请求数据
request_data = {
"search_results": search_results
}
try:
logger.info(f"正在调用搜索结果分析接口: {url}")
logger.debug(f"请求数据大小: {len(search_results)} 条搜索结果")
# 发送POST请求
response = self.session.post(
url=url,
headers=headers,
data=json.dumps(request_data),
timeout=CONFIG["request_timeout"],
)
# 处理响应
if response.status_code == 200:
try:
result = response.json()
logger.info(f"搜索结果分析成功,返回数据类型: {type(result)}")
# 处理标准格式的响应 {code, msg, data}
if isinstance(result, dict) and "code" in result:
# 检查接口返回的业务状态码
if result.get("code") == 200:
data = result.get("data", {})
filtered_results = data.get("filtered_results", [])
logger.info(f"搜索结果分析成功,返回过滤结果数量: {len(filtered_results) if filtered_results else 0}")
# 返回包含filtered_results的标准格式
return filtered_results
else:
# 接口返回了错误状态码,直接返回原始响应
logger.warning(f"接口返回业务错误: code={result.get('code')}, msg={result.get('msg')}")
return result
# 如果接口直接返回数据而不是标准格式,包装成标准格式
if isinstance(result, dict) and "code" not in result:
logger.info("接口返回非标准格式,进行包装处理")
return {"code": 200, "msg": "分析成功", "data": result}
logger.debug(f"返回原始分析结果: {type(result)}")
return result
except ValueError as e:
error_msg = f"响应JSON解析失败: {str(e)}"
logger.error(error_msg)
return {"code": 500, "msg": error_msg, "data": None}
else:
error_msg = f"API请求失败状态码: {response.status_code}, 响应: {response.text}"
logger.error(error_msg)
return {
"code": response.status_code,
"msg": error_msg,
"data": None,
}
except requests.exceptions.Timeout:
error_msg = "搜索结果分析接口请求超时"
logger.error(error_msg)
return {"code": 408, "msg": error_msg, "data": None}
except requests.exceptions.RequestException as e:
error_msg = f"搜索结果分析接口网络请求异常: {str(e)}"
logger.error(error_msg)
return {"code": 500, "msg": error_msg, "data": None}
except Exception as e:
error_msg = f"搜索结果分析接口调用异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {"code": 500, "msg": error_msg, "data": None}
def check_vector_store_status(self, keyId: str) -> bool:
"""
检查向量库状态
参数:
keyId: 企业或项目的唯一标识符
返回:
bool: 向量库是否存在
"""
if not keyId or not isinstance(keyId, str):
logger.error(f"无效的keyId参数: {keyId}")
return False
endpoint = f"{self.base_url}/iot/companies/{keyId}/exists"
# 设置请求头
headers = {
"Accept": "*/*",
}
try:
logger.info(f"检查向量库状态keyId: {keyId}")
# 发送GET请求
response = self.session.get(
endpoint,
headers=headers,
timeout=CONFIG["request_timeout"],
)
# 检查响应状态
response.raise_for_status()
# 从响应中提取exists字段
result = response.json()
exists = result.get("data", {}).get("exists", False)
logger.info(f"向量库状态检查完成keyId: {keyId}, exists: {exists}")
return exists
except requests.exceptions.Timeout:
logger.error(f"检查向量库状态请求超时keyId: {keyId}")
return False
except requests.RequestException as e:
logger.error(f"检查向量库状态失败: {str(e)}")
return False
except Exception as e:
logger.error(f"检查向量库状态发生未知异常: {str(e)}", exc_info=True)
return False
def query_devices_by_location(
self,
keyId: str,
location: Optional[str] = None,
) -> Dict[str, Any]:
"""
根据位置查询设备列表
参数:
keyId: 企业或项目的唯一标识符
location: 设备位置(可选)
返回:
Dict[str, Any]: 包含设备列表的数据字典
"""
logger = _ensure_logger()
if not keyId or not isinstance(keyId, str):
error_msg = "keyId参数是必需的且必须是字符串"
logger.error(error_msg)
return {
"devices": [],
"count": 0,
"location_filter": location or ""
}
endpoint = f"{self.base_url}/iot/companies/{keyId}/devices"
# 构建请求数据
payload = {}
if location:
payload["location"] = location
# 设置请求头
headers = {
"Content-Type": "application/json",
}
try:
logger.info(f"查询设备列表keyId: {keyId}, location: {location}")
# 发送POST请求
response = self.session.post(
endpoint,
headers=headers,
data=json.dumps(payload),
timeout=CONFIG["request_timeout"],
)
# 检查响应状态
response.raise_for_status()
# 解析JSON响应
result = response.json()
# 检查API返回的业务状态码
if result.get("code") == 200:
data = result.get("data", {})
device_count = data.get("count", 0)
logger.info(f"设备列表查询成功,返回设备数量: {device_count}")
# 返回data部分
return data
else:
error_msg = f"API返回业务错误: code={result.get('code')}, msg={result.get('msg')}"
logger.error(error_msg)
return {
"devices": [],
"count": 0,
"location_filter": location or "",
"error": error_msg
}
except requests.exceptions.Timeout:
error_msg = f"查询设备列表请求超时keyId: {keyId}"
logger.error(error_msg)
return {
"devices": [],
"count": 0,
"location_filter": location or "",
"error": error_msg
}
except requests.RequestException as e:
error_msg = f"查询设备列表失败: {str(e)}"
logger.error(error_msg)
return {
"devices": [],
"count": 0,
"location_filter": location or "",
"error": error_msg
}
except Exception as e:
error_msg = f"查询设备列表发生未知异常: {str(e)}"
logger.error(error_msg, exc_info=True)
return {
"devices": [],
"count": 0,
"location_filter": location or "",
"error": error_msg
}
def check_and_filter_devices(
self,
device_list: List[Any],
device_op: Any
) -> Dict[str, Any]:
"""
检查设备分数并过滤有效设备
参数:
device_list: 设备列表,格式为 [[device_info, score, score_details], ...]
device_op: DeviceOperator实例用于调用设备检查和预处理方法
返回:
Dict[str, Any]: 包含检查结果的字典
- 如果有有效设备:返回有效设备列表
- 如果无有效设备:返回候选设备列表和错误码
"""
logger = _ensure_logger()
logger.info("开始检查设备分数并过滤有效设备")
logger.debug(f"输入设备列表数量: {len(device_list) if device_list else 0}")
# 判断设备分数是否达标给每个设备添加checkResult字段
checked_device_list = []
has_valid_device = False
for device_item in device_list:
# deviceList格式: [[device_info, score, score_details], ...]
if isinstance(device_item, list) and len(device_item) >= 3:
device_info, score, score_details = device_item[0], device_item[1], device_item[2]
# 调用check_device_results_score方法检查分数
check_result = device_op.check_device_results_score(score_details)
# 将checkResult添加到device_info中
if isinstance(device_info, dict):
device_info["checkResult"] = check_result.get("checkResult", False)
# 检查是否有通过验证的设备
if device_info["checkResult"]:
has_valid_device = True
# 重新构建device_item
checked_device_list.append([device_info, score, score_details])
else:
# 如果格式不正确保持原样但添加checkResult为False
if isinstance(device_item, dict):
device_item["checkResult"] = False
checked_device_list.append(device_item)
# 如果没有任何设备通过验证,返回候选设备列表
if not has_valid_device:
logger.warning("匹配分数不够准确,返回候选设备列表")
candidate_list = device_op.preprocess_results(checked_device_list)
return {
"code": 400,
"msg": "设备匹配分数不够准确,无法为用户智能操作设备,返回候选设备列表供用户选择; ",
"data": {
"candidates": candidate_list,
"total": len(candidate_list)
},
"is_candidate": True
}
# 过滤出checkResult为True的设备
valid_devices = []
for device_item in checked_device_list:
if isinstance(device_item, list) and len(device_item) >= 1:
device_info = device_item[0]
if isinstance(device_info, dict) and device_info.get("checkResult", False):
valid_devices.append(device_item)
logger.info(f"过滤完成,有效设备数量: {len(valid_devices)}")
return {
"valid_devices": valid_devices,
"is_candidate": False
}
def __del__(self):
"""析构函数,关闭会话"""
if hasattr(self, "session"):
self.session.close()
# 使用示例
if __name__ == "__main__":
# 测试代码只在直接运行此模块时执行,不在导入时执行
# 以下代码已注释避免MCP服务器启动时执行
pass
# vector_service = VectorService(base_url="http://192.168.0.76:5002")
# # 测试初始化向量存储
# # init_result = vector_service.init_vector_store(
# # keyId="1945419433873575938",
# # )
# # print("初始化向量存储结果:", init_result)
# # 测试查询向量420
# # query_result = vector_service.query_vector(
# # **{
# # "keyId": "1932095424144715777",
# # "location": "灵泽办公区",
# # "device": "过道吊灯",
# # "operation": "开关 按键",
# # "top_k": 2,
# # "auto_create": True,
# # }
# # )
# # print("查询向量结果:", query_result)
# # # 检查向量库状态
# # check_result = vector_service.check_vector_store_status("1932095424144715777")
# # logger.info(f"检查向量库状态结果: {check_result}")
# # 测试搜索结果分析接口
# # logger.info("=== 测试搜索结果分析接口 ===")
# sample_search_results = [
# [
# {
# "id": "0899574c-ff51-4983-ae48-667cccc08e9c",
# "location": {
# "key": "35",
# "description": "灵泽展厅;灵泽展区"
# },
# "device": {
# "key": "switch.zimi_cn_1121232402_dhkg05_on_p_2_1",
# "description": "吊灯;灯;照明灯"
# },
# "operation": {
# "key": "turn_off",
# "description": "关闭;关"
# },
# "operation_params": [],
# "deviceType": "light"
# },
# 0.9025000000000001,
# {
# "location_score": 0.95,
# "device_score": 0.75,
# "operation_score": 0.95,
# "operation_param_score": 1.0,
# "combined_score": 0.9025000000000001,
# "search_method": "location_first_strict"
# }
# ],
# [
# {
# "id": "76f71468-356b-4710-a7ea-da7ddac93fab",
# "location": {
# "key": "35",
# "description": "灵泽展厅;灵泽展区"
# },
# "device": {
# "key": "climate.qjiang_cn_741362991_wb20",
# "description": "空调;制冷设备"
# },
# "operation": {
# "key": "turn_off",
# "description": "关空调;关闭空调;关闭;关;闭空调;空调"
# },
# "operation_params": [],
# "deviceType": "airConditioner"
# },
# 0.7150000000000001,
# {
# "location_score": 0.95,
# "device_score": 0.0,
# "operation_score": 0.95,
# "operation_param_score": 1.0,
# "combined_score": 0.7150000000000001,
# "search_method": "location_first_strict"
# }
# ]
# ]
# # analyze_result = vector_service.analyze_search_results(sample_search_results)
# # logger.info(f"搜索结果分析结果: {analyze_result}")

17
lzwcai_mcp_iot/main.py Normal file
View File

@@ -0,0 +1,17 @@
import os
# 设置企业ID必需
os.environ["ENTERPRISE_ID"] = "1952978233106669569"
# 设置API地址
os.environ["DEVICE_API_BASE_URL"] = "http://192.168.2.236:8088"
os.environ["VECTOR_API_BASE_URL"] = "http://192.168.2.236:5002"
# 注意employeeId 已不再需要,现在直接使用 ENTERPRISE_ID
# os.environ["employeeId"] = "1986712221817815042" # 已废弃
# 导入模块
from lzwcai_mcp_iot.iot_device_tool import main
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,63 @@
[build-system]
requires = ["setuptools>=42", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "lzwcai-mcp-iot"
version = "0.3.3"
description = "IoT设备控制服务器使用FastMCP框架提供设备操作功能"
authors = [
{name = "LZWCAI开发团队", email = "dev@lzwcai.com"}
]
readme = "README.md"
requires-python = ">=3.8"
license = "LicenseRef-Proprietary"
classifiers = [
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Operating System :: OS Independent",
]
dependencies = [
"fastmcp>=0.1.0",
"requests"
]
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"black>=23.1.0",
"isort>=5.12.0",
"flake8>=6.0.0",
"mypy>=1.0.0",
]
[project.scripts]
lzwcai-mcp-iot = "lzwcai_mcp_iot.iot_device_tool:main"
[tool.setuptools]
packages = ["lzwcai_mcp_iot", "lzwcai_mcp_iot.src"]
[tool.black]
line-length = 88
target-version = ["py38", "py39", "py310", "py311"]
include = '\.pyi?$'
[tool.isort]
profile = "black"
line_length = 88
[tool.mypy]
python_version = "3.8"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = false
disallow_incomplete_defs = false
[tool.pytest.ini_options]
minversion = "7.0"
testpaths = ["tests"]

4
lzwcai_mcp_iot/setup.cfg Normal file
View File

@@ -0,0 +1,4 @@
[egg_info]
tag_build =
tag_date = 0