"""
producers/youtube_producer.py — Async mock YouTube transcript chunk producer.
Parses WebVTT-style transcript chunks and publishes to 'raw.youtube' at 10 evt/s.
"""
from __future__ import annotations
import asyncio, json, time, os, random
from aiokafka import AIOKafkaProducer

BROKERS = os.getenv("REDPANDA_BROKERS", "localhost:9092")
TOPIC = "raw.youtube"
RATE = 10

MOCK_TRANSCRIPTS = [
    {"id": "yt001", "text": "And according to this study from Harvard Medical School, drinking alkaline water at pH 9.5 increases cellular hydration by 300 percent, which is why I recommend this brand.", "channel": "HealthGuru2024", "verified": False, "vtt_start": "00:04:23.000"},
    {"id": "yt002", "text": "The James Webb Space Telescope has now confirmed the detection of carbon dioxide in the atmosphere of exoplanet WASP-39b, marking a first for atmospheric composition analysis.", "channel": "NASAOfficialChannel", "verified": True, "vtt_start": "00:12:15.000"},
    {"id": "yt003", "text": "So Bitcoin reached an all-time high of $73,000 in March 2024 before pulling back, which technical analysts say set up a classic double-top pattern.", "channel": "CryptoAnalyticsDaily", "verified": False, "vtt_start": "00:02:44.000"},
    {"id": "yt004", "text": "The study published in Nature found that ultra-processed foods were associated with a 50 percent increased risk of cardiovascular disease mortality across the 28-year follow-up period.", "channel": "NutritionScienceReview", "verified": True, "vtt_start": "00:08:30.000"},
    {"id": "yt005", "text": "ChatGPT-4 has an IQ of 155 according to multiple standardized tests I personally administered, making it smarter than 99.9% of humans.", "channel": "AIExpertXYZ", "verified": False, "vtt_start": "00:01:12.000"},
]

async def produce():
    producer = AIOKafkaProducer(
        bootstrap_servers=BROKERS,
        value_serializer=lambda v: json.dumps(v).encode(),
        compression_type="lz4",
    )
    await producer.start()
    print(f"[youtube_producer] Connected to {BROKERS}, publishing to {TOPIC}")
    interval = 1.0 / RATE
    try:
        while True:
            chunk = random.choice(MOCK_TRANSCRIPTS).copy()
            chunk["timestamp"] = time.time()
            await producer.send_and_wait(TOPIC, chunk)
            await asyncio.sleep(interval)
    finally:
        await producer.stop()

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