实现数据接入层功能（REST API + WebSocket + 批量导入）

- 实现REST API服务器，支持IoT数据、手动录入和批量导入接口
- 实现WebSocket服务器，支持实时数据推送和连接管理
- 实现批量导入模块，支持CSV、Excel、JSON多种格式
- 实现数据处理工具，包含字段映射和单位转换功能
- 实现数据模型定义和数据验证机制
- 创建主程序入口和配置文件
- 添加详细的使用文档和API说明

README.md

@@ -0,0 +1,299 @@
+# 水务管理系统 - 数据接入层
+
+## 项目概述
+
+本项目是水务管理系统中的数据接入层，实现了多源数据接入、实时WebSocket推送和批量数据导入功能。
+
+## 功能特性
+
+### 1. REST API 数据接入
+- **IoT设备数据接入**：支持实时接收传感器数据
+- **手动数据录入**：支持人工录入数据
+- **批量API导入**：支持通过API批量导入数据
+- **数据查询接口**：提供按数据类型、时间范围等条件的数据查询
+- **统计分析接口**：提供数据统计和分析功能
+
+### 2. WebSocket 实时推送
+- **实时数据推送**：传感器数据实时推送到客户端
+- **连接管理**：支持多客户端连接和订阅管理
+- **数据历史**：新连接客户端可以获取历史数据
+- **警报推送**：支持实时警报推送
+- **心跳检测**：支持连接状态监控
+
+### 3. 批量数据导入
+- **多格式支持**：支持CSV、Excel、JSON格式导入
+- **数据验证**：内置数据验证和错误处理
+- **字段映射**：支持水利行业标准字段映射
+- **单位转换**：自动进行单位转换和标准化
+- **批量处理**：支持大批量数据处理
+
+## 技术栈
+
+- **后端框架**：FastAPI
+- **WebSocket**：websockets
+- **数据处理**：pandas, numpy
+- **异步处理**：asyncio
+- **数据验证**：pydantic
+- **文件处理**：aiofiles
+
+## 项目结构
+
+```
+water-management-system/
+├── src/
+│   ├── api/                 # REST API模块
+│   │   └── rest_api.py      # 主API服务器
+│   ├── websocket/           # WebSocket模块
+│   │   └── websocket_server.py  # WebSocket服务器
+│   ├── batch/               # 批量导入模块
+│   │   └── batch_import.py  # 批量导入功能
+│   ├── utils/               # 工具模块
+│   │   └── data_utils.py    # 数据处理工具
+│   └── models/              # 数据模型
+│       └── models.py        # 数据模型定义
+├── main.py                  # 主程序入口
+├── requirements.txt        # 依赖文件
+├── config.json             # 配置文件
+└── README.md               # 项目说明
+```
+
+## API 接口文档
+
+### REST API 端点
+
+#### 1. IoT 数据接收
+```
+POST /api/iot/data
+Content-Type: application/json
+
+{
+    "device_id": "device_001",
+    "data_type": "LL",
+    "value": 25.5,
+    "timestamp": "2024-01-01T12:00:00",
+    "location": "A区"
+}
+```
+
+#### 2. 手动数据录入
+```
+POST /api/manual/data
+Content-Type: application/json
+
+{
+    "source": "manual",
+    "data_type": "YL",
+    "value": 0.8,
+    "timestamp": "2024-01-01T12:00:00",
+    "operator": "张三",
+    "notes": "定期数据录入"
+}
+```
+
+#### 3. 批量导入
+```
+POST /api/batch/import
+Content-Type: application/json
+
+{
+    "batch_id": "batch_001",
+    "data_source": "system_import",
+    "records": [
+        {
+            "device_id": "device_002",
+            "data_type": "SW",
+            "value": 5.2,
+            "timestamp": "2024-01-01T12:00:00",
+            "location": "B区"
+        }
+    ]
+}
+```
+
+#### 4. 数据查询
+```
+GET /api/data/{data_type}?limit=100&offset=0
+GET /api/data/recent?hours=24&limit=100
+GET /api/stats
+```
+
+### WebSocket 连接
+
+#### 连接端点
+```
+ws://localhost:8765
+```
+
+#### 消息格式
+
+**发送消息**：
+```json
+{
+    "type": "subscribe",
+    "subscription": "LL"  // 订阅特定类型数据，"all"订阅所有
+}
+```
+
+**接收消息**：
+```json
+{
+    "type": "sensor_data",
+    "data_type": "LL",
+    "device_id": "device_001",
+    "value": 25.5,
+    "location": "A区",
+    "timestamp": "2024-01-01T12:00:00Z"
+}
+```
+
+## 安装和运行
+
+### 1. 安装依赖
+```bash
+pip install -r requirements.txt
+```
+
+### 2. 启动系统
+```bash
+# 普通模式
+python main.py
+
+# 演示模式（自动生成数据）
+python main.py --demo
+
+# 指定配置文件
+python main.py --config custom_config.json
+```
+
+### 3. 初始化项目
+```bash
+python main.py --init
+```
+
+## 配置文件
+
+创建 `config.json` 文件：
+
+```json
+{
+    "api": {
+        "host": "0.0.0.0",
+        "port": 8000
+    },
+    "websocket": {
+        "host": "0.0.0.0",
+        "port": 8765
+    },
+    "batch": {
+        "max_file_size_mb": 100,
+        "supported_formats": ["csv", "excel", "json"]
+    },
+    "demo_mode": true,
+    "logging": {
+        "level": "INFO",
+        "file": "water_management.log"
+    }
+}
+```
+
+## 数据类型说明
+
+支持的水利行业标准数据类型：
+
+| 数据类型 | 描述 | 单位 |
+|---------|------|------|
+| LL | 流量 | m³/h |
+| YL | 压力 | MPa |
+| SW | 水位 | m |
+| ZD | 浊度 | NTU |
+| PH | pH值 | - |
+| WD | 温度 | °C |
+| DD | 电导率 | μS/cm |
+| YD | 硬度 | mg/L |
+
+## 开发规范
+
+### 代码结构
+- 遵循模块化设计，功能解耦
+- 使用异步编程提高性能
+- 统一的错误处理机制
+- 完整的日志记录
+
+### 数据处理
+- 数据验证和清洗
+- 标准化字段映射
+- 单位自动转换
+- 质量评分机制
+
+### 安全考虑
+- 输入数据验证
+- 文件大小限制
+- 连接状态监控
+- 错误信息脱敏
+
+## 测试和验证
+
+### 数据验证
+```python
+from src.utils.data_utils import data_converter
+
+# 验证传感器数据
+validation_result = data_converter.validate_sensor_data({
+    "device_id": "device_001",
+    "data_type": "LL",
+    "value": 25.5,
+    "location": "A区"
+})
+
+print(validation_result)
+```
+
+### 批量导入测试
+```python
+import asyncio
+from src.batch.batch_import import batch_manager
+
+async def test_import():
+    result = await batch_manager.import_file(
+        file_path="data.csv",
+        batch_id="test_batch",
+        data_source="test"
+    )
+    print(result)
+
+asyncio.run(test_import())
+```
+
+## 部署建议
+
+### 1. 生产环境配置
+- 使用反向代理（Nginx）
+- 配置SSL证书
+- 设置防火墙规则
+- 监控系统资源使用
+
+### 2. 性能优化
+- 数据库连接池
+- 缓存机制
+- 异步处理优化
+- 连接数限制
+
+### 3. 监控和日志
+- 应用性能监控
+- 错误日志收集
+- 性能指标统计
+- 告警机制
+
+## 许可证
+
+本项目遵循 MIT 许可证。
+
+## 贡献指南
+
+欢迎提交 Issue 和 Pull Request来贡献代码。
+
+## 联系方式
+
+如有问题，请通过以下方式联系：
+- 邮箱：bot_dev1@xayunmei.com
+- 项目地址：http://git.xayunmei.com/bot_ym/water-management-system

main.py

@@ -0,0 +1,305 @@
+"""
+水务管理系统主程序
+集成REST API、WebSocket和批量导入功能
+"""
+import asyncio
+import logging
+import signal
+import sys
+from pathlib import Path
+import argparse
+from datetime import datetime
+
+# 导入各个模块
+from src.api.rest_api import app as rest_api_app
+from src.websocket.websocket_server import websocket_server
+from src.batch.batch_import import batch_manager
+from src.utils.data_utils import data_converter, data_formatter, quality_checker
+from src.models.models import validator
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler('water_management.log'),
+        logging.StreamHandler(sys.stdout)
+    ]
+)
+logger = logging.getLogger(__name__)
+
+class WaterManagementSystem:
+    """水务管理系统主类"""
+    
+    def __init__(self, config_file: str = "config.json"):
+        self.config_file = config_file
+        self.running = False
+        self.tasks = []
+        self.config = self.load_config()
+        
+    def load_config(self) -> dict:
+        """加载配置文件"""
+        config_path = Path(self.config_file)
+        if config_path.exists():
+            import json
+            with open(config_path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        else:
+            # 默认配置
+            return {
+                "api": {
+                    "host": "0.0.0.0",
+                    "port": 8000
+                },
+                "websocket": {
+                    "host": "0.0.0.0",
+                    "port": 8765
+                },
+                "batch": {
+                    "max_file_size_mb": 100,
+                    "supported_formats": ["csv", "excel", "json"]
+                },
+                "logging": {
+                    "level": "INFO",
+                    "file": "water_management.log"
+                }
+            }
+    
+    async def start_api_server(self):
+        """启动API服务器"""
+        import uvicorn
+        logger.info("启动REST API服务器...")
+        
+        # 在新的事件循环中运行uvicorn
+        api_config = uvicorn.Config(
+            app=rest_api_app,
+            host=self.config["api"]["host"],
+            port=self.config["api"]["port"],
+            log_level="info"
+        )
+        api_server = uvicorn.Server(api_config)
+        
+        # 在后台任务中运行
+        await api_server.serve()
+    
+    async def start_websocket_server(self):
+        """启动WebSocket服务器"""
+        logger.info("启动WebSocket服务器...")
+        
+        # 启动WebSocket服务器
+        server = await websocket_server.start_server()
+        
+        # 添加服务器关闭处理
+        def cleanup():
+            logger.info("关闭WebSocket服务器...")
+            server.close()
+            asyncio.create_task(server.wait_closed())
+        
+        return server, cleanup
+    
+    async def start_data_generator(self):
+        """启动数据生成器（用于演示）"""
+        logger.info("启动数据生成器...")
+        
+        while self.running:
+            # 生成模拟数据
+            import random
+            sensor_types = ["LL", "YL", "SW", "ZD"]
+            sensor_type = random.choice(sensor_types)
+            
+            # 根据传感器类型生成合理的数值范围
+            if sensor_type == "LL":  # 流量
+                value = random.uniform(10, 100)
+            elif sensor_type == "YL":  # 压力
+                value = random.uniform(0.1, 1.0)
+            elif sensor_type == "SW":  # 水位
+                value = random.uniform(0, 10)
+            else:  # ZD 浊度
+                value = random.uniform(0, 50)
+            
+            sensor_data = {
+                "data_type": sensor_type,
+                "device_id": f"device_{random.randint(1, 10)}",
+                "value": round(value, 2),
+                "location": random.choice(["A区", "B区", "C区", "D区"])
+            }
+            
+            # 通过WebSocket发送数据
+            await websocket_server.send_sensor_data(sensor_data)
+            
+            # 等待5秒
+            await asyncio.sleep(5)
+    
+    async def handle_batch_import(self, file_path: str, batch_id: str, data_source: str):
+        """处理批量导入请求"""
+        try:
+            logger.info(f"开始批量导入: {file_path}")
+            
+            # 验证文件
+            validation_result = await batch_manager.validate_file(file_path)
+            if not validation_result["valid"]:
+                raise Exception(f"文件验证失败: {validation_result['error']}")
+            
+            # 导入文件
+            result = await batch_manager.import_file(
+                file_path=file_path,
+                batch_id=batch_id,
+                data_source=data_source,
+                file_type="auto"
+            )
+            
+            logger.info(f"批量导入完成: {result}")
+            return result
+            
+        except Exception as e:
+            logger.error(f"批量导入失败: {str(e)}")
+            raise
+    
+    async def start_system(self):
+        """启动系统"""
+        logger.info("启动水务管理系统...")
+        
+        # 标记系统为运行状态
+        self.running = True
+        
+        try:
+            # 启动WebSocket服务器
+            ws_server, ws_cleanup = await self.start_websocket_server()
+            self.tasks.append(ws_server)
+            
+            # 启动数据生成器（如果启用）
+            if self.config.get("demo_mode", False):
+                generator_task = asyncio.create_task(self.start_data_generator())
+                self.tasks.append(generator_task)
+            
+            # 启动API服务器
+            await self.start_api_server()
+            
+        except Exception as e:
+            logger.error(f"系统启动失败: {str(e)}")
+            await self.stop_system()
+            raise
+    
+    async def stop_system(self):
+        """停止系统"""
+        logger.info("停止水务管理系统...")
+        
+        self.running = False
+        
+        # 取消所有任务
+        for task in self.tasks:
+            if not task.done():
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+        
+        # 清理WebSocket服务器
+        if hasattr(websocket_server, 'server') and websocket_server.server:
+            websocket_server.server.close()
+            await websocket_server.server.wait_closed()
+        
+        logger.info("水务管理系统已停止")
+    
+    async def run(self):
+        """运行系统"""
+        # 设置信号处理
+        def signal_handler():
+            logger.info("收到停止信号...")
+            asyncio.create_task(self.stop_system())
+        
+        for sig in [signal.SIGINT, signal.SIGTERM]:
+            signal.signal(sig, signal_handler)
+        
+        try:
+            await self.start_system()
+            
+            # 保持运行直到收到停止信号
+            while self.running:
+                await asyncio.sleep(1)
+                
+        except KeyboardInterrupt:
+            logger.info("收到键盘中断信号")
+        except Exception as e:
+            logger.error(f"系统运行时出错: {str(e)}")
+        finally:
+            await self.stop_system()
+
+def create_sample_config():
+    """创建示例配置文件"""
+    sample_config = {
+        "api": {
+            "host": "0.0.0.0",
+            "port": 8000
+        },
+        "websocket": {
+            "host": "0.0.0.0",
+            "port": 8765
+        },
+        "batch": {
+            "max_file_size_mb": 100,
+            "supported_formats": ["csv", "excel", "json"]
+        },
+        "demo_mode": True,
+        "logging": {
+            "level": "INFO",
+            "file": "water_management.log"
+        }
+    }
+    
+    import json
+    with open("config.json", 'w', encoding='utf-8') as f:
+        json.dump(sample_config, f, indent=2, ensure_ascii=False)
+    
+    logger.info("示例配置文件已创建: config.json")
+
+def create_requirements():
+    """创建requirements.txt文件"""
+    requirements = [
+        "fastapi==0.104.1",
+        "uvicorn[standard]==0.24.0",
+        "websockets==12.0",
+        "pandas==2.1.3",
+        "openpyxl==3.1.2",
+        "aiofiles==23.2.1",
+        "python-multipart==0.0.6",
+        "jinja2==3.1.2"
+    ]
+    
+    with open("requirements.txt", 'w') as f:
+        f.write('\n'.join(requirements))
+    
+    logger.info("依赖文件已创建: requirements.txt")
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(description="水务管理系统")
+    parser.add_argument("--config", "-c", default="config.json", help="配置文件路径")
+    parser.add_argument("--init", action="store_true", help="初始化项目（创建配置文件和依赖）")
+    parser.add_argument("--demo", action="store_true", help="启动演示模式")
+    
+    args = parser.parse_args()
+    
+    if args.init:
+        create_sample_config()
+        create_requirements()
+        logger.info("项目初始化完成")
+        return
+    
+    # 创建系统实例
+    system = WaterManagementSystem(args.config)
+    
+    # 如果启用演示模式
+    if args.demo:
+        system.config["demo_mode"] = True
+        logger.info("启用演示模式")
+    
+    # 运行系统
+    try:
+        asyncio.run(system.run())
+    except KeyboardInterrupt:
+        logger.info("程序已退出")
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,10 @@
+fastapi==0.104.1
+uvicorn[standard]==0.24.0
+websockets==12.0
+pandas==2.1.3
+openpyxl==3.1.2
+aiofiles==23.2.1
+python-multipart==0.0.6
+jinja2==3.1.2
+requests==2.31.0
+python-dateutil==2.8.2

src/api/rest_api.py

@@ -0,0 +1,181 @@
+"""
+REST API 数据接入模块
+支持 IoT 设备数据、手动录入和 API 批量导入
+"""
+from fastapi import FastAPI, HTTPException, Depends
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import List, Optional, Dict, Any
+import uvicorn
+import asyncio
+import json
+from datetime import datetime
+
+# 创建FastAPI应用
+app = FastAPI(title="Water Management System Data API", version="1.0.0")
+
+# CORS配置
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 数据模型
+class IoTData(BaseModel):
+    device_id: str
+    data_type: str  # "LL", "YL", "SW", "ZD" 等
+    value: float
+    timestamp: datetime
+    location: str
+
+class ManualInputData(BaseModel):
+    source: str
+    data_type: str
+    value: float
+    timestamp: datetime
+    operator: str
+    notes: Optional[str] = None
+
+class BatchImportRequest(BaseModel):
+    batch_id: str
+    data_source: str
+    records: List[Dict[str, Any]]
+
+# 数据存储（示例，实际应该用数据库）
+data_store = []
+
+@app.get("/")
+async def root():
+    """API根路径"""
+    return {"message": "Water Management System API", "version": "1.0.0"}
+
+@app.post("/api/iot/data")
+async def receive_iot_data(data: IoTData):
+    """接收IoT设备数据"""
+    try:
+        data_dict = data.dict()
+        data_store.append({
+            **data_dict,
+            "id": len(data_store) + 1,
+            "type": "iot"
+        })
+        return {"status": "success", "id": len(data_store), "message": "IoT data received"}
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+@app.post("/api/manual/data")
+async def receive_manual_data(data: ManualInputData):
+    """接收手动录入数据"""
+    try:
+        data_dict = data.dict()
+        data_store.append({
+            **data_dict,
+            "id": len(data_store) + 1,
+            "type": "manual"
+        })
+        return {"status": "success", "id": len(data_store), "message": "Manual data received"}
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+@app.post("/api/batch/import")
+async def batch_import(request: BatchImportRequest):
+    """批量导入数据"""
+    try:
+        imported_count = 0
+        failed_count = 0
+        
+        for record in request.records:
+            # 验证记录
+            if not all(k in record for k in ['device_id', 'data_type', 'value']):
+                failed_count += 1
+                continue
+                
+            # 创建数据对象
+            data_record = {
+                "device_id": record['device_id'],
+                "data_type": record['data_type'],
+                "value": float(record['value']),
+                "timestamp": record.get('timestamp', datetime.now()),
+                "location": record.get('location', 'unknown'),
+                "batch_id": request.batch_id,
+                "source": request.data_source,
+                "id": len(data_store) + 1,
+                "type": "batch"
+            }
+            
+            data_store.append(data_record)
+            imported_count += 1
+            
+        return {
+            "status": "success",
+            "imported_count": imported_count,
+            "failed_count": failed_count,
+            "message": f"Batch import completed: {imported_count} records imported, {failed_count} failed"
+        }
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+@app.get("/api/data/{data_type}")
+async def get_data_by_type(data_type: str, limit: int = 100, offset: int = 0):
+    """根据数据类型获取数据"""
+    filtered_data = [
+        item for item in data_store 
+        if item.get('data_type') == data_type
+    ]
+    
+    return {
+        "data": filtered_data[offset:offset+limit],
+        "total": len(filtered_data),
+        "limit": limit,
+        "offset": offset
+    }
+
+@app.get("/api/data/recent")
+async def get_recent_data(hours: int = 24, limit: int = 100):
+    """获取最近的数据"""
+    from datetime import timedelta
+    cutoff_time = datetime.now() - timedelta(hours=hours)
+    
+    recent_data = [
+        item for item in data_store 
+        if item.get('timestamp', datetime.now()) > cutoff_time
+    ]
+    
+    return {
+        "data": recent_data[-limit:],
+        "total": len(recent_data),
+        "hours": hours,
+        "limit": limit
+    }
+
+@app.get("/api/stats")
+async def get_statistics():
+    """获取数据统计信息"""
+    stats = {
+        "total_records": len(data_store),
+        "by_type": {},
+        "by_device": {},
+        "by_hour": {}
+    }
+    
+    for item in data_store:
+        data_type = item.get('data_type', 'unknown')
+        device_id = item.get('device_id', 'unknown')
+        hour = item.get('timestamp', datetime.now()).strftime('%Y-%m-%d %H:00:00')
+        
+        stats['by_type'][data_type] = stats['by_type'].get(data_type, 0) + 1
+        stats['by_device'][device_id] = stats['by_device'].get(device_id, 0) + 1
+        stats['by_hour'][hour] = stats['by_hour'].get(hour, 0) + 1
+    
+    return stats
+
+@app.get("/health")
+async def health_check():
+    """健康检查"""
+    return {"status": "healthy", "timestamp": datetime.now()}
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)

src/batch/batch_import.py

@@ -0,0 +1,336 @@
+"""
+批量数据导入模块
+支持CSV、Excel、JSON等多种格式的批量数据导入
+"""
+import pandas as pd
+import json
+import csv
+import asyncio
+import aiofiles
+from typing import List, Dict, Any, Optional, Union
+from pathlib import Path
+from datetime import datetime
+import logging
+from concurrent.futures import ThreadPoolExecutor
+
+# 配置日志
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class BatchImportError(Exception):
+    """批量导入异常"""
+    pass
+
+class DataValidator:
+    """数据验证器"""
+    
+    # 水利行业标准字段映射
+    STANDARD_FIELDS = {
+        "LL": "流量",
+        "YL": "压力", 
+        "SW": "水位",
+        "ZD": "浊度",
+        "PH": "pH值",
+        "WD": "温度",
+        "DD": "电导率",
+        "YD": "硬度"
+    }
+    
+    # 单位映射
+    UNIT_MAP = {
+        "LL": "m³/h",
+        "YL": "MPa", 
+        "SW": "m",
+        "ZD": "NTU",
+        "PH": "",
+        "WD": "°C",
+        "DD": "μS/cm",
+        "YD": "mg/L"
+    }
+    
+    @classmethod
+    def validate_data_type(cls, data_type: str) -> bool:
+        """验证数据类型是否有效"""
+        return data_type in cls.STANDARD_FIELDS
+    
+    @classmethod
+    def get_field_description(cls, data_type: str) -> str:
+        """获取字段描述"""
+        return cls.STANDARD_FIELDS.get(data_type, "未知类型")
+    
+    @classmethod
+    def get_unit(cls, data_type: str) -> str:
+        """获取单位"""
+        return cls.UNIT_MAP.get(data_type, "")
+    
+    @classmethod
+    def validate_record(cls, record: Dict[str, Any]) -> Dict[str, Any]:
+        """验证单条记录"""
+        errors = []
+        validated_record = {}
+        
+        # 必需字段检查
+        required_fields = ["device_id", "data_type", "value"]
+        for field in required_fields:
+            if field not in record:
+                errors.append(f"缺少必需字段: {field}")
+            else:
+                validated_record[field] = record[field]
+        
+        # 数据类型验证
+        if "data_type" in validated_record:
+            if not cls.validate_data_type(validated_record["data_type"]):
+                errors.append(f"无效的数据类型: {validated_record['data_type']}")
+        
+        # 数值验证
+        if "value" in validated_record:
+            try:
+                validated_record["value"] = float(validated_record["value"])
+            except (ValueError, TypeError):
+                errors.append(f"无效的数值: {validated_record['value']}")
+        
+        # 时间戳处理
+        if "timestamp" in record:
+            try:
+                if isinstance(record["timestamp"], str):
+                    validated_record["timestamp"] = datetime.fromisoformat(record["timestamp"])
+                else:
+                    validated_record["timestamp"] = record["timestamp"]
+            except (ValueError, TypeError):
+                # 如果时间戳无效，使用当前时间
+                validated_record["timestamp"] = datetime.now()
+        else:
+            validated_record["timestamp"] = datetime.now()
+        
+        # 地点字段处理
+        validated_record["location"] = record.get("location", "未知")
+        
+        return {
+            "validated": len(errors) == 0,
+            "record": validated_record,
+            "errors": errors
+        }
+
+class BatchImporter:
+    """批量导入器"""
+    
+    def __init__(self):
+        self.validator = DataValidator()
+    
+    async def import_csv(self, file_path: str, batch_id: str, data_source: str) -> Dict[str, Any]:
+        """导入CSV文件"""
+        try:
+            # 使用线程池执行文件读取
+            loop = asyncio.get_event_loop()
+            with ThreadPoolExecutor() as executor:
+                df = await loop.run_in_executor(
+                    executor, 
+                    lambda: pd.read_csv(file_path, encoding='utf-8')
+                )
+            
+            return await self._process_dataframe(df, batch_id, data_source)
+            
+        except Exception as e:
+            raise BatchImportError(f"CSV文件导入失败: {str(e)}")
+    
+    async def import_excel(self, file_path: str, batch_id: str, data_source: str, sheet_name: str = 0) -> Dict[str, Any]:
+        """导入Excel文件"""
+        try:
+            # 使用线程池执行文件读取
+            loop = asyncio.get_event_loop()
+            with ThreadPoolExecutor() as executor:
+                df = await loop.run_in_executor(
+                    executor,
+                    lambda: pd.read_excel(file_path, sheet_name=sheet_name)
+                )
+            
+            return await self._process_dataframe(df, batch_id, data_source)
+            
+        except Exception as e:
+            raise BatchImportError(f"Excel文件导入失败: {str(e)}")
+    
+    async def import_json(self, file_path: str, batch_id: str, data_source: str) -> Dict[str, Any]:
+        """导入JSON文件"""
+        try:
+            async with aiofiles.open(file_path, 'r', encoding='utf-8') as f:
+                content = await f.read()
+            
+            data = json.loads(content)
+            
+            # 处理不同的JSON格式
+            if isinstance(data, list):
+                return await self._process_records(data, batch_id, data_source)
+            elif isinstance(data, dict):
+                if "records" in data:
+                    return await self._process_records(data["records"], batch_id, data_source)
+                else:
+                    return await self._process_records([data], batch_id, data_source)
+            else:
+                raise BatchImportError("不支持的JSON格式")
+                
+        except Exception as e:
+            raise BatchImportError(f"JSON文件导入失败: {str(e)}")
+    
+    async def _process_dataframe(self, df: pd.DataFrame, batch_id: str, data_source: str) -> Dict[str, Any]:
+        """处理DataFrame数据"""
+        # 转换为字典列表
+        records = df.to_dict('records')
+        return await self._process_records(records, batch_id, data_source)
+    
+    async def _process_records(self, records: List[Dict[str, Any]], batch_id: str, data_source: str) -> Dict[str, Any]:
+        """处理记录列表"""
+        import_count = 0
+        error_count = 0
+        errors = []
+        imported_records = []
+        
+        for i, record in enumerate(records):
+            validation_result = self.validator.validate_record(record)
+            
+            if validation_result["validated"]:
+                # 添加批次信息
+                import_record = {
+                    **validation_result["record"],
+                    "batch_id": batch_id,
+                    "data_source": data_source,
+                    "import_time": datetime.now(),
+                    "type": "batch"
+                }
+                
+                imported_records.append(import_record)
+                import_count += 1
+            else:
+                error_count += 1
+                error_msg = f"记录 {i+1}: {', '.join(validation_result['errors'])}"
+                errors.append(error_msg)
+                logger.warning(error_msg)
+        
+        # 保存导入的记录到文件（实际项目中应该保存到数据库）
+        await self._save_imported_records(imported_records)
+        
+        return {
+            "status": "completed" if error_count == 0 else "completed_with_errors",
+            "imported_count": import_count,
+            "error_count": error_count,
+            "total_count": len(records),
+            "success_rate": import_count / len(records) if records else 0,
+            "batch_id": batch_id,
+            "data_source": data_source,
+            "errors": errors[:10],  # 只返回前10个错误
+            "imported_records": imported_records[:5]  # 返回前5条记录作为示例
+        }
+    
+    async def _save_imported_records(self, records: List[Dict[str, Any]]):
+        """保存导入的记录"""
+        # 这里可以将记录保存到数据库或文件
+        # 为了示例，我们只保存到日志
+        for record in records:
+            logger.info(f"导入记录: {record}")
+    
+    async def get_import_summary(self, batch_id: str) -> Dict[str, Any]:
+        """获取导入摘要"""
+        # 这里应该从数据库查询批次信息
+        # 为了示例，返回一个空摘要
+        return {
+            "batch_id": batch_id,
+            "status": "not_found",
+            "message": "批次信息未找到（示例实现）"
+        }
+
+class BatchImportManager:
+    """批量导入管理器"""
+    
+    def __init__(self):
+        self.importer = BatchImporter()
+    
+    async def import_file(self, file_path: str, batch_id: str, data_source: str, 
+                        file_type: str = "auto", **kwargs) -> Dict[str, Any]:
+        """导入文件"""
+        file_path_obj = Path(file_path)
+        
+        if not file_path_obj.exists():
+            raise BatchImportError(f"文件不存在: {file_path}")
+        
+        # 自动检测文件类型
+        if file_type == "auto":
+            if file_path_obj.suffix.lower() == '.csv':
+                file_type = "csv"
+            elif file_path_obj.suffix.lower() in ['.xlsx', '.xls']:
+                file_type = "excel"
+            elif file_path_obj.suffix.lower() == '.json':
+                file_type = "json"
+            else:
+                raise BatchImportError(f"不支持的文件类型: {file_path_obj.suffix}")
+        
+        logger.info(f"开始导入{file_type}文件: {file_path}")
+        
+        if file_type == "csv":
+            result = await self.importer.import_csv(file_path, batch_id, data_source)
+        elif file_type == "excel":
+            sheet_name = kwargs.get("sheet_name", 0)
+            result = await self.importer.import_excel(file_path, batch_id, data_source, sheet_name)
+        elif file_type == "json":
+            result = await self.importer.import_json(file_path, batch_id, data_source)
+        else:
+            raise BatchImportError(f"不支持的文件类型: {file_type}")
+        
+        logger.info(f"文件导入完成: {result}")
+        return result
+    
+    async def validate_file(self, file_path: str) -> Dict[str, Any]:
+        """验证文件格式"""
+        file_path_obj = Path(file_path)
+        
+        if not file_path_obj.exists():
+            return {"valid": False, "error": "文件不存在"}
+        
+        file_size = file_path_obj.stat().st_size
+        if file_size > 100 * 1024 * 1024:  # 100MB限制
+            return {"valid": False, "error": "文件过大，最大支持100MB"}
+        
+        # 尝试读取文件前几行进行验证
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                first_line = f.readline()
+                if not first_line:
+                    return {"valid": False, "error": "文件为空"}
+        except Exception as e:
+            return {"valid": False, "error": f"无法读取文件: {str(e)}"}
+        
+        return {"valid": True, "size": file_size, "format": file_path_obj.suffix.lower()}
+
+# 全局导入管理器实例
+batch_manager = BatchImportManager()
+
+# 示例用法
+async def example_usage():
+    """示例用法"""
+    import os
+    
+    # 创建示例CSV文件
+    sample_data = [
+        {"device_id": "device_001", "data_type": "LL", "value": 25.5, "location": "A区"},
+        {"device_id": "device_002", "data_type": "YL", "value": 0.8, "location": "B区"},
+        {"device_id": "device_003", "data_type": "SW", "value": 5.2, "location": "C区"}
+    ]
+    
+    sample_file = "/tmp/sample_data.csv"
+    with open(sample_file, 'w', newline='', encoding='utf-8') as f:
+        writer = csv.DictWriter(f, fieldnames=["device_id", "data_type", "value", "location"])
+        writer.writeheader()
+        writer.writerows(sample_data)
+    
+    # 导入文件
+    try:
+        result = await batch_manager.import_file(
+            file_path=sample_file,
+            batch_id="batch_" + datetime.now().strftime("%Y%m%d_%H%M%S"),
+            data_source="manual_test",
+            file_type="csv"
+        )
+        print("导入结果:", result)
+    except Exception as e:
+        print(f"导入失败: {str(e)}")
+
+if __name__ == "__main__":
+    asyncio.run(example_usage())

src/models/models.py

@@ -0,0 +1,317 @@
+"""
+数据模型定义
+定义水务管理系统的各种数据结构
+"""
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Any
+from datetime import datetime
+from enum import Enum
+
+class DataType(Enum):
+    """数据类型枚举"""
+    LL = "LL"  # 流量
+    YL = "YL"  # 压力
+    SW = "SW"  # 水位
+    ZD = "ZD"  # 浊度
+    PH = "PH"  # pH值
+    WD = "WD"  # 温度
+    DD = "DD"  # 电导率
+    YD = "YD"  # 硬度
+
+class AlertLevel(Enum):
+    """警报级别枚举"""
+    INFO = "info"
+    WARNING = "warning"
+    ERROR = "error"
+    CRITICAL = "critical"
+
+@dataclass
+class Device:
+    """设备模型"""
+    id: str
+    name: str
+    device_type: str
+    location: str
+    description: Optional[str] = None
+    install_date: Optional[datetime] = None
+    status: str = "active"  # active, inactive, maintenance
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+@dataclass
+class SensorData:
+    """传感器数据模型"""
+    id: str
+    device_id: str
+    data_type: DataType
+    value: float
+    unit: str
+    timestamp: datetime
+    location: str
+    quality_score: float = 1.0
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "id": self.id,
+            "device_id": self.device_id,
+            "data_type": self.data_type.value,
+            "value": self.value,
+            "unit": self.unit,
+            "timestamp": self.timestamp.isoformat(),
+            "location": self.location,
+            "quality_score": self.quality_score,
+            "metadata": self.metadata
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'SensorData':
+        """从字典创建对象"""
+        return cls(
+            id=data["id"],
+            device_id=data["device_id"],
+            data_type=DataType(data["data_type"]),
+            value=float(data["value"]),
+            unit=data.get("unit", ""),
+            timestamp=datetime.fromisoformat(data["timestamp"]),
+            location=data.get("location", ""),
+            quality_score=float(data.get("quality_score", 1.0)),
+            metadata=data.get("metadata", {})
+        )
+
+@dataclass
+class Alert:
+    """警报模型"""
+    id: str
+    device_id: str
+    alert_type: str
+    level: AlertLevel
+    message: str
+    timestamp: datetime
+    resolved: bool = False
+    resolved_by: Optional[str] = None
+    resolved_at: Optional[datetime] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "id": self.id,
+            "device_id": self.device_id,
+            "alert_type": self.alert_type,
+            "level": self.level.value,
+            "message": self.message,
+            "timestamp": self.timestamp.isoformat(),
+            "resolved": self.resolved,
+            "resolved_by": self.resolved_by,
+            "resolved_at": self.resolved_at.isoformat() if self.resolved_at else None,
+            "metadata": self.metadata
+        }
+
+@dataclass
+class BatchImport:
+    """批量导入记录模型"""
+    id: str
+    batch_id: str
+    data_source: str
+    total_records: int
+    successful_records: int
+    failed_records: int
+    status: str  # pending, processing, completed, failed
+    file_name: Optional[str] = None
+    import_time: Optional[datetime] = None
+    completed_time: Optional[datetime] = None
+    error_messages: List[str] = field(default_factory=list)
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "id": self.id,
+            "batch_id": self.batch_id,
+            "data_source": self.data_source,
+            "total_records": self.total_records,
+            "successful_records": self.successful_records,
+            "failed_records": self.failed_records,
+            "status": self.status,
+            "file_name": self.file_name,
+            "import_time": self.import_time.isoformat() if self.import_time else None,
+            "completed_time": self.completed_time.isoformat() if self.completed_time else None,
+            "error_messages": self.error_messages,
+            "metadata": self.metadata
+        }
+
+@dataclass
+class APIRequest:
+    """API请求模型"""
+    id: str
+    method: str
+    endpoint: str
+    params: Dict[str, Any]
+    headers: Dict[str, Any]
+    body: Optional[Any] = None
+    timestamp: datetime = field(default_factory=datetime.now)
+    response_code: Optional[int] = None
+    response_time_ms: Optional[float] = None
+    response_body: Optional[Any] = None
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "id": self.id,
+            "method": self.method,
+            "endpoint": self.endpoint,
+            "params": self.params,
+            "headers": self.headers,
+            "body": self.body,
+            "timestamp": self.timestamp.isoformat(),
+            "response_code": self.response_code,
+            "response_time_ms": self.response_time_ms,
+            "response_body": self.response_body
+        }
+
+@dataclass
+class WebSocketConnection:
+    """WebSocket连接模型"""
+    id: str
+    client_ip: str
+    connected_at: datetime
+    disconnected_at: Optional[datetime] = None
+    subscriptions: List[str] = field(default_factory=list)
+    message_count: int = 0
+    last_message_at: Optional[datetime] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "id": self.id,
+            "client_ip": self.client_ip,
+            "connected_at": self.connected_at.isoformat(),
+            "disconnected_at": self.disconnected_at.isoformat() if self.disconnected_at else None,
+            "subscriptions": self.subscriptions,
+            "message_count": self.message_count,
+            "last_message_at": self.last_message_at.isoformat() if self.last_message_at else None,
+            "metadata": self.metadata
+        }
+
+@dataclass
+class SystemStats:
+    """系统统计模型"""
+    timestamp: datetime
+    total_records: int
+    total_devices: int
+    active_connections: int
+    api_requests_count: int
+    alerts_count: int
+    data_quality_score: float
+    memory_usage_mb: float
+    cpu_usage_percent: float
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典"""
+        return {
+            "timestamp": self.timestamp.isoformat(),
+            "total_records": self.total_records,
+            "total_devices": self.total_devices,
+            "active_connections": self.active_connections,
+            "api_requests_count": self.api_requests_count,
+            "alerts_count": self.alerts_count,
+            "data_quality_score": self.data_quality_score,
+            "memory_usage_mb": self.memory_usage_mb,
+            "cpu_usage_percent": self.cpu_usage_percent
+        }
+
+class DataValidator:
+    """数据验证器"""
+    
+    @staticmethod
+    def validate_sensor_data(data: Dict[str, Any]) -> List[str]:
+        """验证传感器数据"""
+        errors = []
+        
+        # 必需字段检查
+        required_fields = ["device_id", "data_type", "value", "location"]
+        for field in required_fields:
+            if field not in data:
+                errors.append(f"缺少必需字段: {field}")
+        
+        # 数据类型验证
+        if "data_type" in data:
+            try:
+                DataType(data["data_type"])
+            except ValueError:
+                errors.append(f"无效的数据类型: {data['data_type']}")
+        
+        # 数值验证
+        if "value" in data:
+            try:
+                value = float(data["value"])
+                # 根据数据类型进行数值范围检查
+                data_type = data.get("data_type")
+                if data_type == "LL" and value < 0:
+                    errors.append("流量不能为负数")
+                elif data_type == "YL" and value < 0:
+                    errors.append("压力不能为负数")
+                elif data_type == "SW" and value < 0:
+                    errors.append("水位不能为负数")
+            except (ValueError, TypeError):
+                errors.append(f"无效的数值: {data['value']}")
+        
+        # 时间戳验证
+        if "timestamp" in data:
+            try:
+                if isinstance(data["timestamp"], str):
+                    datetime.fromisoformat(data["timestamp"])
+            except (ValueError, TypeError):
+                errors.append(f"无效的时间戳格式: {data['timestamp']}")
+        
+        return errors
+    
+    @staticmethod
+    def validate_device_data(data: Dict[str, Any]) -> List[str]:
+        """验证设备数据"""
+        errors = []
+        
+        # 必需字段检查
+        required_fields = ["id", "name", "device_type", "location"]
+        for field in required_fields:
+            if field not in data:
+                errors.append(f"缺少必需字段: {field}")
+        
+        # 设备ID格式验证
+        if "id" in data:
+            device_id = data["id"]
+            if not isinstance(device_id, str) or not device_id.strip():
+                errors.append("设备ID不能为空")
+            elif len(device_id) > 50:
+                errors.append("设备ID长度不能超过50个字符")
+        
+        # 状态验证
+        if "status" in data and data["status"] not in ["active", "inactive", "maintenance"]:
+            errors.append("设备状态必须是: active, inactive, maintenance")
+        
+        return errors
+    
+    @staticmethod
+    def validate_alert_data(data: Dict[str, Any]) -> List[str]:
+        """验证警报数据"""
+        errors = []
+        
+        # 必需字段检查
+        required_fields = ["device_id", "alert_type", "level", "message"]
+        for field in required_fields:
+            if field not in data:
+                errors.append(f"缺少必需字段: {field}")
+        
+        # 警报级别验证
+        if "level" in data:
+            try:
+                AlertLevel(data["level"])
+            except ValueError:
+                errors.append(f"无效的警报级别: {data['level']}")
+        
+        return errors
+
+# 全局验证器实例
+validator = DataValidator()

src/utils/data_utils.py

@@ -0,0 +1,409 @@
+"""
+数据处理工具模块
+提供数据验证、转换、格式化等工具函数
+"""
+import json
+import csv
+import pandas as pd
+from typing import Dict, List, Any, Optional, Union
+from datetime import datetime, timedelta
+import hashlib
+import logging
+
+# 配置日志
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class DataConverter:
+    """数据转换器"""
+    
+    # 水利行业标准字段映射
+    FIELD_MAPPING = {
+        "流量": "LL",
+        "压力": "YL",
+        "水位": "SW",
+        "浊度": "ZD",
+        "pH值": "PH",
+        "温度": "WD",
+        "电导率": "DD",
+        "硬度": "YD",
+        # 支持常见的中文字段名
+        "流量计": "LL",
+        "压力表": "YL",
+        "水位计": "SW",
+        "浊度仪": "ZD",
+        "pH计": "PH",
+        "温度计": "WD",
+        "电导率仪": "DD",
+        "硬度计": "YD"
+    }
+    
+    # 单位转换
+    UNIT_CONVERSIONS = {
+        # 流量单位转换 (m³/h)
+        "m³/h": 1.0,
+        "L/s": 3.6,  # L/s = m³/h / 1000 * 3600
+        "m³/d": 1/24,  # m³/d = m³/h / 24
+        "L/min": 1/60,  # L/min = m³/h / 1000 * 60
+        
+        # 压力单位转换 (MPa)
+        "MPa": 1.0,
+        "kPa": 0.001,  # kPa = MPa / 1000
+        "bar": 0.1,  # bar = MPa * 10
+        "kgf/cm²": 0.0980665,  # kgf/cm² = MPa / 0.0980665
+        
+        # 水位单位转换 (m)
+        "m": 1.0,
+        "cm": 0.01,  # cm = m / 100
+        "mm": 0.001,  # mm = m / 1000
+        
+        # 浊度单位转换 (NTU)
+        "NTU": 1.0,
+        "FNU": 1.0,  # FNU ≈ NTU
+        
+        # pH值单位转换
+        "pH": 1.0,
+        
+        # 温度单位转换 (°C)
+        "°C": 1.0,
+        "K": 1.0,  # 相对差值
+        "°F": lambda x: (x - 32) / 1.8,  # °F to °C
+        
+        # 电导率单位转换 (μS/cm)
+        "μS/cm": 1.0,
+        "mS/cm": 1000,  # mS/cm = μS/cm * 1000
+        "S/m": 10000  # S/m = μS/cm * 100
+    }
+    
+    @classmethod
+    def normalize_field_name(cls, field_name: str) -> str:
+        """标准化字段名"""
+        if not field_name:
+            return ""
+        
+        field_name = field_name.strip().upper()
+        
+        # 如果已经是标准格式，直接返回
+        if field_name in cls.FIELD_MAPPING.values():
+            return field_name
+        
+        # 查映射表
+        if field_name in cls.FIELD_MAPPING:
+            return cls.FIELD_MAPPING[field_name]
+        
+        # 英文映射
+        english_mapping = {
+            "flow": "LL",
+            "pressure": "YL", 
+            "level": "SW",
+            "turbidity": "ZD",
+            "ph": "PH",
+            "temperature": "WD",
+            "conductivity": "DD",
+            "hardness": "YD"
+        }
+        
+        if field_name.lower() in english_mapping:
+            return english_mapping[field_name.lower()]
+        
+        return field_name
+    
+    @classmethod
+    def convert_unit(cls, value: float, from_unit: str, to_unit: str) -> float:
+        """单位转换"""
+        if from_unit == to_unit:
+            return value
+        
+        if from_unit not in cls.UNIT_CONVERSIONS:
+            raise ValueError(f"不支持的单位: {from_unit}")
+        
+        if to_unit not in cls.UNIT_CONVERSIONS:
+            raise ValueError(f"不支持的目标单位: {to_unit}")
+        
+        from_conv = cls.UNIT_CONVERSIONS[from_unit]
+        to_conv = cls.UNIT_CONVERSIONS[to_unit]
+        
+        if callable(from_conv):
+            value = from_conv(value)
+        
+        if callable(to_conv):
+            return value / to_conv
+        else:
+            return value * (to_conv / from_conv)
+    
+    @classmethod
+    def validate_sensor_data(cls, data: Dict[str, Any]) -> Dict[str, Any]:
+        """验证传感器数据"""
+        errors = []
+        validated_data = {}
+        
+        # 必需字段验证
+        required_fields = ["device_id", "data_type", "value"]
+        for field in required_fields:
+            if field not in data:
+                errors.append(f"缺少必需字段: {field}")
+            else:
+                validated_data[field] = data[field]
+        
+        # 数据类型验证和标准化
+        if "data_type" in validated_data:
+            original_type = validated_data["data_type"]
+            validated_data["data_type"] = cls.normalize_field_name(original_type)
+            
+            if validated_data["data_type"] != original_type:
+                logger.info(f"字段名标准化: {original_type} -> {validated_data['data_type']}")
+        
+        # 数值验证
+        if "value" in validated_data:
+            try:
+                validated_data["value"] = float(validated_data["value"])
+                # 检查数值范围
+                data_type = validated_data.get("data_type", "")
+                if data_type == "LL" and validated_data["value"] < 0:
+                    errors.append("流量不能为负数")
+                elif data_type == "YL" and validated_data["value"] < 0:
+                    errors.append("压力不能为负数")
+                elif data_type == "SW" and validated_data["value"] < 0:
+                    errors.append("水位不能为负数")
+            except (ValueError, TypeError):
+                errors.append(f"无效的数值: {validated_data['value']}")
+        
+        # 地点验证
+        if "location" not in validated_data or not validated_data["location"]:
+            validated_data["location"] = "未知"
+        
+        # 时间戳处理
+        if "timestamp" in data:
+            try:
+                if isinstance(data["timestamp"], str):
+                    validated_data["timestamp"] = datetime.fromisoformat(data["timestamp"])
+                else:
+                    validated_data["timestamp"] = data["timestamp"]
+            except (ValueError, TypeError):
+                validated_data["timestamp"] = datetime.now()
+        else:
+            validated_data["timestamp"] = datetime.now()
+        
+        return {
+            "valid": len(errors) == 0,
+            "data": validated_data,
+            "errors": errors
+        }
+
+class DataFormatter:
+    """数据格式化器"""
+    
+    @staticmethod
+    def format_sensor_data(data: Dict[str, Any], format_type: str = "json") -> str:
+        """格式化传感器数据"""
+        if format_type == "json":
+            return json.dumps(data, ensure_ascii=False, indent=2)
+        elif format_type == "csv":
+            # CSV格式只包含关键字段
+            csv_fields = ["device_id", "data_type", "value", "location", "timestamp"]
+            csv_data = {k: data.get(k, "") for k in csv_fields}
+            import io
+            output = io.StringIO()
+            writer = csv.DictWriter(output, fieldnames=csv_fields)
+            writer.writeheader()
+            writer.writerow(csv_data)
+            return output.getvalue()
+        else:
+            raise ValueError(f"不支持的格式类型: {format_type}")
+    
+    @staticmethod
+    def format_statistics(stats: Dict[str, Any], format_type: str = "text") -> str:
+        """格式化统计数据"""
+        if format_type == "json":
+            return json.dumps(stats, ensure_ascii=False, indent=2)
+        elif format_type == "text":
+            lines = ["数据统计报告", "=" * 20]
+            lines.append(f"总记录数: {stats.get('total_records', 0)}")
+            
+            if "by_type" in stats:
+                lines.append("\n按数据类型统计:")
+                for data_type, count in stats["by_type"].items():
+                    lines.append(f"  {data_type}: {count} 条")
+            
+            if "by_device" in stats:
+                lines.append("\n按设备统计:")
+                for device_id, count in list(stats["by_device"].items())[:10]:  # 只显示前10个
+                    lines.append(f"  {device_id}: {count} 条")
+            
+            return "\n".join(lines)
+        else:
+            raise ValueError(f"不支持的格式类型: {format_type}")
+
+class DataHasher:
+    """数据哈希工具"""
+    
+    @staticmethod
+    def calculate_data_hash(data: Dict[str, Any]) -> str:
+        """计算数据哈希值"""
+        # 将数据转换为字符串
+        data_str = json.dumps(data, sort_keys=True, ensure_ascii=False)
+        
+        # 计算MD5哈希
+        hash_md5 = hashlib.md5(data_str.encode())
+        return hash_md5.hexdigest()
+    
+    @staticmethod
+    def generate_data_id(device_id: str, data_type: str, timestamp: datetime) -> str:
+        """生成数据ID"""
+        # 使用设备ID、数据类型和时间戳生成唯一ID
+        time_str = timestamp.strftime("%Y%m%d_%H%M%S")
+        hash_input = f"{device_id}_{data_type}_{time_str}"
+        hash_md5 = hashlib.md5(hash_input.encode())
+        return f"{data_type}_{device_id}_{hash_md5.hexdigest()[:8]}"
+
+class DataQualityChecker:
+    """数据质量检查器"""
+    
+    @staticmethod
+    def check_data_quality(records: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """检查数据质量"""
+        quality_report = {
+            "total_records": len(records),
+            "valid_records": 0,
+            "invalid_records": 0,
+            "quality_score": 0,
+            "issues": [],
+            "statistics": {}
+        }
+        
+        if not records:
+            quality_report["quality_score"] = 0
+            return quality_report
+        
+        valid_records = []
+        
+        for record in records:
+            issues = []
+            
+            # 检查必需字段
+            required_fields = ["device_id", "data_type", "value"]
+            for field in required_fields:
+                if field not in record or not record[field]:
+                    issues.append(f"缺少必需字段: {field}")
+            
+            # 检查数据类型
+            if "data_type" in record and record["data_type"]:
+                valid_types = ["LL", "YL", "SW", "ZD", "PH", "WD", "DD", "YD"]
+                if record["data_type"] not in valid_types:
+                    issues.append(f"无效的数据类型: {record['data_type']}")
+            
+            # 检查数值范围
+            if "value" in record and record["value"]:
+                try:
+                    value = float(record["value"])
+                    data_type = record.get("data_type", "")
+                    
+                    if data_type == "LL" and value < 0:
+                        issues.append("流量不能为负数")
+                    elif data_type == "YL" and value < 0:
+                        issues.append("压力不能为负数")
+                    elif data_type == "SW" and value < 0:
+                        issues.append("水位不能为负数")
+                    
+                    # 检查异常值
+                    if data_type == "LL" and value > 10000:
+                        issues.append("流量值异常大")
+                    elif data_type == "YL" and value > 10:
+                        issues.append("压力值异常大")
+                        
+                except (ValueError, TypeError):
+                    issues.append("无效的数值格式")
+            
+            if not issues:
+                valid_records.append(record)
+                quality_report["valid_records"] += 1
+            else:
+                quality_report["invalid_records"] += 1
+                quality_report["issues"].extend(issues)
+        
+        # 计算质量分数
+        quality_report["quality_score"] = quality_report["valid_records"] / len(records)
+        
+        # 统计信息
+        if records:
+            quality_report["statistics"] = {
+                "completeness": quality_report["valid_records"] / len(records),
+                "uniqueness": len(set(r.get("device_id", "") for r in valid_records)) / len(valid_records) if valid_records else 0,
+                "timeliness": quality_report.calculate_timeliness(records)
+            }
+        
+        return quality_report
+    
+    @staticmethod
+    def calculate_timeliness(records: List[Dict[str, Any]]) -> float:
+        """计算数据及时性（24小时内的数据比例）"""
+        if not records:
+            return 0
+        
+        now = datetime.now()
+        recent_count = 0
+        
+        for record in records:
+            timestamp = record.get("timestamp")
+            if timestamp:
+                try:
+                    if isinstance(timestamp, str):
+                        timestamp = datetime.fromisoformat(timestamp)
+                    
+                    time_diff = now - timestamp
+                    if time_diff <= timedelta(hours=24):
+                        recent_count += 1
+                except:
+                    pass
+        
+        return recent_count / len(records)
+
+class DataExporter:
+    """数据导出工具"""
+    
+    @staticmethod
+    def export_to_csv(records: List[Dict[str, Any]], file_path: str) -> bool:
+        """导出为CSV文件"""
+        try:
+            if not records:
+                return False
+            
+            # 获取所有字段
+            all_fields = set()
+            for record in records:
+                all_fields.update(record.keys())
+            
+            # 排序字段
+            field_order = ["device_id", "data_type", "value", "location", "timestamp"]
+            for field in all_fields:
+                if field not in field_order:
+                    field_order.append(field)
+            
+            with open(file_path, 'w', newline='', encoding='utf-8') as csvfile:
+                writer = csv.DictWriter(csvfile, fieldnames=field_order)
+                writer.writeheader()
+                writer.writerows(records)
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"导出CSV失败: {str(e)}")
+            return False
+    
+    @staticmethod
+    def export_to_json(records: List[Dict[str, Any]], file_path: str) -> bool:
+        """导出为JSON文件"""
+        try:
+            with open(file_path, 'w', encoding='utf-8') as jsonfile:
+                json.dump(records, jsonfile, ensure_ascii=False, indent=2, default=str)
+            return True
+        except Exception as e:
+            logger.error(f"导出JSON失败: {str(e)}")
+            return False
+
+# 全局工具实例
+data_converter = DataConverter()
+data_formatter = DataFormatter()
+data_hasher = DataHasher()
+quality_checker = DataQualityChecker()
+data_exporter = DataExporter()

src/websocket/websocket_server.py

@@ -0,0 +1,214 @@
+"""
+WebSocket 实时数据推送服务器
+支持实时数据推送、连接管理和数据广播
+"""
+import asyncio
+import json
+import websockets
+from datetime import datetime
+from typing import Set, Dict, Any
+import logging
+
+# 配置日志
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class WebSocketServer:
+    """WebSocket服务器类"""
+    
+    def __init__(self, host: str = "0.0.0.0", port: int = 8765):
+        self.host = host
+        self.port = port
+        self.clients: Set[websockets.WebSocketServerProtocol] = set()
+        self.data_history: list = []  # 存储最近的数据用于新连接
+        
+    async def register_client(self, websocket: websockets.WebSocketServerProtocol):
+        """注册新客户端"""
+        self.clients.add(websocket)
+        client_ip = websocket.remote_address[0]
+        logger.info(f"新客户端连接: {client_ip}")
+        
+        # 发送历史数据给新连接的客户端
+        if self.data_history:
+            await websocket.send(json.dumps({
+                "type": "history",
+                "data": self.data_history[-50:]  # 发送最近50条数据
+            }))
+        
+        # 发送欢迎消息
+        await websocket.send(json.dumps({
+            "type": "welcome",
+            "message": "已连接到水务管理系统实时数据服务器",
+            "timestamp": datetime.now().isoformat()
+        }))
+    
+    async def unregister_client(self, websocket: websockets.WebSocketServerProtocol):
+        """注销客户端"""
+        if websocket in self.clients:
+            self.clients.remove(websocket)
+            client_ip = websocket.remote_address[0]
+            logger.info(f"客户端断开连接: {client_ip}")
+    
+    async def broadcast_data(self, data: Dict[str, Any]):
+        """广播数据到所有连接的客户端"""
+        if not self.clients:
+            return
+            
+        # 添加时间戳
+        data["timestamp"] = datetime.now().isoformat()
+        
+        # 保存历史数据
+        self.data_history.append(data)
+        if len(self.data_history) > 1000:  # 只保留最近1000条记录
+            self.data_history.pop(0)
+        
+        # 广播数据
+        message = json.dumps(data)
+        disconnected_clients = []
+        
+        for client in self.clients:
+            try:
+                await client.send(message)
+            except websockets.exceptions.ConnectionClosed:
+                disconnected_clients.append(client)
+        
+        # 清理已断开的连接
+        for client in disconnected_clients:
+            await self.unregister_client(client)
+    
+    async def handle_client_message(self, websocket: websockets.WebSocketServerProtocol, message: str):
+        """处理客户端消息"""
+        try:
+            data = json.loads(message)
+            
+            if data.get("type") == "subscribe":
+                # 处理订阅请求
+                subscription_type = data.get("subscription", "all")
+                response = {
+                    "type": "subscription_ack",
+                    "subscription": subscription_type,
+                    "message": f"已订阅 {subscription_type} 类型数据"
+                }
+                await websocket.send(json.dumps(response))
+                logger.info(f"客户端订阅了 {subscription_type} 类型数据")
+            
+            elif data.get("type") == "ping":
+                # 响应心跳检测
+                response = {
+                    "type": "pong",
+                    "timestamp": datetime.now().isoformat()
+                }
+                await websocket.send(json.dumps(response))
+            
+            else:
+                logger.warning(f"未知的消息类型: {data.get('type', 'unknown')}")
+                
+        except json.JSONDecodeError:
+            logger.error("无效的JSON消息")
+        except Exception as e:
+            logger.error(f"处理客户端消息时出错: {str(e)}")
+    
+    async def client_handler(self, websocket: websockets.WebSocketServerProtocol, path: str):
+        """处理客户端连接"""
+        await self.register_client(websocket)
+        
+        try:
+            async for message in websocket:
+                await self.handle_client_message(websocket, message)
+        except websockets.exceptions.ConnectionClosed:
+            pass
+        finally:
+            await self.unregister_client(websocket)
+    
+    async def start_server(self):
+        """启动WebSocket服务器"""
+        logger.info(f"启动WebSocket服务器: {self.host}:{self.port}")
+        
+        # 创建并启动服务器
+        self.server = await websockets.serve(
+            self.client_handler,
+            self.host,
+            self.port
+        )
+        
+        logger.info("WebSocket服务器已启动")
+        return self.server
+    
+    async def send_sensor_data(self, sensor_data: Dict[str, Any]):
+        """发送传感器数据"""
+        data = {
+            "type": "sensor_data",
+            "data_type": sensor_data.get("data_type"),
+            "device_id": sensor_data.get("device_id"),
+            "value": sensor_data.get("value"),
+            "location": sensor_data.get("location"),
+            "timestamp": datetime.now().isoformat()
+        }
+        await self.broadcast_data(data)
+    
+    async def send_alert(self, alert_data: Dict[str, Any]):
+        """发送警报信息"""
+        data = {
+            "type": "alert",
+            "level": alert_data.get("level", "warning"),
+            "message": alert_data.get("message"),
+            "device_id": alert_data.get("device_id"),
+            "timestamp": datetime.now().isoformat()
+        }
+        await self.broadcast_data(data)
+
+# 全局WebSocket服务器实例
+websocket_server = WebSocketServer()
+
+# 示例数据生成器
+async def data_generator():
+    """模拟数据生成器"""
+    import random
+    
+    while True:
+        await asyncio.sleep(5)  # 每5秒发送一次数据
+        
+        # 模拟不同的传感器数据
+        sensor_types = ["LL", "YL", "SW", "ZD"]
+        sensor_type = random.choice(sensor_types)
+        
+        # 根据传感器类型生成合理的数值范围
+        if sensor_type == "LL":  # 流量
+            value = random.uniform(10, 100)
+        elif sensor_type == "YL":  # 压力
+            value = random.uniform(0.1, 1.0)
+        elif sensor_type == "SW":  # 水位
+            value = random.uniform(0, 10)
+        else:  # ZD 浊度
+            value = random.uniform(0, 50)
+        
+        sensor_data = {
+            "data_type": sensor_type,
+            "device_id": f"device_{random.randint(1, 10)}",
+            "value": round(value, 2),
+            "location": random.choice(["A区", "B区", "C区", "D区"])
+        }
+        
+        await websocket_server.send_sensor_data(sensor_data)
+
+# 启动服务器和生成器
+async def main():
+    """主函数"""
+    # 启动WebSocket服务器
+    server = await websocket_server.start_server()
+    
+    # 启动数据生成器
+    generator_task = asyncio.create_task(data_generator())
+    
+    # 保持服务器运行
+    try:
+        await asyncio.Future()  # 永远等待
+    except KeyboardInterrupt:
+        logger.info("收到中断信号，正在关闭服务器...")
+        server.close()
+        await server.wait_closed()
+        generator_task.cancel()
+        await generator_task
+
+if __name__ == "__main__":
+    asyncio.run(main())