From f71f3ce4be0dce9ec4589c9ab595cf1043aa5741 Mon Sep 17 00:00:00 2001 From: Roko Date: Mon, 6 Jul 2026 17:05:42 +0700 Subject: [PATCH] =?UTF-8?q?=F0=9F=8E=89=20initial:=20vision-analyze.py=20v?= =?UTF-8?q?2=20=E2=80=94=20clean=20ABC=20architecture?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Two backends: RouterAI (gpt-4o-mini) + LocalGemma (Jan API) - Smart emoji routing (🌶️/🔞/🤫 → local Gemma, NSFW never hits cloud) - argparse with --help, --local, --model, --prompt - ImagePreprocessor with ffmpeg resize - ProxyHandler({}) for local server bypass - logging, try/finally cleanup, HTTPError/URLError handling - Zero shell injection, zero keys in /tmp - SKILL.md with full documentation for other agents --- SKILL.md | 80 ++++++++++ vision-analyze.py | 387 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 467 insertions(+) create mode 100644 SKILL.md create mode 100644 vision-analyze.py diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..173a645 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,80 @@ +# Vision Analyze — Image Analysis Skill for AI Agents + +Analyze images with two backends: fast cloud (RouterAI gpt-4o-mini) or private local (Gemma 4B via Jan API). Smart routing via emoji flags — NSFW content never touches cloud APIs. + +## Requirements + +- Python 3.10+ +- `ffmpeg` (for image resize) +- API key in `~/.openclaw/openclaw.json` → `models.providers["routerai-ru"].apiKey` +- (optional) Local Jan server (`http://192.168.88.183:1337`) + +## Usage + +```bash +python3 vision-analyze.py +``` + +### Arguments + +| Flag | Description | +|---|---| +| *(none)* | RouterAI gpt-4o-mini, default prompt | +| `--local` | Local Gemma (private, no censorship) | +| `--model=X` | RouterAI model override | +| `--prompt="..."` | Custom prompt | +| `--prompt="🌶 ..."` | Emoji triggers auto-routing to local Gemma | + +### Default Prompt + +``` +Дай последовательное описание композиции: передний план, задний план, +элементы на изображении, люди или персонажи. Пиши живым, человеческим +языком, 3-5 предложений. +``` + +### Routing Rules + +1. **No flags** → RouterAI gpt-4o-mini (fast, ~2-10s, cheap) +2. `--local` → Local Gemma (~30-90s, free, no filters) +3. `--model=X` → RouterAI with model X +4. `🌶️` / `🔞` / `🤫` in `--prompt` → Local Gemma (NSFW never touches RouterAI!) +5. **No auto-fallback!** Error → log & message, never silently route to other backend + +## Architecture + +``` +Backend (ABC) +├── RouterAIBackend — RouterAI API via urllib, HTTPError/URLError handling +└── LocalGemmaBackend — Anthropic /messages, ProxyHandler({}) bypass, 1 retry + +ImagePreprocessor — ffmpeg resize, tempfile, finally cleanup +``` + +### Safety + +- API keys never written to /tmp world-readable +- No `curl`, no `bash -c`, no shell injection vectors +- `ProxyHandler({})` for local requests (bypasses V2Ray) + +## Known Limitations + +- **Local Gemma:** Only one concurrent request. Never interrupt/kill — causes hang. +- **Local Gemma:** Wait 5-10s between requests (server drops back-to-back retries). +- **Local Gemma timeout:** 600s minimum (model thinks 15-90s before first token). +- DeepSeek models cannot see images — do not pass base64 images to them. + +## For Other Agents + +Clone this repo and symlink or copy `vision-analyze.py` into your workspace: + +```bash +git clone https://git.hidosi.ru/Hidosi/agent-scripts-vision-analyze.git /path/to/skill +ln -s /path/to/skill/vision-analyze.py /usr/local/bin/vision-analyze +``` + +Then follow the usage examples above. + +--- + +Created 2026-07-06 by Roko (OpenClaw agent) with OpenCode (kimi-k2.7-code). Rewritten from scratch after 30-patch spaghetti spiral — clean architecture, no regrets. diff --git a/vision-analyze.py b/vision-analyze.py new file mode 100644 index 0000000..c97d8b9 --- /dev/null +++ b/vision-analyze.py @@ -0,0 +1,387 @@ +#!/usr/bin/env python3 +"""vision-analyze-v2.py — анализ изображений через RouterAI или локальную Gemma.""" + +import argparse +import base64 +import json +import logging +import os +import socket +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request +from abc import ABC, abstractmethod + +# --------------------------------------------------------------------------- +# Константы +# --------------------------------------------------------------------------- + +DEFAULT_PROMPT = ( + "Дай последовательное описание композиции: передний план, задний план, " + "элементы на изображении, люди или персонажи. Пиши живым, человеческим языком, 3-5 предложений." +) + +ROUTERAI_URL = "https://routerai.ru/api/v1/chat/completions" +ROUTERAI_DEFAULT_MODEL = "openai/gpt-4o-mini" + +LOCAL_URL = "http://192.168.88.183:1337/messages" +LOCAL_MODEL = "gemma-4-E4B-it-Q8_0" + +ROUTERAI_KEY_PATH = os.path.expanduser("~/.openclaw/openclaw.json") +LOCAL_KEY_PATH = os.path.expanduser("~/.openclaw/workspace/wireguard/jan_api_key.txt") + +SPICY_EMOJIS = frozenset({"🌶️", "🔞", "🤫"}) +ALLOWED_EXTENSIONS = frozenset({".jpg", ".jpeg", ".png", ".gif", ".webp"}) + +ROUTERAI_TIMEOUT = 30 +LOCAL_TIMEOUT = 600 +LOCAL_RETRY_DELAY = 5 + +RESIZE_THRESHOLD_BYTES = 1_000_000 +RESIZE_SCALE = 1024 +RESIZE_QUALITY = 3 + +LOGGER = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Помощники +# --------------------------------------------------------------------------- + + +def _load_routerai_key() -> str: + """Читает API-ключ RouterAI из конфигурации openclaw.""" + try: + with open(ROUTERAI_KEY_PATH, encoding="utf-8") as f: + config = json.load(f) + except FileNotFoundError as exc: + raise ValueError(f"Конфиг RouterAI не найден: {ROUTERAI_KEY_PATH}") from exc + except json.JSONDecodeError as exc: + raise ValueError(f"Невалидный JSON в конфиге RouterAI: {ROUTERAI_KEY_PATH}") from exc + + key = config.get("models", {}).get("providers", {}).get("routerai-ru", {}).get("apiKey") + if not key: + raise ValueError("В конфиге RouterAI отсутствует models.providers['routerai-ru'].apiKey") + return key + + +def _load_local_key() -> str: + """Читает API-ключ локального сервера Gemma.""" + try: + with open(LOCAL_KEY_PATH, encoding="utf-8") as f: + key = f.read().strip() + except FileNotFoundError as exc: + raise ValueError(f"Ключ локального сервера не найден: {LOCAL_KEY_PATH}") from exc + + if not key: + raise ValueError("Файл ключа локального сервера пуст") + return key + + +def _media_type_for_path(path: str) -> str: + """Возвращает MIME-тип изображения по расширению.""" + ext = os.path.splitext(path)[1].lower() + return { + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".gif": "image/gif", + ".webp": "image/webp", + }.get(ext, "image/jpeg") + + +def _encode_image(path: str, media_type: str | None = None) -> tuple[str, str]: + """Кодирует изображение в base64 и возвращает (base64, media_type).""" + with open(path, "rb") as f: + data = f.read() + b64 = base64.b64encode(data).decode("ascii") + return b64, media_type or _media_type_for_path(path) + + +def _contains_spicy(prompt: str) -> bool: + """Проверяет наличие NSFW-эмодзи в промпте.""" + return any(emoji in prompt for emoji in SPICY_EMOJIS) + + +# --------------------------------------------------------------------------- +# ImagePreprocessor +# --------------------------------------------------------------------------- + + +class ImagePreprocessor: + """Подготовка изображений перед отправкой в локальную модель.""" + + @staticmethod + def resize_if_needed(path: str) -> tuple[str, bool]: + """ + Если изображение больше порога, уменьшает его через ffmpeg. + + Возвращает (effective_path, is_temporary). Если is_temporary=True, + вызывающий код должен удалить effective_path после использования. + """ + try: + size = os.path.getsize(path) + except OSError as exc: + raise ValueError(f"Не удалось получить размер файла: {path}") from exc + + if size <= RESIZE_THRESHOLD_BYTES: + return path, False + + tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + try: + tmp.close() + cmd = [ + "ffmpeg", + "-y", + "-i", path, + "-vf", f"scale={RESIZE_SCALE}:-1", + "-q:v", str(RESIZE_QUALITY), + tmp.name, + ] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError(f"ffmpeg завершился с ошибкой: {proc.stderr.strip()}") + if not os.path.isfile(tmp.name) or os.path.getsize(tmp.name) == 0: + raise RuntimeError("ffmpeg создал пустой файл") + return tmp.name, True + except Exception: + if os.path.exists(tmp.name): + os.unlink(tmp.name) + raise + + +# --------------------------------------------------------------------------- +# Бэкенды +# --------------------------------------------------------------------------- + + +class Backend(ABC): + """Абстрактный бэкенд для анализа изображений.""" + + @abstractmethod + def analyze(self, path: str, prompt: str) -> str: + """Анализирует изображение по пути path с заданным prompt.""" + ... + + +class RouterAIBackend(Backend): + """Бэкенд routerai.ru (OpenAI-совместимый API).""" + + def __init__(self, model: str = ROUTERAI_DEFAULT_MODEL) -> None: + self.model = model + self._api_key = _load_routerai_key() + + def analyze(self, path: str, prompt: str) -> str: + img_b64, media_type = _encode_image(path) + + payload = json.dumps({ + "model": self.model, + "modalities": ["image", "text"], + "messages": [{ + "role": "user", + "content": [ + {"type": "text", "text": prompt}, + {"type": "image_url", "image_url": { + "url": f"data:{media_type};base64,{img_b64}" + }}, + ], + }], + "max_tokens": 512, + }).encode("utf-8") + + req = urllib.request.Request( + ROUTERAI_URL, + data=payload, + headers={ + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + }, + ) + + try: + with urllib.request.urlopen(req, timeout=ROUTERAI_TIMEOUT) as resp: + data = json.loads(resp.read()) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + LOGGER.error("RouterAI вернул HTTP %s: %s", exc.code, body) + raise RuntimeError(f"RouterAI HTTP {exc.code}: {body}") from exc + except urllib.error.URLError as exc: + LOGGER.error("Ошибка соединения с RouterAI: %s", exc.reason) + raise RuntimeError(f"Ошибка соединения с RouterAI: {exc.reason}") from exc + except json.JSONDecodeError as exc: + LOGGER.error("Невалидный JSON от RouterAI") + raise RuntimeError("Невалидный JSON от RouterAI") from exc + + content = data.get("choices", [{}])[0].get("message", {}).get("content") + if content: + return content + + LOGGER.error("Неожиданный ответ RouterAI: %s", data) + raise RuntimeError("В ответе RouterAI отсутствует content") + + +class LocalGemmaBackend(Backend): + """Бэкенд локальной Gemma через Anthropic-совместимый endpoint.""" + + def __init__(self) -> None: + self._api_key = _load_local_key() + self._opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + + @staticmethod + def _is_connection_error(exc: urllib.error.URLError) -> bool: + return isinstance(exc.reason, (ConnectionError, TimeoutError, socket.timeout)) + + def _build_request(self, payload: bytes) -> urllib.request.Request: + return urllib.request.Request( + LOCAL_URL, + data=payload, + headers={ + "Authorization": f"Bearer {self._api_key}", + "anthropic-version": "2023-06-01", + "Content-Type": "application/json", + }, + ) + + def analyze(self, path: str, prompt: str) -> str: + effective_path, is_temporary = ImagePreprocessor.resize_if_needed(path) + + try: + img_b64, _ = _encode_image(effective_path, media_type="image/jpeg") + + payload = json.dumps({ + "model": LOCAL_MODEL, + "max_tokens": 1024, + "messages": [{ + "role": "user", + "content": [ + {"type": "text", "text": prompt}, + {"type": "image", "source": { + "type": "base64", + "media_type": "image/jpeg", + "data": img_b64, + }}, + ], + }], + }).encode("utf-8") + + attempts = 0 + while True: + req = self._build_request(payload) + try: + with self._opener.open(req, timeout=LOCAL_TIMEOUT) as resp: + data = json.loads(resp.read()) + break + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + LOGGER.error("Локальный сервер вернул HTTP %s: %s", exc.code, body) + raise RuntimeError(f"Локальный сервер HTTP {exc.code}: {body}") from exc + except urllib.error.URLError as exc: + attempts += 1 + if attempts <= 1 and self._is_connection_error(exc): + LOGGER.warning( + "Соединение с локальным сервером сброшено, повторная попытка через %s сек...", + LOCAL_RETRY_DELAY, + ) + time.sleep(LOCAL_RETRY_DELAY) + continue + LOGGER.error("Ошибка соединения с локальным сервером: %s", exc.reason) + raise RuntimeError(f"Ошибка соединения с локальным сервером: {exc.reason}") from exc + except json.JSONDecodeError as exc: + LOGGER.error("Невалидный JSON от локального сервера") + raise RuntimeError("Невалидный JSON от локального сервера") from exc + + for block in data.get("content", []): + if block.get("type") == "text": + return block.get("text", "") + + for block in data.get("content", []): + if block.get("type") == "thinking": + return block.get("thinking", "Только размышления, ответ не сформирован") + + LOGGER.error("Неожиданный ответ локального сервера: %s", data) + raise RuntimeError("В ответе локального сервера отсутствуют текстовые блоки") + finally: + if is_temporary and os.path.exists(effective_path): + os.unlink(effective_path) + + +# --------------------------------------------------------------------------- +# CLI и маршрутизация +# --------------------------------------------------------------------------- + + +def _resolve_backend(args: argparse.Namespace) -> Backend: + """Выбирает бэкенд на основе аргументов и содержимого промпта.""" + if args.local: + LOGGER.info("Форсирован локальный бэкенд Gemma") + return LocalGemmaBackend() + + if args.model: + LOGGER.info("Используется RouterAI с моделью %s", args.model) + return RouterAIBackend(model=args.model) + + if _contains_spicy(args.prompt): + LOGGER.info("Обнаружен spicy-флаг в промпте, используется локальный бэкенд Gemma") + return LocalGemmaBackend() + + LOGGER.info("Используется RouterAI с моделью по умолчанию %s", ROUTERAI_DEFAULT_MODEL) + return RouterAIBackend() + + +def _validate_path(path: str) -> None: + """Проверяет, что путь указывает на разрешённое изображение.""" + if not os.path.isfile(path): + raise ValueError(f"Файл не найден: {path}") + + ext = os.path.splitext(path)[1].lower() + if ext not in ALLOWED_EXTENSIONS: + raise ValueError( + f"Неподдерживаемый формат файла: {ext}. " + f"Разрешены: {', '.join(sorted(ALLOWED_EXTENSIONS))}" + ) + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Анализ изображений через RouterAI или локальную Gemma." + ) + parser.add_argument("path", help="Путь к изображению") + parser.add_argument("--local", action="store_true", help="Форсировать локальную Gemma") + parser.add_argument("--model", help="Модель для RouterAI (например, openai/gpt-4o-mini)") + parser.add_argument("--prompt", default=DEFAULT_PROMPT, help="Пользовательский промпт") + return parser + + +def main() -> int: + """Точка входа скрипта.""" + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + parser = _build_parser() + args = parser.parse_args() + + try: + _validate_path(args.path) + except ValueError as exc: + LOGGER.error("%s", exc) + return 1 + + try: + backend = _resolve_backend(args) + result = backend.analyze(args.path, args.prompt) + except (RuntimeError, ValueError) as exc: + LOGGER.error("%s", exc) + return 1 + + print(result) + return 0 + + +if __name__ == "__main__": + sys.exit(main())