hao/skills
1
0
派生 0
代码 工单 合并请求 工作流 软件包 项目 发布 百科 活动

Add CN86 SMS keyword verification skill and README

浏览代码
这个提交包含在：
X
2026-03-06 01:13:09 -08:00
父节点 be5f7c8808
当前提交 71e6c95e23
共修改 5 个文件，包含 563 行新增和 0 行删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件

127
cn86-sms-keyword-verification/SKILL.md 普通文件
查看文件

@@ -0,0 +1,127 @@
---
name: cn86-sms-keyword-verification
description: 86手机号关键词验证码流程，基于 LubanSMS 通用短信接收 API 完成取号、轮询短信、提取验证码、释放号码，并可用“千问”等关键词做短信登录/注册 demo。
type: workflow
domain: utilities
version: 1.0.0
tags: [sms, verification, phone, keyword, china, qwen, lubansms]
triggers:
keywords:
primary: [86 手机号, 86手机号, 手机号获取, 短信验证码, 关键词验证码, 获取验证码, 接码]
secondary: [千问验证码, qwen sms, lubansms, getKeywordNumber, getKeywordSms, keywordSmsHistory]
context_boost: [注册, 登录, 验证码, 短信, 手机号, 关键词]
priority: high
---
# CN86 SMS Keyword Verification
86 手机号 + 关键词获取验证码的标准流程技能。
本技能使用 `https://lubansms.com/v2/api` 的“通用短信接收”接口做 demo。你给的内部参考文档 `docs/sms-register-qwen.md` 里，附录实际也是按这组 LubanSMS 接口在写；文档里提到的千问（Qwen）示例关键词是 `千问`。
如需稳定执行，优先使用打包好的 `scripts/lubansms_keyword_cli.py`，不要手写临时 curl 再去人肉判断状态。
## 什么时候用
- 需要先拿一个 86 号码，再根据短信关键词轮询验证码
- 目标站点会把验证码发到中国手机号，但平台只给“按关键词取短信”的能力
- 要做可重复的注册/登录自动化 demo，例如千问短信登录
- 需要历史记录、余额检查、释放号码这些配套动作
## 需要的输入
- `keyword`：短信中能稳定命中的关键词，例如 `千问`
- `LUBAN_SMS_APIKEY`：运行时环境变量，不要写死进仓库
- 可选 `phone`：想复用已有号码时传入；留空则随机取号
- 可选 `timeout` / `interval`：轮询等待时长和间隔
## API Key 处理
优先使用环境变量：
```bash
export LUBAN_SMS_APIKEY='<your_api_key>'
export LUBAN_SMS_API_BASE='https://lubansms.com/v2/api'
```
只在一次性调试时才用 `--api-key` 直传。
## 标准流程
1. 先查余额，避免轮询到一半才发现 key 无效或余额不足。
2. 调 `getKeywordNumber` 申请号码。
3. 规范化手机号：
- API 内部继续使用原始数字串，例如 `16741251148`
- 对外展示可拼成 `+8616741251148`
- 如果目标站点像千问一样把国家码拆开填，就传 `phoneCode=86` + 原始手机号
4. 在目标站点触发发送短信。
5. 调 `getKeywordSms` 按关键词轮询短信。
6. 从返回短信正文中提取验证码。
7. 成功或失败后都调用 `delKeywordNumber` 释放号码。
8. 排查问题时再查 `keywordSmsHistory`。
## 平台状态判断
- `code=0`：成功
- `code=400` 且 `msg=尚未收到短信,请稍后重试`：可继续轮询
- `code=400` 且 `msg=不正确的apikey`：立即停止，检查 key
- 其他 `code!=0`：视为 API 失败，不要盲目重试到超时
## 千问（Qwen）Demo
千问短信内容在参考文档里使用的关键词是 `千问`。
### 1. 先检查余额
```bash
export LUBAN_SMS_APIKEY='<your_api_key>'
python3 scripts/lubansms_keyword_cli.py balance
```
### 2. 一次性 demo：取号 → 等短信 → 提取验证码 → 释放号码
```bash
python3 scripts/lubansms_keyword_cli.py demo \
--keyword 千问 \
--timeout 300 \
--interval 5
```
这个命令会：
- 申请一个随机 86 号码
- 输出原始号码和 `+86` 格式
- 等你在目标站点触发短信发送
- 轮询关键词短信并提取验证码
- 默认在结束时释放号码
### 3. 拆开执行
```bash
python3 scripts/lubansms_keyword_cli.py request-number
python3 scripts/lubansms_keyword_cli.py wait-code --phone 16741251148 --keyword 千问
python3 scripts/lubansms_keyword_cli.py release --phone 16741251148
```
## 常见坑
- 不要把 `LUBAN_SMS_APIKEY` 写进提交文件。
- 不要把 `+86` 或 `86` 前缀后的完整国际格式直接塞回 `phone=` 参数；平台接口通常要原始号码数字串。
- 不要忘记释放号码；无论成功、失败、超时都应该释放。
- 不要把“尚未收到短信”当成致命错误；这是正常轮询态。
- 不要只看关键词命中，不提取验证码；很多自动化链路最后需要明确的 OTP 数值。
## 推荐脚本
使用 `scripts/lubansms_keyword_cli.py`：
- `balance`：查余额
- `request-number`：申请号码
- `get-sms`：查一次关键词短信
- `wait-code`：轮询直到拿到验证码或超时
- `release`：释放号码
- `history`：查历史记录
- `demo`：完整演示流程
## 参考资料
- `references/lubansms-and-qwen-notes.md`

4
cn86-sms-keyword-verification/agents/openai.yaml 普通文件
查看文件

@@ -0,0 +1,4 @@
interface:
display_name: "CN86 SMS Keyword Verification"
short_description: "Get 86 phone numbers and OTP codes via LubanSMS keyword APIs"
default_prompt: "Use LubanSMS keyword APIs to request an 86 phone number, poll SMS by keyword, extract the OTP code, and release the number for flows like Qwen SMS login."

96
cn86-sms-keyword-verification/references/lubansms-and-qwen-notes.md 普通文件
查看文件

@@ -0,0 +1,96 @@
# LubanSMS / Qwen Notes
## 来源
这份参考整理自两部分：
1. 官方 API 文档：`https://lubansms.com/api_docs/`
2. 内部参考：`hao/one` 仓库里的 `docs/sms-register-qwen.md`
## 关键结论
- 这次要做的“86 手机号 + 关键词获取验证码”流程，落地接口是 LubanSMS 的 **通用短信接收** 系列。
- 内部参考文档把千问（Qwen）作为示例站点，关键词给的是 `千问`。
- 运行时应该使用环境变量 `LUBAN_SMS_APIKEY`，而不是把真实 key 提交到仓库。
## 官方接口
### 1. 查询余额
`GET /getBalance?apikey=YOUR_APIKEY`
成功示例：
```json
{"code":0,"msg":"","balance":"96.72"}
```
### 2. 请求号码
`GET /getKeywordNumber?apikey=YOUR_APIKEY&phone=&cardType=全部`
成功示例：
```json
{"code":0,"msg":"","phone":"18888888888"}
```
说明：
- `phone` 留空表示随机号码
- `cardType` 在官方文档里标成“已弃用”，但仍可兼容传 `全部`
### 3. 获取关键词短信
`GET /getKeywordSms?apikey=YOUR_APIKEY&phone=<手机号>&keyword=<关键词>`
等待中：
```json
{"code":400,"msg":"尚未收到短信,请稍后重试"}
```
收到短信：
```json
{"code":0,"msg":"【百度】验证码xxxx,您正在进行登陆验证."}
```
### 4. 释放号码
`GET /delKeywordNumber?apikey=YOUR_APIKEY&phone=<手机号>`
成功示例：
```json
{"code":0,"msg":""}
```
### 5. 查询关键词短信历史
`GET /keywordSmsHistory?apikey=YOUR_APIKEY&page=1`
用途：
- 验证关键词是否正确
- 排查目标站点是否实际发过短信
- 辅助确认短信模板和验证码格式
## 千问专用备注
内部参考文档给出的要点：
- 千问示例关键词：`千问`
- 站点如果把国家码和手机号拆开，使用 `phoneCode=86` + 原始手机号
- 接码平台 API 内部仍然按原始手机号查询，不要拼上 `+86`
## 2026-03-06 验证结论
已验证：
- `getBalance` 可正常返回余额
- `keywordSmsHistory` 可正常返回最近的千问短信历史
- 说明本技能里的示例流程和关键词方向是成立的
未在仓库中提交：
- 实际 API key
- 实际号码或敏感历史记录

290
cn86-sms-keyword-verification/scripts/lubansms_keyword_cli.py 可执行文件
查看文件

@@ -0,0 +1,290 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import time
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Any, Dict, Optional
DEFAULT_BASE_URL = os.getenv("LUBAN_SMS_API_BASE", "https://lubansms.com/v2/api")
DEFAULT_HTTP_TIMEOUT = 30
WAITING_MESSAGES = (
"尚未收到短信,请稍后重试",
"wait",
)
class LubanSMSError(RuntimeError):
pass
@dataclass
class PhoneFormats:
raw: str
e164: str
class LubanSMSClient:
def __init__(self, api_key: str, base_url: str = DEFAULT_BASE_URL, http_timeout: int = DEFAULT_HTTP_TIMEOUT) -> None:
if not api_key:
raise LubanSMSError("missing api key: set LUBAN_SMS_APIKEY or pass --api-key")
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.http_timeout = http_timeout
def _request(self, path: str, **params: Any) -> Dict[str, Any]:
query = {"apikey": self.api_key}
for key, value in params.items():
if value is None:
continue
query[key] = value
url = f"{self.base_url}/{path}?{urllib.parse.urlencode(query)}"
request = urllib.request.Request(
url,
headers={
"Accept": "application/json",
"User-Agent": "codex-cn86-sms-keyword-verification/1.0",
},
)
try:
with urllib.request.urlopen(request, timeout=self.http_timeout) as response:
raw = response.read().decode("utf-8", errors="replace")
except Exception as exc:
raise LubanSMSError(f"request failed for {path}: {exc}") from exc
try:
payload = json.loads(raw)
except json.JSONDecodeError as exc:
raise LubanSMSError(f"invalid json from {path}: {raw}") from exc
return payload
def get_balance(self) -> Dict[str, Any]:
return self._request("getBalance")
def request_keyword_number(self, phone: str = "", card_type: str = "全部") -> Dict[str, Any]:
return self._request("getKeywordNumber", phone=normalize_phone_input(phone), cardType=card_type)
def get_keyword_sms(self, phone: str, keyword: str) -> Dict[str, Any]:
return self._request("getKeywordSms", phone=normalize_phone_input(phone), keyword=keyword)
def release_keyword_number(self, phone: str) -> Dict[str, Any]:
return self._request("delKeywordNumber", phone=normalize_phone_input(phone))
def keyword_sms_history(self, page: int = 1) -> Dict[str, Any]:
return self._request("keywordSmsHistory", page=page)
def normalize_phone_input(phone: str) -> str:
digits = re.sub(r"\D", "", phone or "")
if digits.startswith("86") and len(digits) > 11:
digits = digits[2:]
return digits
def format_cn86_phone(phone: str) -> PhoneFormats:
raw = normalize_phone_input(phone)
if not raw:
raise LubanSMSError("phone is required")
return PhoneFormats(raw=raw, e164=f"+86{raw}")
def extract_verification_code(message: str) -> Optional[str]:
patterns = [
r"验证码(?:为|是|[:：])?\s*([A-Za-z0-9]{4,8})",
r"code(?: is|[:：])?\s*([A-Za-z0-9]{4,8})",
r"\b(\d{6})\b",
r"\b(\d{4,8})\b",
]
for pattern in patterns:
match = re.search(pattern, message, flags=re.IGNORECASE)
if match:
return match.group(1)
return None
def is_waiting_payload(payload: Dict[str, Any]) -> bool:
if payload.get("code") == 0:
return False
message = str(payload.get("msg", ""))
return any(token in message for token in WAITING_MESSAGES)
def require_success(payload: Dict[str, Any], *, allow_wait: bool = False) -> Dict[str, Any]:
if payload.get("code") == 0:
return payload
if allow_wait and is_waiting_payload(payload):
return payload
raise LubanSMSError(f"api error: code={payload.get('code')} msg={payload.get('msg')}")
def wait_for_code(client: LubanSMSClient, phone: str, keyword: str, timeout: int, interval: int) -> Dict[str, Any]:
formats = format_cn86_phone(phone)
started = time.monotonic()
attempts = 0
last_payload: Dict[str, Any] | None = None
while time.monotonic() - started < timeout:
attempts += 1
payload = client.get_keyword_sms(formats.raw, keyword)
last_payload = payload
if payload.get("code") == 0:
message = str(payload.get("msg", ""))
return {
"status": "success",
"keyword": keyword,
"phone": formats.raw,
"phone_e164": formats.e164,
"attempts": attempts,
"elapsed_seconds": round(time.monotonic() - started, 2),
"message": message,
"verification_code": extract_verification_code(message),
}
if not is_waiting_payload(payload):
raise LubanSMSError(f"unexpected sms response: code={payload.get('code')} msg={payload.get('msg')}")
time.sleep(interval)
return {
"status": "timeout",
"keyword": keyword,
"phone": formats.raw,
"phone_e164": formats.e164,
"attempts": attempts,
"elapsed_seconds": round(time.monotonic() - started, 2),
"last_payload": last_payload,
"verification_code": None,
}
def print_json(data: Dict[str, Any]) -> None:
json.dump(data, sys.stdout, ensure_ascii=False, indent=2)
sys.stdout.write("\n")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="LubanSMS CN86 keyword SMS CLI")
parser.add_argument("--api-key", default=os.getenv("LUBAN_SMS_APIKEY", ""), help="LubanSMS API key; defaults to LUBAN_SMS_APIKEY")
parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="API base URL")
parser.add_argument("--http-timeout", type=int, default=DEFAULT_HTTP_TIMEOUT, help="HTTP request timeout in seconds")
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("balance", help="Query account balance")
request_number = subparsers.add_parser("request-number", help="Request a keyword phone number")
request_number.add_argument("--phone", default="", help="Reuse a specific phone if supported")
request_number.add_argument("--card-type", default="全部", help="Card type; official docs mark this deprecated")
get_sms = subparsers.add_parser("get-sms", help="Fetch SMS once by keyword")
get_sms.add_argument("--phone", required=True, help="Raw CN phone number or +86 number")
get_sms.add_argument("--keyword", required=True, help="Keyword to match, e.g. 千问")
wait_code = subparsers.add_parser("wait-code", help="Poll until OTP arrives or times out")
wait_code.add_argument("--phone", required=True, help="Raw CN phone number or +86 number")
wait_code.add_argument("--keyword", required=True, help="Keyword to match, e.g. 千问")
wait_code.add_argument("--timeout", type=int, default=300, help="Polling timeout in seconds")
wait_code.add_argument("--interval", type=int, default=5, help="Polling interval in seconds")
wait_code.add_argument("--release-on-exit", action="store_true", help="Release the number after polling finishes")
release = subparsers.add_parser("release", help="Release a phone number")
release.add_argument("--phone", required=True, help="Raw CN phone number or +86 number")
history = subparsers.add_parser("history", help="Fetch keyword SMS history")
history.add_argument("--page", type=int, default=1, help="Page number")
demo = subparsers.add_parser("demo", help="Request number, wait for keyword SMS, then release")
demo.add_argument("--keyword", required=True, help="Keyword to match, e.g. 千问")
demo.add_argument("--phone", default="", help="Reuse a specific phone if supported")
demo.add_argument("--card-type", default="全部", help="Card type; official docs mark this deprecated")
demo.add_argument("--timeout", type=int, default=300, help="Polling timeout in seconds")
demo.add_argument("--interval", type=int, default=5, help="Polling interval in seconds")
demo.add_argument("--keep-number", action="store_true", help="Do not release the number at the end")
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
client = LubanSMSClient(api_key=args.api_key, base_url=args.base_url, http_timeout=args.http_timeout)
try:
if args.command == "balance":
print_json(require_success(client.get_balance()))
return 0
if args.command == "request-number":
payload = require_success(client.request_keyword_number(phone=args.phone, card_type=args.card_type))
formats = format_cn86_phone(payload.get("phone", ""))
payload["phone"] = formats.raw
payload["phone_e164"] = formats.e164
print_json(payload)
return 0
if args.command == "get-sms":
payload = require_success(client.get_keyword_sms(phone=args.phone, keyword=args.keyword), allow_wait=True)
if payload.get("code") == 0:
payload["verification_code"] = extract_verification_code(str(payload.get("msg", "")))
print_json(payload)
return 0
if args.command == "wait-code":
result = wait_for_code(client, phone=args.phone, keyword=args.keyword, timeout=args.timeout, interval=args.interval)
if args.release_on_exit:
try:
result["release"] = require_success(client.release_keyword_number(args.phone))
except Exception as exc:
result["release_error"] = str(exc)
print_json(result)
return 0 if result.get("status") == "success" else 2
if args.command == "release":
print_json(require_success(client.release_keyword_number(args.phone)))
return 0
if args.command == "history":
print_json(require_success(client.keyword_sms_history(page=args.page)))
return 0
if args.command == "demo":
allocation = require_success(client.request_keyword_number(phone=args.phone, card_type=args.card_type))
formats = format_cn86_phone(allocation.get("phone", ""))
result: Dict[str, Any] = {
"status": "allocated",
"keyword": args.keyword,
"phone": formats.raw,
"phone_e164": formats.e164,
"allocation": allocation,
"notes": [
"Trigger the target site to send SMS after allocation.",
"For sites like Qwen, pass phoneCode=86 and the raw phone digits.",
],
}
try:
poll_result = wait_for_code(client, phone=formats.raw, keyword=args.keyword, timeout=args.timeout, interval=args.interval)
result.update(poll_result)
except LubanSMSError as exc:
result["status"] = "error"
result["error"] = str(exc)
finally:
if not args.keep_number:
try:
result["release"] = require_success(client.release_keyword_number(formats.raw))
except Exception as exc:
result["release_error"] = str(exc)
print_json(result)
return 0 if result.get("status") == "success" else 2
parser.error(f"unsupported command: {args.command}")
return 1
except LubanSMSError as exc:
print_json({"error": str(exc), "command": args.command})
return 1
if __name__ == "__main__":
sys.exit(main())