import base64 from google import genai from google.genai import types SYSTEM_PROMPT_RU = ( "Ты анализируешь кадры с Twitch-стрима. Дай подробное описание всего, что видишь на экране:\n" "1. Какая игра/приложение на экране, жанр, сеттинг\n" "2. Что конкретно происходит: действия персонажа, ситуация в игре, этап (меню, геймплей, катсцена, лобби)\n" "3. Элементы интерфейса: HUD, здоровье, инвентарь, мини-карта, счёт, таймеры\n" "4. Камера стримера: что видно, эмоции, жесты (если есть)\n" "5. Оверлеи: донаты, алерты, виджеты, чат\n" "6. Текст на экране: любой читаемый текст, названия, никнеймы\n" "Пиши развёрнуто (5-10 предложений). Описание должно быть достаточно детальным, " "чтобы другая AI-модель могла полностью понять контекст происходящего без просмотра изображения.\n" "Если сцена похожа на предыдущую, опиши только изменения, но подробно." ) SYSTEM_PROMPT_EN = ( "You are analyzing frames from a Twitch stream. Give a detailed description of everything on screen:\n" "1. What game/application is shown, genre, setting\n" "2. What exactly is happening: character actions, game situation, stage (menu, gameplay, cutscene, lobby)\n" "3. UI elements: HUD, health, inventory, minimap, score, timers\n" "4. Streamer camera: what's visible, emotions, gestures (if present)\n" "5. Overlays: donations, alerts, widgets, chat\n" "6. On-screen text: any readable text, names, nicknames\n" "Write in detail (5-10 sentences). The description must be detailed enough " "for another AI model to fully understand the context without seeing the image.\n" "If the scene is similar to the previous one, describe only the changes, but in detail." ) class VisionAnalyzer: def __init__(self, api_key: str, base_url: str | None = None, lang: str = "ru"): client_kwargs = {"api_key": api_key} if base_url: client_kwargs["http_options"] = types.HttpOptions(base_url=base_url) self.client = genai.Client(**client_kwargs) self.model = "gemini-2.0-flash" self.system_prompt = SYSTEM_PROMPT_RU if lang == "ru" else SYSTEM_PROMPT_EN self.previous_description: str | None = None async def analyze_frame(self, frame_data: bytes) -> str: b64_image = base64.b64encode(frame_data).decode("utf-8") contents = [] if self.previous_description: contents.append( types.Content( role="user", parts=[ types.Part.from_text( text=f"Предыдущее описание: {self.previous_description}" ) ], ) ) contents.append( types.Content( role="model", parts=[types.Part.from_text(text="Понял, учту контекст.")], ) ) contents.append( types.Content( role="user", parts=[ types.Part.from_bytes(data=frame_data, mime_type="image/jpeg"), types.Part.from_text(text="Опиши что сейчас происходит на стриме."), ], ) ) response = await self.client.aio.models.generate_content( model=self.model, contents=contents, config=types.GenerateContentConfig( system_instruction=self.system_prompt, max_output_tokens=1000, temperature=0.3, ), ) description = response.text or "(нет описания)" self.previous_description = description return description