feat(plugin): 集成 OpenClaw Bridge + QQ 操作技能套件
新增 openclaw_bridge 插件,实现 QQ ↔ OpenClaw Gateway 桥接: src/process.py: - QQ 消息自动转发 OpenClaw Gateway - 内置安全检测(高危词、管理员白名单、MC指令过滤) - 高危词库外部化到 config.toml [security] section - 群昵称自动解析 - Session 隔离(按群×用户) config/openclawbridge/config.toml: - 插件配置模板(已脱敏:占位符替代个人信息) scripts/qq_*.py (×15): - 消息发送/文件传输/群管理/好友管理 - 信息查询/历史回溯/OCR/点赞 - 全部脱敏(QQ号→YOUR_ADMIN_QQ, IP→YOUR_NAPCAT_HOST等) skills/*/SKILL.md (×8): - AgentSkills 使用指导 - qq-messenger / qq-management / qq-resolver / qq-napcat-extras - browser / file-process / mc-query / nix-helper SKILL.md: - AgentSkill 入口,完整架构说明和部署步骤
This commit is contained in:
199
scripts/qq_resolve_name.py
Normal file
199
scripts/qq_resolve_name.py
Normal file
@ -0,0 +1,199 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
QQ 号/群号 ↔ 用户名/群名 转换脚本
|
||||
|
||||
将数字 ID 转为可读的名称。通过 NapCat (OneBot) HTTP API 查询。
|
||||
|
||||
用法:
|
||||
# 解析单个 QQ 号
|
||||
python3 qq_resolve_name.py --uid YOUR_ADMIN_QQ
|
||||
python3 qq_resolve_name.py --uid YOUR_ADMIN_QQ --json
|
||||
|
||||
# 解析单个群号
|
||||
python3 qq_resolve_name.py --gid YOUR_GROUP_ID
|
||||
python3 qq_resolve_name.py --gid YOUR_GROUP_ID --json
|
||||
|
||||
# 批量解析
|
||||
python3 qq_resolve_name.py --uid YOUR_ADMIN_QQ --uid 12345678 --gid YOUR_GROUP_ID
|
||||
python3 qq_resolve_name.py --uid YOUR_ADMIN_QQ --gid YOUR_GROUP_ID --json
|
||||
|
||||
# 从文件读取待解析的 ID(每行一个:uid:123456 或 gid:YOUR_GROUP_ID)
|
||||
python3 qq_resolve_name.py --from-file /tmp/ids.txt
|
||||
python3 qq_resolve_name.py --from-file /tmp/ids.txt --json
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import argparse
|
||||
import requests
|
||||
|
||||
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
|
||||
|
||||
|
||||
def get_stranger_info(user_id):
|
||||
"""通过 OneBot API 获取陌生人/好友信息"""
|
||||
url = f"{CQHTTP_URL}/get_stranger_info"
|
||||
payload = {"user_id": int(user_id), "no_cache": True}
|
||||
resp = requests.post(url, json=payload, timeout=10)
|
||||
data = resp.json()
|
||||
if data.get("status") == "ok":
|
||||
return data.get("data", {})
|
||||
return None
|
||||
|
||||
|
||||
def get_group_info(group_id):
|
||||
"""通过 OneBot API 获取群信息"""
|
||||
url = f"{CQHTTP_URL}/get_group_info"
|
||||
payload = {"group_id": int(group_id), "no_cache": True}
|
||||
resp = requests.post(url, json=payload, timeout=10)
|
||||
data = resp.json()
|
||||
if data.get("status") == "ok":
|
||||
return data.get("data", {})
|
||||
return None
|
||||
|
||||
|
||||
def format_user_info(info, uid):
|
||||
"""格式化用户信息"""
|
||||
if not info:
|
||||
return f"• {uid} → 未找到(可能不是好友或不存在)"
|
||||
uid = info.get("user_id", uid)
|
||||
nickname = info.get("nickname", "?")
|
||||
remark = info.get("remark", "")
|
||||
level = info.get("level", 0)
|
||||
sex = {"male": "男", "female": "女", "unknown": "未知"}.get(info.get("sex", ""), "?")
|
||||
age = info.get("age", "?")
|
||||
remark_str = f"(备注:{remark})" if remark else ""
|
||||
return f"• {nickname} ({uid}){remark_str} [{sex}/{age}岁/等级{level}]"
|
||||
|
||||
|
||||
def format_group_info(info, gid):
|
||||
"""格式化群信息"""
|
||||
if not info:
|
||||
return f"• {gid} → 未找到(群不存在或 bot 未加入)"
|
||||
gid = info.get("group_id", gid)
|
||||
name = info.get("group_name", "?")
|
||||
member_count = info.get("member_count", "?")
|
||||
max_member = info.get("max_member_count", "?")
|
||||
level = info.get("group_level", 0)
|
||||
owner = info.get("owner_id", "?")
|
||||
return f"• {name} ({gid}) — {member_count}/{max_member} 人 | 群主:{owner} | 等级:{level}"
|
||||
|
||||
|
||||
def resolve_uids(uids):
|
||||
"""批量解析多个 QQ 号"""
|
||||
results = {"users": []}
|
||||
for uid in uids:
|
||||
info = get_stranger_info(uid)
|
||||
results["users"].append({
|
||||
"user_id": uid,
|
||||
"nickname": info.get("nickname", "?") if info else None,
|
||||
"remark": info.get("remark", "") if info else None,
|
||||
"found": info is not None,
|
||||
"raw": info
|
||||
})
|
||||
return results
|
||||
|
||||
|
||||
def resolve_gids(gids):
|
||||
"""批量解析多个群号"""
|
||||
results = {"groups": []}
|
||||
for gid in gids:
|
||||
info = get_group_info(gid)
|
||||
results["groups"].append({
|
||||
"group_id": gid,
|
||||
"group_name": info.get("group_name", "?") if info else None,
|
||||
"found": info is not None,
|
||||
"raw": info
|
||||
})
|
||||
return results
|
||||
|
||||
|
||||
def parse_id_line(line):
|
||||
"""解析行格式 uid:123456 或 gid:123456"""
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
return None
|
||||
if ":" in line:
|
||||
kind, val = line.split(":", 1)
|
||||
return kind.strip().lower(), val.strip()
|
||||
# 纯数字行:默认当作 uid
|
||||
if line.isdigit():
|
||||
return "uid", line
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="QQ 号/群号 ↔ 用户名/群名 转换")
|
||||
parser.add_argument("--uid", action="append", type=str, dest="uids",
|
||||
help="待解析的 QQ 号(可多次使用)")
|
||||
parser.add_argument("--gid", action="append", type=str, dest="gids",
|
||||
help="待解析的群号(可多次使用)")
|
||||
parser.add_argument("--from-file", type=str,
|
||||
help="从文件读取 ID(每行 uid:123456 或 gid:123456)")
|
||||
parser.add_argument("--json", action="store_true", help="输出 JSON 格式")
|
||||
args = parser.parse_args()
|
||||
|
||||
uids = args.uids or []
|
||||
gids = args.gids or []
|
||||
|
||||
if args.from_file:
|
||||
with open(args.from_file, "r") as f:
|
||||
for line in f:
|
||||
parsed = parse_id_line(line)
|
||||
if parsed:
|
||||
kind, val = parsed
|
||||
if kind in ("uid", "user", "qq"):
|
||||
uids.append(val)
|
||||
elif kind in ("gid", "group"):
|
||||
gids.append(val)
|
||||
|
||||
if not uids and not gids:
|
||||
print("❌ 请指定 --uid 或 --gid 或 --from-file")
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
results = {}
|
||||
if uids:
|
||||
results["users"] = []
|
||||
for uid in uids:
|
||||
info = get_stranger_info(uid)
|
||||
results["users"].append({
|
||||
"user_id": uid,
|
||||
"info": info
|
||||
})
|
||||
|
||||
if gids:
|
||||
results["groups"] = []
|
||||
for gid in gids:
|
||||
info = get_group_info(gid)
|
||||
results["groups"].append({
|
||||
"group_id": gid,
|
||||
"info": info
|
||||
})
|
||||
|
||||
if args.json:
|
||||
print(json.dumps(results, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
lines = []
|
||||
if "users" in results:
|
||||
lines.append("📋 用户信息:")
|
||||
for item in results["users"]:
|
||||
lines.append(format_user_info(item["info"], item["user_id"]))
|
||||
if "groups" in results:
|
||||
if lines:
|
||||
lines.append("")
|
||||
lines.append("📋 群信息:")
|
||||
for item in results["groups"]:
|
||||
lines.append(format_group_info(item["info"], item["group_id"]))
|
||||
print("\n".join(lines))
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
print(f"❌ 连接失败: 无法连接到 NapCat ({CQHTTP_URL})")
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"❌ 查询失败: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user