添加websocket功能

app/config/config.yaml

@@ -7,6 +7,11 @@ kangda:
   deviceId: pc
   lang: zh_CN 
 
+ws: 
+  url: rest.concoai.com/imserver/
+  account: rbsstaff1
+  tenantInfoId: 4fff5d4bcc4b4239941ff077a0da8958
+
 url_login : "http://erpapi.concoai.com/basis-api/user/login"
 url_event_list : "http://erpapi.concoai.com/robot/event/page"
 url_event_detail: "http://erpapi.concoai.com/robot/event/"  

app/core/database.py

@@ -32,7 +32,7 @@ engine = create_async_engine(
 #     echo=False
 # )
 
-# 创建异步会话工厂
+# 创建异步会话工厂, expire_on_commit 禁用提交后过期对象--> 提交后原对象仍然可以继续使用.
 async_session = sessionmaker(
     engine, class_=AsyncSession, expire_on_commit=False
 )

app/main.py

@@ -1,7 +1,8 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, WebSocket
 from fastapi.middleware.cors import CORSMiddleware
 from app.api.v1.api import api_router
 from app.core.config import settings
+from app.services.websocket_service import websocket_service
 
 app = FastAPI(
     title="温度数据采集系统",
@@ -21,6 +22,11 @@ app.add_middleware(
 # 注册路由
 app.include_router(api_router)
 
+# @app.websocket("/ws")
+# async def websocket_endpoint(websocket: WebSocket):
+#     """WebSocket端点"""
+#     await websocket_service.handle_websocket(websocket)
+
 @app.get("/")
 async def root():
     return {"message": "温度数据采集系统API服务正在运行"} 

app/models/models.py

@@ -91,7 +91,7 @@ class Temperature(Base):
     imageId = Column(BigInteger, ForeignKey('image.imageId'), nullable=False, comment='关联图片ID')
     temperature = Column(String(20), nullable=False, comment='温度值')
     # status = Column(String(2), comment='温度是否正常')
-    confidence = Column(String(20), nullable=False, comment='识别置信度')
+    confidence = Column(String(40), nullable=False, comment='识别置信度')
     createTime = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
 
     # 关系

app/services/event_sync_service.py

@@ -160,7 +160,14 @@ async def run_sync():
             print(f"同步过程出错: {str(e)}")
         
         # 等待5分钟
-        await asyncio.sleep(300)
+        await asyncio.sleep(6000)
+        
+async def run_sync_once():
+    service = EventSyncService()
+    try:
+        await service.sync_events()
+    except Exception as e:
+        print(f"同步过程出错: {str(e)}")
 
 if __name__ == "__main__":
     asyncio.run(run_sync()) 

app/services/websocket_service.py

@@ -0,0 +1,137 @@
+import asyncio
+import json
+import os
+from datetime import datetime
+import websockets
+import aiofiles
+from app.kangdaApi.parseConfig import parse_config
+from app.services.event_sync_service import run_sync_once
+
+class WebSocketClient:
+    def __init__(self):
+        self.config = parse_config("app/config/config.yaml")
+        self.ws_config = self.config.get("ws", {})
+        self.message_dir = "logs/websocket_messages"
+        self.reconnect_delay = 5  # 重连延迟（秒）
+        self.is_connected = False
+        self.file_lock = asyncio.Lock()  # 文件锁
+        os.makedirs(self.message_dir, exist_ok=True)
+
+    def get_ws_url(self) -> str:
+        """生成WebSocket连接URL"""
+        timestamp = int(datetime.now().timestamp() * 1000)
+        user_id = f"{self.ws_config.get('account')}_{timestamp}_{self.ws_config.get('tenantInfoId')}"
+        return f"wss://{self.ws_config.get('url')}{user_id}"
+
+    async def save_message(self, message: str):
+        """保存接收到的消息到文件"""
+        # 获取当前日期作为文件名
+        current_date = datetime.now().strftime("%Y%m%d")
+        filename = f"{self.message_dir}/messages_{current_date}.json"
+        
+        # 准备要保存的消息数据
+        message_data = {
+            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            "content": message
+        }
+        
+        # 使用文件锁确保并发安全
+        async with self.file_lock:
+            try:
+                # 检查文件是否存在
+                if os.path.exists(filename):
+                    # 如果文件存在，读取现有内容
+                    async with aiofiles.open(filename, 'r', encoding='utf-8') as f:
+                        content = await f.read()
+                        try:
+                            messages = json.loads(content)
+                        except json.JSONDecodeError:
+                            messages = []
+                else:
+                    messages = []
+                
+                # 添加新消息
+                messages.append(message_data)
+                
+                # 保存更新后的内容
+                async with aiofiles.open(filename, 'w', encoding='utf-8') as f:
+                    await f.write(json.dumps(messages, ensure_ascii=False, indent=2))
+                
+                print(f"消息已保存到文件: {filename}")
+                
+            except Exception as e:
+                print(f"保存消息时出错: {str(e)}")
+
+    async def process_message(self, message: str):
+        """处理接收到的消息"""
+        try:
+            # 保存消息
+            await self.save_message(message)
+            
+            # 尝试解析JSON消息
+            message_dict = json.loads(message)
+            
+            # 检查是否需要触发事件同步
+            if self._should_trigger_sync(message_dict):
+                print("触发事件同步...")
+                await run_sync_once()
+                
+        except json.JSONDecodeError:
+            print("收到非JSON格式消息")
+        except Exception as e:
+            print(f"处理消息时出错: {str(e)}")
+
+    def _should_trigger_sync(self, message: dict) -> bool:
+        """判断是否需要触发事件同步"""
+        try:
+            # 示例：当消息类型为"event_update"时触发同步
+            # 根据实际业务需求修改判断条件
+            return message.get("type") == "event_update"
+        except Exception as e:
+            print(f"检查触发条件时出错: {str(e)}")
+            return False
+
+    async def connect_and_listen(self):
+        """连接到WebSocket服务器并监听消息"""
+        while True:
+            try:
+                url = self.get_ws_url()
+                print(f"正在连接到WebSocket服务器: {url}")
+                
+                async with websockets.connect(url) as websocket:
+                    self.is_connected = True
+                    print("WebSocket连接已建立")
+                    
+                    while True:
+                        try:
+                            # 接收消息
+                            message = await websocket.recv()
+                            print(f"收到消息: {message[:100]}...")  # 只打印前100个字符
+                            
+                            # 处理消息
+                            await self.process_message(message)
+                            
+                        except websockets.exceptions.ConnectionClosed:
+                            print("WebSocket连接已关闭")
+                            break
+                        except Exception as e:
+                            print(f"处理消息时出错: {str(e)}")
+                            
+            except Exception as e:
+                self.is_connected = False
+                print(f"WebSocket连接出错: {str(e)}")
+                print(f"将在 {self.reconnect_delay} 秒后重试...")
+                await asyncio.sleep(self.reconnect_delay)
+
+    async def start(self):
+        """启动WebSocket客户端"""
+        print("启动WebSocket客户端...")
+        await self.connect_and_listen()
+
+async def run_websocket_client():
+    """运行WebSocket客户端"""
+    client = WebSocketClient()
+    await client.start()
+
+if __name__ == "__main__":
+    asyncio.run(run_websocket_client()) 

requirements.txt

@@ -1,4 +1,5 @@
 sqlalchemy
 pymysql
 pydantic-settings
-aiomysql
+aiomysql
+aiofiles

run_websocket.py

@@ -0,0 +1,6 @@
+import asyncio
+from app.services.websocket_service import run_websocket_client
+
+if __name__ == "__main__":
+    print("启动WebSocket客户端服务...")
+    asyncio.run(run_websocket_client())