C:/Program Files/Git/api/events加入socketio机制——更新超时时间

2025-10-21 15:00:13 +08:00
parent 4b6d86e923
commit cfb00ba895
3 changed files with 405 additions and 5 deletions
--- a/test_create_event.py
+++ b/test_create_event.py
@@ -0,0 +1,109 @@
+"""
+测试脚本：手动创建事件到数据库
+用于测试 WebSocket 实时推送功能
+"""
+
+import sys
+from datetime import datetime
+from sqlalchemy import create_engine, Column, Integer, String, Text, Float, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+# 数据库连接（从 app.py 复制）
+DATABASE_URI = 'mysql+pymysql://root:Zzl5588161!@111.198.58.126:33060/stock?charset=utf8mb4'
+
+engine = create_engine(DATABASE_URI, echo=False)
+Session = sessionmaker(bind=engine)
+session = Session()
+
+Base = declarative_base()
+
+# Event 模型（简化版，只包含必要字段）
+class Event(Base):
+    __tablename__ = 'events'
+
+    id = Column(Integer, primary_key=True)
+    title = Column(String(500), nullable=False)
+    description = Column(Text)
+    event_type = Column(String(100))
+    importance = Column(String(10))
+    status = Column(String(50), default='active')
+    hot_score = Column(Float, default=0)
+    view_count = Column(Integer, default=0)
+    created_at = Column(DateTime, default=datetime.now)
+    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
+
+
+def create_test_event():
+    """创建一个测试事件"""
+
+    import random
+
+    event_types = ['policy', 'market', 'tech', 'industry', 'finance']
+    importances = ['S', 'A', 'B', 'C']
+
+    test_event = Event(
+        title=f'测试事件 - {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}',
+        description=f'这是一个用于测试 WebSocket 实时推送的事件，创建于 {datetime.now()}',
+        event_type=random.choice(event_types),
+        importance=random.choice(importances),
+        status='active',
+        hot_score=round(random.uniform(50, 100), 2),
+        view_count=random.randint(100, 1000)
+    )
+
+    try:
+        session.add(test_event)
+        session.commit()
+
+        print("✅ 测试事件创建成功!")
+        print(f"   ID: {test_event.id}")
+        print(f"   标题: {test_event.title}")
+        print(f"   类型: {test_event.event_type}")
+        print(f"   重要性: {test_event.importance}")
+        print(f"   热度: {test_event.hot_score}")
+        print(f"\n💡 提示: 轮询将在 2 分钟内检测到此事件并推送到前端")
+        print(f"   （如果需要立即推送，请将轮询间隔改为更短）")
+
+        return test_event.id
+
+    except Exception as e:
+        session.rollback()
+        print(f"❌ 创建事件失败: {e}")
+        return None
+    finally:
+        session.close()
+
+
+def create_multiple_events(count=3):
+    """创建多个测试事件"""
+    print(f"正在创建 {count} 个测试事件...\n")
+
+    for i in range(count):
+        event_id = create_test_event()
+        if event_id:
+            print(f"[{i+1}/{count}] 事件 #{event_id} 创建成功\n")
+        else:
+            print(f"[{i+1}/{count}] 创建失败\n")
+
+    print(f"\n✅ 完成！共创建 {count} 个事件")
+
+
+if __name__ == '__main__':
+    print("=" * 60)
+    print("WebSocket 事件推送测试 - 手动创建事件")
+    print("=" * 60)
+    print()
+
+    if len(sys.argv) > 1:
+        try:
+            count = int(sys.argv[1])
+            create_multiple_events(count)
+        except ValueError:
+            print("❌ 参数必须是数字")
+            print("用法: python test_create_event.py [数量]")
+    else:
+        # 默认创建 1 个事件
+        create_test_event()
+
+    print("\n" + "=" * 60)