完整 AgentSkill:词库外部化 + 全部脚本/Skill 脱敏上传

- 高危词库从 process.py 抽到 config.toml [security] section
- process.py 改为 _get_high_risk_words() 动态加载(配置优先)
- 新增 15 个 QQ 操作脚本(已脱敏:QQ号/IP/Token → 占位符)
- 新增 8 个 SKILL.md(qq-messenger/management/resolver/napcat-extras 等)
- 新增 SKILL.md 入口(完整部署方案文档)
- 保留 SDK 框架(plugin_modules.py, file_store_api.py, package.py 等)
This commit is contained in:
Claw
2026-05-04 16:23:59 +08:00
parent d7d33c39a4
commit e1fbd31c64
35 changed files with 5058 additions and 276 deletions

58
scripts/ask_nix.py Executable file
View File

@ -0,0 +1,58 @@
#!/usr/bin/env python3
"""
qq-agent → Nix 求助工具(同步 Gateway API 调用)
直接发问题到 Gateway,等 Nix 回复,不需要文件桥接。
模仿 openclaw_bridge 插件的 `_send_to_openclaw` 实现。
用法: python3 ask_nix.py "问题内容"
返回: Nix 的回复,或超时错误
"""
import sys
import json
import requests
GATEWAY_URL = "http://127.0.0.1:18789/v1/chat/completions"
GATEWAY_TOKEN = "YOUR_GATEWAY_TOKEN"
NIX_MODEL = "openclaw/ops-manager"
def ask(question: str, timeout: int = 120) -> str:
"""同步调用 Gateway,等 Nix 回复"""
payload = {
"model": NIX_MODEL,
"messages": [{"role": "user", "content": question}],
"stream": False,
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {GATEWAY_TOKEN}",
}
try:
resp = requests.post(
GATEWAY_URL,
json=payload,
headers=headers,
timeout=timeout,
)
if resp.status_code == 200:
data = resp.json()
choices = data.get("choices", [])
if choices:
return choices[0].get("message", {}).get("content", "")
return "Nix 返回为空"
return f"请求失败({resp.status_code}): {resp.text[:200]}"
except requests.exceptions.Timeout:
return "[timeout] Nix 没回 超时了 要么他不在线 要么忙着呢 老板稍后再试吧"
except requests.exceptions.ConnectionError:
return "Nix 连接失败,请检查 Gateway 服务状态"
except Exception as e:
return f"请求异常: {str(e)}"
if __name__ == "__main__":
question = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read().strip()
if not question:
print("错误:未提供问题", file=sys.stderr)
sys.exit(1)
print(ask(question))

75
scripts/browse.py Normal file
View File

@ -0,0 +1,75 @@
#!/usr/bin/env python3
"""
Headless browser tool for qq-agent.
Usage:
python3 browse.py https://example.com # 获取页面文字内容
python3 browse.py https://example.com --screenshot # 截图保存
python3 browse.py https://example.com --wait 3 # 等待3秒再抓取(用于JS渲染页面)
"""
import sys
import os
import argparse
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SAVE_DIR = os.path.join(BASE_DIR, "files")
CHROMIUM_PATH = "/usr/local/bin/chromium"
def browse(url: str, wait_sec: int = 0, screenshot: bool = False):
with sync_playwright() as p:
browser = p.chromium.launch(
executable_path=CHROMIUM_PATH,
headless=True,
args=["--no-sandbox", "--disable-setuid-sandbox"]
)
page = browser.new_page(viewport={"width": 1280, "height": 720})
page.set_default_timeout(15000)
try:
page.goto(url, wait_until="domcontentloaded")
except PlaywrightTimeout:
print("⚠️ 页面加载超时,使用已获取的内容")
if wait_sec > 0:
page.wait_for_timeout(wait_sec * 1000)
title = page.title()
print(f"标题: {title}")
print(f"URL: {url}")
print()
if screenshot:
os.makedirs(SAVE_DIR, exist_ok=True)
safe_name = "".join(c if c.isalnum() or c in '-_' else '_' for c in url[:50])
path = os.path.join(SAVE_DIR, f"screenshot_{safe_name}.png")
page.screenshot(path=path, full_page=True)
print(f"📸 截图已保存: {path}")
print()
# 提取正文文字
content = page.inner_text("body")
# 清理过长的空白行
lines = [l.strip() for l in content.split("\n")]
text = "\n".join(l for l in lines if l)
if len(text) > 5000:
print(text[:5000])
print(f"\n...(内容过长,仅显示前5000字符,共{len(text)}字符)")
else:
print(text)
browser.close()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="无头浏览器工具")
parser.add_argument("url", help="要访问的网页URL")
parser.add_argument("--screenshot", action="store_true", help="截图保存")
parser.add_argument("--wait", type=int, default=0, help="等待秒数(用于JS渲染页面)")
args = parser.parse_args()
browse(args.url, wait_sec=args.wait, screenshot=args.screenshot)

50
scripts/mc_clear_items.py Normal file
View File

@ -0,0 +1,50 @@
#!/usr/bin/env python3
"""清理Minecraft地面掉落物(含倒计时,通过RCON通知玩家)"""
import sys, os, re, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mc_config import RCON_HOST, RCON_PORT, RCON_PASSWORD
from mc_rcon import RCONClient
def clear_items():
killed = 0
error = None
client = RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD)
try:
client.connect()
# 验证连接
resp = client.send_command("list")
if resp is None:
raise Exception("无法连接到Minecraft RCON服务")
# 30秒倒计时
for i in range(30, 0, -1):
if i in (30, 20, 10, 5, 4, 3, 2, 1):
client.send_command(f'tellraw @a ["\\u00a7e\\u26a0\\ufe0f 将在 {i} 秒后清理地面掉落物,请捡起你的物品!"]')
time.sleep(1)
client.send_command('tellraw @a ["\\u00a7c\\ud83e\\uddf9 正在清理地面掉落物..."]')
time.sleep(0.5)
kill_resp = client.send_command("kill @e[type=item]")
body = kill_resp.get("body", "") if kill_resp else ""
m = re.search(r'Killed (\d+) entities', body)
killed = int(m.group(1)) if m else 0
client.send_command(f'tellraw @a ["\\u00a7a\\u2705 已清理 {killed} 个掉落物"]')
except Exception as e:
error = str(e)
try:
client.send_command('tellraw @a ["\\u00a7c清理掉落物时发生错误"]')
except Exception:
pass
finally:
client.close()
import json
result = {"killed": killed, "error": error}
print(json.dumps(result, ensure_ascii=False))
return result
if __name__ == "__main__":
clear_items()

8
scripts/mc_config.py Normal file
View File

@ -0,0 +1,8 @@
#!/usr/bin/env python3
"""MC RCON 共享配置"""
RCON_HOST = "127.0.0.1"
RCON_PORT = 25575
RCON_PASSWORD = "233719"
MC_SERVER_DIR = "/home/minecraft"
PLAYTIME_DB = "/home/qqrebot/playtime.json"

210
scripts/mc_players.py Normal file
View File

@ -0,0 +1,210 @@
#!/usr/bin/env python3
"""获取Minecraft玩家活动信息(JSON输出)"""
import sys, os, re, json
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mc_config import MC_SERVER_DIR, PLAYTIME_DB
def get_recent_events(limit=5):
log_path = os.path.join(MC_SERVER_DIR, "logs", "latest.log")
if not os.path.exists(log_path):
return []
events = []
patterns = [
(re.compile(r'(\w+)\s+joined\s+the\s+game'), 'join'),
(re.compile(r'(\w+)\s+left\s+the\s+game'), 'leave'),
]
skip = {"Server", "RCON", "Thread", "Generic", "World"}
try:
with open(log_path, "r", errors="ignore") as f:
lines = f.readlines()
for line in lines[-5000:]:
for pattern, etype in patterns:
m = pattern.search(line)
if m:
name = m.group(1)
if name in skip:
continue
events.append({"type": etype, "player": name})
break
seen = set()
unique = []
for e in reversed(events):
key = (e["player"], e["type"])
if key not in seen:
seen.add(key)
unique.append(e)
if len(unique) >= limit:
break
return list(reversed(unique))
except Exception:
return []
def get_playtime_db():
if os.path.exists(PLAYTIME_DB):
try:
with open(PLAYTIME_DB, "r") as f:
return json.load(f)
except Exception:
pass
return {"players": {}, "last_log_pos": 0}
def save_playtime_db(db):
try:
with open(PLAYTIME_DB, "w") as f:
json.dump(db, f, indent=2, ensure_ascii=False)
except Exception:
pass
def update_playtime():
db = get_playtime_db()
log_path = os.path.join(MC_SERVER_DIR, "logs", "latest.log")
if not os.path.exists(log_path):
return db
try:
with open(log_path, "rb") as f:
f.seek(0, 2)
size = f.tell()
start = min(db.get("last_log_pos", 0), size)
if start > size:
start = 0
with open(log_path, "r", errors="ignore") as f:
f.seek(start)
lines = f.readlines()
now = datetime.now()
join_p = re.compile(r'joined the game')
leave_p = re.compile(r'left the game')
name_p = re.compile(r'\]:\s+(\w+)\s+(?:joined|left)')
for line in lines:
m = name_p.search(line)
if not m:
continue
player = m.group(1)
if join_p.search(line):
if player not in db["players"]:
db["players"][player] = {"total_seconds": 0, "last_join": None}
db["players"][player]["last_join"] = now.isoformat()
elif leave_p.search(line):
if player in db["players"] and db["players"][player].get("last_join"):
join_time = datetime.fromisoformat(db["players"][player]["last_join"])
sec = (now - join_time).total_seconds()
if 0 < sec < 86400:
db["players"][player]["total_seconds"] += sec
db["players"][player]["last_join"] = None
db["last_log_pos"] = size
save_playtime_db(db)
except Exception:
pass
return db
def format_time(seconds):
h = seconds // 3600
m = (seconds % 3600) // 60
if h >= 24:
d = h // 24
h = h % 24
return f"{d}天{h}小时{m}分"
elif h > 0:
return f"{h}小时{m}分"
return f"{m}分钟"
def get_playtime_leaderboard(top_n=10):
db = update_playtime()
players = db.get("players", {})
lb = []
for name, data in players.items():
total = data.get("total_seconds", 0)
if data.get("last_join"):
try:
jt = datetime.fromisoformat(data["last_join"])
s = (datetime.now() - jt).total_seconds()
if 0 < s < 86400:
total += s
except Exception:
pass
if total > 0:
lb.append({"name": name, "seconds": int(total), "formatted": format_time(int(total))})
lb.sort(key=lambda x: -x["seconds"])
return lb[:top_n]
def _load_uuid_to_name():
"""从 usercache.json 加载 UUID->名字映射,回退 ops.json/whitelist.json"""
mapping = {}
# 主源:usercache.json(包含所有登录过的玩家)
for src_file in ["usercache.json", "ops.json", "whitelist.json"]:
path = os.path.join(MC_SERVER_DIR, src_file)
try:
with open(path, "r") as f:
for entry in json.load(f):
if 'uuid' in entry and 'name' in entry:
mapping[entry['uuid']] = entry['name']
except Exception:
pass
return mapping
def get_death_leaderboard(top_n=10):
uuid_to_name = _load_uuid_to_name()
stats_dir = os.path.join(MC_SERVER_DIR, "world", "stats")
if not os.path.exists(stats_dir):
return []
deaths = {}
for fname in os.listdir(stats_dir):
if not fname.endswith('.json'):
continue
uuid = fname.replace('.json', '')
fpath = os.path.join(stats_dir, fname)
try:
with open(fpath, "r") as f:
data = json.load(f)
dc = data.get('stats', {}).get('minecraft:custom', {}).get('minecraft:deaths', 0)
if dc > 0:
name = uuid_to_name.get(uuid, uuid[:8])
deaths[name] = dc
except Exception:
continue
return [{"name": n, "deaths": d} for n, d in sorted(deaths.items(), key=lambda x: -x[1])[:top_n]]
def get_achievement_leaderboard(top_n=10):
uuid_to_name = _load_uuid_to_name()
adv_dir = os.path.join(MC_SERVER_DIR, "world", "advancements")
if not os.path.exists(adv_dir):
return []
ach = {}
for fname in os.listdir(adv_dir):
if not fname.endswith('.json'):
continue
uuid = fname.replace('.json', '')
fpath = os.path.join(adv_dir, fname)
try:
with open(fpath, "r") as f:
data = json.load(f)
name = uuid_to_name.get(uuid)
if not name:
name = uuid[:8] # 名字查不到就用 UUID 前缀
done = 0
for k, v in data.items():
if not isinstance(v, dict):
continue
if 'recipes/' in k or ':recipes/' in k:
continue
if v.get('done'):
done += 1
if done > 0:
ach[name] = done
except Exception:
continue
return [{"name": n, "achievements": a} for n, a in sorted(ach.items(), key=lambda x: -x[1])[:top_n]]
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--top", type=int, default=10)
args = parser.parse_args()
result = {
"recent_events": get_recent_events(5),
"playtime": get_playtime_leaderboard(args.top),
"deaths": get_death_leaderboard(args.top),
"achievements": get_achievement_leaderboard(args.top),
}
print(json.dumps(result, ensure_ascii=False, indent=2))

162
scripts/mc_query.py Normal file
View File

@ -0,0 +1,162 @@
#!/usr/bin/env python3
"""统一入口:根据参数调用不同MC查询脚本"""
import sys, os, json, subprocess, textwrap
SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
# 纯文本格式化函数(禁止Markdown)
def fmt_status(data):
lines = []
s = data
if not s.get("online"):
lines.append("服务器已离线")
return "\n".join(lines)
players = s.get("players", {})
lines.append(f"在线玩家: {players.get('online', 0)}/{players.get('max', 0)}")
plist = s.get("player_list", [])
if plist:
lines.append("玩家列表: " + ", ".join(plist))
mem = s.get("memory_mb")
if mem:
lines.append(f"内存使用: {mem['mb']}MB ({mem['gb']}GB)")
cpu = s.get("cpu_percent")
if cpu is not None:
lines.append(f"CPU占用: {cpu}%")
uptime = s.get("uptime")
if uptime:
lines.append(f"运行时长: {uptime}")
tps = s.get("tps")
if tps:
lines.append(f"TPS: {tps.get('tps_1m', 'N/A')} (1分钟)")
lines.append(f"版本: {s.get('version', '未知')}")
ws = s.get("world_sizes", {})
if ws.get("total"):
lines.append(f"存档总大小: {ws['total']}")
if ws.get("disk_pct") is not None:
lines.append(f"磁盘使用: {ws['disk_used_gb']}GB / {ws['disk_total_gb']}GB ({ws['disk_pct']}%)")
load = s.get("system_load")
if load:
lines.append(f"系统负载: {load['load_1']} ({load['load_pct']}% @ {load['cpu_count']}核)")
return "\n".join(lines)
def fmt_players(data):
lines = []
events = data.get("recent_events", [])
if events:
lines.append("最近动态:")
for e in events:
icon = "加入" if e["type"] == "join" else "离开"
lines.append(f" {e['player']}: {icon}")
pt = data.get("playtime", [])
if pt:
lines.append("")
lines.append("在线时间排行:")
for i, p in enumerate(pt[:5], 1):
lines.append(f" {i}. {p['name']}: {p['formatted']}")
deaths = data.get("deaths", [])
if deaths:
lines.append("")
lines.append("死亡次数排行:")
for i, p in enumerate(deaths[:5], 1):
lines.append(f" {i}. {p['name']}: {p['deaths']}次")
ach = data.get("achievements", [])
if ach:
lines.append("")
lines.append("成就数排行:")
for i, p in enumerate(ach[:5], 1):
lines.append(f" {i}. {p['name']}: {p['achievements']}个")
return "\n".join(lines) if lines else "暂无玩家数据"
def fmt_world(data):
lines = []
gt = data.get("game_time", {})
if gt:
lines.append(f"世界时间: 第{gt.get('day', 1)}天 {gt.get('time', '00:00')} ({gt.get('phase', '')})")
es = data.get("entity_stats", {})
if es:
lines.append(f"实体总数: {es.get('total', 0)} (敌对:{es.get('hostile', 0)} 友好:{es.get('passive', 0)} 其他:{es.get('other', 0)})")
top = es.get("top", [])
if top:
lines.append("实体排行: " + ", ".join([f"{name}({cnt})" for name, cnt in top]))
net = data.get("network", {})
if net:
lines.append(f"网络流量: RX {net.get('rx', '?')} TX {net.get('tx', '?')}")
mods = data.get("active_mods", [])
if mods:
lines.append(f"活跃模组: {', '.join(mods)}")
return "\n".join(lines) if lines else "暂无世界数据"
def run_script(name):
script = os.path.join(SCRIPTS_DIR, name)
try:
result = subprocess.run(
[sys.executable, script],
capture_output=True, text=True, timeout=30
)
if result.returncode != 0:
return {"error": f"脚本错误: {result.stderr[:200]}"}
return json.loads(result.stdout)
except subprocess.TimeoutExpired:
return {"error": "查询超时"}
except Exception as e:
return {"error": str(e)}
COMMANDS = {
"status": ("mc_status.py", fmt_status, "服务器状态"),
"players": ("mc_players.py", fmt_players, "玩家信息"),
"world": ("mc_world.py", fmt_world, "世界详情"),
"clear": ("mc_clear_items.py", None, "清理掉落物"),
}
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Minecraft 信息查询工具")
parser.add_argument("command", choices=list(COMMANDS.keys()) + ["all"], help="查询类型")
parser.add_argument("--json", action="store_true", help="输出JSON而非纯文本")
args = parser.parse_args()
if args.command == "all":
results = {}
for key, (script, fmt, desc) in COMMANDS.items():
if key == "clear":
continue
data = run_script(script)
results[key] = {"data": data, "text": fmt(data) if fmt else str(data)}
if args.json:
print(json.dumps(results, ensure_ascii=False, indent=2))
else:
for key, val in results.items():
print(f"=== {COMMANDS[key][2]} ===")
print(val["text"])
print("")
else:
script, fmt, desc = COMMANDS[args.command]
data = run_script(script)
if args.json:
print(json.dumps(data, ensure_ascii=False, indent=2))
else:
if "error" in data:
print(f"查询失败: {data['error']}")
elif fmt:
print(fmt(data))
else:
print(str(data))

116
scripts/mc_rcon.py Normal file
View File

@ -0,0 +1,116 @@
#!/usr/bin/env python3
"""MC RCON 客户端"""
import socket
import struct
class RCONClient:
MAX_RETRIES = 3
RETRY_DELAY = 2
CONNECT_TIMEOUT = 5
SOCKET_TIMEOUT = 10
def __init__(self, host, port, password):
self.host = host
self.port = port
self.password = password
self.sock = None
self.packet_id = 1
self.authenticated = False
def connect(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(self.CONNECT_TIMEOUT)
self.sock.connect((self.host, self.port))
self.authenticate()
def authenticate(self):
packet = self._build_packet(self.packet_id, 3, self.password)
self.sock.sendall(packet)
response = self._read_response()
if not response or response.get("id") == -1:
raise Exception("RCON authentication failed")
self.authenticated = True
self.packet_id += 1
def is_connected(self):
if not self.sock or not self.authenticated:
return False
try:
self.sock.send(b"")
return True
except Exception:
return False
def send_command(self, command, max_retries=None):
if max_retries is None:
max_retries = self.MAX_RETRIES
last_error = None
for attempt in range(max_retries):
try:
if not self.is_connected():
self.close()
self.connect()
packet = self._build_packet(self.packet_id, 2, command)
self.sock.sendall(packet)
self.packet_id += 1
response = self._read_response()
return response
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError, OSError) as e:
last_error = e
self.close()
if attempt < max_retries - 1:
import time
time.sleep(self.RETRY_DELAY)
except Exception as e:
last_error = e
self.close()
if attempt < max_retries - 1:
import time
time.sleep(self.RETRY_DELAY)
return None
def _build_packet(self, packet_id, packet_type, body_str):
body = body_str.encode("utf-8") + b"\x00\x00"
payload = struct.pack("<ii", packet_id, packet_type) + body
length = struct.pack("<i", len(payload))
return length + payload
def _read_response(self):
try:
self.sock.settimeout(self.SOCKET_TIMEOUT)
length_data = self.sock.recv(4)
except (ConnectionResetError, ConnectionAbortedError, OSError, socket.timeout):
return None
if not length_data:
return None
length = struct.unpack("<i", length_data)[0]
if length <= 0:
return None
remaining = b""
while len(remaining) < length:
try:
chunk = self.sock.recv(length - len(remaining))
if not chunk:
break
remaining += chunk
except (ConnectionResetError, ConnectionAbortedError, OSError, socket.timeout):
break
if len(remaining) < 8:
return None
packet_id, packet_type = struct.unpack("<ii", remaining[:8])
body = remaining[8:-2] if len(remaining) > 8 else b""
return {"id": packet_id, "type": packet_type, "body": body.decode("utf-8", errors="replace")}
def close(self):
if self.sock:
self.sock.close()
self.sock = None
self.authenticated = False
def __enter__(self):
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()

191
scripts/mc_status.py Normal file
View File

@ -0,0 +1,191 @@
#!/usr/bin/env python3
"""获取Minecraft服务器运行状态(JSON输出)"""
import sys, os, subprocess, re
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mc_config import RCON_HOST, RCON_PORT, RCON_PASSWORD, MC_SERVER_DIR
from mc_rcon import RCONClient
def get_mc_pid():
try:
result = subprocess.run(
["pgrep", "-f", "java.*(minecraft|forge|neoforge)"],
capture_output=True, text=True, timeout=5
)
if result.returncode == 0 and result.stdout.strip():
return int(result.stdout.strip().split()[0])
except Exception:
pass
return None
def get_memory(pid):
if not pid:
return None
try:
with open(f"/proc/{pid}/status", "r") as f:
for line in f:
if line.startswith("VmRSS:"):
kb = int(line.split()[1])
return {"mb": round(kb/1024), "gb": round(kb/1024/1024, 1)}
except Exception:
pass
return None
def get_cpu(pid):
if not pid:
return None
try:
with open(f"/proc/{pid}/stat", "r") as f:
stat = f.read().split()
utime, stime = int(stat[13]), int(stat[14])
clk_tck = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
starttime = int(stat[21])
with open("/proc/uptime", "r") as f:
uptime = float(f.read().split()[0])
proc_uptime = uptime - (starttime / clk_tck)
if proc_uptime > 0:
return round(((utime + stime) / clk_tck) / proc_uptime * 100, 1)
except Exception:
pass
return None
def get_uptime(pid):
if not pid:
return None
try:
with open(f"/proc/{pid}/stat", "r") as f:
stat = f.read().split()
clk_tck = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
starttime = int(stat[21])
with open("/proc/uptime", "r") as f:
uptime = float(f.read().split()[0])
sec = uptime - (starttime / clk_tck)
d, h, m = int(sec//86400), int((sec%86400)//3600), int((sec%3600)//60)
parts = []
if d: parts.append(f"{d}天")
if h: parts.append(f"{h}小时")
parts.append(f"{m}分钟")
return "".join(parts)
except Exception:
return None
def get_tps():
try:
with RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD) as client:
client.send_command("spark tps")
import time
time.sleep(3)
log_path = os.path.join(MC_SERVER_DIR, "logs", "latest.log")
if not os.path.exists(log_path):
return None
with open(log_path, "r", errors="ignore") as f:
lines = f.readlines()
for line in reversed(lines[-500:]):
if "tps" not in line.lower():
continue
m = re.search(r'\*?([\d.]+),\s*\*?([\d.]+),\s*\*?([\d.]+),\s*\*?([\d.]+),\s*\*?([\d.]+)', line)
if m:
return {
"tps_1m": float(m.group(3)),
"raw": f"5s:{m.group(1)} 10s:{m.group(2)} 1m:{m.group(3)} 5m:{m.group(4)} 15m:{m.group(5)}"
}
except Exception:
pass
return None
def get_version():
try:
for vendor, name in [("neoforged", "NeoForge"), ("minecraftforge", "Forge")]:
vdir = os.path.join(MC_SERVER_DIR, "libraries", "net", vendor, name.lower())
if os.path.exists(vdir):
vers = [d for d in os.listdir(vdir) if os.path.isdir(os.path.join(vdir, d))]
if vers:
return f"{name} {vers[0]}"
except Exception:
pass
return "未知"
def get_world_sizes():
sizes = {}
for w in ["world", "world_nether", "world_the_end"]:
p = os.path.join(MC_SERVER_DIR, w)
if os.path.exists(p):
try:
r = subprocess.run(["du", "-sh", p], capture_output=True, text=True, timeout=10)
if r.returncode == 0:
sizes[w] = r.stdout.split()[0]
except Exception:
sizes[w] = "?"
try:
r = subprocess.run(["du", "-sh", MC_SERVER_DIR], capture_output=True, text=True, timeout=30)
if r.returncode == 0:
sizes["total"] = r.stdout.split()[0]
except Exception:
pass
try:
st = os.statvfs("/")
total = st.f_frsize * st.f_blocks
used = st.f_frsize * (st.f_blocks - st.f_bfree)
sizes["disk_total_gb"] = round(total/1024**3, 1)
sizes["disk_used_gb"] = round(used/1024**3, 1)
sizes["disk_pct"] = round((used/total)*100, 1)
except Exception:
pass
return sizes
def get_system_load():
try:
with open("/proc/loadavg", "r") as f:
parts = f.read().split()
cpu_count = os.cpu_count() or 1
return {
"load_1": float(parts[0]),
"load_5": float(parts[1]),
"load_15": float(parts[2]),
"cpu_count": cpu_count,
"load_pct": round((float(parts[0])/cpu_count)*100, 1)
}
except Exception:
return None
def get_player_list():
try:
with RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD) as client:
resp = client.send_command("list")
body = resp.get("body", "") if resp else ""
m = re.search(r'players online:\s*(.+)?$', body, re.IGNORECASE)
if m and m.group(1):
return [p.strip() for p in m.group(1).split(",") if p.strip()]
except Exception:
pass
return []
def get_player_count():
try:
with RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD) as client:
resp = client.send_command("list")
body = resp.get("body", "") if resp else ""
m = re.search(r'There are (\d+) of a max of (\d+) players', body)
if m:
return {"online": int(m.group(1)), "max": int(m.group(2))}
except Exception:
pass
return {"online": 0, "max": 0}
if __name__ == "__main__":
import json
pid = get_mc_pid()
result = {
"online": bool(pid),
"pid": pid,
"players": get_player_count(),
"player_list": get_player_list(),
"memory_mb": get_memory(pid),
"cpu_percent": get_cpu(pid),
"uptime": get_uptime(pid),
"tps": get_tps(),
"version": get_version(),
"world_sizes": get_world_sizes(),
"system_load": get_system_load(),
}
print(json.dumps(result, ensure_ascii=False, indent=2))

121
scripts/mc_world.py Normal file
View File

@ -0,0 +1,121 @@
#!/usr/bin/env python3
"""获取Minecraft世界详细信息(JSON输出)"""
import sys, os, subprocess, re
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mc_config import RCON_HOST, RCON_PORT, RCON_PASSWORD, MC_SERVER_DIR
from mc_rcon import RCONClient
def get_entity_stats():
try:
with RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD) as client:
resp = client.send_command("forge entity list")
body = resp.get("body", "") if resp else ""
if not body:
return {}
m = re.search(r'Total:\s*(\d+)', body)
total = int(m.group(1)) if m else 0
entity_map = {}
for line in body.strip().split("\n")[1:]:
line = line.strip()
em = re.match(r'(\d+):\s*(.+)', line)
if em:
entity_map[em.group(2).strip()] = int(em.group(1))
hostile_types = {"zombie", "skeleton", "creeper", "spider", "enderman", "witch", "slime", "phantom", "husk", "stray", "drowned", "pillager", "vindicator", "evoker", "ravager", "vex", "hoglin", "zoglin", "piglin", "piglin_brute", "wither_skeleton", "blaze", "ghast", "magma_cube", "guardian", "elder_guardian", "shulker", "endermite", "silverfish"}
passive_types = {"sheep", "cow", "pig", "chicken", "horse", "donkey", "mule", "rabbit", "fox", "bee", "cat", "wolf", "parrot", "turtle", "panda", "polar_bear", "llama", "mooshroom", "bat", "squid", "dolphin", "cod", "salmon", "tropical_fish", "pufferfish", "villager", "wandering_trader", "iron_golem", "snow_golem", "frog", "axolotl", "goat", "sniffer", "camel"}
hostile, passive, other = 0, 0, 0
for name, count in entity_map.items():
nl = name.replace("minecraft:", "").lower()
if nl in hostile_types:
hostile += count
elif nl in passive_types:
passive += count
else:
other += count
top = sorted(entity_map.items(), key=lambda x: -x[1])[:5]
return {"total": total, "hostile": hostile, "passive": passive, "other": other, "top": top}
except Exception:
return {}
def get_game_time():
try:
with RCONClient(RCON_HOST, RCON_PORT, RCON_PASSWORD) as client:
resp_day = client.send_command("time query day")
resp_daytime = client.send_command("time query daytime")
body_day = resp_day.get("body", "") if resp_day else ""
body_daytime = resp_daytime.get("body", "") if resp_daytime else ""
m_day = re.search(r'The time is (\d+)', body_day)
m_daytime = re.search(r'The time is (\d+)', body_daytime)
day = int(m_day.group(1)) + 1 if m_day else 1
daytime = int(m_daytime.group(1)) if m_daytime else 0
years = day // 365
months = (day % 365) // 30
days = (day % 365) % 30
hours = int(daytime / 1000)
minutes = int((daytime % 1000) * 60 / 1000)
if 0 <= daytime < 1000:
phase = "日出"
elif 1000 <= daytime < 12000:
phase = "白天"
elif 12000 <= daytime < 13000:
phase = "日落"
elif 13000 <= daytime < 23000:
phase = "夜晚"
else:
phase = "黎明"
return {"day": day, "years": years, "months": months, "days": days, "time": f"{hours:02d}:{minutes:02d}", "phase": phase}
except Exception:
return {}
def get_network_stats():
try:
with open("/proc/net/dev", "r") as f:
lines = f.readlines()
total_rx, total_tx = 0, 0
for line in lines[2:]:
parts = line.strip().split()
if len(parts) >= 10 and not parts[0].startswith("lo:"):
total_rx += int(parts[1])
total_tx += int(parts[9])
def fmt(b):
if b >= 1024**3:
return f"{b/1024**3:.1f}GB"
elif b >= 1024**2:
return f"{b/1024**2:.0f}MB"
elif b >= 1024:
return f"{b/1024:.0f}KB"
return f"{b}B"
return {"rx": fmt(total_rx), "tx": fmt(total_tx)}
except Exception:
return {}
def get_active_mods(limit=3):
log_path = os.path.join(MC_SERVER_DIR, "logs", "debug.log")
if not os.path.exists(log_path):
return []
skip = ["net.minecraft", "net.minecraftforge", "Server thread", "RCON", "Thread", "Generic", "mojang", "java", "sun.", "com.google", "org.apache", "io.netty", "org.mtr", "Mixin", "dev.architectury", "libIPN", "MinecraftMappings", "Yggdrasil", "DataFixer"]
counts = {}
try:
with open(log_path, "r", errors="ignore") as f:
lines = f.readlines()
for line in lines[-2000:]:
m = re.search(r'\[([^\/\]]+)/\]', line)
if not m:
continue
mod = m.group(1)
if any(mod.startswith(s) for s in skip) or len(mod) < 3:
continue
counts[mod] = counts.get(mod, 0) + 1
return [m[0] for m in sorted(counts.items(), key=lambda x: -x[1])[:limit]]
except Exception:
return []
if __name__ == "__main__":
import json
result = {
"entity_stats": get_entity_stats(),
"game_time": get_game_time(),
"network": get_network_stats(),
"active_mods": get_active_mods(),
}
print(json.dumps(result, ensure_ascii=False, indent=2))

206
scripts/qq_friend_action.py Normal file
View File

@ -0,0 +1,206 @@
#!/usr/bin/env python3
"""
QQ 好友管理动作 — 通过 NapCat (OneBot) HTTP API
支持操作:
- 删除好友(delete friend)
- 拉黑用户(delete friend + 从所有群踢出+禁止加群)
- 同意好友请求(approve friend request)
- 拒绝好友请求(reject friend request)
- 列出好友列表
权限规则:
- 删除好友、拉黑:只有 boss(YOUR_ADMIN_QQ)批准后才能执行
- 例外:检测到攻击性信息时,**自动删除+拉黑一条龙**,无需等待批准
- 同意/拒绝好友请求:必须问 boss
用法:
python3 qq_friend_action.py --delete 12345678 # 删除好友
python3 qq_friend_action.py --block 12345678 # 拉黑用户(删好友+从所有群踢出)
python3 qq_friend_action.py --block 12345678 --gid YOUR_GROUP_ID # 从指定群踢出+拒绝加群
python3 qq_friend_action.py --approve-friend <flag> # 同意好友请求
python3 qq_friend_action.py --reject-friend <flag> # 拒绝好友请求
python3 qq_friend_action.py --list-friends # 列出好友
"""
import json
import sys
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
def delete_friend(user_id: int) -> dict:
"""删除好友"""
resp = requests.post(f"{CQHTTP_URL}/delete_friend",
json={"user_id": user_id}, timeout=10)
return resp.json()
def get_group_list():
"""获取群列表(用于踢出)"""
resp = requests.post(f"{CQHTTP_URL}/get_group_list", json={}, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", [])
return []
def set_group_kick(group_id: int, user_id: int) -> dict:
"""从群踢出用户并拒绝加群"""
# 注意:NapCat 可能需要 uid 格式,try-except 兜底
resp = requests.post(f"{CQHTTP_URL}/set_group_kick", json={
"group_id": group_id,
"user_id": user_id,
"reject_add_request": True
}, timeout=10)
return resp.json()
def set_group_ban(group_id: int, user_id: int, duration: int = 2592000) -> dict:
"""禁言用户(30天)"""
resp = requests.post(f"{CQHTTP_URL}/set_group_ban", json={
"group_id": group_id,
"user_id": user_id,
"duration": duration
}, timeout=10)
return resp.json()
def handle_friend_request(flag: str, approve: bool, remark: str = "") -> dict:
"""处理好友请求"""
params = {
"flag": flag,
"approve": approve,
}
if approve and remark:
params["remark"] = remark
resp = requests.post(f"{CQHTTP_URL}/set_friend_add_request",
json=params, timeout=10)
return resp.json()
def get_friend_list():
"""获取好友列表"""
resp = requests.post(f"{CQHTTP_URL}/get_friend_list", json={}, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", [])
return []
def format_friends(friends: list) -> str:
if not friends:
return "暂无好友数据"
lines = [f"共 {len(friends)} 个好友\n"]
for f in friends:
uid = f.get("user_id", "?")
nickname = f.get("nickname", "?")
remark = f.get("remark", "")
remark_str = f"(备注:{remark})" if remark else ""
lines.append(f"• {nickname} ({uid}){remark_str}")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="QQ 好友管理操作")
parser.add_argument("--delete", type=int, metavar="QQ号", help="删除好友")
parser.add_argument("--block", type=int, metavar="QQ号", help="拉黑用户(删好友+踢出所有群)")
parser.add_argument("--gid", type=int, help="仅从指定群踢出(配合 --block)")
parser.add_argument("--approve-friend", type=str, help="同意好友请求(输入 flag)")
parser.add_argument("--reject-friend", type=str, help="拒绝好友请求(输入 flag)")
parser.add_argument("--remark", type=str, default="", help="好友备注(可选,仅同意时生效)")
parser.add_argument("--list-friends", action="store_true", help="列出好友列表")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
args = parser.parse_args()
try:
if args.list_friends:
friends = get_friend_list()
if args.json:
print(json.dumps(friends, ensure_ascii=False, indent=2))
else:
print(format_friends(friends))
return
if args.delete:
user_id = args.delete
result = delete_friend(user_id)
if result.get("status") == "ok":
print(f"✅ 已删除好友 {user_id}")
else:
print(f"❌ 删除好友失败: {result.get('message', '未知错误')}")
sys.exit(1)
return
if args.block:
user_id = args.block
results = []
# 1. 删除好友
r1 = delete_friend(user_id)
deleted = r1.get("status") == "ok"
if deleted:
results.append(f"✅ 已删除好友 {user_id}")
else:
results.append(f"❌ 删除好友失败: {r1.get('message', '未知错误')}")
# 2. 从群踢出
if args.gid:
groups = [{"group_id": args.gid, "group_name": ""}]
else:
groups = get_group_list()
kicked_groups = []
for g in groups:
gid = g.get("group_id")
if not gid:
continue
try:
r_kick = set_group_kick(gid, user_id)
if r_kick.get("status") == "ok":
gname = g.get("group_name", str(gid))
kicked_groups.append(gname)
except Exception:
pass
if kicked_groups:
results.append(f"✅ 已从 {len(kicked_groups)} 个群踢出:{', '.join(kicked_groups[:3])}")
else:
results.append("ℹ️ 未执行踢出操作(可能已不是好友/不在群中)")
print("\n".join(results))
return
if args.approve_friend:
result = handle_friend_request(args.approve_friend, approve=True,
remark=args.remark)
if result.get("status") == "ok":
print(f"✅ 已同意好友请求 {args.approve_friend}")
else:
print(f"❌ 同意失败: {result.get('message', '未知错误')}")
sys.exit(1)
return
if args.reject_friend:
result = handle_friend_request(args.reject_friend, approve=False)
if result.get("status") == "ok":
print(f"✅ 已拒绝好友请求 {args.reject_friend}")
else:
print(f"❌ 拒绝失败: {result.get('message', '未知错误')}")
sys.exit(1)
return
parser.print_help()
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()

103
scripts/qq_get_file.py Normal file
View File

@ -0,0 +1,103 @@
#!/usr/bin/env python3
"""
从 QQ 接收文件/图片 - 通过 NapCat (OneBot) HTTP API
自动下载到 qqagent 工作区的 files/ 目录,供 agent 直接使用。
用法:
python3 qq_get_file.py --file <file_id>
→ 下载到 files/ 目录
→ 输出保存路径、文件名、大小、类型
python3 qq_get_file.py --file <file_id> --info --json
→ 只查文件信息(不下)
"""
import json
import sys
import os
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
FILES_DIR = "YOUR_WORKSPACE_PATH/files"
def get_file(file_id: str) -> dict:
resp = requests.post(
f"{CQHTTP_URL}/get_file",
json={"file_id": file_id},
timeout=15
)
resp.raise_for_status()
data = resp.json()
if data.get("status") != "ok":
raise Exception(f"API error: {data}")
return data["data"]
def save_file(file_info: dict) -> tuple[str, str]:
"""下载文件到 FILES_DIR,返回 (本地路径, 文件名)"""
os.makedirs(FILES_DIR, exist_ok=True)
filename = file_info.get("file_name", "unknown")
local_path = os.path.join(FILES_DIR, filename)
if file_info.get("base64"):
import base64
with open(local_path, "wb") as f:
f.write(base64.b64decode(file_info["base64"]))
else:
url = file_info.get("url", "")
if not url:
raise Exception("No base64 or url available")
resp = requests.get(url, timeout=30)
resp.raise_for_status()
with open(local_path, "wb") as f:
f.write(resp.content)
return local_path, filename
def detect_type(filename: str) -> str:
ext = filename.lower().rsplit(".", 1)[-1] if "." in filename else ""
image_exts = {"jpg", "jpeg", "png", "gif", "bmp", "webp"}
return "图片" if ext in image_exts else "文件"
def main():
parser = argparse.ArgumentParser(description="从 QQ 接收文件")
parser.add_argument("--file", required=True, help="文件 ID(file_id)")
parser.add_argument("--output", help="保存目录(默认 files/)")
parser.add_argument("--info", action="store_true", help="只查文件信息,不下")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
args = parser.parse_args()
try:
file_info = get_file(args.file)
if args.info or args.json:
info = {
"file_id": args.file,
"name": file_info.get("file_name"),
"size": file_info.get("file_size"),
"has_base64": bool(file_info.get("base64")),
}
print(json.dumps(info, ensure_ascii=False, indent=2))
return
save_dir = args.output or FILES_DIR
local_path, filename = save_file(file_info)
ftype = detect_type(filename)
print(f"✅ 已保存到 {local_path}")
print(f" 文件名: {filename}")
print(f" 大小: {file_info.get('file_size')} bytes")
print(f" 类型: {ftype}")
except Exception as e:
print(f"❌ 错误: {e}")
sys.exit(1)
if __name__ == "__main__":
main()

88
scripts/qq_get_friends.py Normal file
View File

@ -0,0 +1,88 @@
#!/usr/bin/env python3
"""
获取好友列表 - 通过 NapCat (OneBot) HTTP API
用法:
python3 qq_get_friends.py # 输出格式化好友列表
python3 qq_get_friends.py --json # 输出 JSON(供 agent 使用)
python3 qq_get_friends.py --keyword 张三 # 搜索昵称/备注包含"张三"的好友
"""
import json
import sys
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
def get_friend_list():
"""获取好友列表,返回好友字典列表"""
url = f"{CQHTTP_URL}/get_friend_list"
resp = requests.post(url, json={}, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", [])
raise Exception(f"API 返回异常: {json.dumps(data, ensure_ascii=False)}")
def format_friends(friends, keyword=None):
"""格式化为可读文本"""
if keyword:
keyword = keyword.lower()
filtered = []
for f in friends:
uid = str(f.get("user_id", ""))
nickname = f.get("nickname", "")
remark = f.get("remark", "")
if (keyword in uid or keyword in nickname.lower()
or keyword in remark.lower()):
filtered.append(f)
friends = filtered
if not friends:
return "暂无好友数据"
lines = [f"共 {len(friends)} 个好友\n"]
for f in friends:
uid = f.get("user_id", "?")
nickname = f.get("nickname", "?")
remark = f.get("remark", "")
remark_str = f"(备注:{remark})" if remark else ""
lines.append(f"• {nickname} ({uid}){remark_str}")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="获取 QQ 好友列表")
parser.add_argument("--json", action="store_true", help="输出 JSON 格式")
parser.add_argument("--keyword", type=str, help="按昵称/备注关键词筛选")
args = parser.parse_args()
try:
friends = get_friend_list()
if args.keyword:
kw = args.keyword.lower()
friends = [
f for f in friends
if kw in str(f.get("user_id", ""))
or kw in f.get("nickname", "").lower()
or kw in f.get("remark", "").lower()
]
if args.json:
print(json.dumps(friends, ensure_ascii=False, indent=2))
else:
print(format_friends(friends, args.keyword))
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()

View File

@ -0,0 +1,219 @@
#!/usr/bin/env python3
"""
群文件查询与下载 — 通过 NapCat (OneBot) HTTP API
支持操作:
- 列出群根目录文件 + 文件夹结构
- 按关键字搜索文件
- 查看文件夹内文件
- 下载文件到本地
用法:
python3 qq_get_group_files.py --gid YOUR_GROUP_ID # 列出群根目录文件
python3 qq_get_group_files.py --gid YOUR_GROUP_ID --folder <folder_id> # 查看文件夹内文件
python3 qq_get_group_files.py --gid YOUR_GROUP_ID --keyword 编译 # 搜索文件名
python3 qq_get_group_files.py --gid YOUR_GROUP_ID --download <file_id> # 下载指定文件
python3 qq_get_group_files.py --gid YOUR_GROUP_ID --all --download <dir> # 批量下载所有文件(未实现)
python3 qq_get_group_files.py --gid YOUR_GROUP_ID --json # JSON 输出
"""
import json
import sys
import os
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
FILES_DIR = "YOUR_WORKSPACE_PATH/files"
DEFAULT_GID = YOUR_GROUP_ID # Liquid Studio 群
def get_root_files(group_id: int) -> dict:
"""获取群根目录文件列表"""
resp = requests.post(f"{CQHTTP_URL}/get_group_root_files",
json={"group_id": group_id}, timeout=15)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", {})
raise Exception(f"API 异常: {json.dumps(data, ensure_ascii=False)}")
def get_folder_files(group_id: int, folder_id: str) -> dict:
"""获取文件夹内文件列表"""
resp = requests.post(f"{CQHTTP_URL}/get_group_files_by_folder", json={
"group_id": group_id,
"folder_id": folder_id
}, timeout=15)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", {})
raise Exception(f"API 异常: {json.dumps(data, ensure_ascii=False)}")
def download_file(group_id: int, file_id: str, filename: str) -> str:
"""通过 NapCat /get_group_file_url 获取下载链接并保存文件
"""
os.makedirs(FILES_DIR, exist_ok=True)
# 1. 获取下载 URL
resp = requests.post(f"{CQHTTP_URL}/get_group_file_url", json={
"group_id": group_id,
"file_id": file_id
}, timeout=15)
data = resp.json()
if data.get("status") != "ok":
raise Exception(f"获取下载链接失败: {data.get('wording', data.get('message', '未知错误'))}")
dl_url = data["data"]["url"]
if not dl_url:
raise Exception("获取下载链接为空")
# 2. 构造完整 URL(文件名加在 ?fname= 后面)
if "?fname=" in dl_url:
full_url = dl_url + filename
else:
full_url = dl_url
# 3. 下载文件
save_path = os.path.join(FILES_DIR, filename)
try:
dl = requests.get(full_url, timeout=60, allow_redirects=True)
if dl.status_code != 200:
raise Exception(f"下载失败 (HTTP {dl.status_code})")
with open(save_path, "wb") as f:
f.write(dl.content)
size = len(dl.content)
return f"✅ 已下载: {save_path} ({size / 1024:.1f} KB)"
except requests.exceptions.ConnectionError:
raise Exception("下载连接失败(文件服务器不可达)")
except requests.exceptions.ReadTimeout:
raise Exception("下载超时")
def format_size(size_bytes: int) -> str:
"""友好显示文件大小"""
if size_bytes < 1024:
return f"{size_bytes} B"
elif size_bytes < 1024 * 1024:
return f"{size_bytes / 1024:.1f} KB"
else:
return f"{size_bytes / (1024 * 1024):.1f} MB"
def format_file_info(file: dict, indent: str = "") -> str:
"""格式化为可读文本"""
name = file.get("file_name", "?")
size = format_size(file.get("size", file.get("file_size", 0)))
uploader = file.get("uploader_name", "?")
downloads = file.get("download_times", 0)
fid = file.get("file_id", "?")
return (f"{indent}📄 {name}\n"
f"{indent} 大小: {size} | 上传者: {uploader} | 下载: {downloads} 次\n"
f"{indent} 文件ID: {fid}\n")
def format_folder_info(folder: dict, indent: str = "") -> str:
"""格式化为可读文本"""
name = folder.get("folder_name", "?")
fid = folder.get("folder_id", folder.get("folder", "?"))
creator = folder.get("creator_name", "?")
count = folder.get("total_file_count", "?")
return (f"{indent}📁 {name} ({count} 个文件)\n"
f"{indent} 文件夹ID: {fid}\n")
def main():
parser = argparse.ArgumentParser(description="群文件查询与下载")
parser.add_argument("--gid", type=int, default=DEFAULT_GID, help=f"群号 (默认: {DEFAULT_GID})")
parser.add_argument("--folder", type=str, help="查看指定文件夹内容(传入 folder_id)")
parser.add_argument("--keyword", type=str, help="搜索文件名(不区分大小写)")
parser.add_argument("--download", type=str, help="下载文件(传入 file_id)")
parser.add_argument("--name", type=str, help="下载时指定文件名(可选)")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
args = parser.parse_args()
try:
# 下载模式
if args.download:
file_id = args.download
filename = args.name or f"group_file_{file_id[:16]}"
if not args.json:
print(f"⏳ 正在下载文件...")
try:
result = download_file(args.gid, file_id, filename)
if args.json:
print(json.dumps({"file_id": file_id, "status": "ok", "path": result.split(": ")[-1] if ": " in result else result}))
else:
print(result)
except Exception as e:
if args.json:
print(json.dumps({"file_id": file_id, "status": "error", "error": str(e)}))
else:
print(f"❌ 下载失败: {e}")
sys.exit(1)
return
# 获取文件列表
if args.folder:
data = get_folder_files(args.gid, args.folder)
else:
data = get_root_files(args.gid)
files = data.get("files", [])
folders = data.get("folders", [])
if args.json:
output = {
"group_id": args.gid,
"files": files,
"folders": folders
}
# 搜索过滤
if args.keyword:
kw = args.keyword.lower()
output["files"] = [f for f in files if kw in f.get("file_name", "").lower()]
output["folders"] = [f for f in folders if kw in f.get("folder_name", "").lower()]
print(json.dumps(output, ensure_ascii=False, indent=2))
return
# 文本格式输出
# 搜索过滤
if args.keyword:
kw = args.keyword.lower()
matched_files = [f for f in files if kw in f.get("file_name", "").lower()]
matched_folders = [f for f in folders if kw in f.get("folder_name", "").lower()]
files = matched_files
folders = matched_folders
lines = [f"📁 群 {args.gid} 文件列表\n"]
if args.folder:
lines.append(f"(文件夹: {args.folder})\n")
if folders:
lines.append(f"📂 文件夹 ({len(folders)} 个)\n")
for fol in folders:
lines.append(format_folder_info(fol))
lines.append("")
if files:
lines.append(f"📄 文件 ({len(files)} 个)\n")
for f in files:
lines.append(format_file_info(f))
lines.append("")
if not folders and not files:
lines.append("暂无文件")
if args.keyword:
lines.append(f"(未找到包含「{args.keyword}」的文件)")
print("".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()

85
scripts/qq_get_groups.py Normal file
View File

@ -0,0 +1,85 @@
#!/usr/bin/env python3
"""
获取群聊列表 - 通过 NapCat (OneBot) HTTP API
用法:
python3 qq_get_groups.py # 输出格式化群列表
python3 qq_get_groups.py --json # 输出 JSON(供 agent 使用)
python3 qq_get_groups.py --keyword 我的世界 # 搜索群名包含"我的世界"的群
"""
import json
import sys
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
def get_group_list():
"""获取群聊列表,返回群字典列表"""
url = f"{CQHTTP_URL}/get_group_list"
resp = requests.post(url, json={}, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", [])
raise Exception(f"API 返回异常: {json.dumps(data, ensure_ascii=False)}")
def format_groups(groups, keyword=None):
"""格式化为可读文本"""
if keyword:
keyword = keyword.lower()
filtered = []
for g in groups:
gid = str(g.get("group_id", ""))
name = g.get("group_name", "")
if keyword in gid or keyword in name.lower():
filtered.append(g)
groups = filtered
if not groups:
return "暂无群聊数据" if not keyword else f"未找到包含「{keyword}」的群"
lines = [f"共 {len(groups)} 个群\n"]
for g in groups:
gid = g.get("group_id", "?")
name = g.get("group_name", "?")
member_count = g.get("member_count", "?")
max_member = g.get("max_member_count", "?")
lines.append(f"• {name} ({gid}) — {member_count}/{max_member} 人")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="获取 QQ 群聊列表")
parser.add_argument("--json", action="store_true", help="输出 JSON 格式")
parser.add_argument("--keyword", type=str, help="按群名关键词筛选")
args = parser.parse_args()
try:
groups = get_group_list()
if args.keyword:
kw = args.keyword.lower()
groups = [
g for g in groups
if kw in str(g.get("group_id", ""))
or kw in g.get("group_name", "").lower()
]
if args.json:
print(json.dumps(groups, ensure_ascii=False, indent=2))
else:
print(format_groups(groups, args.keyword))
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()

113
scripts/qq_get_history.py Normal file
View File

@ -0,0 +1,113 @@
#!/usr/bin/env python3
"""
获取历史消息 - 通过 NapCat (OneBot) HTTP API
用于 qq-agent 回忆之前和某个群/用户聊过什么。当用户提到当前 session
不清楚的内容时,调用此工具补充上下文。
用法:
# 获取群聊历史消息
python3 qq_get_history.py --gid 812704915 --num 10
# 获取私聊历史消息
python3 qq_get_history.py --uid YOUR_ADMIN_QQ --num 10
# JSON 格式输出(供 agent 解析)
python3 qq_get_history.py --gid 812704915 --num 5 --json
python3 qq_get_history.py --uid YOUR_ADMIN_QQ --num 20 --json
"""
import json
import sys
import argparse
import requests
from datetime import datetime
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
def get_group_msg_history(group_id: str, count: int) -> list[dict]:
"""获取群聊历史消息"""
resp = requests.post(
f"{CQHTTP_URL}/get_group_msg_history",
json={"group_id": int(group_id), "count": count},
timeout=10
)
resp.raise_for_status()
data = resp.json()
if data.get("status") != "ok":
raise Exception(f"API error: {data}")
return data.get("data", {}).get("messages", [])
def get_private_msg_history(user_id: str, count: int) -> list[dict]:
"""获取私聊历史消息"""
resp = requests.post(
f"{CQHTTP_URL}/get_friend_msg_history",
json={"user_id": int(user_id), "count": count},
timeout=10
)
resp.raise_for_status()
data = resp.json()
if data.get("status") != "ok":
raise Exception(f"API error: {data}")
return data.get("data", {}).get("messages", [])
def format_messages(messages: list[dict]) -> list[dict]:
"""提取消息中的关键字段,便于 agent 解析"""
formatted = []
for msg in messages:
sender = msg.get("sender", {})
ts = msg.get("time", 0)
time_str = datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
formatted.append({
"time": time_str,
"sender_id": sender.get("user_id"),
"sender_name": sender.get("nickname", ""),
"sender_card": sender.get("card", ""),
"message": msg.get("raw_message", ""),
"message_type": msg.get("message_type", ""),
})
return formatted
def print_readable(messages: list[dict]):
"""可读格式输出到终端"""
for m in messages:
name = m["sender_card"] or m["sender_name"]
print(f"[{m['time']}] {name}({m['sender_id']}): {m['message']}")
def main():
parser = argparse.ArgumentParser(description="获取 QQ 历史消息")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--gid", help="群号")
group.add_argument("--uid", help="QQ号(私聊历史)")
parser.add_argument("--num", type=int, default=10, help="拉取消息数量")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
args = parser.parse_args()
try:
if args.gid:
messages = get_group_msg_history(args.gid, args.num)
else:
messages = get_private_msg_history(args.uid, args.num)
formatted = format_messages(messages)
if args.json:
print(json.dumps(formatted, ensure_ascii=False, indent=2))
else:
print_readable(formatted)
except requests.exceptions.ConnectionError:
print("错误:无法连接到 NapCat,检查 CQHTTP_URL 是否正确")
sys.exit(1)
except Exception as e:
print(f"错误:{e}")
sys.exit(1)
if __name__ == "__main__":
main()

205
scripts/qq_group_action.py Normal file
View File

@ -0,0 +1,205 @@
#!/usr/bin/env python3
"""
QQ 群管理动作 — 通过 NapCat (OneBot) HTTP API
支持操作:
- 退群(leave group)
- 按群号加群(join group)
- 接受进群邀请(approve invite)
- 拒绝进群邀请(reject invite)
- 处理加群请求(approve/reject group request)
注意:
- 进群/退群操作 **必须** 先问老板,老板同意才能执行
- 紧急情况(如检测到攻击性信息)除外
- 退群操作不可逆,执行前二次确认
- 加群需要有人邀请或群主通过群链接/名片形式邀请,无法通过纯 API 直接加入
用法:
python3 qq_group_action.py --leave YOUR_BOT_QQ # 退出群
python3 qq_group_action.py --join YOUR_BOT_QQ # 按群号尝试加群(需有邀请)
python3 qq_group_action.py --approve-invite <flag> # 接受进群邀请
python3 qq_group_action.py --reject-invite <flag> # 拒绝进群邀请
python3 qq_group_action.py --list-groups # 列出当前群列表
"""
import json
import sys
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
def get_group_list():
"""获取当前已加入的群列表"""
resp = requests.post(f"{CQHTTP_URL}/get_group_list", json={}, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", [])
raise Exception(f"API 异常: {json.dumps(data, ensure_ascii=False)}")
def get_group_system_msgs():
"""获取系统消息,包括待处理的进群邀请"""
resp = requests.post(f"{CQHTTP_URL}/get_group_system_msg", json={}, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return data.get("data", {})
return {}
def leave_group(group_id: int) -> dict:
"""退出指定群聊。不可逆!"""
resp = requests.post(f"{CQHTTP_URL}/set_group_leave",
json={"group_id": group_id}, timeout=10)
return resp.json()
def join_group_via_invite(group_id: int) -> str:
"""
尝试通过群号加群。
策略:
1. 检查是否已在群里 → 直接返回
2. 检查是否有该群的待处理邀请 → 接受邀请
3. 没有邀请 → 告知需要通过邀请方式加群
"""
# 1. 检查是否已在群
groups = get_group_list()
for g in groups:
if g.get("group_id") == group_id:
return f"✅ 已经在群里: {g.get('group_name', '未知')} ({group_id})"
# 2. 检查待处理邀请
sys_msgs = get_group_system_msgs()
invites = sys_msgs.get("invited_requests", [])
for inv in invites:
if inv.get("group_id") == group_id:
# 接受邀请
flag = str(inv.get("request_id", ""))
resp = requests.post(f"{CQHTTP_URL}/set_group_add_request", json={
"flag": flag,
"sub_type": "invite",
"approve": True,
}, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return f"✅ 已接受群邀请,成功加入: {inv.get('group_name', '?')} ({group_id})"
else:
return f"❌ 接受邀请失败: {data.get('message', '未知错误')}"
# 3. 没有待处理的邀请
group_name = "(未知)"
try:
resp = requests.post(f"{CQHTTP_URL}/get_group_info", json={"group_id": group_id}, timeout=5)
d = resp.json()
if d.get("status") == "ok" and d.get("data"):
group_name = d["data"].get("group_name", group_name)
except:
pass
return (
f"⚠️ 无法自动加入群 {group_name} ({group_id})\n"
f" QQ 协议不提供通过 API 直接加群的能力。\n"
f" 需要有人邀请 bot 进群,bot 收到邀请后自动处理。\n"
f" 请让群管理用 QQ 客户端邀请 bot(QQ号: YOUR_ADMIN_QQ),\n"
f" 或发送群邀请链接/二维码给 bot,bot 会自动接受。"
)
def handle_group_request(flag: str, approve: bool, reason: str = "") -> dict:
"""处理加群请求 / 群邀请"""
params = {
"flag": flag,
"sub_type": "invite" if "invite" in flag else "add",
"approve": approve,
}
if not approve and reason:
params["reason"] = reason
resp = requests.post(f"{CQHTTP_URL}/set_group_add_request",
json=params, timeout=10)
return resp.json()
def format_group_info(groups: list) -> str:
"""格式化群列表为可读文本"""
if not groups:
return "当前没有加入任何群"
lines = [f"共 {len(groups)} 个群\n"]
for g in groups:
gid = g.get("group_id", "?")
name = g.get("group_name", "?")
mc = g.get("member_count", "?")
mm = g.get("max_member_count", "?")
lines.append(f"• {name} ({gid}) — {mc}/{mm} 人")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="QQ 群管理操作")
parser.add_argument("--leave", type=int, help="退出指定群(输入群号)")
parser.add_argument("--join", type=int, help="尝试加入指定群(输入群号)")
parser.add_argument("--approve-invite", type=str, help="接受进群邀请(输入 flag)")
parser.add_argument("--reject-invite", type=str, help="拒绝进群邀请(输入 flag)")
parser.add_argument("--reason", type=str, default="", help="拒绝理由(可选)")
parser.add_argument("--list-groups", action="store_true", help="列出当前加入的群")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
args = parser.parse_args()
try:
if args.list_groups:
groups = get_group_list()
if args.json:
print(json.dumps(groups, ensure_ascii=False, indent=2))
else:
print(format_group_info(groups))
return
if args.leave:
result = leave_group(args.leave)
if result.get("status") == "ok":
print(f"✅ 已退出群 {args.leave}")
else:
print(f"❌ 退群失败: {result.get('message', '未知错误')}")
sys.exit(1)
return
if args.join:
result = join_group_via_invite(args.join)
success = result.startswith("✅")
print(result)
sys.exit(0 if success else 1)
return
if args.approve_invite:
result = handle_group_request(args.approve_invite, approve=True)
if result.get("status") == "ok":
print(f"✅ 已接受群邀请 {args.approve_invite}")
else:
print(f"❌ 接受邀请失败: {result.get('message', '未知错误')}")
sys.exit(1)
return
if args.reject_invite:
result = handle_group_request(args.reject_invite, approve=False,
reason=args.reason)
if result.get("status") == "ok":
print(f"✅ 已拒绝群邀请 {args.reject_invite}")
else:
print(f"❌ 拒绝邀请失败: {result.get('message', '未知错误')}")
sys.exit(1)
return
parser.print_help()
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()

591
scripts/qq_group_manage.py Normal file
View File

@ -0,0 +1,591 @@
#!/usr/bin/env python3
"""
QQ 群综合管理 — 通过 NapCat (OneBot v11) HTTP API
支持操作分类:
[信息查询] group-list | group-info | member-list | member-info | at-all-remain
[成员管理] set-card | set-admin | set-title | kick | ban | unban
[群设置] rename | mute-all | set-portrait
[文件操作] list-files | folder-create | file-url
[消息操作] msg-history | recall | pin-msg
[系统操作] leave | pending-requests
用法:
python3 qq_group_manage.py group-list
python3 qq_group_manage.py group-info --gid 123456
python3 qq_group_manage.py member-list --gid 123456
python3 qq_group_manage.py member-info --gid 123456 --uid 789
python3 qq_group_manage.py set-card --gid 123456 --uid 789 --card "新昵称"
python3 qq_group_manage.py kick --gid 123456 --uid 789
python3 qq_group_manage.py ban --gid 123456 --uid 789 --minutes 10
python3 qq_group_manage.py unban --gid 123456 --uid 789
python3 qq_group_manage.py at-all-remain --gid 123456
python3 qq_group_manage.py msg-history --gid 123456 --count 10
python3 qq_group_manage.py recall --mid 123456
python3 qq_group_manage.py pin-msg --mid 123456
python3 qq_group_manage.py list-files --gid 123456
python3 qq_group_manage.py list-files --gid 123456 --folder_id xxx
python3 qq_group_manage.py file-url --gid 123456 --file_id xxx
python3 qq_group_manage.py rename --gid 123456 --name "新群名"
python3 qq_group_manage.py mute-all --gid 123456 --enable true
python3 qq_group_manage.py leave --gid 123456
python3 qq_group_manage.py pending-requests
输出: 所有操作可用 --json 参数输出 JSON 格式
"""
import json
import sys
import os
import argparse
import requests
from datetime import datetime
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
def api_call(endpoint: str, params: dict = None) -> dict:
"""调用 NapCat HTTP API"""
if params is None:
params = {}
resp = requests.post(f"{CQHTTP_URL}/{endpoint}", json=params, timeout=15)
data = resp.json()
return data
def api_ok(data: dict) -> bool:
"""检查 API 返回是否成功"""
return data.get("status") == "ok" and data.get("retcode") == 0
def handle_api_error(data: dict, endpoint: str) -> str:
"""格式化 API 错误信息"""
msg = data.get("message", "") or data.get("wording", "") or "未知错误"
return f"❌ {endpoint} 失败: {msg}"
# ==================== 信息查询 ====================
def cmd_group_list(json_output: bool):
"""列出所有已加入的群"""
data = api_call("get_group_list")
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_list")}
groups = data.get("data", [])
if json_output:
return {"ok": True, "data": groups}
lines = [f"📋 共 {len(groups)} 个群\n"]
for g in sorted(groups, key=lambda x: x.get("group_id", 0)):
gid = g.get("group_id", "?")
name = g.get("group_name", "?")
mc = g.get("member_count", "?")
mm = g.get("max_member_count", "?")
remark = g.get("group_remark", "")
remark_str = f" [{remark}]" if remark else ""
lines.append(f" [{gid}] {name}{remark_str} — {mc}/{mm} 人")
return {"ok": True, "text": "\n".join(lines)}
def cmd_group_info(gid: int, json_output: bool):
"""获取群详细信息"""
data = api_call("get_group_info", {"group_id": gid})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_info")}
info = data.get("data", {})
if json_output:
return {"ok": True, "data": info}
shutdown = "🟢 全员禁言中" if info.get("group_all_shut") else "🔊 全员可发言"
lines = [
f"📊 群信息",
f" 群名: {info.get('group_name', '?')}",
f" 群号: {info.get('group_id', '?')} ({info.get('group_remark', '')})",
f" 成员: {info.get('member_count', '?')}/{info.get('max_member_count', '?')}",
f" 状态: {shutdown}",
]
return {"ok": True, "text": "\n".join(lines)}
def cmd_member_list(gid: int, json_output: bool):
"""列出群成员"""
data = api_call("get_group_member_list", {"group_id": gid})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_member_list")}
members = data.get("data", [])
if json_output:
return {"ok": True, "data": members}
# 按角色排序: owner > admin > member
role_order = {"owner": 0, "admin": 1, "member": 2}
members.sort(key=lambda m: (role_order.get(m.get("role", "member"), 3), m.get("user_id", 0)))
lines = [f"👥 群成员 ({len(members)} 人)\n"]
role_icons = {"owner": "👑", "admin": "🛡️", "member": "👤"}
for m in members:
uid = m.get("user_id", "?")
name = m.get("card", "") or m.get("nickname", "")
role = m.get("role", "member")
icon = role_icons.get(role, "👤")
lines.append(f" {icon} {name} ({uid})")
return {"ok": True, "text": "\n".join(lines)}
def cmd_member_info(gid: int, uid: int, json_output: bool):
"""获取单个成员信息"""
data = api_call("get_group_member_info", {"group_id": gid, "user_id": uid})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_member_info")}
info = data.get("data", {})
if json_output:
return {"ok": True, "data": info}
role_names = {"owner": "群主", "admin": "管理员", "member": "成员"}
join_time = datetime.fromtimestamp(info.get("join_time", 0)).strftime("%Y-%m-%d %H:%M")
lines = [
f"👤 成员信息",
f" QQ: {info.get('user_id', '?')}",
f" 昵称: {info.get('nickname', '?')}",
f" 群名片: {info.get('card') or '无'}",
f" 角色: {role_names.get(info.get('role', 'member'), '?')}",
f" 头衔: {info.get('title') or '无'}",
f" 入群时间: {join_time}",
f" 禁言至: {'是' if info.get('shut_up_timestamp', 0) > 0 else '否'}",
]
return {"ok": True, "text": "\n".join(lines)}
def cmd_at_all_remain(gid: int, json_output: bool):
"""查询 @全体成员 剩余次数"""
data = api_call("get_group_at_all_remain", {"group_id": gid})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_at_all_remain")}
info = data.get("data", {})
if json_output:
return {"ok": True, "data": info}
can = "可" if info.get("can_at_all") else "不可"
lines = [
f"📢 @全体成员 剩余",
f" 状态: {can}使用",
f" 群剩余: {info.get('remain_at_all_count_for_group', '?')} 次",
f" 你剩余: {info.get('remain_at_all_count_for_uin', '?')} 次",
]
return {"ok": True, "text": "\n".join(lines)}
# ==================== 成员管理 ====================
def cmd_set_card(gid: int, uid: int, card: str, json_output: bool):
"""设置群名片(管理员或群主权限)"""
data = api_call("set_group_card", {"group_id": gid, "user_id": uid, "card": card})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_card")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已设置 {uid} 的群名片为: {card}"}
def cmd_set_admin(gid: int, uid: int, enable: bool, json_output: bool):
"""设置/取消管理员(仅群主有权限)"""
data = api_call("set_group_admin", {"group_id": gid, "user_id": uid, "enable": enable})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_admin")}
action = "设为管理员" if enable else "取消管理员"
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已{action} ({uid})"}
def cmd_set_title(gid: int, uid: int, title: str, json_output: bool):
"""设置群头衔(管理员或群主权限)"""
data = api_call("set_group_special_title",
{"group_id": gid, "user_id": uid, "special_title": title})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_special_title")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已设置 {uid} 的头衔为: {title}"}
def cmd_kick(gid: int, uid: int, reject_add: bool, json_output: bool):
"""踢出群成员(管理员或群主权限)"""
data = api_call("set_group_kick",
{"group_id": gid, "user_id": uid, "reject_add_request": reject_add})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_kick")}
add_str = "并拒绝加群申请" if reject_add else ""
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已踢出 {uid}{add_str}"}
def cmd_ban(gid: int, uid: int, minutes: int, json_output: bool):
"""禁言成员(管理员或群主权限)。minutes=0 解禁"""
data = api_call("set_group_ban",
{"group_id": gid, "user_id": uid, "duration": minutes * 60})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_ban")}
if minutes <= 0:
return {"ok": True, "text": f"✅ 已解除 {uid} 的禁言"}
mins_str = f"{minutes}分钟"
if minutes >= 1440:
mins_str = f"{minutes//1440}天{minutes%1440//60}小时"
elif minutes >= 60:
mins_str = f"{minutes//60}小时{minutes%60}分钟"
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已禁言 {uid} ({mins_str})"}
# ==================== 群设置 ====================
def cmd_rename(gid: int, name: str, json_output: bool):
"""修改群名称(管理员或群主权限)"""
data = api_call("set_group_name", {"group_id": gid, "group_name": name})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_name")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 群名已改为: {name}"}
def cmd_mute_all(gid: int, enable: bool, json_output: bool):
"""全员禁言(管理员或群主权限)"""
data = api_call("set_group_whole_ban", {"group_id": gid, "enable": enable})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_whole_ban")}
action = "全员禁言" if enable else "解除全员禁言"
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已{action}"}
def cmd_set_portrait(gid: int, file_path: str, json_output: bool):
"""设置群头像(管理员或群主权限)"""
data = api_call("set_group_portrait", {"group_id": gid, "file": file_path})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_portrait")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 群头像已更换"}
# ==================== 文件操作 ====================
def cmd_list_files(gid: int, folder_id: str, json_output: bool):
"""列出群文件"""
params = {"group_id": gid}
if folder_id:
params["folder_id"] = folder_id
data = api_call("get_group_files_by_folder", params)
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_files_by_folder")}
info = data.get("data", {})
files = info.get("files", [])
folders = info.get("folders", [])
if json_output:
return {"ok": True, "data": info}
lines = [f"📁 群文件 "]
if folder_id:
lines[0] += f"(文件夹: {folder_id})"
if not files and not folders:
lines.append(" (空)")
if folders:
lines.append(f"\n📂 文件夹 ({len(folders)}):")
for f in folders:
fid = f.get("folder_id", "?")
fname = f.get("folder_name", "?")
fcnt = f.get("total_file_count", 0)
lines.append(f" 📁 {fname} (id:{fid[:12]}..., {fcnt}文件)")
if files:
lines.append(f"\n📄 文件 ({len(files)}):")
for f in files:
fn = f.get("file_name", "?")
fs = f.get("file_size", 0)
fsize = f"{fs/1024/1024:.1f}MB" if fs > 1024*1024 else f"{fs/1024:.1f}KB"
fid = f.get("file_id", "?")
lines.append(f" 📄 {fn} ({fsize}) id:{fid[:12]}...")
return {"ok": True, "text": "\n".join(lines)}
def cmd_file_url(gid: int, file_id: str, json_output: bool):
"""获取文件下载链接"""
data = api_call("get_group_file_url", {"group_id": gid, "file_id": file_id})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_file_url")}
info = data.get("data", {})
if json_output:
return {"ok": True, "data": info}
url = info.get("url", "?")
return {"ok": True, "text": f"🔗 下载链接: {url}"}
def cmd_create_folder(gid: int, name: str, parent: str, json_output: bool):
"""创建群文件文件夹(管理员或群主权限)"""
data = api_call("create_group_file_folder",
{"group_id": gid, "name": name, "parent_folder_id": parent})
if not api_ok(data):
# 检查业务错误
result = data.get("data", {}).get("result", {})
msg = result.get("clientWording", "")
if msg:
return {"ok": False, "error": f"❌ {msg}"}
return {"ok": False, "error": handle_api_error(data, "create_group_file_folder")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 文件夹「{name}」已创建"}
# ==================== 消息操作 ====================
def cmd_msg_history(gid: int, count: int, json_output: bool):
"""获取群消息历史"""
data = api_call("get_group_msg_history",
{"group_id": gid, "count": min(count, 50)})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_msg_history")}
msgs = data.get("data", {}).get("messages", [])
if json_output:
return {"ok": True, "data": msgs}
if not msgs:
return {"ok": True, "text": "📭 最近没有消息"}
lines = [f"📜 最近 {len(msgs)} 条消息\n"]
for m in reversed(msgs):
uid = m.get("user_id", "?")
text = m.get("message", "")
ts = datetime.fromtimestamp(m.get("time", 0)).strftime("%H:%M")
# 简化消息内容
if isinstance(text, list):
parts = []
for item in text:
if isinstance(item, dict):
t = item.get("type", "")
d = item.get("data", {})
if t == "text":
parts.append(d.get("text", ""))
elif t == "image":
parts.append("[图片]")
elif t == "face":
parts.append("[表情]")
elif t == "at":
parts.append(f"@{d.get('qq', '?')}")
elif t == "reply":
parts.append(f"[回复:{d.get('id','?')}]")
else:
parts.append(f"[{t}]")
else:
parts.append(str(item))
text = "".join(parts)
lines.append(f" [{ts}] {uid}: {str(text)[:80]}")
return {"ok": True, "text": "\n".join(lines)}
def cmd_recall(mid: int, json_output: bool):
"""撤回消息"""
data = api_call("delete_msg", {"message_id": mid})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "delete_msg")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已撤回消息 {mid}"}
def cmd_pin_msg(mid: int, json_output: bool):
"""精华消息(需消息 + 是群主/管理员)"""
data = api_call("set_essence_msg", {"message_id": mid})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_essence_msg")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已设 {mid} 为精华消息"}
# ==================== 系统操作 ====================
def cmd_leave(gid: int, json_output: bool):
"""退出群聊(不可逆!必须先问老板)"""
data = api_call("set_group_leave", {"group_id": gid})
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "set_group_leave")}
if json_output:
return {"ok": True, "data": data}
return {"ok": True, "text": f"✅ 已退出群 {gid}"}
def cmd_pending_requests(json_output: bool):
"""查看待处理的加群请求/群邀请"""
data = api_call("get_group_system_msg")
if not api_ok(data):
return {"ok": False, "error": handle_api_error(data, "get_group_system_msg")}
info = data.get("data", {})
if json_output:
return {"ok": True, "data": info}
lines = ["📬 待处理请求"]
has_pending = False
for key, label in [("invited_requests", "群邀请"),
("join_requests", "加群请求")]:
items = info.get(key, [])
if items:
has_pending = True
lines.append(f"\n {label} ({len(items)}):")
for item in items:
gid = item.get("group_id", "?")
uid = item.get("user_id", "?")
nick = item.get("nickname", "?")
flag = item.get("request_id", "?")
lines.append(f" [{gid}] {nick}({uid}) — flag:{str(flag)[:20]}")
if not has_pending:
lines.append(" 暂无待处理的请求")
return {"ok": True, "text": "\n".join(lines)}
# ==================== 主入口 ====================
def main():
parser = argparse.ArgumentParser(
description="QQ 群综合管理工具")
parser.add_argument("command", nargs="?", help="操作命令")
parser.add_argument("--gid", type=int, help="群号")
parser.add_argument("--uid", type=int, help="用户 QQ")
parser.add_argument("--mid", type=int, help="消息 ID")
parser.add_argument("--card", type=str, help="群名片")
parser.add_argument("--title", type=str, help="群头衔")
parser.add_argument("--name", type=str, help="群名称")
parser.add_argument("--file_path", type=str, help="本地文件路径")
parser.add_argument("--file_id", type=str, help="文件 ID")
parser.add_argument("--folder_id", type=str, help="文件夹 ID", default="")
parser.add_argument("--parent", type=str, help="父文件夹 ID", default="/")
parser.add_argument("--count", type=int, default=10, help="消息数量")
parser.add_argument("--minutes", type=int, default=10, help="禁言分钟数(0=解禁)")
parser.add_argument("--enable", type=str, choices=["true", "false"], help="启用/禁用")
parser.add_argument("--reject_add", action="store_true", help="踢出时拒绝加群申请")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
args = parser.parse_args()
if not args.command:
parser.print_help()
sys.exit(1)
try:
cmd = args.command
jo = args.json
# 信息查询
if cmd == "group-list":
result = cmd_group_list(jo)
elif cmd == "group-info":
if not args.gid:
return fail("需要 --gid")
result = cmd_group_info(args.gid, jo)
elif cmd == "member-list":
if not args.gid:
return fail("需要 --gid")
result = cmd_member_list(args.gid, jo)
elif cmd == "member-info":
if not args.gid or not args.uid:
return fail("需要 --gid 和 --uid")
result = cmd_member_info(args.gid, args.uid, jo)
elif cmd == "at-all-remain":
if not args.gid:
return fail("需要 --gid")
result = cmd_at_all_remain(args.gid, jo)
# 成员管理
elif cmd == "set-card":
if not args.gid or not args.uid or args.card is None:
return fail("需要 --gid, --uid, --card")
result = cmd_set_card(args.gid, args.uid, args.card, jo)
elif cmd == "set-admin":
if not args.gid or not args.uid or args.enable is None:
return fail("需要 --gid, --uid, --enable(true/false)")
result = cmd_set_admin(args.gid, args.uid, args.enable == "true", jo)
elif cmd == "set-title":
if not args.gid or not args.uid or args.title is None:
return fail("需要 --gid, --uid, --title")
result = cmd_set_title(args.gid, args.uid, args.title, jo)
elif cmd == "kick":
if not args.gid or not args.uid:
return fail("需要 --gid 和 --uid")
result = cmd_kick(args.gid, args.uid, args.reject_add, jo)
elif cmd in ("ban", "unban"):
if not args.gid or not args.uid:
return fail("需要 --gid 和 --uid")
mins = 0 if cmd == "unban" else args.minutes
result = cmd_ban(args.gid, args.uid, mins, jo)
# 群设置
elif cmd == "rename":
if not args.gid or not args.name:
return fail("需要 --gid 和 --name")
result = cmd_rename(args.gid, args.name, jo)
elif cmd == "mute-all":
if not args.gid or args.enable is None:
return fail("需要 --gid 和 --enable(true/false)")
result = cmd_mute_all(args.gid, args.enable == "true", jo)
elif cmd == "set-portrait":
if not args.gid or not args.file_path:
return fail("需要 --gid 和 --file_path")
result = cmd_set_portrait(args.gid, args.file_path, jo)
# 文件操作
elif cmd == "list-files":
if not args.gid:
return fail("需要 --gid")
result = cmd_list_files(args.gid, args.folder_id, jo)
elif cmd == "file-url":
if not args.gid or not args.file_id:
return fail("需要 --gid 和 --file_id")
result = cmd_file_url(args.gid, args.file_id, jo)
elif cmd == "folder-create":
if not args.gid or not args.name:
return fail("需要 --gid 和 --name")
result = cmd_create_folder(args.gid, args.name, args.parent, jo)
# 消息操作
elif cmd == "msg-history":
if not args.gid:
return fail("需要 --gid")
result = cmd_msg_history(args.gid, args.count, jo)
elif cmd == "recall":
if not args.mid:
return fail("需要 --mid(消息ID)")
result = cmd_recall(args.mid, jo)
elif cmd == "pin-msg":
if not args.mid:
return fail("需要 --mid(消息ID)")
result = cmd_pin_msg(args.mid, jo)
# 系统操作
elif cmd == "leave":
if not args.gid:
return fail("需要 --gid")
result = cmd_leave(args.gid, jo)
elif cmd == "pending-requests":
result = cmd_pending_requests(jo)
else:
print(f"❌ 未知命令: {cmd}")
sys.exit(1)
# 输出
if jo:
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
if result.get("text"):
print(result["text"])
elif result.get("ok"):
print("✅ 完成")
else:
print(result.get("error", "❌ 操作失败"))
sys.exit(0 if result.get("ok") else 1)
except requests.exceptions.ConnectionError:
print(f"❌ 连接失败: 无法连接到 NapCat ({CQHTTP_URL})")
sys.exit(1)
except Exception as e:
print(f"❌ 异常: {e}")
sys.exit(1)
def fail(msg: str):
print(f"❌ {msg}")
sys.exit(1)
if __name__ == "__main__":
main()

234
scripts/qq_ocr_image.py Normal file
View File

@ -0,0 +1,234 @@
#!/usr/bin/env python3
"""
QQ 图片文字识别 (OCR) — 本地 Tesseract + NapCat 回退
使用策略:
1. 本地 Tesseract OCR(快速、可靠、无需 GUI)
2. 如果 Tesseract 不可用,回退到 NapCat 的 /ocr_image 端点
跨主机文件传递(NapCat 回退路径):
宿主机 YOUR_SHARED_DIR/ → NapCat 容器内 /app/files/
用法:
python3 qq_ocr_image.py /tmp/screenshot.png # 本地图片
python3 qq_ocr_image.py --url https://example.com/a.png # 远程图片
python3 qq_ocr_image.py /tmp/image.png --lang eng # 指定语言
语言参数:
chi_sim 简体中文(默认)
chi_tra 繁体中文
eng 英文
chi_sim+eng 中英文混合(推荐)
返回:
JSON: {status, data: {texts: [{text, confidence}], full_text}}
"""
import json
import sys
import os
import argparse
import shutil
import time
# === 本地 Tesseract OCR ===
try:
from PIL import Image
import pytesseract
TESSERACT_AVAILABLE = True
except ImportError:
TESSERACT_AVAILABLE = False
# === NapCat 回退 ===
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
SHARED_DIR = "YOUR_SHARED_DIR"
CONTAINER_FILES = "/app/files"
OCR_TIMEOUT = 60
def ocr_local(image_path: str, lang: str) -> dict:
"""
使用本地 Tesseract OCR
"""
if not TESSERACT_AVAILABLE:
return None # 走回退
if not os.path.exists(image_path):
return {
"status": "failed",
"message": f"文件不存在: {image_path}"
}
try:
img = Image.open(image_path)
full_text = pytesseract.image_to_string(img, lang=lang)
# 也拿到详细数据(带置信度)
data = pytesseract.image_to_data(img, lang=lang, output_type=pytesseract.Output.DICT)
texts = []
for i in range(len(data["text"])):
text = data["text"][i].strip()
conf = data["conf"][i]
if text and conf >= 0: # conf = -1 表示该区域无文本
texts.append({
"text": text,
"confidence": int(conf),
"bbox": {
"x": data["left"][i],
"y": data["top"][i],
"w": data["width"][i],
"h": data["height"][i]
}
})
if not texts:
# 没识别到文字,但 full_text 可能有内容
lines = [l.strip() for l in full_text.strip().split("\n") if l.strip()]
texts = [{"text": l, "confidence": 0} for l in lines]
return {
"status": "ok",
"source": "tesseract",
"data": {
"texts": texts,
"full_text": full_text.strip(),
"language": lang
}
}
except Exception as e:
return {
"status": "failed",
"source": "tesseract",
"message": f"Tesseract OCR 失败: {e}"
}
def ocr_napcat(image_source: str) -> dict:
"""
回退到 NapCat /ocr_image
"""
# 本地文件 → 共享目录桥接
if not (image_source.startswith("http://") or
image_source.startswith("https://") or
image_source.startswith("file://")):
abs_path = os.path.abspath(image_source)
if not os.path.exists(abs_path):
return {
"status": "failed",
"message": f"文件不存在: {abs_path}"
}
ts = int(time.time())
basename = os.path.basename(abs_path)
target_name = f"{ts}_ocr_{basename}"
target_path = os.path.join(SHARED_DIR, target_name)
try:
shutil.copy2(abs_path, target_path)
image_source = f"file://{CONTAINER_FILES}/{target_name}"
except Exception as e:
return {
"status": "failed",
"message": f"复制文件到共享目录失败: {e}"
}
# 调用 NapCat API
try:
resp = requests.post(
f"{CQHTTP_URL}/ocr_image",
json={"image": image_source},
timeout=OCR_TIMEOUT
)
resp_data = resp.json()
if resp_data.get("status") == "ok":
resp_data["source"] = "napcat"
return resp_data
except requests.exceptions.Timeout:
return {
"status": "failed",
"source": "napcat",
"message": f"NapCat OCR 超时({OCR_TIMEOUT}秒),在 Docker 无 GUI 环境中不可用"
}
except requests.exceptions.ConnectionError as e:
return {
"status": "failed",
"source": "napcat",
"message": f"无法连接到 NapCat ({CQHTTP_URL}): {e}"
}
def main():
parser = argparse.ArgumentParser(description="QQ 图片文字识别 (OCR)")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("image", type=str, nargs="?",
help="本地图片路径或 URL")
group.add_argument("--file", type=str,
help="本地图片文件路径")
group.add_argument("--url", type=str,
help="远程图片 URL")
parser.add_argument("--lang", type=str, default="chi_sim+eng",
help="识别语言(默认 chi_sim+eng,支持 eng / chi_sim / chi_tra / chi_sim+eng)")
parser.add_argument("--force-napcat", action="store_true",
help="强制使用 NapCat OCR(跳过本地 Tesseract)")
args = parser.parse_args()
if args.file:
source = args.file
elif args.url:
source = args.url
else:
source = args.image
if not source:
print(json.dumps({
"status": "failed",
"message": "请指定图片路径或 URL"
}))
sys.exit(1)
# === 执行 OCR ===
result = None
# 优先本地 Tesseract(除非 --force-napcat 或远程 URL)
if not args.force_napcat and TESSERACT_AVAILABLE:
if source.startswith("http://") or source.startswith("https://"):
# 下载远程图片到本地
try:
import requests as req
r = req.get(source, timeout=15)
ext = source.split(".")[-1].split("?")[0][:4] if "." in source else "png"
tmp_path = f"/tmp/_ocr_dl_{int(time.time())}.{ext}"
with open(tmp_path, "wb") as f:
f.write(r.content)
result = ocr_local(tmp_path, args.lang)
os.remove(tmp_path)
except Exception as e:
result = {
"status": "failed",
"source": "tesseract",
"message": f"下载远程图片失败: {e}"
}
else:
result = ocr_local(source, args.lang)
# 如果本地失败或不可用,走 NapCat 回退
if result is None or result.get("status") != "ok":
napcat_result = ocr_napcat(source)
if napcat_result and napcat_result.get("status") == "ok":
result = napcat_result
elif result and result.get("status") != "ok":
result["napcat_fallback"] = napcat_result.get("message") if napcat_result else None
if result is None:
result = {"status": "failed", "message": "所有 OCR 方式均失败"}
print(json.dumps(result, ensure_ascii=False, indent=2))
if result.get("status") != "ok":
sys.exit(1)
if __name__ == "__main__":
main()

199
scripts/qq_resolve_name.py Normal file
View 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()

176
scripts/qq_send_file.py Normal file
View File

@ -0,0 +1,176 @@
#!/usr/bin/env python3
"""
QQ 主动发文件脚本 - 后台发送
原理:
1. 立即返回(fire-and-forget),不阻塞 agent
2. 后台独立进程:复制 → 删除旧文件 → 通过 NapCat API 发文件
3. 文件发送即完成,不再额外通知 agent
用法:
python3 qq_send_file.py --private YOUR_ADMIN_QQ /path/to/file
python3 qq_send_file.py --group YOUR_GROUP_ID --name "报告.txt" /tmp/report.txt
python3 qq_send_file.py --group YOUR_GROUP_ID --image /tmp/screenshot.png
"""
import shutil
import os
import sys
import time
import json
import argparse
import subprocess
import requests
TARGET_DIR = "YOUR_SHARED_DIR"
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
HISTORY_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".send_history.json")
SCRIPT_PATH = os.path.abspath(__file__)
GATEWAY_URL = "http://127.0.0.1:18789"
GATEWAY_TOKEN = "YOUR_GATEWAY_TOKEN"
def log(msg):
sys.stderr.write(f"[qq_send_file] {msg}\n")
def load_history():
try:
if os.path.exists(HISTORY_FILE):
with open(HISTORY_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
log(f"读取历史失败: {e}")
return {"files": []}
def save_history(history):
try:
with open(HISTORY_FILE, "w", encoding="utf-8") as f:
json.dump(history, f, ensure_ascii=False, indent=2)
except Exception as e:
log(f"保存历史失败: {e}")
def cleanup_old_files(history):
for old_file in history.get("files", []):
if os.path.exists(old_file):
try:
os.remove(old_file)
log(f"已删除旧文件: {old_file}")
except Exception as e:
log(f"删除旧文件失败: {old_file} - {e}")
def _fmt_size(size):
for unit in ["B", "KB", "MB", "GB"]:
if size < 1024:
return f"{size:.1f}{unit}"
size /= 1024
return f"{size:.1f}TB"
def notify_agent_via_gateway(api, target_id, display_name, src_size, ok):
"""已禁用。文件发送完成后不再通知 agent,减少 Gateway 负载。"""
pass
def background_work(api, target_id, src_path, display_name, is_image):
"""后台进程执行的完整工作流程"""
try:
if not os.path.exists(src_path):
notify_agent_via_gateway(api, target_id, display_name, 0, False)
return 1
src_size = os.path.getsize(src_path)
# 1. 清理旧文件
history = load_history()
cleanup_old_files(history)
# 2. 复制到共享目录
ts = int(time.time())
basename = display_name or os.path.basename(src_path)
target_filename = f"{ts}_{basename}"
target_path = os.path.join(TARGET_DIR, target_filename)
shutil.copy2(src_path, target_path)
log(f"已复制: {target_path} ({src_size} bytes)")
# 3. 发送文件
file_uri = f"file:///app/files/{target_filename}"
if is_image:
msg = f"[CQ:image,file={file_uri}]"
else:
msg = f"[CQ:file,file={file_uri},title={display_name or basename}]"
if api == "group":
url = f"{CQHTTP_URL}/send_group_msg"
payload = {"group_id": int(target_id), "message": msg}
else:
url = f"{CQHTTP_URL}/send_private_msg"
payload = {"user_id": int(target_id), "message": msg}
resp = requests.post(url, json=payload, timeout=120)
data = resp.json()
if data.get("status") == "ok":
msg_id = data.get("data", {}).get("message_id", "unknown")
log(f"发送成功, message_id={msg_id}")
save_history({"files": [target_path]})
notify_agent_via_gateway(api, target_id, display_name, src_size, True)
else:
log(f"发送失败: {json.dumps(data, ensure_ascii=False)}")
notify_agent_via_gateway(api, target_id, display_name, src_size, False)
except Exception as e:
log(f"后台工作异常: {e}")
try:
notify_agent_via_gateway(api, target_id, display_name, 0, False)
except:
pass
return 0
def main():
parser = argparse.ArgumentParser(description="发送 QQ 文件(后台发送)")
target = parser.add_mutually_exclusive_group(required=True)
target.add_argument("--group", type=str, help="目标群号")
target.add_argument("--private", type=str, help="目标用户 QQ 号")
parser.add_argument("--name", type=str, help="显示的文件名")
parser.add_argument("--image", action="store_true", help="作为图片发送")
parser.add_argument("file_path", help="本地文件路径")
args = parser.parse_args()
if not os.path.exists(args.file_path):
print(f"❌ 文件不存在: {args.file_path}")
sys.exit(1)
api = "group" if args.group else "private"
target_id = args.group or args.private
display_name = args.name or os.path.basename(args.file_path)
subprocess.Popen(
[sys.executable, SCRIPT_PATH, "--bgworker",
api, target_id, args.file_path, display_name,
str(int(args.image))],
start_new_session=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdin=subprocess.DEVNULL,
)
print("✅ 已触发后台发送")
if __name__ == "__main__":
if "--bgworker" in sys.argv:
_, api, target_id, file_path, display_name, is_image_str = sys.argv[1:]
is_image = is_image_str == "1"
rc = background_work(api, target_id, file_path, display_name, is_image)
sys.exit(rc)
else:
main()

81
scripts/qq_send_like.py Normal file
View File

@ -0,0 +1,81 @@
#!/usr/bin/env python3
"""
QQ 好友点赞 — 通过 NapCat (OneBot) HTTP API
调用 NapCat 的 /send_like 端点给好友点赞。
用法:
python3 qq_send_like.py <user_id> # 给指定 QQ 号点赞
python3 qq_send_like.py <user_id> <times> # 点赞指定次数(最大20)
示例:
python3 qq_send_like.py YOUR_ADMIN_QQ # 给老板点个赞
python3 qq_send_like.py 12345678 10 # 给好友点10个赞
返回:
JSON: {"status": "ok", "retcode": 0, ...} 或错误信息
"""
import json
import sys
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
def send_like(user_id: int, times: int = 1) -> dict:
"""
给好友点赞
Args:
user_id: 目标 QQ 号
times: 点赞次数,1-20,默认为1
Returns:
API 响应 JSON
"""
resp = requests.post(
f"{CQHTTP_URL}/send_like",
json={"user_id": user_id, "times": times},
timeout=10
)
return resp.json()
def main():
parser = argparse.ArgumentParser(description="给 QQ 好友点赞")
parser.add_argument("user_id", type=int, help="目标 QQ 号")
parser.add_argument("times", type=int, nargs="?", default=1,
help="点赞次数 (1-20, 默认1)")
args = parser.parse_args()
if args.times < 1 or args.times > 20:
print(json.dumps({
"status": "failed",
"message": f"点赞次数必须在 1-20 之间, 当前: {args.times}"
}))
sys.exit(1)
try:
result = send_like(args.user_id, args.times)
print(json.dumps(result, ensure_ascii=False))
if result.get("status") != "ok":
sys.exit(1)
except requests.exceptions.Timeout:
print(json.dumps({
"status": "failed",
"message": "请求超时,NapCat 可能未运行"
}))
sys.exit(1)
except requests.exceptions.ConnectionError as e:
print(json.dumps({
"status": "failed",
"message": f"无法连接到 NapCat: {e}"
}))
sys.exit(1)
if __name__ == "__main__":
main()

111
scripts/qq_send_msg.py Normal file
View File

@ -0,0 +1,111 @@
#!/usr/bin/env python3
"""
QQ 主动发信脚本 - 通过 go-cqhttp HTTP API 发送消息到 QQ
用于 qq-agent 主动推送消息(非回复场景)
用法:
# 发送私聊消息
python3 qq_send_msg.py --private YOUR_ADMIN_QQ --message "服务器已重启完成"
# 发送群聊消息
python3 qq_send_msg.py --group YOUR_BOT_QQ --message "系统维护通知: ..."
# 从文件读取消息内容
python3 qq_send_msg.py --private YOUR_ADMIN_QQ --file /tmp/report.txt
# 快速发送(私聊+短时间内多条消息带上合并开关)
python3 qq_send_msg.py --private YOUR_ADMIN_QQ --message "你好" --auto_escape
注意:
- 管理员 QQ 号: YOUR_ADMIN_QQ
- 默认发送到管理员私聊
"""
import json
import sys
import argparse
import requests
import os
# go-cqhttp HTTP API 地址(与 qqrebot 配置文件一致)
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
# 默认接收用户(管理员)
DEFAULT_USER = "YOUR_ADMIN_QQ"
def send_private_msg(user_id, message, auto_escape=False):
"""发送私聊消息"""
url = f"{CQHTTP_URL}/send_private_msg"
payload = {
"user_id": int(user_id),
"message": message,
"auto_escape": auto_escape,
}
resp = requests.post(url, json=payload, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return True, data
return False, data
def send_group_msg(group_id, message, auto_escape=False):
"""发送群聊消息"""
url = f"{CQHTTP_URL}/send_group_msg"
payload = {
"group_id": int(group_id),
"message": message,
"auto_escape": auto_escape,
}
resp = requests.post(url, json=payload, timeout=10)
data = resp.json()
if data.get("status") == "ok":
return True, data
return False, data
def main():
parser = argparse.ArgumentParser(description="发送 QQ 消息")
target = parser.add_mutually_exclusive_group(required=False)
target.add_argument("--private", type=str, default=DEFAULT_USER, nargs="?",
const=DEFAULT_USER, help="接收用户 QQ 号 (默认: 管理员)")
target.add_argument("--group", type=str, help="目标群号")
content = parser.add_mutually_exclusive_group(required=True)
content.add_argument("--message", help="消息内容")
content.add_argument("--file", help="从文件读取消息内容")
parser.add_argument("--auto_escape", action="store_true",
help="是否转义 CQ 码 (默认不转义)")
args = parser.parse_args()
if args.file:
with open(args.file, "r") as f:
message = f.read()
else:
message = args.message
try:
if args.group:
ok, result = send_group_msg(args.group, message, args.auto_escape)
target_desc = f"群 {args.group}"
else:
ok, result = send_private_msg(args.private, message, args.auto_escape)
target_desc = f"用户 {args.private}"
if ok:
print(f"✅ 消息已发送到 {target_desc}")
else:
print(f"❌ 发送失败 ({target_desc}): {json.dumps(result, ensure_ascii=False)}")
sys.exit(1)
except requests.exceptions.ConnectionError:
print(f"❌ 连接失败: 无法连接到 go-cqhttp ({CQHTTP_URL})")
sys.exit(1)
except Exception as e:
print(f"❌ 发送异常: {e}")
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,89 @@
#!/usr/bin/env python3
"""
群文件上传 — 通过 NapCat 消息 API 上传文件到群
用法:
python3 qq_upload_group_file.py --gid YOUR_GROUP_ID --file /path/to/file.txt
python3 qq_upload_group_file.py --gid YOUR_GROUP_ID --file ./doc.md --name 文档.md
python3 qq_upload_group_file.py --gid YOUR_GROUP_ID --file ./image.png --json
"""
import json
import sys
import os
import base64
import argparse
import requests
CQHTTP_URL = "http://YOUR_NAP CAT_HOST:25570"
DEFAULT_GID = YOUR_GROUP_ID # Liquid Studio 群
def upload_file(group_id: int, file_path: str, file_name: str = None) -> dict:
"""上传文件到群,通过 base64:// 方式发送文件消息"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"文件不存在: {file_path}")
file_size = os.path.getsize(file_path)
file_name = file_name or os.path.basename(file_path)
# 读取并 base64 编码
with open(file_path, "rb") as f:
b64_content = base64.b64encode(f.read()).decode()
resp = requests.post(f"{CQHTTP_URL}/send_group_msg", json={
"group_id": group_id,
"message": [
{
"type": "file",
"data": {
"file": f"base64://{b64_content}",
"name": file_name
}
}
]
}, timeout=60)
data = resp.json()
if data.get("status") != "ok":
raise Exception(data.get("message", data.get("wording", "上传失败")))
return {
"message_id": data["data"]["message_id"],
"file_name": file_name,
"file_size": file_size,
"group_id": group_id
}
def main():
parser = argparse.ArgumentParser(description="群文件上传")
parser.add_argument("--gid", type=int, default=DEFAULT_GID, help=f"群号 (默认: {DEFAULT_GID})")
parser.add_argument("--file", type=str, required=True, help="要上传的文件路径")
parser.add_argument("--name", type=str, help="文件名(默认使用原文件名)")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
args = parser.parse_args()
try:
result = upload_file(args.gid, args.file, args.name)
if args.json:
print(json.dumps(result, ensure_ascii=False))
else:
size_str = f"{result['file_size'] / 1024:.1f} KB" if result['file_size'] >= 1024 else f"{result['file_size']} B"
print(f"✅ 已上传: {result['file_name']} ({size_str})")
print(f" message_id: {result['message_id']}")
except FileNotFoundError as e:
print(f"❌ {e}")
sys.exit(1)
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()

View File

@ -0,0 +1,259 @@
#!/usr/bin/env python3
"""
视频链接解析与下载工具 — 基于 you-get
场景:
AI 在聊天中检测到视频分享链接(B站等),调用此脚本下载到本地,
然后用 qq_upload_group_file.py 发送到对应群/私聊。
用法:
# 查看视频信息(不下载)
python3 qq_video_download.py --url "https://www.bilibili.com/video/BVxxxx" --info
# 下载视频(自动选择最佳可用画质)
python3 qq_video_download.py --url "https://www.bilibili.com/video/BVxxxx"
# 指定画质下载
python3 qq_video_download.py --url "..." --format dash-flv480-AVC
# JSON 输出(供 AI 解析)
python3 qq_video_download.py --url "..." --info --json
"""
import json
import sys
import os
import re
import subprocess
import argparse
import requests
from urllib.parse import urlparse
FILE_DIR = os.path.dirname(os.path.abspath(__file__))
DEFAULT_OUTPUT = os.path.join(FILE_DIR, "..", "files", "videos")
os.makedirs(DEFAULT_OUTPUT, exist_ok=True)
# B站短链接域名
BILIBILI_SHORT_DOMAINS = ["b23.tv", "bili22.cn", "bili33.cn"]
def normalize_url(url: str) -> str:
"""归一化视频链接:自动解析短链接到标准地址"""
url = url.strip().strip('"').strip("'")
# 短链接才需要解析,标准 URL 跳过
parsed = urlparse(url)
if parsed.netloc not in BILIBILI_SHORT_DOMAINS:
return url
try:
resp = requests.head(url, allow_redirects=True, timeout=10, headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
})
final_url = resp.url
if final_url and final_url != url:
return final_url
except Exception:
pass
return url
# 安全文件名:去掉不安全的字符
def safe_filename(name: str) -> str:
name = re.sub(r'[<>:"/\\|?*]', '_', name)
name = re.sub(r'\s+', ' ', name).strip()
return name or "video"
def run_you_get(args: list, timeout=120) -> dict:
"""运行 you-get 并返回结构化结果"""
cmd = ["you-get"] + args
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
except subprocess.TimeoutExpired:
return {"ok": False, "error": "下载超时,视频太大或网络太慢"}
stdout = proc.stdout or ""
stderr = proc.stderr or ""
exit_code = proc.returncode
if exit_code != 0:
# 有些 you-get 非零退出但实际成功了,检查输出
error_msg = stderr.strip() or stdout.strip() or f"you-get 退出码 {exit_code}"
return {"ok": False, "error": error_msg}
return {"ok": True, "exit_code": exit_code, "stdout": stdout, "stderr": stderr}
def parse_info_json(raw_json: str) -> dict:
"""解析 you-get --json 输出"""
try:
data = json.loads(raw_json)
except json.JSONDecodeError:
return None
info = {
"site": data.get("site", ""),
"title": data.get("title", ""),
"url": data.get("url", ""),
"streams": []
}
for fmt_id, stream in data.get("streams", {}).items():
info["streams"].append({
"id": fmt_id,
"container": stream.get("container", ""),
"quality": stream.get("quality", ""),
"size": stream.get("size", 0),
"size_human": f"{stream.get('size', 0) / 1024 / 1024:.1f} MB" if stream.get("size", 0) > 0 else "未知"
})
return info
def cmd_info(url: str, json_output: bool) -> dict:
"""获取视频信息"""
url = normalize_url(url)
result = run_you_get(["--json", url], timeout=30)
if not result["ok"]:
return result
info = parse_info_json(result["stdout"])
if not info:
return {"ok": False, "error": "无法解析视频信息"}
result["info"] = info
if not json_output:
# 人类可读格式
lines = [f"🎬 {info['title']}", f" 来源: {info['site']}", f" 链接: {info['url']}", ""]
for s in info["streams"]:
lines.append(f" [{s['id']}] {s['quality']} | {s['container']} | {s['size_human']}")
result["text"] = "\n".join(lines)
return result
def _pick_best_avc_stream(info: dict) -> str | None:
"""从可用流中挑选最佳的 AVC (H.264) 格式,确保 QQ 客户端可播放"""
avc_streams = [s for s in info.get("streams", []) if "AVC" in s.get("id", "")]
if not avc_streams:
return None
# 按质量降序(480 比 360 高),取第一个
avc_streams.sort(key=lambda s: s.get("id", ""), reverse=True)
return avc_streams[0]["id"]
def cmd_download(url: str, output_dir: str, filename: str, fmt: str,
no_merge: bool, no_caption: bool, json_output: bool) -> dict:
"""下载视频"""
url = normalize_url(url)
# 未指定格式时,自动选最佳的 AVC (H.264) 流(QQ 播放器不支持 AV1/HEVC)
if not fmt:
info_result = cmd_info(url, json_output=True)
if info_result.get("ok"):
best = _pick_best_avc_stream(info_result["info"])
if best:
fmt = best
else:
# 没有 AVC 流,用第一个
streams = info_result["info"].get("streams", [])
if streams:
fmt = streams[0]["id"]
args = ["--output-dir", output_dir, "--force"]
if no_merge:
args.append("--no-merge")
if no_caption:
args.append("--no-caption")
if filename:
args.extend(["--output-filename", filename])
if fmt:
args.extend(["--format", fmt])
args.append(url)
result = run_you_get(args, timeout=300) # 5 分钟超时
if not result["ok"]:
return result
# 解析下载后的文件
stderr = result.get("stderr", "")
stdout = result.get("stdout", "")
# you-get 会在 stdout/stderr 输出 Merged into xxx.mp4
merged_match = re.search(r'Merged into (.+\.mp4)', stderr or stdout)
if merged_match:
final_path = os.path.join(output_dir, merged_match.group(1))
else:
# 查找输出目录中最新添加的 mp4
mp4_files = sorted(
[f for f in os.listdir(output_dir) if f.endswith(".mp4")],
key=lambda f: os.path.getmtime(os.path.join(output_dir, f)),
reverse=True
)
final_path = os.path.join(output_dir, mp4_files[0]) if mp4_files else ""
file_size = os.path.getsize(final_path) if final_path and os.path.exists(final_path) else 0
result["file"] = {
"path": final_path,
"filename": os.path.basename(final_path) if final_path else "",
"size": file_size,
"size_human": f"{file_size / 1024 / 1024:.1f} MB" if file_size > 0 else "未知"
}
if not json_output:
lines = [f"✅ 下载完成: {result['file']['filename']}"]
lines.append(f" 大小: {result['file']['size_human']}")
lines.append(f" 路径: {result['file']['path']}")
result["text"] = "\n".join(lines)
return result
def main():
parser = argparse.ArgumentParser(description="视频链接解析与下载")
parser.add_argument("--url", type=str, required=True, help="视频分享链接")
parser.add_argument("--info", action="store_true", help="仅查看信息,不下载")
parser.add_argument("--output", type=str, default=DEFAULT_OUTPUT, help="下载目录")
parser.add_argument("--name", type=str, help="输出文件名(不含扩展名)")
parser.add_argument("--format", type=str, help="画质格式 ID(如 dash-flv480-AVC)")
parser.add_argument("--json", action="store_true", help="JSON 格式输出")
parser.add_argument("--no-merge", action="store_true", help="不合并 DASH 流")
parser.add_argument("--no-caption", action="store_true", help="不下字幕/弹幕")
args = parser.parse_args()
try:
if args.info:
result = cmd_info(args.url, args.json)
else:
result = cmd_download(args.url, args.output, args.name,
args.format, args.no_merge, args.no_caption, args.json)
if args.json:
# JSON 输出中去掉 stdout/stderr 这种大字段
output = {k: v for k, v in result.items() if k not in ("stdout", "stderr", "text")}
print(json.dumps(output, ensure_ascii=False, indent=2))
else:
if result.get("text"):
print(result["text"])
elif not result.get("ok"):
print(f"❌ {result.get('error', '未知错误')}")
else:
print(result.get("text", "✅ 完成"))
sys.exit(0 if result.get("ok") else 1)
except Exception as e:
if args.json:
print(json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False))
else:
print(f"❌ 错误: {e}")
sys.exit(1)
if __name__ == "__main__":
main()