chatrebot_aireply_plug/scripts/qq_get_group_files.py

#!/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_NAPCAT_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()