local_worker/role_backfiller.py

import os
import sys
import time
import requests
from dotenv import load_dotenv

# Load local environment settings
load_dotenv()

API_URL = os.getenv("API_URL", "http://localhost:3000")
API_KEY = os.getenv("API_KEY", "")

if not API_KEY:
    print("[Error] API_KEY is missing in .env! Cannot push data to backend.")
    sys.exit(1)

headers = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json"
}

def get_all_streams():
    """Fetch all streams from the backend"""
    try:
        res = requests.get(f"{API_URL}/api/streams", headers=headers, timeout=10)
        if res.status_code == 200:
            data = res.json()
            # API returns { streams: [...] } or just [...]
            if isinstance(data, list):
                return data
            return data.get("streams", [])
    except Exception as e:
        print(f"[Error] Не удалось получить список стримов: {e}")
    return []

def find_vod_id_by_stream_id(stream_id):
    """Try to find a VOD ID for a given Twitch stream ID using GQL"""
    url = "https://gql.twitch.tv/gql"
    gql_headers = {
        "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko",
        "User-Agent": "Mozilla/5.0"
    }
    payload = {
        "query": f"""
        query {{
            channel(name: "{os.getenv('TWITCH_CHANNEL', 'winx_prinx')}") {{
                videos(first: 30, type: ARCHIVE) {{
                    edges {{
                        node {{
                            id
                            broadcastType
                            lengthSeconds
                            publishedAt
                            stream {{
                                id
                            }}
                        }}
                    }}
                }}
            }}
        }}
        """
    }
    try:
        res = requests.post(url, json=payload, headers=gql_headers, timeout=10)
        if res.status_code == 200:
            data = res.json()
            edges = data.get("data", {}).get("channel", {}).get("videos", {}).get("edges", [])
            for edge in edges:
                node = edge.get("node", {})
                stream_info = node.get("stream") or {}
                if stream_info.get("id") == str(stream_id):
                    return node.get("id")
    except Exception as e:
        pass
    return None

def extract_vod_id(stream):
    """Extract VOD ID from stream record"""
    twitch_id = str(stream.get("twitch_stream_id", ""))
    if twitch_id.startswith("vod-"):
        return twitch_id[4:]
    # Live stream ID — try to find corresponding VOD
    print(f"   -> Поиск VOD для live stream ID {twitch_id}...")
    vod_id = find_vod_id_by_stream_id(twitch_id)
    if vod_id:
        print(f"   -> Найден VOD ID: {vod_id}")
    else:
        print(f"   -> VOD не найден (стрим мог быть удалён или VOD недоступен)")
    return vod_id

def download_vod_roles(vod_id):
    """Download chat log of a past VOD and extract only roles, with retry on failure"""
    roles_dict = {}
    url = "https://gql.twitch.tv/gql"
    gql_headers = {
        "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko",
        "User-Agent": "Mozilla/5.0"
    }

    current_offset = 0
    total_messages = 0
    MAX_RETRIES = 5

    with requests.Session() as session:
        while True:
            payload = {
                "operationName": "VideoCommentsByOffsetOrCursor",
                "variables": {
                    "videoID": str(vod_id),
                    "contentOffsetSeconds": current_offset
                },
                "extensions": {
                    "persistedQuery": {
                        "version": 1,
                        "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a"
                    }
                }
            }

            success = False
            for attempt in range(MAX_RETRIES):
                try:
                    res = session.post(url, json=payload, headers=gql_headers, timeout=15)
                    if res.status_code == 429:
                        wait = 2 ** attempt
                        print(f"\n   [!] Rate limited (429). Повтор через {wait}с...")
                        time.sleep(wait)
                        continue
                    if res.status_code != 200:
                        print(f"\n   [!] Ошибка GQL статус {res.status_code}")
                        break

                    data = res.json()
                    if isinstance(data, list):
                        data = data[0]

                    video = data.get("data", {}).get("video", {})
                    if not video:
                        return list(roles_dict.values())  # VOD ended or not found

                    edges = (video.get("comments") or {}).get("edges") or []
                    if not edges:
                        return list(roles_dict.values())  # Reached end

                    for edge in edges:
                        if not edge:
                            continue
                        node = edge.get("node")
                        if not node:
                            continue
                        commenter = node.get("commenter")
                        if not commenter:
                            continue
                        user = commenter.get("login")
                        if not user:
                            continue

                        display_name = commenter.get("displayName", user)
                        message = node.get("message")
                        if not message:
                            continue

                        timestamp = node.get("createdAt")
                        user_badges = message.get("userBadges") or []
                        badges = [b.get("setID") for b in user_badges if b]

                        # Debug: print first badge encounter to verify GQL structure
                        if total_messages < 3 and user_badges:
                            print(f"\n   [DEBUG] userBadges raw: {user_badges}")
                            print(f"   [DEBUG] parsed badges: {badges}")

                        is_mod = "moderator" in badges or "broadcaster" in badges
                        is_sub = "subscriber" in badges or "founder" in badges
                        is_vip = "vip" in badges

                        # Only store users who have at least one badge
                        if is_mod or is_sub or is_vip:
                            roles_dict[user.lower()] = {
                                "username": user,
                                "displayName": display_name,
                                "isMod": is_mod,
                                "isSub": is_sub,
                                "isVip": is_vip,
                                "timestamp": timestamp
                            }
                        total_messages += 1

                    print(f"   -> Обработано {total_messages} сообщений, найдено {len(roles_dict)} уникальных пользователей...", end="\r")

                    last_offset = edges[-1].get("node", {}).get("contentOffsetSeconds")
                    if last_offset is not None:
                        next_offset = last_offset + 1
                        if next_offset <= current_offset:
                            next_offset = current_offset + 30
                        current_offset = next_offset
                    else:
                        return list(roles_dict.values())

                    # Tiny sleep to avoid aggressive spamming
                    time.sleep(0.1)
                    success = True
                    break  # Success — go to next page

                except Exception as e:
                    wait = 2 ** attempt  # 1, 2, 4, 8, 16 seconds
                    print(f"\n   [!] Ошибка (попытка {attempt+1}/{MAX_RETRIES}): {e}")
                    if attempt < MAX_RETRIES - 1:
                        print(f"   [!] Повтор через {wait}с...")
                        time.sleep(wait)

            if not success:
                print(f"\n   [!] Не удалось получить данные после {MAX_RETRIES} попыток. Останавливаемся на offset={current_offset}.")
                break

    return list(roles_dict.values())


def upload_roles(roles, vod_id):
    """Send roles to backend in batches"""
    batch_size = 500
    for i in range(0, len(roles), batch_size):
        batch = roles[i:i+batch_size]
        try:
            res = requests.post(f"{API_URL}/api/log/roles", json={"roles": batch}, headers=headers, timeout=10)
            if res.status_code == 200:
                print(f"   -> Обновлено пользователей: {i + len(batch)} / {len(roles)}", end="\r")
            else:
                print(f"\n   [!] Ошибка отправки: {res.status_code}")
        except Exception as e:
            print(f"\n   [!] Ошибка сети: {e}")
    print()

if __name__ == "__main__":
    print("=========================================================")
    print("      Ultra-Fast Role Backfiller (AUTO MODE)            ")
    print("=========================================================")
    print(f"Backend: {API_URL}")
    print()

    # Fetch all streams
    print("[1] Получение списка стримов с сервера...")
    streams = get_all_streams()

    if not streams:
        print("[Error] Список стримов пустой или не удалось получить!")
        sys.exit(1)

    # Filter only VODs (twitch_stream_id starts with "vod-")
    vod_streams = [s for s in streams if str(s.get("twitch_stream_id", "")).startswith("vod-")]

    print(f"[+] Найдено стримов всего: {len(streams)}")
    print()

    # Process each stream
    for idx, stream in enumerate(streams, 1):
        vod_id = extract_vod_id(stream)
        title = stream.get("title", "Без названия")
        print(f"[{idx}/{len(streams)}] Stream — «{title[:50]}»")

        if not vod_id:
            print("   -> Пропуск: нет VOD ID")
            continue

        roles = download_vod_roles(vod_id)
        print(f"\n   -> Найдено пользователей с бейджами: {len(roles)}")

        if roles:
            upload_roles(roles, vod_id)
            print(f"   -> ✅ Готово!")
        else:
            print(f"   -> Чат пустой или VOD недоступен, пропуск.")
        print()

    print("=========================================================")
    print("       Восстановление ролей завершено!                  ")
    print("=========================================================")