feat: 合并发送图片技能、新增发送文件技能
This commit is contained in:
parent
8b99623429
commit
7898db9291
82
skills/send-file/SKILL.md
Normal file
82
skills/send-file/SKILL.md
Normal file
@ -0,0 +1,82 @@
|
||||
---
|
||||
name: send-file
|
||||
description: "发送本地或远程文件到当前微信会话。用户要求发送、转发或分享一个或多个已有文档、压缩包及其他普通文件，且输入是机器人可访问的本地路径或 http/https 文件 URL 时使用；图片应优先使用 send-image。"
|
||||
---
|
||||
|
||||
# 发送文件
|
||||
|
||||
使用 `scripts/send_file.py` 调用机器人客户端文件接口，将一个或多个本地文件、远程文件 URL 发送到当前微信会话。
|
||||
|
||||
## 入参
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"file_path": {
|
||||
"type": "string",
|
||||
"description": "要发送的本地文件路径，必须是机器人运行环境可访问的文件。"
|
||||
},
|
||||
"file_paths": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "要发送的多个本地文件路径。"
|
||||
},
|
||||
"file_url": {
|
||||
"type": "string",
|
||||
"description": "要发送的远程文件 URL，必须以 http 或 https 开头。"
|
||||
},
|
||||
"file_urls": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "要发送的多个远程文件 URL。"
|
||||
}
|
||||
},
|
||||
"anyOf": [
|
||||
{ "required": ["file_path"] },
|
||||
{ "required": ["file_paths"] },
|
||||
{ "required": ["file_url"] },
|
||||
{ "required": ["file_urls"] }
|
||||
],
|
||||
"additionalProperties": false
|
||||
}
|
||||
```
|
||||
|
||||
可同时提供本地路径和远程 URL。对应命令行参数：
|
||||
|
||||
- `--file_path <本地文件路径>`：可重复传入。
|
||||
- `--file_paths <JSON数组>`：一次传入多个本地文件路径。
|
||||
- `--file_url <远程文件URL>`：可重复传入。
|
||||
- `--file_urls <JSON数组>`：一次传入多个远程文件 URL。
|
||||
|
||||
## 执行
|
||||
|
||||
在技能目录运行脚本：
|
||||
|
||||
```bash
|
||||
python3 scripts/send_file.py --file_path '/tmp/report.pdf'
|
||||
```
|
||||
|
||||
发送远程文件：
|
||||
|
||||
```bash
|
||||
python3 scripts/send_file.py --file_urls '["https://example.com/a.pdf","https://example.com/b.zip"]'
|
||||
```
|
||||
|
||||
本地文件逐个调用 `POST /api/v1/robot/message/send/file/local`；远程文件合并调用 `POST /api/v1/robot/message/send/file/url`。接收方固定读取 `ROBOT_FROM_WX_ID`，客户端端口读取 `ROBOT_WECHAT_CLIENT_PORT`。
|
||||
|
||||
## 约束
|
||||
|
||||
- 至少提供一个非空本地路径或远程 URL。
|
||||
- 本地路径必须指向已存在、非空且不超过 25MB 的文件，不能把 URL 作为本地路径传入。
|
||||
- 远程 URL 必须包含有效的 `http` 或 `https` 协议及主机名；客户端下载后的文件也不能超过 25MB。
|
||||
- 脚本按首次出现顺序去重；所有参数通过校验后，单个发送失败时会继续尝试其余文件，最终返回全部失败原因。
|
||||
|
||||
## 输出
|
||||
|
||||
- 全部成功时输出「文件发送成功」，无需再发送额外消息。
|
||||
- 任一失败时返回具体错误信息，不要声称文件已全部发送。
|
||||
4
skills/send-file/agents/openai.yaml
Normal file
4
skills/send-file/agents/openai.yaml
Normal file
@ -0,0 +1,4 @@
|
||||
interface:
|
||||
display_name: "发送文件"
|
||||
short_description: "将一个或多个本地文件或远程文件链接发送到当前微信会话"
|
||||
default_prompt: "使用 $send-file 将这些本地文件或远程文件发送到当前微信会话。"
|
||||
199
skills/send-file/scripts/send_file.py
Normal file
199
skills/send-file/scripts/send_file.py
Normal file
@ -0,0 +1,199 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
sys.stderr = sys.stdout
|
||||
|
||||
MAX_FILE_SIZE = 25 * 1024 * 1024
|
||||
|
||||
|
||||
def _raise_for_client_error(response: dict) -> None:
|
||||
if not response:
|
||||
return
|
||||
|
||||
code = response.get("code")
|
||||
if code is not None and str(code) != "200":
|
||||
message = str(response.get("message") or f"客户端返回错误码 {code}")
|
||||
raise RuntimeError(message)
|
||||
|
||||
data = response.get("data")
|
||||
if isinstance(data, list):
|
||||
failures = [
|
||||
item.strip()
|
||||
for item in data
|
||||
if isinstance(item, str) and item.lstrip().startswith("失败:")
|
||||
]
|
||||
if failures:
|
||||
raise RuntimeError("；".join(failures))
|
||||
|
||||
|
||||
def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
|
||||
data = json.dumps(body).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
text = resp.read().decode("utf-8")
|
||||
if not text.strip():
|
||||
return {}
|
||||
|
||||
response = json.loads(text)
|
||||
if not isinstance(response, dict):
|
||||
raise RuntimeError("客户端返回格式不正确")
|
||||
_raise_for_client_error(response)
|
||||
return response
|
||||
|
||||
|
||||
def _expand_json_array_values(values: list[str], label: str) -> list[str]:
|
||||
expanded: list[str] = []
|
||||
for value in values:
|
||||
stripped = value.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
if stripped.startswith("["):
|
||||
parsed = json.loads(stripped)
|
||||
if not isinstance(parsed, list) or not all(
|
||||
isinstance(item, str) for item in parsed
|
||||
):
|
||||
raise ValueError(f"{label} 必须是字符串数组")
|
||||
expanded.extend(item.strip() for item in parsed if item.strip())
|
||||
continue
|
||||
expanded.append(stripped)
|
||||
return expanded
|
||||
|
||||
|
||||
def _dedupe(values: list[str]) -> list[str]:
|
||||
deduped: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for value in values:
|
||||
if value not in seen:
|
||||
seen.add(value)
|
||||
deduped.append(value)
|
||||
return deduped
|
||||
|
||||
|
||||
def _parse_cli_params(argv: list[str]) -> tuple[list[str], list[str]]:
|
||||
parser = argparse.ArgumentParser(add_help=False)
|
||||
parser.add_argument("--file_path", action="append", default=[])
|
||||
parser.add_argument("--file_paths", action="append", default=[])
|
||||
parser.add_argument("--file_url", action="append", default=[])
|
||||
parser.add_argument("--file_urls", action="append", default=[])
|
||||
|
||||
namespace, unknown = parser.parse_known_args(argv)
|
||||
if unknown:
|
||||
raise ValueError(f"存在不支持的参数: {' '.join(unknown)}")
|
||||
|
||||
file_paths = _expand_json_array_values(
|
||||
namespace.file_path + namespace.file_paths, "file_paths"
|
||||
)
|
||||
file_urls = _expand_json_array_values(
|
||||
namespace.file_url + namespace.file_urls, "file_urls"
|
||||
)
|
||||
return _dedupe(file_paths), _dedupe(file_urls)
|
||||
|
||||
|
||||
def _is_remote_url(value: str) -> bool:
|
||||
return urllib.parse.urlparse(value).scheme.lower() in {"http", "https"}
|
||||
|
||||
|
||||
def _normalize_local_file_path(value: str) -> str:
|
||||
if _is_remote_url(value):
|
||||
raise ValueError(f"本地文件路径不能是远程 URL: {value}")
|
||||
|
||||
path = Path(value).expanduser()
|
||||
if not path.is_file():
|
||||
raise ValueError(f"本地文件不存在: {value}")
|
||||
|
||||
file_size = path.stat().st_size
|
||||
if file_size <= 0:
|
||||
raise ValueError(f"本地文件不能为空: {value}")
|
||||
if file_size > MAX_FILE_SIZE:
|
||||
raise ValueError(f"本地文件大小不能超过 25MB: {value}")
|
||||
return str(path.resolve())
|
||||
|
||||
|
||||
def _validate_remote_file_url(value: str) -> str:
|
||||
parsed = urllib.parse.urlparse(value)
|
||||
if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc:
|
||||
raise ValueError(f"远程文件 URL 格式不正确: {value}")
|
||||
return value
|
||||
|
||||
|
||||
def _send_local_file(client_port: str, to_wxid: str, file_path: str) -> None:
|
||||
send_url = (
|
||||
f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/file/local"
|
||||
)
|
||||
_http_post_json(send_url, {"to_wxid": to_wxid, "file_path": file_path})
|
||||
|
||||
|
||||
def _send_remote_files(
|
||||
client_port: str, to_wxid: str, file_urls: list[str]
|
||||
) -> None:
|
||||
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/file/url"
|
||||
_http_post_json(send_url, {"to_wxid": to_wxid, "file_urls": file_urls})
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) < 2:
|
||||
sys.stdout.write("缺少本地文件路径或远程文件 URL\n")
|
||||
return 1
|
||||
|
||||
try:
|
||||
raw_file_paths, raw_file_urls = _parse_cli_params(sys.argv[1:])
|
||||
if not raw_file_paths and not raw_file_urls:
|
||||
sys.stdout.write("缺少本地文件路径或远程文件 URL\n")
|
||||
return 1
|
||||
file_paths = _dedupe(
|
||||
[_normalize_local_file_path(value) for value in raw_file_paths]
|
||||
)
|
||||
file_urls = [
|
||||
_validate_remote_file_url(value) for value in raw_file_urls
|
||||
]
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
sys.stdout.write(f"参数格式错误: {exc}\n")
|
||||
return 1
|
||||
|
||||
client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
|
||||
if not client_port:
|
||||
sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n")
|
||||
return 1
|
||||
|
||||
to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip()
|
||||
if not to_wxid:
|
||||
sys.stdout.write("环境变量 ROBOT_FROM_WX_ID 未配置\n")
|
||||
return 1
|
||||
|
||||
failures: list[str] = []
|
||||
for file_path in file_paths:
|
||||
try:
|
||||
_send_local_file(client_port, to_wxid, file_path)
|
||||
except Exception as exc:
|
||||
failures.append(f"本地文件 {file_path}: {exc}")
|
||||
|
||||
if file_urls:
|
||||
try:
|
||||
_send_remote_files(client_port, to_wxid, file_urls)
|
||||
except Exception as exc:
|
||||
failures.append(f"远程文件: {exc}")
|
||||
|
||||
if failures:
|
||||
sys.stdout.write(f"文件发送失败: {'；'.join(failures)}\n")
|
||||
return 1
|
||||
|
||||
sys.stdout.write("文件发送成功\n")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
82
skills/send-image/SKILL.md
Normal file
82
skills/send-image/SKILL.md
Normal file
@ -0,0 +1,82 @@
|
||||
---
|
||||
name: send-image
|
||||
description: "发送本地或远程图片到当前微信会话。用户要求发送、转发或分享一个或多个已有图片文件/图片 URL，或其他技能产出了需要发送的本地图片路径时使用；不要用于生成或修改图片。"
|
||||
---
|
||||
|
||||
# 发送图片
|
||||
|
||||
使用 `scripts/send_image.py` 调用机器人客户端接口，将一个或多个本地图片文件、远程图片 URL 发送到当前微信会话。
|
||||
|
||||
## 入参
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"file_path": {
|
||||
"type": "string",
|
||||
"description": "要发送的本地图片路径，必须是机器人运行环境可访问的文件。"
|
||||
},
|
||||
"file_paths": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "要发送的多个本地图片路径。"
|
||||
},
|
||||
"image_url": {
|
||||
"type": "string",
|
||||
"description": "要发送的远程图片 URL，必须以 http 或 https 开头。"
|
||||
},
|
||||
"image_urls": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "要发送的多个远程图片 URL。"
|
||||
}
|
||||
},
|
||||
"anyOf": [
|
||||
{ "required": ["file_path"] },
|
||||
{ "required": ["file_paths"] },
|
||||
{ "required": ["image_url"] },
|
||||
{ "required": ["image_urls"] }
|
||||
],
|
||||
"additionalProperties": false
|
||||
}
|
||||
```
|
||||
|
||||
可同时提供本地路径和远程 URL。对应命令行参数：
|
||||
|
||||
- `--file_path <本地图片路径>`：可重复传入。
|
||||
- `--file_paths <JSON数组>`：一次传入多个本地图片路径。
|
||||
- `--image_url <远程图片URL>`：可重复传入。
|
||||
- `--image_urls <JSON数组>`：一次传入多个远程图片 URL。
|
||||
|
||||
## 执行
|
||||
|
||||
在技能目录运行脚本：
|
||||
|
||||
```bash
|
||||
python3 scripts/send_image.py --file_path '/tmp/example.png'
|
||||
```
|
||||
|
||||
发送远程图片：
|
||||
|
||||
```bash
|
||||
python3 scripts/send_image.py --image_urls '["https://example.com/a.png","https://example.com/b.jpg"]'
|
||||
```
|
||||
|
||||
本地图片逐个调用 `POST /api/v1/robot/message/send/image/local`；远程图片合并调用 `POST /api/v1/robot/message/send/image/url`。接收方固定读取 `ROBOT_FROM_WX_ID`，客户端端口读取 `ROBOT_WECHAT_CLIENT_PORT`。
|
||||
|
||||
## 约束
|
||||
|
||||
- 至少提供一个非空本地路径或远程 URL。
|
||||
- 本地路径必须指向已存在且非空的文件，不能把 URL 作为本地路径传入。
|
||||
- 远程 URL 必须包含有效的 `http` 或 `https` 协议及主机名。
|
||||
- 脚本按首次出现顺序去重；所有参数通过校验后，单个发送失败时会继续尝试其余图片，最终返回全部失败原因。
|
||||
|
||||
## 输出
|
||||
|
||||
- 全部成功时输出「图片发送成功」，无需再发送额外消息。
|
||||
- 任一失败时返回具体错误信息，不要声称图片已全部发送。
|
||||
4
skills/send-image/agents/openai.yaml
Normal file
4
skills/send-image/agents/openai.yaml
Normal file
@ -0,0 +1,4 @@
|
||||
interface:
|
||||
display_name: "发送图片"
|
||||
short_description: "将一个或多个本地图片文件或远程图片链接发送到当前微信会话"
|
||||
default_prompt: "使用 $send-image 将这些本地图片或远程图片发送到当前微信会话。"
|
||||
193
skills/send-image/scripts/send_image.py
Normal file
193
skills/send-image/scripts/send_image.py
Normal file
@ -0,0 +1,193 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
sys.stderr = sys.stdout
|
||||
|
||||
|
||||
def _raise_for_client_error(response: dict) -> None:
|
||||
if not response:
|
||||
return
|
||||
|
||||
code = response.get("code")
|
||||
if code is not None and str(code) != "200":
|
||||
message = str(response.get("message") or f"客户端返回错误码 {code}")
|
||||
raise RuntimeError(message)
|
||||
|
||||
data = response.get("data")
|
||||
if isinstance(data, list):
|
||||
failures = [
|
||||
item.strip()
|
||||
for item in data
|
||||
if isinstance(item, str) and item.lstrip().startswith("失败:")
|
||||
]
|
||||
if failures:
|
||||
raise RuntimeError("；".join(failures))
|
||||
|
||||
|
||||
def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
|
||||
data = json.dumps(body).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
text = resp.read().decode("utf-8")
|
||||
if not text.strip():
|
||||
return {}
|
||||
|
||||
response = json.loads(text)
|
||||
if not isinstance(response, dict):
|
||||
raise RuntimeError("客户端返回格式不正确")
|
||||
_raise_for_client_error(response)
|
||||
return response
|
||||
|
||||
|
||||
def _expand_json_array_values(values: list[str], label: str) -> list[str]:
|
||||
expanded: list[str] = []
|
||||
for value in values:
|
||||
stripped = value.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
if stripped.startswith("["):
|
||||
parsed = json.loads(stripped)
|
||||
if not isinstance(parsed, list) or not all(
|
||||
isinstance(item, str) for item in parsed
|
||||
):
|
||||
raise ValueError(f"{label} 必须是字符串数组")
|
||||
expanded.extend(item.strip() for item in parsed if item.strip())
|
||||
continue
|
||||
expanded.append(stripped)
|
||||
return expanded
|
||||
|
||||
|
||||
def _dedupe(values: list[str]) -> list[str]:
|
||||
deduped: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for value in values:
|
||||
if value not in seen:
|
||||
seen.add(value)
|
||||
deduped.append(value)
|
||||
return deduped
|
||||
|
||||
|
||||
def _parse_cli_params(argv: list[str]) -> tuple[list[str], list[str]]:
|
||||
parser = argparse.ArgumentParser(add_help=False)
|
||||
parser.add_argument("--file_path", action="append", default=[])
|
||||
parser.add_argument("--file_paths", action="append", default=[])
|
||||
parser.add_argument("--image_url", action="append", default=[])
|
||||
parser.add_argument("--image_urls", action="append", default=[])
|
||||
|
||||
namespace, unknown = parser.parse_known_args(argv)
|
||||
if unknown:
|
||||
raise ValueError(f"存在不支持的参数: {' '.join(unknown)}")
|
||||
|
||||
file_paths = _expand_json_array_values(
|
||||
namespace.file_path + namespace.file_paths, "file_paths"
|
||||
)
|
||||
image_urls = _expand_json_array_values(
|
||||
namespace.image_url + namespace.image_urls, "image_urls"
|
||||
)
|
||||
return _dedupe(file_paths), _dedupe(image_urls)
|
||||
|
||||
|
||||
def _is_remote_url(value: str) -> bool:
|
||||
return urllib.parse.urlparse(value).scheme.lower() in {"http", "https"}
|
||||
|
||||
|
||||
def _normalize_local_image_path(value: str) -> str:
|
||||
if _is_remote_url(value):
|
||||
raise ValueError(f"本地图片路径不能是远程 URL: {value}")
|
||||
|
||||
path = Path(value).expanduser()
|
||||
if not path.is_file():
|
||||
raise ValueError(f"本地图片文件不存在: {value}")
|
||||
if path.stat().st_size <= 0:
|
||||
raise ValueError(f"本地图片文件不能为空: {value}")
|
||||
return str(path.resolve())
|
||||
|
||||
|
||||
def _validate_remote_image_url(value: str) -> str:
|
||||
parsed = urllib.parse.urlparse(value)
|
||||
if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc:
|
||||
raise ValueError(f"远程图片 URL 格式不正确: {value}")
|
||||
return value
|
||||
|
||||
|
||||
def _send_local_image(client_port: str, to_wxid: str, file_path: str) -> None:
|
||||
send_url = (
|
||||
f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/local"
|
||||
)
|
||||
_http_post_json(send_url, {"to_wxid": to_wxid, "file_path": file_path})
|
||||
|
||||
|
||||
def _send_remote_images(
|
||||
client_port: str, to_wxid: str, image_urls: list[str]
|
||||
) -> None:
|
||||
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/url"
|
||||
_http_post_json(send_url, {"to_wxid": to_wxid, "image_urls": image_urls})
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) < 2:
|
||||
sys.stdout.write("缺少本地图片路径或远程图片 URL\n")
|
||||
return 1
|
||||
|
||||
try:
|
||||
raw_file_paths, raw_image_urls = _parse_cli_params(sys.argv[1:])
|
||||
if not raw_file_paths and not raw_image_urls:
|
||||
sys.stdout.write("缺少本地图片路径或远程图片 URL\n")
|
||||
return 1
|
||||
file_paths = _dedupe(
|
||||
[_normalize_local_image_path(value) for value in raw_file_paths]
|
||||
)
|
||||
image_urls = [
|
||||
_validate_remote_image_url(value) for value in raw_image_urls
|
||||
]
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
sys.stdout.write(f"参数格式错误: {exc}\n")
|
||||
return 1
|
||||
|
||||
client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
|
||||
if not client_port:
|
||||
sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n")
|
||||
return 1
|
||||
|
||||
to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip()
|
||||
if not to_wxid:
|
||||
sys.stdout.write("环境变量 ROBOT_FROM_WX_ID 未配置\n")
|
||||
return 1
|
||||
|
||||
failures: list[str] = []
|
||||
for file_path in file_paths:
|
||||
try:
|
||||
_send_local_image(client_port, to_wxid, file_path)
|
||||
except Exception as exc:
|
||||
failures.append(f"本地图片 {file_path}: {exc}")
|
||||
|
||||
if image_urls:
|
||||
try:
|
||||
_send_remote_images(client_port, to_wxid, image_urls)
|
||||
except Exception as exc:
|
||||
failures.append(f"远程图片: {exc}")
|
||||
|
||||
if failures:
|
||||
sys.stdout.write(f"图片发送失败: {'；'.join(failures)}\n")
|
||||
return 1
|
||||
|
||||
sys.stdout.write("图片发送成功\n")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@ -1,67 +0,0 @@
|
||||
---
|
||||
name: send-local-image
|
||||
description: "发送本地图片技能。当你需要将本地图片文件发送到当前微信会话时使用。"
|
||||
---
|
||||
|
||||
# Send Local Image Skill
|
||||
|
||||
## 描述
|
||||
|
||||
这是一个发送本地图片文件到当前微信会话的技能。
|
||||
|
||||
技能脚本位于 `scripts/send_local_image.py`，会直接调用机器人客户端接口发送图片，不会修改或生成图片。
|
||||
|
||||
## 触发条件
|
||||
|
||||
- 当你有本地图片地址，需要将图片发送到当前微信会话时触发。
|
||||
|
||||
## 入参规范
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"file_path": {
|
||||
"type": "string",
|
||||
"description": "要发送的本地图片文件路径。必须是机器人运行环境可访问的路径。"
|
||||
},
|
||||
"file_paths": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "可选，要发送的多个本地图片文件路径。"
|
||||
}
|
||||
},
|
||||
"anyOf": [{ "required": ["file_path"] }, { "required": ["file_paths"] }],
|
||||
"additionalProperties": false
|
||||
}
|
||||
```
|
||||
|
||||
对应命令行参数：
|
||||
|
||||
- `--file_path <本地图片路径>` 必填或可重复传入
|
||||
- `--file_paths <JSON数组>` 可选，用于一次传入多个本地图片路径
|
||||
|
||||
## 执行步骤
|
||||
|
||||
1. 仅当用户明确要发送已有本地图片文件时触发该技能。
|
||||
2. 从用户输入或上下文中提取本地图片路径，不要把远程 URL 当成本地路径。
|
||||
3. 在仓库根目录执行脚本，例如：
|
||||
|
||||
```bash
|
||||
python3 scripts/send_local_image.py --file_path '/tmp/example.png'
|
||||
```
|
||||
|
||||
4. 脚本会调用客户端接口 `POST http://127.0.0.1:{ROBOT_WECHAT_CLIENT_PORT}/api/v1/robot/message/send/image/local` 将图片发送给当前会话。
|
||||
|
||||
## 校验规则
|
||||
|
||||
- `file_path` 不能为空。
|
||||
- 传入的路径必须指向一个已存在的本地文件。
|
||||
- 远程 `http` 或 `https` 图片地址应使用 `send-remote-image` 技能。
|
||||
|
||||
## 回复要求
|
||||
|
||||
- 成功时，脚本输出「图片发送成功」，表示图片已通过客户端接口直接发送，无需 AI 智能体再做额外处理。
|
||||
- 失败时，返回脚本输出的具体错误信息。
|
||||
@ -1,125 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
sys.stderr = sys.stdout
|
||||
|
||||
|
||||
def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
|
||||
data = json.dumps(body).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
text = resp.read().decode("utf-8")
|
||||
if not text.strip():
|
||||
return {}
|
||||
return json.loads(text)
|
||||
|
||||
|
||||
def _expand_json_array_values(values: list[str]) -> list[str]:
|
||||
expanded: list[str] = []
|
||||
for value in values:
|
||||
stripped = value.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
if stripped.startswith("["):
|
||||
parsed = json.loads(stripped)
|
||||
if not isinstance(parsed, list):
|
||||
raise ValueError("file_paths 必须是字符串数组")
|
||||
for item in parsed:
|
||||
if not isinstance(item, str):
|
||||
raise ValueError("file_paths 必须是字符串数组")
|
||||
if item.strip():
|
||||
expanded.append(item.strip())
|
||||
continue
|
||||
expanded.append(stripped)
|
||||
return expanded
|
||||
|
||||
|
||||
def _parse_cli_params(argv: list[str]) -> list[str]:
|
||||
parser = argparse.ArgumentParser(add_help=False)
|
||||
parser.add_argument("--file_path", action="append", default=[])
|
||||
parser.add_argument("--file_paths", action="append", default=[])
|
||||
|
||||
namespace, unknown = parser.parse_known_args(argv)
|
||||
if unknown:
|
||||
raise ValueError(f"存在不支持的参数: {' '.join(unknown)}")
|
||||
|
||||
file_paths = _expand_json_array_values(namespace.file_path + namespace.file_paths)
|
||||
deduped: list[str] = []
|
||||
seen = set()
|
||||
for file_path in file_paths:
|
||||
if file_path not in seen:
|
||||
seen.add(file_path)
|
||||
deduped.append(file_path)
|
||||
return deduped
|
||||
|
||||
|
||||
def _is_remote_url(value: str) -> bool:
|
||||
return urllib.parse.urlparse(value).scheme in {"http", "https"}
|
||||
|
||||
|
||||
def _normalize_local_file_path(value: str) -> str:
|
||||
if _is_remote_url(value):
|
||||
raise ValueError("本地图片技能不支持远程 URL，请使用 send-remote-image")
|
||||
|
||||
path = Path(value).expanduser()
|
||||
if not path.is_file():
|
||||
raise ValueError(f"本地图片文件不存在: {value}")
|
||||
return str(path.resolve())
|
||||
|
||||
|
||||
def _send_local_image(client_port: str, to_wxid: str, file_path: str) -> None:
|
||||
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/local"
|
||||
_http_post_json(send_url, {"to_wxid": to_wxid, "file_path": file_path})
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) < 2:
|
||||
sys.stdout.write("缺少本地图片路径\n")
|
||||
return 1
|
||||
|
||||
try:
|
||||
raw_file_paths = _parse_cli_params(sys.argv[1:])
|
||||
if not raw_file_paths:
|
||||
sys.stdout.write("缺少本地图片路径\n")
|
||||
return 1
|
||||
file_paths = [_normalize_local_file_path(value) for value in raw_file_paths]
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
sys.stdout.write(f"参数格式错误: {exc}\n")
|
||||
return 1
|
||||
|
||||
client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
|
||||
if not client_port:
|
||||
sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n")
|
||||
return 1
|
||||
|
||||
to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip()
|
||||
if not to_wxid:
|
||||
sys.stdout.write("环境变量 ROBOT_FROM_WX_ID 未配置\n")
|
||||
return 1
|
||||
|
||||
try:
|
||||
for file_path in file_paths:
|
||||
_send_local_image(client_port, to_wxid, file_path)
|
||||
sys.stdout.write("图片发送成功\n")
|
||||
return 0
|
||||
except Exception as exc:
|
||||
sys.stdout.write(f"图片发送失败: {exc}\n")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@ -1,67 +0,0 @@
|
||||
---
|
||||
name: send-remote-image
|
||||
description: "发送远程图片技能。当你需要将一个或多个远程图片 URL 发送到当前微信会话时使用。"
|
||||
---
|
||||
|
||||
# Send Remote Image Skill
|
||||
|
||||
## 描述
|
||||
|
||||
这是一个发送远程图片 URL 到当前微信会话的技能。
|
||||
|
||||
技能脚本位于 `scripts/send_remote_image.py`，会直接调用机器人客户端接口发送图片，不会下载、修改或生成图片。
|
||||
|
||||
## 触发条件
|
||||
|
||||
- 当你有远程图片 URL，需要将图片发送到当前微信会话时触发。
|
||||
|
||||
## 入参规范
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"image_url": {
|
||||
"type": "string",
|
||||
"description": "要发送的远程图片 URL，必须以 http 或 https 开头。"
|
||||
},
|
||||
"image_urls": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "可选，要发送的多个远程图片 URL。"
|
||||
}
|
||||
},
|
||||
"anyOf": [{ "required": ["image_url"] }, { "required": ["image_urls"] }],
|
||||
"additionalProperties": false
|
||||
}
|
||||
```
|
||||
|
||||
对应命令行参数：
|
||||
|
||||
- `--image_url <远程图片URL>` 必填或可重复传入
|
||||
- `--image_urls <JSON数组>` 可选，用于一次传入多个远程图片 URL
|
||||
|
||||
## 执行步骤
|
||||
|
||||
1. 仅当用户明确要发送已有远程图片 URL 时触发该技能。
|
||||
2. 从用户输入或上下文中提取图片 URL，不要把本地路径当成远程图片。
|
||||
3. 在仓库根目录执行脚本，例如：
|
||||
|
||||
```bash
|
||||
python3 scripts/send_remote_image.py --image_url 'https://example.com/image.png'
|
||||
```
|
||||
|
||||
4. 脚本会调用客户端接口 `POST http://127.0.0.1:{ROBOT_WECHAT_CLIENT_PORT}/api/v1/robot/message/send/image/url` 将图片发送给当前会话。
|
||||
|
||||
## 校验规则
|
||||
|
||||
- `image_url` 不能为空。
|
||||
- 图片 URL 必须以 `http` 或 `https` 开头。
|
||||
- 本地图片文件路径应使用 `send-local-image` 技能。
|
||||
|
||||
## 回复要求
|
||||
|
||||
- 成功时，脚本输出「图片发送成功」，表示图片已通过客户端接口直接发送，无需 AI 智能体再做额外处理。
|
||||
- 失败时，返回脚本输出的具体错误信息。
|
||||
@ -1,116 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
sys.stderr = sys.stdout
|
||||
|
||||
|
||||
def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict:
|
||||
data = json.dumps(body).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
text = resp.read().decode("utf-8")
|
||||
if not text.strip():
|
||||
return {}
|
||||
return json.loads(text)
|
||||
|
||||
|
||||
def _expand_json_array_values(values: list[str]) -> list[str]:
|
||||
expanded: list[str] = []
|
||||
for value in values:
|
||||
stripped = value.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
if stripped.startswith("["):
|
||||
parsed = json.loads(stripped)
|
||||
if not isinstance(parsed, list):
|
||||
raise ValueError("image_urls 必须是字符串数组")
|
||||
for item in parsed:
|
||||
if not isinstance(item, str):
|
||||
raise ValueError("image_urls 必须是字符串数组")
|
||||
if item.strip():
|
||||
expanded.append(item.strip())
|
||||
continue
|
||||
expanded.append(stripped)
|
||||
return expanded
|
||||
|
||||
|
||||
def _parse_cli_params(argv: list[str]) -> list[str]:
|
||||
parser = argparse.ArgumentParser(add_help=False)
|
||||
parser.add_argument("--image_url", action="append", default=[])
|
||||
parser.add_argument("--image_urls", action="append", default=[])
|
||||
|
||||
namespace, unknown = parser.parse_known_args(argv)
|
||||
if unknown:
|
||||
raise ValueError(f"存在不支持的参数: {' '.join(unknown)}")
|
||||
|
||||
image_urls = _expand_json_array_values(namespace.image_url + namespace.image_urls)
|
||||
deduped: list[str] = []
|
||||
seen = set()
|
||||
for image_url in image_urls:
|
||||
if image_url not in seen:
|
||||
seen.add(image_url)
|
||||
deduped.append(image_url)
|
||||
return deduped
|
||||
|
||||
|
||||
def _validate_remote_image_url(value: str) -> str:
|
||||
parsed = urllib.parse.urlparse(value)
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||||
raise ValueError(f"远程图片 URL 格式不正确: {value}")
|
||||
return value
|
||||
|
||||
|
||||
def _send_remote_images(client_port: str, to_wxid: str, image_urls: list[str]) -> None:
|
||||
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/url"
|
||||
_http_post_json(send_url, {"to_wxid": to_wxid, "image_urls": image_urls})
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) < 2:
|
||||
sys.stdout.write("缺少远程图片 URL\n")
|
||||
return 1
|
||||
|
||||
try:
|
||||
raw_image_urls = _parse_cli_params(sys.argv[1:])
|
||||
if not raw_image_urls:
|
||||
sys.stdout.write("缺少远程图片 URL\n")
|
||||
return 1
|
||||
image_urls = [_validate_remote_image_url(value) for value in raw_image_urls]
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
sys.stdout.write(f"参数格式错误: {exc}\n")
|
||||
return 1
|
||||
|
||||
client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
|
||||
if not client_port:
|
||||
sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n")
|
||||
return 1
|
||||
|
||||
to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip()
|
||||
if not to_wxid:
|
||||
sys.stdout.write("环境变量 ROBOT_FROM_WX_ID 未配置\n")
|
||||
return 1
|
||||
|
||||
try:
|
||||
_send_remote_images(client_port, to_wxid, image_urls)
|
||||
sys.stdout.write("图片发送成功\n")
|
||||
return 0
|
||||
except Exception as exc:
|
||||
sys.stdout.write(f"图片发送失败: {exc}\n")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@ -245,5 +245,5 @@ tsx scripts/web_page.ts --url 'https://example.com' --mode screenshot --screensh
|
||||
- `content` 模式成功时，智能体应基于脚本输出给用户总结、解释或回答问题。
|
||||
- 自动化动作失败时，脚本会输出具体失败在第几个 action 及失败原因。
|
||||
- `screenshot` 模式成功且已发送图片时，脚本输出「页面截图已发送」。
|
||||
- `screenshot` 模式成功但未发送图片时，脚本输出本地 PNG 路径，智能体可继续用 `send-local-image` 技能发送。
|
||||
- `screenshot` 模式成功但未发送图片时，脚本输出本地 PNG 路径，智能体可继续用 `send-image` 技能发送。
|
||||
- 失败时，返回脚本输出的具体错误信息。
|
||||
|
||||
Loading…
Reference in New Issue
Block a user