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