#!/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 # 查看文件夹内文件 python3 qq_get_group_files.py --gid YOUR_GROUP_ID --keyword 编译 # 搜索文件名 python3 qq_get_group_files.py --gid YOUR_GROUP_ID --download # 下载指定文件 python3 qq_get_group_files.py --gid YOUR_GROUP_ID --all --download # 批量下载所有文件（未实现） 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()