From 65d8dd11940a02adf2b94fd72ada70ce5027b96d Mon Sep 17 00:00:00 2001 From: bradautomates Date: Thu, 17 Sep 2026 13:46:22 +1000 Subject: [PATCH] feat: Gemini interaction adapter with categorized, key-safe failures Co-Authored-By: Claude Fable 5.1 --- skills/watch/scripts/gemini.py | 110 ++++++++++++++++++++++++++ tests/conftest.py | 4 + tests/test_gemini.py | 138 +++++++++++++++++++++++++++++++++ 3 files changed, 252 insertions(+) create mode 100644 skills/watch/scripts/gemini.py create mode 100644 tests/test_gemini.py diff --git a/skills/watch/scripts/gemini.py b/skills/watch/scripts/gemini.py new file mode 100644 index 0000000..056fe5a --- /dev/null +++ b/skills/watch/scripts/gemini.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +"""Gemini engine: Google watches the video (agentic processing) and answers. + +Standard library only. The key travels in the x-goog-api-key header and is +scrubbed from every diagnostic. +""" +from __future__ import annotations + +import json +import math +import re +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +from runtime import diagnostic + +API = 'https://generativelanguage.googleapis.com' +INTERACTIONS = f'{API}/v1beta/interactions' +_YOUTUBE = re.compile( + r'^https?://(?:(?:www\.|m\.|music\.)?youtube\.com/(?:watch\?\S*\bv=|shorts/|live/)|youtu\.be/)\S+$', re.I) +_HINTS = { + 'auth': 'check GEMINI_API_KEY (create one at https://aistudio.google.com/apikey)', + 'quota': 'the key is rate-limited or out of quota; wait or check its plan', + 'rejected': 'Gemini refused this request; the video may be private, unsupported, or too long', + 'service': 'Gemini had a server error; retry shortly', + 'network': 'could not reach generativelanguage.googleapis.com; check network, proxy, or egress rules', + 'response': 'Gemini returned something this version cannot read', +} +DEFAULT_QUESTION = ('Give a thorough summary of this video: what is shown, any on-screen text or code, ' + 'and what is said, in chronological order.') +PROMPT_SUFFIX = ('\n\nCite an MM:SS (or H:MM:SS) timestamp for every claim about a specific moment. ' + 'Separate what is seen from what is heard when it matters. If the video does not ' + 'show or say something needed to answer, say so plainly instead of guessing.') + + +def is_youtube(source: str) -> bool: + return bool(_YOUTUBE.match(source or '')) + + +def build_prompt(question: str | None) -> str: + return ((question or '').strip() or DEFAULT_QUESTION) + PROMPT_SUFFIX + + +def _fail(category: str, detail: str, key: str) -> SystemExit: + detail = diagnostic(detail, 600).replace(key, '[redacted]') if key else diagnostic(detail, 600) + return SystemExit(f'Gemini {category}: {_HINTS[category]}. Detail: {detail or "none"}. ' + 'No local fallback was attempted; rerun with --engine local to use frames + transcript.') + + +def _error_message(body: bytes) -> str: + text = body.decode('utf-8', errors='replace') + try: + data = json.loads(text) + if isinstance(data, list) and data: # Auth errors arrive wrapped in a one-element array. + data = data[0] + return str(data['error']['message']) + except (ValueError, KeyError, TypeError): + return text + + +def _call(method: str, url: str, key: str, *, data=None, headers=None, timeout: float = 60.0): + request = Request(url, data=data, method=method, headers={'x-goog-api-key': key, **(headers or {})}) + try: + with urlopen(request, timeout=timeout) as response: + return response.status, dict(response.headers), response.read() + except HTTPError as exc: + category = ('auth' if exc.code in (401, 403) else 'quota' if exc.code == 429 + else 'service' if exc.code >= 500 else 'rejected') + raise _fail(category, f'HTTP {exc.code}: {_error_message(exc.read())}', key) from None + except (URLError, TimeoutError, OSError) as exc: + raise _fail('network', f'{type(exc).__name__}: {getattr(exc, "reason", exc)}', key) from None + + +def _clock(seconds: int) -> str: + hours, rest = divmod(seconds, 3600) + minutes, secs = divmod(rest, 60) + return f'{hours}:{minutes:02d}:{secs:02d}' if hours else f'{minutes:02d}:{secs:02d}' + + +def _processing(clip): + if not clip or (clip[0] is None and clip[1] is None): + return 'agentic', 'agentic' + start, end = clip + config = {'type': 'static'} # Offsets are only accepted in static mode, as duration strings. + if start is not None: + config['start_offset'] = f'{int(math.floor(start))}s' + if end is not None: + config['end_offset'] = f'{int(math.ceil(end))}s' + label = (f'static clip {_clock(int(math.floor(start or 0)))}–' + f'{_clock(int(math.ceil(end))) if end is not None else "end"}') + return config, label + + +def ask(video: dict, question: str | None, *, model: str, key: str, clip=None, timeout: float = 600.0) -> dict: + processing, label = _processing(clip) + payload = {'model': model, 'input': [{'type': 'video', **video, 'processing': processing}, + {'type': 'text', 'text': build_prompt(question)}]} + _, _, body = _call('POST', INTERACTIONS, key, data=json.dumps(payload).encode('utf-8'), + headers={'Content-Type': 'application/json'}, timeout=timeout) + try: + data = json.loads(body) + parts = [part['text'] for step in data['steps'] if step.get('type') == 'model_output' + for part in step.get('content', []) if part.get('type') == 'text' and part.get('text')] + tokens = (data.get('usage') or {}).get('total_tokens') + except (ValueError, KeyError, TypeError, AttributeError): + raise _fail('response', body[:300].decode('utf-8', errors='replace'), key) from None + if not parts: + raise _fail('response', f'no answer text (status {data.get("status")})', key) + return {'text': '\n'.join(parts).strip(), 'model': model, 'processing': label, + 'total_tokens': tokens if isinstance(tokens, int) else None} diff --git a/tests/conftest.py b/tests/conftest.py index e93c196..dcb5db7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -40,6 +40,10 @@ def isolated_user(monkeypatch, tmp_path): def no_upload(*args, **kwargs): raise AssertionError("Tests must mock cloud uploads") monkeypatch.setattr(whisper, "urlopen", no_upload) + import gemini + def no_gemini(*args, **kwargs): + raise AssertionError("Tests must mock Gemini calls") + monkeypatch.setattr(gemini, "urlopen", no_gemini) # 14 visually distinct fills → 14 abrupt cuts → x264 emits a keyframe per cut. COLORS = [ diff --git a/tests/test_gemini.py b/tests/test_gemini.py new file mode 100644 index 0000000..9de3e5d --- /dev/null +++ b/tests/test_gemini.py @@ -0,0 +1,138 @@ +"""Gemini adapter: payloads, parsing, and failures against a fake urlopen. No network.""" +from __future__ import annotations + +import io +import json +from urllib.error import HTTPError, URLError + +import pytest + +import gemini + +KEY = 'unit-test-key-DO-NOT-LEAK' + + +class FakeResponse: + def __init__(self, body=b'{}', status=200, headers=None): + self.status, self.headers, self._body = status, headers or {}, body + def read(self): + return self._body + def __enter__(self): + return self + def __exit__(self, *exc): + return False + + +def http_error(code, body): + return HTTPError('https://example.invalid', code, 'error', {}, io.BytesIO(body)) + + +def answer(*texts): + steps = [{'type': 'processing_call', 'id': 'c1'}, {'type': 'processing_result', 'call_id': 'c1'}, {'type': 'thought'}] + steps.append({'type': 'model_output', 'content': [{'type': 'text', 'text': t} for t in texts]}) + return json.dumps({'status': 'completed', 'steps': steps, 'usage': {'total_tokens': 4348}}).encode() + + +@pytest.fixture +def calls(monkeypatch): + """Record every request; replies are popped from calls.replies in order.""" + class Calls(list): + replies = [] + recorded = Calls() + def fake(request, timeout=None): + recorded.append(request) + reply = recorded.replies.pop(0) + if isinstance(reply, Exception): + raise reply + return reply + monkeypatch.setattr(gemini, 'urlopen', fake) + return recorded + + +@pytest.mark.parametrize('source,expected', [ + ('https://www.youtube.com/watch?v=rlOpbu3Enkw', True), + ('https://youtu.be/rlOpbu3Enkw', True), + ('https://m.youtube.com/shorts/abc123', True), + ('https://www.youtube.com/playlist?list=PL1', False), + ('https://vimeo.com/123', False), + ('/tmp/clip.mp4', False), + ('https://notyoutube.com/watch?v=x', False), +]) +def test_is_youtube(source, expected): + assert gemini.is_youtube(source) is expected + + +def test_prompt_always_demands_timestamps(): + assert 'What is on the whiteboard?' in gemini.build_prompt('What is on the whiteboard?') + assert 'MM:SS' in gemini.build_prompt('What is on the whiteboard?') + default = gemini.build_prompt(None) + assert 'summary' in default.lower() and 'MM:SS' in default + + +def test_agentic_youtube_payload_and_header_only_key(calls): + calls.replies = [FakeResponse(answer('At 00:10 the screen turns blue.'))] + result = gemini.ask({'uri': 'https://youtu.be/abc'}, 'When does it change?', model='gemini-3.7-flash', key=KEY) + request = calls[0] + assert request.full_url == 'https://generativelanguage.googleapis.com/v1beta/interactions' + assert KEY not in request.full_url + assert request.get_header('X-goog-api-key') == KEY + payload = json.loads(request.data) + assert payload['model'] == 'gemini-3.7-flash' + assert payload['input'][0] == {'type': 'video', 'uri': 'https://youtu.be/abc', 'processing': 'agentic'} + assert payload['input'][1]['type'] == 'text' and 'When does it change?' in payload['input'][1]['text'] + assert result == {'text': 'At 00:10 the screen turns blue.', 'model': 'gemini-3.7-flash', + 'processing': 'agentic', 'total_tokens': 4348} + + +def test_clip_uses_static_processing_with_duration_strings(calls): + calls.replies = [FakeResponse(answer('Blue.'))] + result = gemini.ask({'uri': 'https://generativelanguage.googleapis.com/v1beta/files/x', 'mime_type': 'video/mp4'}, + None, model='m', key=KEY, clip=(1200.4, 1500.2)) + video = json.loads(calls[0].data)['input'][0] + assert video['mime_type'] == 'video/mp4' + assert video['processing'] == {'type': 'static', 'start_offset': '1200s', 'end_offset': '1501s'} + assert result['processing'] == 'static clip 20:00–25:01' + + +def test_open_ended_clip_omits_the_missing_bound(calls): + calls.replies = [FakeResponse(answer('ok'))] + gemini.ask({'uri': 'https://youtu.be/abc'}, None, model='m', key=KEY, clip=(90.0, None)) + assert json.loads(calls[0].data)['input'][0]['processing'] == {'type': 'static', 'start_offset': '90s'} + + +def test_multiple_text_parts_are_joined(calls): + calls.replies = [FakeResponse(answer('First.', 'Second.'))] + assert gemini.ask({'uri': 'https://youtu.be/abc'}, None, model='m', key=KEY)['text'] == 'First.\nSecond.' + + +def test_empty_answer_is_a_failure(calls): + calls.replies = [FakeResponse(json.dumps({'steps': [{'type': 'thought'}]}).encode())] + with pytest.raises(SystemExit, match='Gemini response'): + gemini.ask({'uri': 'https://youtu.be/abc'}, None, model='m', key=KEY) + + +@pytest.mark.parametrize('reply,category', [ + (http_error(401, b'[{"error":{"code":401,"message":"Request had invalid authentication credentials.","status":"UNAUTHENTICATED"}}]'), 'auth'), + (http_error(403, b'{"error":{"message":"API key not valid","code":"permission_denied"}}'), 'auth'), + (http_error(429, b'{"error":{"message":"Quota exceeded","code":"resource_exhausted"}}'), 'quota'), + (http_error(400, b'{"error":{"message":"Invalid input at \'input[0].processing\'.","code":"invalid_request"}}'), 'rejected'), + (http_error(503, b'upstream unavailable'), 'service'), + (URLError('certificate verify failed'), 'network'), + (TimeoutError('timed out'), 'network'), + (FakeResponse(b'not json'), 'response'), +]) +def test_failures_are_categorized_and_never_leak_the_key(calls, reply, category): + calls.replies = [reply] + with pytest.raises(SystemExit) as caught: + gemini.ask({'uri': 'https://youtu.be/abc'}, None, model='m', key=KEY) + message = str(caught.value) + assert message.startswith(f'Gemini {category}:') + assert KEY not in message + assert '--engine local' in message + + +def test_key_echoed_by_the_server_is_redacted(calls): + calls.replies = [http_error(400, json.dumps({'error': {'message': f'bad key {KEY}'}}).encode())] + with pytest.raises(SystemExit) as caught: + gemini.ask({'uri': 'https://youtu.be/abc'}, None, model='m', key=KEY) + assert KEY not in str(caught.value) and '[redacted]' in str(caught.value)