WebSocket Real-Time Notifications for Video Platforms with Python

Cron Fetcher (PHP) ──→ Redis Pub/Sub ──→ Python WS Server ──→ Browser clients

pip install websockets redis asyncio

import asyncio
import json
import logging
import redis.asyncio as aioredis
import websockets
from websockets.server import WebSocketServerProtocol

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

connected: set[WebSocketServerProtocol] = set()

async def broadcast(message: str) -> None:
    if not connected:
        return
    dead = set()
    results = await asyncio.gather(
        *(ws.send(message) for ws in connected),
        return_exceptions=True,
    )
    for ws, result in zip(list(connected), results):
        if isinstance(result, Exception):
            dead.add(ws)
    connected.difference_update(dead)

async def handler(websocket: WebSocketServerProtocol) -> None:
    connected.add(websocket)
    region = websocket.request_headers.get('X-Region', 'US')
    logger.info(f'Client connected (region={region}), total={len(connected)}')
    try:
        await websocket.wait_closed()
    finally:
        connected.discard(websocket)

async def redis_subscriber() -> None:
    client = aioredis.from_url('redis://localhost:6379', decode_responses=True)
    pubsub = client.pubsub()
    await pubsub.subscribe('topvideohub:trending')

    async for message in pubsub.listen():
        if message['type'] != 'message':
            continue
        try:
            payload = json.loads(message['data'])
            payload['title'] = payload.get('title', '')
            await broadcast(json.dumps(payload, ensure_ascii=False))
        except (json.JSONDecodeError, KeyError) as e:
            logger.warning(f'Bad message: {e}')

async def main() -> None:
    server = await websockets.serve(handler, '0.0.0.0', 8765)
    logger.info('WebSocket server listening on :8765')
    await asyncio.gather(server.wait_closed(), redis_subscriber())

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

<?php
declare(strict_types=1);

class NotificationPublisher
{
    private \Redis $redis;

    public function __construct()
    {
        $this->redis = new \Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }

    public function publishTrending(array $video, string $region): void
    {
        $payload = json_encode([
            'event'     => 'new_trending',
            'region'    => $region,
            'video_id'  => $video['video_id'],
            'title'     => $video['title'],
            'thumbnail' => $video['thumbnail_url'],
            'channel'   => $video['channel_title'],
            'ts'        => time(),
        ], JSON_UNESCAPED_UNICODE);

        $this->redis->publish('topvideohub:trending', $payload);
    }
}

// In the fetch loop:
$publisher = new NotificationPublisher();
foreach ($newVideos as $video) {
    $db->insertVideo($video, $region);
    $publisher->publishTrending($video, $region);
}

class TrendingNotifier {
  constructor(region = 'JP') {
    this.region = region;
    this.ws = null;
    this.reconnectDelay = 1000;
    this.connect();
  }

  connect() {
    this.ws = new WebSocket('wss://ws.topvideohub.com/');
    this.ws.onopen = () => { this.reconnectDelay = 1000; };
    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.event === 'new_trending') this.showToast(data);
    };
    this.ws.onclose = () => {
      setTimeout(() => this.connect(), this.reconnectDelay);
      this.reconnectDelay = Math.min(this.reconnectDelay * 2, 30000);
    };
  }

  showToast(video) {
    const toast = document.createElement('div');
    toast.className = 'trending-toast';
    toast.innerHTML = `<img src="${video.thumbnail}" alt=""><div><strong>New in ${video.region}</strong><span>${video.title}</span></div>`;
    document.body.appendChild(toast);
    setTimeout(() => toast.remove(), 5000);
  }
}

const notifier = new TrendingNotifier(userRegion);