From 78254fa071fc99c78173f00375233c2188a3dcfc Mon Sep 17 00:00:00 2001 From: haotian <2421912570@qq.com> Date: Thu, 22 May 2025 16:21:08 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0websocket=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/config/config.yaml | 5 ++ app/core/database.py | 2 +- app/main.py | 8 +- app/models/models.py | 2 +- app/services/event_sync_service.py | 9 +- app/services/websocket_service.py | 137 +++++++++++++++++++++++++++++ requirements.txt | 3 +- run_websocket.py | 6 ++ 8 files changed, 167 insertions(+), 5 deletions(-) create mode 100644 app/services/websocket_service.py create mode 100644 run_websocket.py diff --git a/app/config/config.yaml b/app/config/config.yaml index f4ed249..5d0daff 100644 --- a/app/config/config.yaml +++ b/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/" diff --git a/app/core/database.py b/app/core/database.py index b2eb2cd..d753eda 100644 --- a/app/core/database.py +++ b/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 ) diff --git a/app/main.py b/app/main.py index b591bfc..585318c 100644 --- a/app/main.py +++ b/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服务正在运行"} \ No newline at end of file diff --git a/app/models/models.py b/app/models/models.py index a60ca6b..651ae00 100644 --- a/app/models/models.py +++ b/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='创建时间') # 关系 diff --git a/app/services/event_sync_service.py b/app/services/event_sync_service.py index b94f941..48a90c3 100644 --- a/app/services/event_sync_service.py +++ b/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()) \ No newline at end of file diff --git a/app/services/websocket_service.py b/app/services/websocket_service.py new file mode 100644 index 0000000..489252f --- /dev/null +++ b/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()) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 4b79e1a..7c7f76b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ sqlalchemy pymysql pydantic-settings -aiomysql \ No newline at end of file +aiomysql +aiofiles \ No newline at end of file diff --git a/run_websocket.py b/run_websocket.py new file mode 100644 index 0000000..1d9a447 --- /dev/null +++ b/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()) \ No newline at end of file