import asyncio import websockets import logging import ssl from dotenv import load_dotenv import os from aiohttp import web # --- 設定 --- # 載入 .env 檔案中的環境變數 # 請確認路徑正確 load_dotenv(dotenv_path='/home/winway/web/broadcast/.env') # 設定日誌 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # 從 .env 讀取設定 SSL_KEYFILE = os.getenv('SSL_KEYFILE') SSL_CERTFILE = os.getenv('SSL_CERTFILE') ALLOWED_ORIGINS = [o for o in os.getenv('ALLOWED_ORIGINS', '').split(',') if o] # 存放所有已連線的客戶端 CONNECTED_CLIENTS = set() # 專門存放廣播者 BROADCASTER_WS = None # 存放 WebM 標頭，以便傳送給新連線的客戶端 WEBM_HEADER = None # --- WebSocket 處理邏輯 --- async def ws_handler(websocket): """ 處理每一個 WebSocket 連線 """ global BROADCASTER_WS, WEBM_HEADER origin = websocket.request.headers.get('Origin') if not origin or origin not in ALLOWED_ORIGINS: logging.warning(f"拒絕來自不允許或未知來源的連線: {origin}") return logging.info(f"新客戶端連線: {websocket.remote_address} (來自: {origin})") CONNECTED_CLIENTS.add(websocket) # 如果 header 尚未完成，將 client 加入等待清單，header 完成後主動補發 if BROADCASTER_WS and WEBM_HEADER: # 只有 header 完成才發送，且長度必須大於 4096 bytes try: assert len(WEBM_HEADER) > 4096, f"WEBM_HEADER 長度異常: {len(WEBM_HEADER)} bytes，拒絕發送！" logging.info(f"正在傳送已儲存的標頭給新客戶端 {websocket.remote_address}，header 長度: {len(WEBM_HEADER)} bytes") assert isinstance(WEBM_HEADER, bytes), "WEBM_HEADER 必須是 bytes" await websocket.send(WEBM_HEADER) except AssertionError as e: logging.error(f"[BUG] 嘗試發送不完整 header: {e}") return except websockets.ConnectionClosed: logging.warning(f"傳送標頭時，客戶端 {websocket.remote_address} 已斷線。") CONNECTED_CLIENTS.remove(websocket) return else: # header 尚未完成，將 client 加入等待清單 if not hasattr(ws_handler, 'pending_clients'): ws_handler.pending_clients = set() ws_handler.pending_clients.add(websocket) try: # --- 廣播者 header 累積 --- header_buffer = b'' received_header = False def is_webm_header_complete(buf): # EBML header 通常以 0x1A45DFA3 開頭，且包含 'webm'，長度通常大於 4096 bytes return buf.startswith(b'\x1A\x45\xDF\xA3') and b'webm' in buf and len(buf) > 4096 async for message in websocket: # 如果還沒有廣播者，第一位傳送訊息的客戶端就成為廣播者 if not BROADCASTER_WS: logging.info(f"第一個傳送訊息的 client: {websocket.remote_address}，成為廣播者") BROADCASTER_WS = websocket header_buffer += message if is_webm_header_complete(header_buffer): WEBM_HEADER = header_buffer received_header = True logging.info(f"新廣播者指定為: {websocket.remote_address}. 已儲存 header，大小: {len(WEBM_HEADER)} bytes.") logging.info(f"header 前 32 bytes: {WEBM_HEADER[:32].hex(' ')}") # header 完成後，主動補發給所有等待中的 client if hasattr(ws_handler, 'pending_clients'): for pending_ws in list(ws_handler.pending_clients): try: assert len(WEBM_HEADER) > 4096, f"WEBM_HEADER 長度異常: {len(WEBM_HEADER)} bytes，拒絕補發！" logging.info(f"主動補發 header 給等待中的 client {pending_ws.remote_address}，header 長度: {len(WEBM_HEADER)} bytes") await pending_ws.send(WEBM_HEADER) except AssertionError as e: logging.error(f"[BUG] 嘗試補發不完整 header: {e}") except Exception as e: logging.warning(f"補發 header 給 {pending_ws.remote_address} 失敗: {e}") finally: ws_handler.pending_clients.discard(pending_ws) else: logging.info(f"累積 header 中，目前 {len(header_buffer)} bytes... 前 32 bytes: {header_buffer[:32].hex(' ')}") # 不廣播 header 之前的資料 if not received_header: continue # 只有來自廣播者的訊息才需要被轉發 if websocket is BROADCASTER_WS: if not received_header: # header 尚未累積完成，繼續累積 header_buffer += message if is_webm_header_complete(header_buffer): WEBM_HEADER = header_buffer received_header = True logging.info(f"header 累積完成，大小: {len(WEBM_HEADER)} bytes.") logging.info(f"header 前 32 bytes: {WEBM_HEADER[:32].hex(' ')}") else: logging.info(f"累積 header 中，目前 {len(header_buffer)} bytes... 前 32 bytes: {header_buffer[:32].hex(' ')}") continue logging.info(f"從廣播者 {websocket.remote_address} 收到訊息，長度: {len(message)} bytes") # 廣播訊息給所有其他已連線的客戶端 (收聽者) listeners = {client for client in CONNECTED_CLIENTS if client is not BROADCASTER_WS} if listeners: logging.info(f"正在廣播訊息給 {len(listeners)} 個客戶端...") await asyncio.gather( *[client.send(message) for client in listeners], return_exceptions=True ) else: logging.info("沒有其他客戶端可以廣播。") else: logging.warning(f"收到來自非廣播者客戶端 {websocket.remote_address} 的訊息，已忽略。") except websockets.ConnectionClosed as e: logging.info(f"客戶端斷開連線: {websocket.remote_address} - {e.code} {e.reason}") except Exception as e: logging.error(f"處理 {websocket.remote_address} 時發生錯誤: {e}", exc_info=True) finally: # 如果是廣播者斷線，則重設狀態 if websocket is BROADCASTER_WS: logging.info(f"廣播者 {websocket.remote_address} 已斷線。重設廣播狀態。") BROADCASTER_WS = None WEBM_HEADER = None # 從等待清單移除（避免 zombie client） if hasattr(ws_handler, 'pending_clients'): ws_handler.pending_clients.discard(websocket) CONNECTED_CLIENTS.remove(websocket) logging.info(f"已移除客戶端: {websocket.remote_address}. 目前連線數: {len(CONNECTED_CLIENTS)}") # --- HTTP 處理邏輯 (用於狀態查詢) --- async def get_connected_clients(request): """ HTTP 處理函數，回傳連線中的客戶端列表 """ client_list = [str(client.remote_address) for client in CONNECTED_CLIENTS] return web.json_response({ 'count': len(client_list), 'clients': client_list }) async def main(): if not all([SSL_KEYFILE, SSL_CERTFILE]): logging.error("SSL_KEYFILE 或 SSL_CERTFILE 未在 .env 中設定。") return if not os.path.exists(SSL_KEYFILE) or not os.path.exists(SSL_CERTFILE): logging.error(f"SSL 憑證檔案不存在。請檢查路徑: KEY={SSL_KEYFILE}, CERT={SSL_CERTFILE}") return ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) ssl_context.load_cert_chain(SSL_CERTFILE, SSL_KEYFILE) # --- 啟動 aiohttp HTTP 伺服器 (在 8766 port) --- app = web.Application() app.router.add_get('/clients', get_connected_clients) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, '0.0.0.0', 8766) await site.start() logging.info("HTTP 狀態伺服器已在 http://0.0.0.0:8766 啟動") # --- 啟動 WebSocket 伺服器 (在 8765 port) --- async with websockets.serve(ws_handler, "0.0.0.0", 8765, ssl=ssl_context, origins=ALLOWED_ORIGINS): logging.info(f"WebSocket 伺服器已在 wss://0.0.0.0:8765 啟動，允許的來源: {ALLOWED_ORIGINS}") await asyncio.Future() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n伺服器正在關閉...") except Exception as e: logging.error(f"伺服器啟動失敗: {e}", exc_info=True)