大屏可视化系统：WebRTC视频流与WebSocket实时数据集成方案

// 前端 WebRTC 配置示例 (基于资料中的代码模式)
const peerConnectionConfig = {
  iceServers: [
    // 公共STUN服务器，用于获取公网地址
    { urls: 'stun:stun.l.google.com:19302' },
    // 自建或第三方TURN服务器，用于在对称NAT/防火墙下中继流量（关键）
    // {
    //   urls: 'turn:your-turn-server.com:3478',
    //   username: 'username',
    //   credential: 'credential'
    // }
  ]
};
const peerConnection = new RTCPeerConnection(peerConnectionConfig);

// src/services/webrtc.service.ts
export class WebRTCService {
  private peerConnection: RTCPeerConnection | null = null;
  private signalingSocket: WebSocket; // 假设已通过WebSocket服务注入

  // ICE服务器配置（与前序配置一致）
  private readonly rtcConfig: RTCConfiguration = {
    iceServers: [
      { urls: 'stun:stun.l.google.com:19302' }, // 公共STUN
      { 
        urls: 'turn:your-turn-server.com:3478', // 预留TURN服务器地址
        username: 'your-username',
        credential: 'your-credential'
      }
    ]
  };

  constructor(signalingService: any) {
    this.signalingSocket = signalingService.getSocket(); // 获取已建立的WebSocket连接
    this.setupSignalingHandlers();
  }
}

// 在 WebRTCService 类中
public async startConnection(streamId: string): Promise<void> {
  // 1. 创建PeerConnection实例
  this.peerConnection = new RTCPeerConnection(this.rtcConfig);

  // 2. 处理ICE候选，并通过WebSocket发送给信令服务器
  this.peerConnection.onicecandidate = (event) => {
    if (event.candidate) {
      this.signalingSocket.send(JSON.stringify({
        type: 'webrtc_signal',
        target: streamId, // 指定目标视频源或信令服务器
        signal: { iceCandidate: event.candidate }
      }));
    }
  };

  // 3. 处理接收到的远端媒体流，并注入到统一数据适配层
  this.peerConnection.ontrack = (event) => {
    const remoteStream = event.streams[0];
    // 关键：将原始MediaStream传递给统一数据适配层进行处理
    window.dispatchEvent(new CustomEvent('webrtc-stream-received', { 
      detail: { streamId, mediaStream: remoteStream }
    }));
    // 同时，也可直接绑定到video元素进行预览（如需要）
    const videoElement = document.getElementById(`video-${streamId}`) as HTMLVideoElement;
    if (videoElement && videoElement.srcObject !== remoteStream) {
      videoElement.srcObject = remoteStream;
    }
  };

  // 4. 创建Offer，设置本地描述，并发送
  try {
    const offer = await this.peerConnection.createOffer();
    await this.peerConnection.setLocalDescription(offer);
    this.signalingSocket.send(JSON.stringify({
      type: 'webrtc_signal',
      target: streamId,
      signal: { sdp: this.peerConnection.localDescription }
    }));
  } catch (error) {
    console.error('创建Offer失败:', error);
  }
}

// 处理从WebSocket收到的远端信令（Answer或ICE候选）
private async handleRemoteSignal(signal: any): Promise<void> {
  if (!this.peerConnection) return;

  if (signal.sdp) {
    const remoteDesc = new RTCSessionDescription(signal.sdp);
    await this.peerConnection.setRemoteDescription(remoteDesc);
    // 如果收到的是Offer（作为接收端），则需要创建Answer
    if (signal.sdp.type === 'offer') {
      const answer = await this.peerConnection.createAnswer();
      await this.peerConnection.setLocalDescription(answer);
      this.signalingSocket.send(JSON.stringify({
        type: 'webrtc_signal',
        signal: { sdp: this.peerConnection.localDescription }
      }));
    }
  } else if (signal.iceCandidate) {
    await this.peerConnection.addIceCandidate(new RTCIceCandidate(signal.iceCandidate));
  }
}

// 前端示例：建立连接与消息分发
class WebSocketService {
  constructor(url) {
    this.socket = new WebSocket(url);
    this.setupEventListeners();
  }

  setupEventListeners() {
    this.socket.onopen = () => {
      console.log('WebSocket连接已建立，可进行身份认证或订阅');
      // 触发连接成功事件，供其他模块响应
      eventBus.emit('WS_CONNECTION_ESTABLISHED');
    };

    this.socket.onmessage = async (event) => {
      try {
        const rawData = JSON.parse(event.data);
        // 将原始数据送入统一数据适配层进行处理
        const processedData = await dataAdapter.process(rawData);
        
        // 根据处理后的数据类型，更新对应的 Zustand Store 或触发事件
        switch(processedData.type) {
          case 'business_data':
            // 更新业务数据Store，例如更新订单量、在线人数
            useRealtimeDashboardStore.getState().updateMetrics(processedData.payload);
            // 同时发布事件，供插件化组件订阅
            eventBus.emit('BUSINESS_DATA_UPDATED', processedData.payload);
            break;
          case 'control_command':
            // 执行控制命令，如切换视图、布局
            executeControlCommand(processedData.payload);
            eventBus.emit('CONTROL_COMMAND_RECEIVED', processedData.payload);
            break;
          case 'webrtc_signal':
            // 将WebRTC信令传递给WebRTC管理模块
            webRTCManager.handleSignal(processedData);
            break;
        }
      } catch (error) {
        console.error('WebSocket消息处理失败:', error);
        // 触发错误事件，可由全局拦截器处理
        eventBus.emit('DATA_PROCESSING_ERROR', { error, rawData: event.data });
      }
    };

    this.socket.onerror = (error) => {
      console.error('WebSocket错误:', error);
      eventBus.emit('WS_CONNECTION_ERROR', error);
    };
  }

  send(data) {
    if (this.socket.readyState === WebSocket.OPEN) {
      this.socket.send(JSON.stringify(data));
    }
  }
}

# Python后端示例 (简化核心逻辑)
import asyncio, websockets, json
from typing import Set

class SignalingServer:
    def __init__(self):
        self.connected_clients = {}  # client_id: websocket
        self.rooms = {}  # room_id: Set[client_id]

    async def register(self, websocket, client_id):
        """客户端注册"""
        self.connected_clients[client_id] = websocket
        print(f"客户端 {client_id} 已连接")

    async def broadcast_business_data(self):
        """模拟广播业务数据（如从消息队列Kafka中读取）"""
        while True:
            if self.connected_clients:
                # 模拟获取实时业务数据
                mock_data = {
                    "type": "business_data",
                    "payload": {
                        "timestamp": time.time(),
                        "orders_last_minute": random.randint(100, 200),
                        "online_users": random.randint(5000, 6000),
                        "gmv": random.uniform(100000, 200000)
                    }
                }
                message = json.dumps(mock_data)
                # 广播给所有连接的客户端（或特定房间）
                tasks = [client.send(message) for client in self.connected_clients.values()]
                await asyncio.gather(*tasks, return_exceptions=True)
            await asyncio.sleep(1)  # 每秒广播一次

    async def handler(self, websocket, path):
        """处理客户端连接"""
        client_id = await self.authenticate(websocket)  # 身份认证
        await self.register(websocket, client_id)
        try:
            async for message in websocket:
                data = json.loads(message)
                # 1. 业务数据请求：直接回复或广播
                if data.get('type') == 'subscribe_metrics':
                    await self.handle_subscription(websocket, data)
                # 2. WebRTC信令：根据target进行点对点转发
                elif data.get('type') == 'webrtc_signal':
                    target_client = self.connected_clients.get(data['target'])
                    if target_client:
                        await target_client.send(json.dumps(data))
                # 3. 控制命令：广播给所有大屏或特定组
                elif data.get('type') == 'control_command':
                    await self.broadcast_to_dashboard_clients(data)
        finally:
            # 连接断开处理
            self.connected_clients.pop(client_id, None)
            print(f"客户端 {client_id} 已断开")

async def main():
    server = SignalingServer()
    start_server = websockets.serve(server.handler, "0.0.0.0", 8765)
    # 并行运行服务器和业务数据广播任务
    await asyncio.gather(start_server, server.broadcast_business_data())

src/
├── main.ts                      # 应用入口
├── App.vue                      # 根组件
├── index.html
├── vite.config.ts               # Vite 配置
├── styles/
│   └── global.css               # 全局样式与主题变量
├── services/
│   ├── websocket.service.ts     # WebSocket 服务封装
│   └── webrtc.service.ts        # WebRTC 服务封装
├── stores/
│   └── dashboard.store.ts       # Zustand 状态管理
├── components/
│   ├── VideoStream.vue          # 视频流展示组件
│   ├── DataDashboard.vue        # 数据可视化仪表盘组件
│   └── LayoutContainer.vue     # 大屏布局容器
└── plugins/
    └── echarts.plugin.ts        # ECharts 插件注册与主题适配

import { defineConfig } from 'vite'
import vue from '@vitejs/plugin-vue'
import path from 'path'

export default defineConfig({
  plugins: [vue()],
  resolve: {
    alias: {
      '@': path.resolve(__dirname, './src'),
    },
  },
  server: {
    port: 5173,
    host: true,
  },
  build: {
    target: 'es2020',
    rollupOptions: {
      output: {
        manualChunks: {
          echarts: ['echarts'],
          'vue-router': ['vue-router'],
        },
      },
    },
  },
})

import mitt, { Emitter } from 'mitt'

type WebSocketMessage = {
  type: 'business_data' | 'webrtc_signal' | 'control_command'
  target?: string
  payload: any
}

type WebSocketEvents = {
  WS_CONNECTION_ESTABLISHED: void
  BUSINESS_DATA_UPDATED: { orders: number; onlineUsers: number; timestamp: number }
  webrtc_signal_received: any
  connection_error: Error
}

export class WebSocketService {
  private socket: WebSocket | null = null
  private eventBus: Emitter<WebSocketEvents> = mitt<WebSocketEvents>()
  private reconnectAttempts = 0
  private readonly maxReconnectAttempts = 5
  private reconnectTimeout: number | null = null

  constructor(private url: string = 'ws://localhost:8765') {}

  connect(): Promise<void> {
    return new Promise((resolve, reject) => {
      if (this.socket?.readyState === WebSocket.OPEN) {
        resolve()
        return
      }

      try {
        this.socket = new WebSocket(this.url)
      } catch (error) {
        reject(error)
        return
      }

      this.socket.onopen = () => {
        console.log('WebSocket连接已建立')
        this.reconnectAttempts = 0
        this.eventBus.emit('WS_CONNECTION_ESTABLISHED')
        resolve()
      }

      this.socket.onmessage = (event) => {
        try {
          const data: WebSocketMessage = JSON.parse(event.data)
          this.handleMessage(data)
        } catch (error) {
          console.error('解析WebSocket消息失败:', error)
        }
      }

      this.socket.onerror = (error) => {
        console.error('WebSocket错误:', error)
        this.eventBus.emit('connection_error', new Error('WebSocket连接错误'))
      }

      this.socket.onclose = (event) => {
        console.log(`WebSocket连接关闭，代码: ${event.code}, 原因: ${event.reason}`)
        this.attemptReconnect()
      }
    })
  }

  private handleMessage(data: WebSocketMessage): void {
    switch (data.type) {
      case 'business_data':
        // 假设 payload 格式为 { orders: number, onlineUsers: number }
        this.eventBus.emit('BUSINESS_DATA_UPDATED', data.payload)
        break
      case 'webrtc_signal':
        this.eventBus.emit('webrtc_signal_received', data.payload)
        break
      case 'control_command':
        console.log('收到控制命令:', data.payload)
        // 可根据命令类型分发到不同处理器
        break
      default:
        console.warn('未知消息类型:', data.type)
    }
  }

  send(data: object): void {
    if (this.socket?.readyState === WebSocket.OPEN) {
      this.socket.send(JSON.stringify(data))
    } else {
      console.warn('WebSocket未连接，消息发送失败:', data)
    }
  }

  private attemptReconnect(): void {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error('达到最大重连次数，停止重连')
      return
    }

    if (this.reconnectTimeout) {
      clearTimeout(this.reconnectTimeout)
    }

    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000)
    this.reconnectAttempts++

    console.log(`将在 ${delay}ms 后尝试第 ${this.reconnectAttempts} 次重连...`)
    this.reconnectTimeout = window.setTimeout(() => {
      this.connect().catch((err) => console.error('重连失败:', err))
    }, delay)
  }

  getEventBus(): Emitter<WebSocketEvents> {
    return this.eventBus
  }

  disconnect(): void {
    if (this.reconnectTimeout) {
      clearTimeout(this.reconnectTimeout)
      this.reconnectTimeout = null
    }
    if (this.socket) {
      this.socket.close(1000, '客户端主动断开')
      this.socket = null
    }
  }
}

// 导出单例
export const webSocketService = new WebSocketService()

import { webSocketService, WebSocketService } from './websocket.service'

interface RTCConfig {
  iceServers: RTCIceServer[]
}

export class WebRTCService {
  private peerConnection: RTCPeerConnection | null = null
  private localStream: MediaStream | null = null
  private remoteStream: MediaStream | null = null
  private dataChannel: RTCDataChannel | null = null
  private readonly config: RTCConfig = {
    iceServers: [
      { urls: 'stun:stun.l.google.com:19302' },
      // 生产环境需配置 TURN 服务器
      // { urls: 'turn:your-turn-server.com:3478', username: 'xxx', credential: 'xxx' }
    ],
  }

  constructor(private signalingService: WebSocketService = webSocketService) {
    this.setupSignalingHandlers()
  }

  private setupSignalingHandlers(): void {
    const eventBus = this.signalingService.getEventBus()
    eventBus.on('webrtc_signal_received', this.handleRemoteSignal.bind(this))
  }

  // 作为接收方，发起连接请求
  async startConnection(streamId: string): Promise<void> {
    if (this.peerConnection) {
      console.warn('已有存在的WebRTC连接，先关闭')
      this.closeConnection()
    }

    try {
      this.peerConnection = new RTCPeerConnection(this.config)

      // 设置 ICE 候选处理
      this.peerConnection.onicecandidate = (event) => {
        if (event.candidate) {
          this.signalingService.send({
            type: 'webrtc_signal',
            target: streamId,
            payload: { iceCandidate: event.candidate },
          })
        }
      }

      // 接收远端流
      this.peerConnection.ontrack = (event) => {
        console.log('收到远端视频流轨道')
        if (event.streams && event.streams[0]) {
          this.remoteStream = event.streams[0]
          // 触发自定义事件，让组件可以获取流
          const streamEvent = new CustomEvent('webrtc-stream-received', {
            detail: { stream: this.remoteStream },
          })
          window.dispatchEvent(streamEvent)
        }
      }

      // 创建数据通道（可选，用于传输控制指令等）
      this.dataChannel = this.peerConnection.createDataChannel('controlChannel')
      this.setupDataChannel()

      // 创建 Offer
      const offer = await this.peerConnection.createOffer()
      await this.peerConnection.setLocalDescription(offer)

      // 发送 Offer 到信令服务器
      this.signalingService.send({
        type: 'webrtc_signal',
        target: streamId,
        payload: { sdp: this.peerConnection.localDescription },
      })

      console.log('WebRTC连接已发起，等待远端应答...')
    } catch (error) {
      console.error('创建WebRTC连接失败:', error)
      throw error
    }
  }

  // 处理远端信令（SDP Answer 或 ICE Candidate）
  async handleRemoteSignal(signal: any): Promise<void> {
    if (!this.peerConnection) {
      console.warn('收到信令时，PeerConnection 未初始化')
      return
    }

    try {
      if (signal.sdp) {
        const remoteDesc = new RTCSessionDescription(signal.sdp)
        await this.peerConnection.setRemoteDescription(remoteDesc)
        console.log('已设置远端SDP描述')

        // 如果收到的是Offer，需要创建Answer（本例中我们是接收方，通常只处理Answer）
        if (signal.sdp.type === 'offer') {
          const answer = await this.peerConnection.createAnswer()
          await this.peerConnection.setLocalDescription(answer)
          this.signalingService.send({
            type: 'webrtc_signal',
            target: 'sender', // 应替换为实际发送方ID
            payload: { sdp: this.peerConnection.localDescription },
          })
        }
      } else if (signal.iceCandidate) {
        await this.peerConnection.addIceCandidate(new RTCIceCandidate(signal.iceCandidate))
        console.log('已添加ICE候选')
      }
    } catch (error) {
      console.error('处理远端信令失败:', error)
    }
  }

  private setupDataChannel(): void {
    if (!this.dataChannel) return

    this.dataChannel.onopen = () => {
      console.log('RTCDataChannel 已打开')
      // 可以发送控制指令
      this.dataChannel?.send(JSON.stringify({ type: 'handshake', message: '通道就绪' }))
    }

    this.dataChannel.onmessage = (event) => {
      console.log('收到DataChannel消息:', event.data)
      // 处理来自远端的控制指令或数据
      try {
        const data = JSON.parse(event.data)
        // 分发处理...
      } catch (e) {
        console.log('收到非JSON消息:', event.data)
      }
    }

    this.dataChannel.onerror = (error) => {
      console.error('DataChannel错误:', error)
    }
  }

  sendDataViaChannel(data: object): void {
    if (this.dataChannel?.readyState === 'open') {
      this.dataChannel.send(JSON.stringify(data))
    } else {
      console.warn('DataChannel未就绪，消息发送失败')
    }
  }

  closeConnection(): void {
    if (this.dataChannel) {
      this.dataChannel.close()
      this.dataChannel = null
    }
    if (this.peerConnection) {
      this.peerConnection.close()
      this.peerConnection = null
    }
    if (this.localStream) {
      this.localStream.getTracks().forEach((track) => track.stop())
      this.localStream = null
    }
    this.remoteStream = null
    console.log('WebRTC连接已关闭')
  }

  getRemoteStream(): MediaStream | null {
    return this.remoteStream
  }
}

// 导出单例
export const webRTCService = new WebRTCService()

import { create } from 'zustand'
import { subscribeWithSelector } from 'zustand/middleware'

interface DashboardState {
  // 业务数据
  orderCount: number
  onlineUserCount: number
  lastUpdateTime: number | null
  // 系统状态
  isWebSocketConnected: boolean
  isVideoStreamActive: boolean
  currentLayout: 'grid' | 'focus' | 'custom'
  // 筛选条件
  timeRange: 'realtime' | 'hourly' | 'daily'
  selectedRegion: string | null
}

interface DashboardActions {
  updateBusinessData: (orders: number, onlineUsers: number) => void
  setWebSocketStatus: (connected: boolean) => void
  setVideoStreamStatus: (active: boolean) => void
  switchLayout: (layout: DashboardState['currentLayout']) => void
  setTimeRange: (range: DashboardState['timeRange']) => void
  setSelectedRegion: (region: string | null) => void
  reset: () => void
}

const initialState: DashboardState = {
  orderCount: 0,
  onlineUserCount: 0,
  lastUpdateTime: null,
  isWebSocketConnected: false,
  isVideoStreamActive: false,
  currentLayout: 'grid',
  timeRange: 'realtime',
  selectedRegion: null,
}

export const useDashboardStore = create<DashboardState & DashboardActions>()(
  subscribeWithSelector((set) => ({
    ...initialState,

    updateBusinessData: (orders, onlineUsers) =>
      set({
        orderCount: orders,
        onlineUserCount: onlineUsers,
        lastUpdateTime: Date.now(),
      }),

    setWebSocketStatus: (connected) => set({ isWebSocketConnected: connected }),

    setVideoStreamStatus: (active) => set({ isVideoStreamActive: active }),

    switchLayout: (layout) => set({ currentLayout: layout }),

    setTimeRange: (range) => set({ timeRange: range }),

    setSelectedRegion: (region) => set({ selectedRegion: region }),

    reset: () => set(initialState),
  }))
)

// 可选：订阅状态变化，用于持久化或日志
useDashboardStore.subscribe(
  (state) => [state.orderCount, state.onlineUserCount],
  ([orders, users]) => {
    console.log(`业务数据更新 - 订单: ${orders}, 在线用户: ${users}`)
  }
)

<template>
  <div class="video-stream-container">
    <div class="video-header">
      <h3>{{ title }}</h3>
      <div class="status-indicator" :class="{ active: isStreamActive }"></div>
    </div>
    <div class="video-wrapper">
      <video
        ref="videoElement"
        autoplay
        playsinline
        muted
        class="video-element"
        :class="{ 'has-stream': isStreamActive }"
      ></video>
      <div v-if="!isStreamActive" class="video-placeholder">
        <div class="placeholder-icon">📹</div>
        <p>等待视频流连接...</p>
        <button v-if="showConnectButton" @click="emit('connect')" class="connect-btn">
          连接视频流
        </button>
      </div>
    </div>
    <div v-if="showStats" class="video-stats">
      <span>分辨率: {{ videoStats.resolution }}</span>
      <span>帧率: {{ videoStats.frameRate }} fps</span>
    </div>
  </div>
</template>

<script setup lang="ts">
import { ref, onMounted, onUnmounted, watch } from 'vue'
import { webRTCService } from '@/services/webrtc.service'

interface Props {
  title?: string
  streamId?: string
  showConnectButton?: boolean
  showStats?: boolean
}

const props = withDefaults(defineProps<Props>(), {
  title: '实时视频流',
  streamId: 'default-stream',
  showConnectButton: true,
  showStats: true,
})

const emit = defineEmits<{
  connect: []
  streamActive: [isActive: boolean]
}>()

const videoElement = ref<HTMLVideoElement | null>(null)
const isStreamActive = ref(false)
const videoStats = ref({
  resolution: 'N/A',
  frameRate: 0,
})

let statsInterval: number | null = null

const handleStreamReceived = (event: Event) => {
  const customEvent = event as CustomEvent<{ stream: MediaStream }>
  if (videoElement.value && customEvent.detail?.stream) {
    videoElement.value.srcObject = customEvent.detail.stream
    isStreamActive.value = true
    emit('streamActive', true)
    startStatsMonitoring(customEvent.detail.stream)
  }
}

const startStatsMonitoring = (stream: MediaStream) => {
  if (statsInterval) clearInterval(statsInterval)

  statsInterval = window.setInterval(() => {
    if (videoElement.value && videoElement.value.videoWidth) {
      videoStats.value = {
        resolution: `${videoElement.value.videoWidth}x${videoElement.value.videoHeight}`,
        frameRate: Math.round(getFrameRate()),
      }
    }
  }, 1000)
}

const getFrameRate = (): number => {
  // 简化实现，实际应使用 VideoFrameCallback API 或计算时间差
  return 30 // 默认值
}

const connectToStream = async () => {
  try {
    await webRTCService.startConnection(props.streamId)
  } catch (error) {
    console.error('连接视频流失败:', error)
    alert('无法连接视频流，请检查网络和后端服务。')
  }
}

onMounted(() => {
  window.addEventListener('webrtc-stream-received', handleStreamReceived)
  // 组件挂载时自动连接（可选）
  // connectToStream()
})

onUnmounted(() => {
  window.removeEventListener('webrtc-stream-received', handleStreamReceived)
  if (statsInterval) clearInterval(statsInterval)
  webRTCService.closeConnection()
  isStreamActive.value = false
  emit('streamActive', false)
})

watch(
  () => props.streamId,
  (newId) => {
    if (newId && isStreamActive.value) {
      // 如果streamId变化且当前有活动流，重新连接
      webRTCService.closeConnection()
      connectToStream()
    }
  }
)
</script>

<template>
  <div class="data-dashboard">
    <div class="dashboard-header">
      <h2>业务运营概览</h2>
      <div class="controls">
        <select v-model="selectedTimeRange" @change="handleTimeRangeChange">
          <option value="realtime">实时</option>
          <option value="hourly">小时</option>
          <option value="daily">日度</option>
        </select>
        <button @click="refreshData" class="refresh-btn" :disabled="isLoading">
          {{ isLoading ? '更新中...' : '刷新' }}
        </button>
      </div>
    </div>

    <div class="stats-cards">
      <div class="stat-card">
        <div class="stat-icon">📈</div>
        <div class="stat-content">
          <div class="stat-label">今日订单量</div>
          <div class="stat-value">{{ formatNumber(orderCount) }}</div>
          <div class="stat-trend" :class="orderTrendClass">
            {{ orderTrend }}%
          </div>
        </div>
      </div>
      <div class="stat-card">
        <div class="stat-icon">👥</div>
        <div class="stat-content">
          <div class="stat-label">当前在线人数</div>
          <div class="stat-value">{{ formatNumber(onlineUserCount) }}</div>
          <div class="stat-trend" :class="userTrendClass">
            {{ userTrend }}%
          </div>
        </div>
      </div>
      <div class="stat-card">
        <div class="stat-icon">💰</div>
        <div class="stat-content">
          <div class="stat-label">实时成交额</div>
          <div class="stat-value">¥{{ formatNumber(realtimeAmount) }}</div>
          <div class="stat-sub">每分钟更新</div>
        </div>
      </div>
    </div>

    <div class="charts-container">
      <div class="chart-wrapper">
        <div ref="ordersChartRef" class="chart"></div>
      </div>
      <div class="chart-wrapper">
        <div ref="usersChartRef" class="chart"></div>
      </div>
    </div>

    <div class="last-update">
      最后更新: {{ lastUpdateTime ? formatTime(lastUpdateTime) : '--' }}
    </div>
  </div>
</template>

<script setup lang="ts">
import { ref, onMounted, onUnmounted, watch, nextTick } from 'vue'
import * as echarts from 'echarts'
import { useDashboardStore } from '@/stores/dashboard.store'
import { webSocketService } from '@/services/websocket.service'

// 状态管理
const dashboardStore = useDashboardStore()
const orderCount = ref(0)
const onlineUserCount = ref(0)
const lastUpdateTime = ref<number | null>(null)
const realtimeAmount = ref(0)

// 图表相关
const ordersChartRef = ref<HTMLElement | null>(null)
const usersChartRef = ref<HTMLElement | null>(null)
let ordersChart: echarts.ECharts | null = null
let usersChart: echarts.ECharts | null = null

// UI状态
const selectedTimeRange = ref<'realtime' | 'hourly' | 'daily'>('realtime')
const isLoading = ref(false)
const orderTrend = ref(0)
const userTrend = ref(0)

const orderTrendClass = computed(() => (orderTrend.value >= 0 ? 'positive' : 'negative'))
const userTrendClass = computed(() => (userTrend.value >= 0 ? 'positive' : 'negative'))

// 模拟历史数据（实际应从后端获取）
const historicalOrders = ref<number[]>([120, 135, 118, 145, 160, 155, 170, 165, 180, 175])
const historicalUsers = ref<number[]>([850, 920, 880, 950, 1000, 980, 1050, 1020, 1100, 1080])

onMounted(() => {
  initCharts()
  setupWebSocketListener()
  // 模拟初始数据
  simulateDataUpdate()
})

onUnmounted(() => {
  if (ordersChart) ordersChart.dispose()
  if (usersChart) usersChart.dispose()
})

const initCharts = () => {
  nextTick(() => {
    if (ordersChartRef.value) {
      ordersChart = echarts.init(ordersChartRef.value)
      updateOrdersChart()
    }
    if (usersChartRef.value) {
      usersChart = echarts.init(usersChartRef.value)
      updateUsersChart()
    }

    // 响应窗口大小变化
    window.addEventListener('resize', handleResize)
  })
}

const handleResize = () => {
  ordersChart?.resize()
  usersChart?.resize()
}

const setupWebSocketListener = () => {
  const eventBus = webSocketService.getEventBus()
  eventBus.on('BUSINESS_DATA_UPDATED', handleBusinessDataUpdate)
}

const handleBusinessDataUpdate = (data: { orders: number; onlineUsers: number; timestamp: number }) => {
  orderCount.value = data.orders
  onlineUserCount.value = data.onlineUsers
  lastUpdateTime.value = data.timestamp

  // 更新趋势（简化计算）
  if (historicalOrders.value.length > 0) {
    const lastOrder = historicalOrders.value[historicalOrders.value.length - 1]
    orderTrend.value = ((data.orders - lastOrder) / lastOrder) * 100
  }
  if (historicalUsers.value.length > 0) {
    const lastUser = historicalUsers.value[historicalUsers.value.length - 1]
    userTrend.value = ((data.onlineUsers - lastUser) / lastUser) * 100
  }

  // 更新历史数据（模拟）
  historicalOrders.value.push(data.orders)
  historicalUsers.value.push(data.onlineUsers)
  if (historicalOrders.value.length > 20) {
    historicalOrders.value.shift()
    historicalUsers.value.shift()
  }

  // 更新图表
  updateOrdersChart()
  updateUsersChart()

  // 模拟实时成交额变化
  realtimeAmount.value = data.orders * 158 // 假设平均客单价
}

const updateOrdersChart = () => {
  if (!ordersChart) return

  const option: echarts.EChartsOption = {
    backgroundColor: 'transparent',
    tooltip: {
      trigger: 'axis',
      formatter: '{b}<br/>订单量: {c}',
    },
    grid: {
      left: '3%',
      right: '4%',
      bottom: '3%',
      top: '10%',
      containLabel: true,
    },
    xAxis: {
      type: 'category',
      data: historicalOrders.value.map((_, i) => `T-${historicalOrders.value.length - i}`),
      axisLine: { lineStyle: { color: '#666' } },
      axisLabel: { color: '#999' },
    },
    yAxis: {
      type: 'value',
      axisLine: { lineStyle: { color: '#666' } },
      axisLabel: { color: '#999' },
      splitLine: { lineStyle: { color: '#333', type: 'dashed' } },
    },
    series: [
      {
        name: '订单量',
        type: 'line',
        data: historicalOrders.value,
        smooth: true,
        lineStyle: { color: '#5470c6', width: 3 },
        itemStyle: { color: '#5470c6' },
        areaStyle: {
          color: new echarts.graphic.LinearGradient(0, 0, 0, 1, [
            { offset: 0, color: 'rgba(84, 112, 198, 0.5)' },
            { offset: 1, color: 'rgba(84, 112, 198, 0.1)' },
          ]),
        },
      },
    ],
  }

  ordersChart.setOption(option)
}

const updateUsersChart = () => {
  if (!usersChart) return

  const option: echarts.EChartsOption = {
    backgroundColor: 'transparent',
    tooltip: {
      trigger: 'axis',
      formatter: '{b}<br/>在线人数: {c}',
    },
    grid: {
      left: '3%',
      right: '4%',
      bottom: '3%',
      top: '10%',
      containLabel: true,
    },
    xAxis: {
      type: 'category',
      data: historicalUsers.value.map((_, i) => `T-${historicalUsers.value.length - i}`),
      axisLine: { lineStyle: { color: '#666' } },
      axisLabel: { color: '#999' },
    },
    yAxis: {
      type: 'value',
      axisLine: { lineStyle: { color: '#666' } },
      axisLabel: { color: '#999' },
      splitLine: { lineStyle: { color: '#333', type: 'dashed' } },
    },
    series: [
      {
        name: '在线人数',
        type: 'bar',
        data: historicalUsers.value,
        itemStyle: {
          color: new echarts.graphic.LinearGradient(0, 0, 0, 1, [
            { offset: 0, color: '#91cc75' },
            { offset: 1, color: '#fac858' },
          ]),
        },
      },
    ],
  }

  usersChart.setOption(option)
}

const handleTimeRangeChange = () => {
  isLoading.value = true
  // 模拟根据时间范围获取不同数据
  setTimeout(() => {
    // 这里实际应调用API获取对应时间范围的数据
    console.log(`切换时间范围到: ${selectedTimeRange.value}`)
    isLoading.value = false
  }, 500)
}

const refreshData = () => {
  isLoading.value = true
  // 模拟手动刷新
  setTimeout(() => {
    simulateDataUpdate()
    isLoading.value = false
  }, 800)
}

const simulateDataUpdate = () => {
  // 模拟WebSocket数据更新
  const mockData = {
    orders: Math.floor(Math.random() * 200) + 100, // 100-300
    onlineUsers: Math.floor(Math.random() * 500) + 800, // 800-1300
    timestamp: Date.now(),
  }
  handleBusinessDataUpdate(mockData)
}

const formatNumber = (num: number): string => {
  return num.toLocaleString('zh-CN')
}

const formatTime = (timestamp: number): string => {
  const date = new Date(timestamp)
  return `${date.getHours().toString().padStart(2, '0')}:${date.getMinutes().toString().padStart(2, '0')}:${date.getSeconds().toString().padStart(2, '0')}`
}
</script>

<style scoped>
.data-dashboard {
  background: var(--card-bg-color);
  border-radius: var(--border-radius-base);
  padding: var(--spacing-base);
  height: 100%;
  display: flex;
  flex-direction: column;
}

.dashboard-header {
  display: flex;
  justify-content: space-between;
  align-items: center;
  margin-bottom: 24px;
}

.dashboard-header h2 {
  font-size: 1.3rem;
  font-weight: 600;
}

.controls {
  display: flex;
  gap: 12px;
  align-items: center;
}

.controls select {
  padding: 6px 12px;
  background-color: #2a3a52;
  color: var(--text-color-primary);
  border: 1px solid #3a4a62;
  border-radius: 4px;
  font-size: 0.9rem;
}

.refresh-btn {
  padding: 6px 16px;
  background-color: var(--primary-color);
  color: white;
  border: none;
  border-radius: 4px;
  cursor: pointer;
  font-size: 0.9rem;
  transition: background-color 0.2s;
}

.refresh-btn:hover:not(:disabled) {
  background-color: #1a7ad9;
}

.refresh-btn:disabled {
  opacity: 0.6;
  cursor: not-allowed;
}

.stats-cards {
  display: grid;
  grid-template-columns: repeat(3, 1fr);
  gap: 16px;
  margin-bottom: 24px;
}

.stat-card {
  background: linear-gradient(135deg, #1e2a3e 0%, #253044 100%);
  border-radius: 8px;
  padding: 20px;
  display: flex;
  align-items: center;
  border: 1px solid #2d3a4f;
}

.stat-icon {
  font-size: 2rem;
  margin-right: 16px;
}

.stat-content {
  flex: 1;
}

.stat-label {
  font-size: 0.9rem;
  color: var(--text-color-secondary);
  margin-bottom: 4px;
}

.stat-value {
  font-size: 1.8rem;
  font-weight: 700;
  margin-bottom: 4px;
}

.stat-trend {
  font-size: 0.85rem;
  font-weight: 600;
}

.stat-trend.positive {
  color: #52c41a;
}

.stat-trend.negative {
  color: #ff4d4f;
}

.stat-sub {
  font-size: 0.8rem;
  color: var(--text-color-secondary);
}

.charts-container {
  display: grid;
  grid-template-columns: 1fr 1fr;
  gap: 16px;
  flex: 1;
  min-height: 0;
}

.chart-wrapper {
  background: #1a2332;
  border-radius: 8px;
  padding: 12px;
}

.chart {
  width: 100%;
  height: 250px;
}

.last-update {
  margin-top: 16px;
  text-align: right;
  font-size: 0.85rem;
  color: var(--text-color-secondary);
}
</style>

<template>
  <LayoutContainer />
</template>

<script setup lang="ts">
import LayoutContainer from './components/LayoutContainer.vue'
</script>

<style>
#app {
  width: 100vw;
  height: 100vh;
  overflow: hidden;
}
</style>

import { createApp } from 'vue'
import App from './App.vue'
import './styles/global.css'

// 注册ECharts插件
import { initECharts } from './plugins/echarts.plugin'
initECharts()

const app = createApp(App)
app.mount('#app')

import * as echarts from 'echarts'

export function initECharts() {
  // 注册主题（可选）
  echarts.registerTheme('dashboard-dark', {
    backgroundColor: 'transparent',
    textStyle: {
      color: '#e5e6eb',
    },
    title: {
      textStyle: {
        color: '#e5e6eb',
      },
    },
    line: {
      itemStyle: {
        borderWidth: 2,
      },
      lineStyle: {
        width: 3,
      },
      symbolSize: 8,
      symbol: 'circle',
    },
  })

  // 全局配置
  echarts.setOptions({
    useUTC: false,
    animationDuration: 300,
    animationEasing: 'cubicOut',
  })

  console.log('ECharts插件已初始化，主题已注册')
}

#!/usr/bin/env python3
"""
简易WebSocket服务器，模拟业务数据推送和WebRTC信令转发
运行: python server.py
前端连接: ws://localhost:8765
"""
import asyncio
import json
import random
import websockets
from datetime import datetime
from typing import Set

# 存储所有连接的客户端
connected_clients: Set[websockets.WebSocketServerProtocol] = set()

async def broadcast_business_data():
    """每秒广播一次模拟的业务数据"""
    while True:
        if connected_clients:
            data = {
                "type": "business_data",
                "payload": {
                    "orders": random.randint(100, 300),
                    "onlineUsers": random.randint(800, 1300),
                    "timestamp": int(datetime.now().timestamp() * 1000)
                }
            }
            message = json.dumps(data)
            # 广播给所有客户端
            await asyncio.gather(
                *[client.send(message) for client in connected_clients],
                return_exceptions=True
            )
        await asyncio.sleep(1)  # 每秒更新一次

async def handle_client(websocket):
    """处理单个客户端连接"""
    # 注册新客户端
    connected_clients.add(websocket)
    client_id = id(websocket)
    print(f"客户端 {client_id} 已连接，当前连接数: {len(connected_clients)}")
    
    try:
        # 发送欢迎消息
        welcome_msg = {
            "type": "control_command",
            "payload": {
                "command": "welcome",
                "message": f"已连接到服务器，您的ID: {client_id}",
                "timestamp": int(datetime.now().timestamp() * 1000)
            }
        }
        await websocket.send(json.dumps(welcome_msg))
        
        # 处理来自客户端的消息
        async for message in websocket:
            try:
                data = json.loads(message)
                print(f"收到来自客户端 {client_id} 的消息: {data['type']}")
                
                # 根据消息类型处理
                if data["type"] == "webrtc_signal":
                    # WebRTC信令消息，广播给所有其他客户端（简单示例）
                    # 实际应用中应根据target字段定向转发
                    data["sender"] = client_id
                    broadcast_msg = json.dumps(data)
                    tasks = []
                    for client in connected_clients:
                        if client != websocket:
                            tasks.append(client.send(broadcast_msg))
                    if tasks:
                        await asyncio.gather(*tasks, return_exceptions=True)
                        
                elif data["type"] == "control_command":
                    # 控制命令，记录日志
                    print(f"控制命令: {data['payload']}")
                    # 可以在这里处理特定命令并广播响应
                    
            except json.JSONDecodeError:
                print(f"客户端 {client_id} 发送了无效的JSON消息")
            except KeyError:
                print(f"客户端 {client_id} 发送的消息格式错误")
                
    except websockets.exceptions.ConnectionClosed:
        print(f"客户端 {client_id} 连接已关闭")
    finally:
        # 移除断开连接的客户端
        connected_clients.remove(websocket)
        print(f"客户端 {client_id} 已断开，当前连接数: {len(connected_clients)}")

async def main():
    """启动WebSocket服务器"""
    # 启动业务数据广播任务
    broadcast_task = asyncio.create_task(broadcast_business_data())
    
    # 启动WebSocket服务器
    server = await websockets.serve(
        handle_client,
        "0.0.0.0",  # 监听所有接口
        8765,        # 端口
        ping_interval=20,   # 每20秒发送一次ping
        ping_timeout=40      # 40秒无响应则断开
    )
    
    print("WebSocket服务器已启动，监听 ws://0.0.0.0:8765")
    print("按 Ctrl+C 停止服务器")
    
    try:
        await server.wait_closed()
    except KeyboardInterrupt:
        print("\n正在关闭服务器...")
    finally:
        # 取消广播任务
        broadcast_task.cancel()
        try:
            await broadcast_task
        except asyncio.CancelledError:
            pass
        
        # 关闭所有客户端连接
        if connected_clients:
            print(f"正在关闭 {len(connected_clients)} 个客户端连接...")
            await asyncio.gather(
                *[client.close() for client in connected_clients],
                return_exceptions=True
            )

if __name__ == "__main__":
    asyncio.run(main())