Real-Time Intelligence in Action : Tracking Prime Time News Channel's YouTube Live Streams with Microsoft Fabric RTI.py

Real-Time Intelligence in Action : Tracking Prime Time News Channel's YouTube Live Streams with Microsoft Fabric RTI.py

#Author : Nalaka W -01.05.2025 # 

from googleapiclient.discovery import build
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from datetime import datetime, timedelta
import pytz
import json
import time
import requests
import re
import sys

# === CONFIGURATION ===

DURATION_MINUTES = 100
FABRIC_CONNECTION_STRING = "Endpoint=sb://"
YOUTUBE_API_KEY = ""
SRI_LANKA_TZ = pytz.timezone("Asia/Colombo")
CHANNEL_HANDLES = ["@AdaDeranaNews", "@HiruNewsOfficial", "@newsfirstsrilanka", "@swarnavahininews_live", "@SiyathaNews", "@ITNNewsOnline"]

# ADD AS MANY VIDEO IDS TO SKIP AS YOU WANT HERE
SKIP_VIDEO_IDS = {"sjYsrbWNTkk", "WdNmIgTxOYA"}

# === INIT YOUTUBE SERVICE ===

youtube = build("youtube", "v3", developerKey=YOUTUBE_API_KEY)

# === MEMORY TRACKING FOR ACTIVE STREAMS ===

active_streams = {}

# === FETCH LIVE VIDEO ID FROM CHANNEL URL ===

def get_live_video_id_from_channel(channel_handle):
    url = f"https://www.youtube.com/{channel_handle}/streams"
    try:
        response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"})
        response.raise_for_status()
        match = re.search(r'"videoId":"(.*?)"', response.text)
        video_id = match.group(1) if match else None
        return None if video_id in SKIP_VIDEO_IDS else video_id
    except Exception as e:
        print(f"❌ Error getting video for {channel_handle}: {e}")
        return None

# === FETCH STREAM DETAILS VIA API ===

def fetch_stream_details(video_ids):
    try:
        video_ids = [vid for vid in video_ids if vid not in SKIP_VIDEO_IDS]
        if not video_ids:
            return []

        request = youtube.videos().list(
            part="snippet,liveStreamingDetails",
            id=",".join(video_ids)
        )
        response = request.execute()
        results = []
        now = datetime.now(SRI_LANKA_TZ)

        for item in response.get("items", []):
            vid = item["id"]
            snippet = item["snippet"]
            live_details = item.get("liveStreamingDetails", {})

            is_live = "actualEndTime" not in live_details
            ended_time = live_details.get("actualEndTime", "")

            results.append({
                "platform": "YouTube",
                "stream_id": vid,
                "Title": snippet.get("title", ""),
                "Channel_Title": snippet.get("channelTitle", ""),
                "published_at": snippet.get("publishedAt", ""),
                "video_url": f"https://www.youtube.com/watch?v={vid}",
                "Live_Viewers": live_details.get("concurrentViewers", "0"),
                "retrieval_time": now.strftime('%Y-%m-%d %H:%M:%S'),
                "retrieval_date": now.strftime('%Y-%m-%d'),
                "retrieval_time_only": now.strftime('%H:%M:%S'),
                "channel_id": snippet.get("channelId", ""),
                "status": "Live" if is_live else "Ended",
                "actual_end_time": ended_time
            })
        return results
    except Exception as e:
        print(f"❌ Error fetching stream details: {e}")
        return []

# === SEND TO FABRIC EVENTSTREAM ===

def send_to_eventstream(messages, connection_string):
    entity_path = next((p.split("=")[1] for p in connection_string.split(";") if p.startswith("EntityPath=")), None)
    if not entity_path:
        raise ValueError("EntityPath not found")
    servicebus_client = ServiceBusClient.from_connection_string(connection_string)
    try:
        with servicebus_client.get_queue_sender(entity_path) as sender:
            batch = [ServiceBusMessage(json.dumps(msg)) for msg in messages]
            sender.send_messages(batch)
            print(f"📤 Sent {len(messages)} messages to EventStream")
    except Exception as e:
        print(f"❌ Error sending to EventStream: {e}")
    finally:
        servicebus_client.close()

# === CONTINUOUS TRACKING LOOP WITH TIMEOUT ===

def run_tracker():
    start_time = datetime.now()
    end_time = start_time + timedelta(minutes=DURATION_MINUTES)

    print(f"🔁 Starting YouTube Stream Tracker for {DURATION_MINUTES} minutes...")

    while True:
        try:
            if datetime.now() >= end_time:
                print("🛑 Time limit reached. Stopping execution.")
                sys.exit(0)

            current_live_streams = {}

            for handle in CHANNEL_HANDLES:
                video_id = get_live_video_id_from_channel(handle)
                if video_id:
                    current_live_streams[video_id] = handle

            all_tracked_ids = list(set(current_live_streams.keys()).union(active_streams.keys()))
            all_tracked_ids = [vid for vid in all_tracked_ids if vid not in SKIP_VIDEO_IDS]

            if all_tracked_ids:
                stream_data = fetch_stream_details(all_tracked_ids)
                messages_to_send = []

                for stream in stream_data:
                    vid = stream["stream_id"]
                    title = stream.get("Title", "").lower()

                    # ✅ Filter: Only include streams with "news" or "ප්‍රවෘත්ති" in the title.
                    # If you want case-insensitive for all, use keywords list in lower-case only.
                    if not any(keyword in title for keyword in ["news", "ප්‍රවෘත්ති"]):
                        continue

                    if vid in SKIP_VIDEO_IDS:
                        continue

                    is_live = stream["status"] == "Live"

                    if is_live:
                        active_streams[vid] = stream
                    elif vid in active_streams:
                        active_streams.pop(vid)

                    messages_to_send.append(stream)

                if messages_to_send:
                    send_to_eventstream(messages_to_send, FABRIC_CONNECTION_STRING)

            else:
                print("⚠️ No live videos found.")

            time.sleep(60)

        except Exception as err:
            print(f"❌ Loop error: {err}")
            time.sleep(60)

# === EXECUTE THE SCRIPT ===

if __name__ == "__main__":
    run_tracker()

#Assets - https://github.com/nalakan/DijiConnect_ISSDemo/tree/main/Other