feat: 图片识别技能

skills/image-recognition/SKILL.md

@@ -0,0 +1,83 @@
+---
+name: image-recognition
+description: "AI 图像识别工具。当用户提供图片并希望识别、描述、提取文字、分析画面内容或回答图片相关问题时使用。"
+argument-hint: "需要 prompt（识别提示词）和 image_url（图片 URL）"
+---
+
+# Image Recognition Skill
+
+## 描述
+
+这是一个 AI 图像识别技能，输入一张图片 URL 或本地图片路径和识别提示词，输出模型对图片的识别、描述或分析结果。
+
+该技能从数据库读取当前会话的聊天 AI 配置：`chat_base_url`、`chat_api_key` 和 `image_recognition_model`，并调用 OpenAI 兼容的多模态 Chat Completions 接口完成图片理解。
+
+这个仓库里额外提供了一个可执行脚本 `scripts/image_recognition.py`，方便宿主机器人直接调用。
+
+## 触发条件
+
+- 用户发送图片并要求描述图片内容。
+- 用户说「识别这张图」「看看图片里有什么」「分析一下这张图片」。
+- 用户要求从图片中提取文字、物体、场景、人物动作、票据内容等信息。
+- 用户基于图片提问，例如「这是什么」「图片里的文字是什么」「这张图哪里不对」。
+
+## 参数说明（JSON Schema）
+
+调用脚本时，需要通过 shell 风格参数传入，参数结构如下：
+
+```json
+{
+  "type": "object",
+  "properties": {
+    "prompt": {
+      "type": "string",
+      "description": "图像识别提示词，用户想从图片中得到什么信息。"
+    },
+    "image_url": {
+      "type": "string",
+      "description": "图片的 URL 地址、本地图片路径，或 file:// 本地图片地址。"
+    }
+  },
+  "required": ["prompt", "image_url"],
+  "additionalProperties": false
+}
+```
+
+对应的命令行参数为：
+
+- `--prompt <识别提示词>` 必填
+- `--image_url <图片 URL 或本地图片路径>` 必填
+
+## 依赖安装
+
+- 脚本首次运行时会自动创建虚拟环境并安装依赖，无需手动执行。
+- 如需手动重新安装，可执行：`python3 scripts/bootstrap.py`
+
+## 执行步骤
+
+1. 当用户提供图片并要求识别、描述、分析或提取信息时触发该技能。
+2. 从用户输入中提取 `prompt` 和 `image_url`，不要改写用户真正想问图片的问题。
+3. 在仓库根目录执行脚本，例如：
+
+```bash
+python3 scripts/image_recognition.py --prompt '请描述这张图片' --image_url 'https://example.com/image.jpg'
+```
+
+本地图片示例：
+
+```bash
+python3 scripts/image_recognition.py --prompt '请提取图片里的文字' --image_url '/tmp/example.jpg'
+```
+
+4. 脚本将图片和提示词一起发送给 OpenAI 兼容的多模态模型。远程图片会直接传 URL；本地图片会转成 `data:image/...;base64,...` 后传入 `image_url.url`。
+5. 成功时，脚本输出图像识别结果文本，宿主机器人可直接作为消息回复给用户。
+
+## 校验规则
+
+- `prompt` 不能为空。
+- `image_url` 不能为空，支持 `http://`、`https://`、`file://` 和本地图片路径。
+
+## 回复要求
+
+- 成功时，脚本输出图片识别结果。
+- 失败时，返回脚本输出的具体错误信息。

skills/image-recognition/scripts/bootstrap.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import hashlib
+import subprocess
+import sys
+import traceback
+from pathlib import Path
+
+sys.stderr = sys.stdout
+
+
+def _skill_root_from(script_dir: Path) -> Path:
+    return script_dir.parent
+
+
+def _venv_dir(script_dir: Path) -> Path:
+    return _skill_root_from(script_dir) / ".venv"
+
+
+def _venv_python(venv_dir: Path) -> Path:
+    if sys.platform == "win32":
+        return venv_dir / "Scripts" / "python.exe"
+    return venv_dir / "bin" / "python"
+
+
+def _stamp_file(venv_dir: Path) -> Path:
+    return venv_dir / ".req_hash"
+
+
+def _file_hash(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _deps_up_to_date(requirements_file: Path, venv_dir: Path) -> bool:
+    stamp = _stamp_file(venv_dir)
+    if not stamp.is_file():
+        return False
+    return stamp.read_text().strip() == _file_hash(requirements_file)
+
+
+def _write_stamp(requirements_file: Path, venv_dir: Path) -> None:
+    _stamp_file(venv_dir).write_text(_file_hash(requirements_file))
+
+
+def _ensure_venv(venv_dir: Path, venv_python: Path) -> int:
+    if venv_python.is_file():
+        return 0
+
+    sys.stdout.write(f"未检测到技能虚拟环境，正在创建: {venv_dir}\n")
+    import shutil
+    py = sys.executable or next(
+        (shutil.which(candidate) for candidate in ("python3", "python") if shutil.which(candidate)), None
+    )
+    if not py:
+        raise RuntimeError("无法找到 Python 解释器路径")
+
+    try:
+        subprocess.run([py, "-m", "venv", str(venv_dir)], check=True, stdout=sys.stdout, stderr=sys.stdout)
+    except subprocess.CalledProcessError as exc:
+        sys.stdout.write(f"创建虚拟环境失败，退出码: {exc.returncode}\n")
+        return exc.returncode or 1
+
+    return 0
+
+
+def main() -> int:
+    script_dir = Path(__file__).resolve().parent
+    requirements_file = script_dir / "requirements.txt"
+    venv_dir = _venv_dir(script_dir)
+    venv_python = _venv_python(venv_dir)
+
+    if not requirements_file.is_file():
+        sys.stdout.write(f"未找到依赖文件: {requirements_file}\n")
+        return 1
+
+    ensure_result = _ensure_venv(venv_dir, venv_python)
+    if ensure_result != 0:
+        return ensure_result
+
+    if _deps_up_to_date(requirements_file, venv_dir):
+        sys.stdout.write("依赖已是最新，跳过安装\n")
+        return 0
+
+    try:
+        subprocess.run([str(venv_python), "-m", "pip", "install", "--upgrade", "pip"], check=True, stdout=sys.stdout, stderr=sys.stdout)
+    except subprocess.CalledProcessError as exc:
+        sys.stdout.write(f"升级 pip 失败，退出码: {exc.returncode}\n")
+        return exc.returncode or 1
+
+    try:
+        subprocess.run([str(venv_python), "-m", "pip", "install", "-r", str(requirements_file)], check=True, stdout=sys.stdout, stderr=sys.stdout)
+    except subprocess.CalledProcessError as exc:
+        sys.stdout.write(f"安装依赖失败，退出码: {exc.returncode}\n")
+        return exc.returncode or 1
+
+    _write_stamp(requirements_file, venv_dir)
+    sys.stdout.write(f"依赖安装完成，当前技能虚拟环境: {venv_dir}\n")
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except SystemExit:
+        raise
+    except Exception:
+        traceback.print_exc(file=sys.stdout)
+        raise SystemExit(1)

skills/image-recognition/scripts/image_recognition.py

@@ -0,0 +1,331 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import argparse
+import base64
+import mimetypes
+import os
+import re
+import subprocess
+import sys
+import traceback
+from pathlib import Path
+from urllib.parse import urlparse, unquote
+
+sys.stderr = sys.stdout
+
+
+def _skill_root() -> Path:
+    return Path(__file__).resolve().parent.parent
+
+
+def _skill_venv_python() -> Path:
+    venv_dir = _skill_root() / ".venv"
+    if sys.platform == "win32":
+        return venv_dir / "Scripts" / "python.exe"
+    return venv_dir / "bin" / "python"
+
+
+def _get_python_executable() -> str:
+    if sys.executable:
+        return sys.executable
+    import shutil
+    for candidate in ("python3", "python"):
+        found = shutil.which(candidate)
+        if found:
+            return found
+    raise RuntimeError("无法找到 Python 解释器路径")
+
+
+def _run_bootstrap() -> None:
+    bootstrap = Path(__file__).resolve().parent / "bootstrap.py"
+    result = subprocess.run([_get_python_executable(), str(bootstrap)])
+    if result.returncode != 0:
+        raise SystemExit(result.returncode)
+
+
+def _ensure_skill_venv_python() -> None:
+    venv_python = _skill_venv_python()
+    if not venv_python.is_file():
+        _run_bootstrap()
+        venv_python = _skill_venv_python()
+        if not venv_python.is_file():
+            sys.stdout.write("bootstrap 后仍未找到虚拟环境\n")
+            raise SystemExit(1)
+
+    venv_dir = _skill_root() / ".venv"
+    if Path(sys.prefix) == venv_dir.resolve():
+        return
+
+    os.execv(str(venv_python), [str(venv_python), str(Path(__file__).resolve()), *sys.argv[1:]])
+
+
+_ensure_skill_venv_python()
+
+try:
+    import pymysql  # type: ignore  # noqa: E402
+    from openai import OpenAI  # type: ignore  # noqa: E402
+except ModuleNotFoundError:
+    _run_bootstrap()
+    _py = _get_python_executable()
+    os.execv(_py, [_py, str(Path(__file__).resolve()), *sys.argv[1:]])
+
+
+def _mysql_connect():
+    host = os.environ.get("MYSQL_HOST", "127.0.0.1")
+    port = int(os.environ.get("MYSQL_PORT", "3306"))
+    user = os.environ.get("MYSQL_USER", "root")
+    password = os.environ.get("MYSQL_PASSWORD", "")
+    database = os.environ.get("ROBOT_CODE", "")
+    if not database:
+        raise RuntimeError("环境变量 ROBOT_CODE 未配置")
+
+    return pymysql.connect(
+        host=host,
+        port=port,
+        user=user,
+        password=password,
+        database=database,
+        charset="utf8mb4",
+        connect_timeout=10,
+        read_timeout=30,
+    )
+
+
+def _query_one(conn, sql: str, params: tuple = ()) -> dict | None:
+    cur = conn.cursor()
+    cur.execute(sql, params)
+    columns = [desc[0] for desc in cur.description] if cur.description else []
+    row = cur.fetchone()
+    cur.close()
+    if row is None:
+        return None
+    return dict(zip(columns, row))
+
+
+def _clean_text(value: object) -> str:
+    if isinstance(value, (bytes, bytearray)):
+        value = value.decode("utf-8")
+    if isinstance(value, str):
+        return value.strip()
+    return ""
+
+
+def _extract_model(record: dict | None) -> str:
+    if record:
+        model = _clean_text(record.get("image_recognition_model"))
+        if model:
+            return model
+    return ""
+
+
+def _normalize_ai_base_url(base_url: str) -> str:
+    normalized = base_url.rstrip("/")
+    if normalized and not re.search(r"/v\d+$", normalized):
+        normalized += "/v1"
+    return normalized
+
+
+def load_image_recognition_config(conn, from_wx_id: str) -> dict:
+    global_fields = "chat_base_url, chat_api_key, image_recognition_model"
+    global_record = _query_one(conn, f"SELECT {global_fields} FROM global_settings LIMIT 1")
+
+    config = {"base_url": "", "api_key": "", "model": ""}
+    if global_record:
+        base_url = _clean_text(global_record.get("chat_base_url"))
+        api_key = _clean_text(global_record.get("chat_api_key"))
+        if base_url:
+            config["base_url"] = base_url
+        if api_key:
+            config["api_key"] = api_key
+        model = _extract_model(global_record)
+        if model:
+            config["model"] = model
+
+    if from_wx_id.endswith("@chatroom"):
+        override_fields = "chat_base_url, chat_api_key, image_recognition_model"
+        override = _query_one(
+            conn,
+            f"SELECT {override_fields} FROM chat_room_settings WHERE chat_room_id = %s LIMIT 1",
+            (from_wx_id,),
+        )
+    else:
+        override_fields = "chat_base_url, chat_api_key, image_recognition_model"
+        override = _query_one(
+            conn,
+            f"SELECT {override_fields} FROM friend_settings WHERE wechat_id = %s LIMIT 1",
+            (from_wx_id,),
+        )
+
+    if override:
+        base_url = _clean_text(override.get("chat_base_url"))
+        api_key = _clean_text(override.get("chat_api_key"))
+        if base_url:
+            config["base_url"] = base_url
+        if api_key:
+            config["api_key"] = api_key
+        model = _extract_model(override)
+        if model:
+            config["model"] = model
+
+    config["base_url"] = _normalize_ai_base_url(config["base_url"])
+    return config
+
+
+def _local_image_path(value: str) -> Path:
+    parsed = urlparse(value)
+    if parsed.scheme == "file":
+        if parsed.netloc not in ("", "localhost"):
+            raise ValueError("不支持非本机 file URL")
+        return Path(unquote(parsed.path)).expanduser()
+
+    path = Path(value).expanduser()
+    if not path.is_absolute():
+        path = Path.cwd() / path
+    return path
+
+
+def _local_image_to_data_url(value: str) -> str:
+    path = _local_image_path(value)
+    if not path.is_file():
+        raise ValueError(f"本地图片不存在: {path}")
+
+    mime_type, _ = mimetypes.guess_type(str(path))
+    if not mime_type or not mime_type.startswith("image/"):
+        raise ValueError(f"无法识别本地图片类型: {path}")
+
+    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
+    return f"data:{mime_type};base64,{encoded}"
+
+
+def _resolve_image_url(value: str) -> str:
+    parsed = urlparse(value)
+    if parsed.scheme in {"http", "https"} and parsed.netloc:
+        return value
+    if parsed.scheme == "data" and value.startswith("data:image/"):
+        return value
+    if parsed.scheme and parsed.scheme != "file":
+        raise ValueError(f"不支持的图片地址协议: {parsed.scheme}")
+    return _local_image_to_data_url(value)
+
+
+def _extract_response_text(response) -> str:
+    if not response.choices:
+        return ""
+
+    content = response.choices[0].message.content
+    if isinstance(content, str):
+        return content.strip()
+    if isinstance(content, list):
+        texts: list[str] = []
+        for item in content:
+            text = getattr(item, "text", None)
+            if isinstance(text, str) and text.strip():
+                texts.append(text.strip())
+            elif isinstance(item, dict) and isinstance(item.get("text"), str) and item["text"].strip():
+                texts.append(item["text"].strip())
+        return "\n".join(texts)
+    return ""
+
+
+def recognize_image(prompt: str, image_url: str, config: dict) -> str:
+    api_key = config.get("api_key", "")
+    base_url = config.get("base_url", "")
+    model = config.get("model", "")
+    if not api_key or not base_url or not model:
+        raise RuntimeError("AI图片识别未配置，请联系管理员进行配置")
+
+    resolved_image_url = _resolve_image_url(image_url)
+    client = OpenAI(api_key=api_key, base_url=base_url)
+    response = client.chat.completions.create(
+        model=model,
+        messages=[
+            {
+                "role": "user",
+                "content": [
+                    {"type": "image_url", "image_url": {"url": resolved_image_url}},
+                    {"type": "text", "text": prompt},
+                ],
+            }
+        ],
+        stream=False,
+    )
+    content = _extract_response_text(response)
+    if not content:
+        raise RuntimeError("图片识别失败，返回了空内容")
+    return content
+
+
+def _parse_cli_params(argv: list[str]) -> dict:
+    parser = argparse.ArgumentParser(add_help=False)
+    parser.add_argument("--prompt", default="")
+    parser.add_argument("--image_url", default="")
+
+    namespace, unknown = parser.parse_known_args(argv)
+    if unknown:
+        raise ValueError(f"存在不支持的参数: {' '.join(unknown)}")
+
+    return {"prompt": namespace.prompt, "image_url": namespace.image_url}
+
+
+def main() -> int:
+    if len(sys.argv) < 2:
+        sys.stdout.write("缺少输入参数\n")
+        return 1
+
+    try:
+        params = _parse_cli_params(sys.argv[1:])
+    except ValueError as exc:
+        sys.stdout.write(f"参数格式错误: {exc}\n")
+        return 1
+
+    prompt = params.get("prompt", "").strip()
+    image_url = params.get("image_url", "").strip()
+    if not prompt:
+        sys.stdout.write("缺少图像识别提示词\n")
+        return 1
+    if not image_url:
+        sys.stdout.write("缺少图片 URL\n")
+        return 1
+
+    from_wx_id = os.environ.get("ROBOT_FROM_WX_ID", "").strip()
+    if not from_wx_id:
+        sys.stdout.write("环境变量 ROBOT_FROM_WX_ID 未配置\n")
+        return 1
+
+    try:
+        conn = _mysql_connect()
+    except Exception as exc:
+        sys.stdout.write(f"数据库连接失败: {exc}\n")
+        return 1
+
+    try:
+        config = load_image_recognition_config(conn, from_wx_id)
+    except Exception as exc:
+        sys.stdout.write(f"加载图像识别配置失败: {exc}\n")
+        return 1
+    finally:
+        try:
+            conn.close()
+        except Exception:
+            pass
+
+    try:
+        content = recognize_image(prompt, image_url, config)
+    except Exception as exc:
+        sys.stdout.write(f"图片识别失败: {exc}\n")
+        return 1
+
+    sys.stdout.write(f"{content}\n")
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except SystemExit:
+        raise
+    except Exception:
+        traceback.print_exc(file=sys.stdout)
+        raise SystemExit(1)

skills/image-recognition/scripts/requirements.txt

@@ -0,0 +1,3 @@
+cryptography
+openai>=2.34.0
+pymysql>=1.1,<2

skills/stocks/SKILL.md

@@ -17,10 +17,11 @@ argument-hint: "无需参数，直接调用即可"
 ## 执行步骤
 
 1. 使用 `web-page` 这个网页内容读取、自动化交互和截图技能，访问 `https://quote.eastmoney.com/stockhotmap/` 这个网页。
-2. 网页上有这个类名(`topzs`)的 div 元素，展示了`上证指数`、`深证成指`、`创业板指`、`科创综指`、`北证50`的涨跌情况。你需要这些信息的时候，就从这个元素里提取出来。
+2. 这个网页是 ajax 动态数据由 javascript 动态渲染的，访问后等待 5 秒钟再提取内容。
-3. 网页上有这个类名(`stockmap`)的 div 元素，展示了各个板块的涨跌情况。你需要这些信息的时候，就从这个元素里提取出来。
+3. 如果需要截图，截图前先将带有这个类名(`popwscps_d`)的元素删除
-4. 这个网页是 ajax 动态渲染的，访问后等待 5 秒钟再提取内容。
+4. 网页上有这个类名(`topzs`)的 div 元素，展示了`上证指数`、`深证成指`、`创业板指`、`科创综指`、`北证50`的涨跌情况。优先截图这个元素，然后使用图像识别技能识别具体内容。
-5. 如果需要截图，截图前先将带有这个类名(`em_widget em_show`)的元素删除
+5. 网页上有这个 id(`stockmap`)的 div 元素，展示了各个板块的涨跌情况。如果用户想看板块涨跌情况，直接截图并使用发送图片技能发送图片。
+6. 如果没有获取到具体涨跌情况，则截图 id(`stockmap`) 的 div 元素并使用发送图片技能发送图片。
 
 ## 回复要求