114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
#!/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_NAPCAT_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()
|