110 lines
3.4 KiB
Python
110 lines
3.4 KiB
Python
"""
|
|
测试脚本：手动创建事件到数据库
|
|
用于测试 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)
|