创建一个文件集合-中文名称乱码 · Issue #2282 · labring/FastGPT 这个 Issue是Java代码。
下面是Python代码的写的FastGPT文件上传，帮助大家少踩坑。

1. 创建一个文件集合-中文名称乱码

1. FastGPT 的编码预期（来自 Issue #2282）

服务端期望接收 URL 编码（Percent-encoding） 的文件名，而不是 RFC 2231 或原始 UTF-8 字节。

2. 踩坑对比

2. FastGPT 上传文件Python代码直接使用

import requests
import os
import time
import json
from datetime import datetime
from urllib.parse import quote_plus  # 关键：模拟 Java 的 URLEncoder

# -------------------------- 基础配置 --------------------------
API_KEY = "your_api_key" # API_KEY
DATASET_ID = "your_dataset_id" #知识库 ID
API_URL = "https://cloud.fastgpt.cn/api/core/dataset/collection/create/localFile" # 上传路径
LOCAL_PDF_DIR = r"D:\download\pdf-book"  #本地上传的文件夹路径，改为你自己的
TRAINING_TYPE = "chunk"
RETRY_TIMES = 3
REQUEST_INTERVAL = 2
LOG_FILE = "pdf_import_log.txt"  #日志文件


# -------------------------- 核心函数 --------------------------
def write_log(content):
    log_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(f"[{log_time}] {content}\n")
    print(f"[{log_time}] {content}")


def import_single_pdf(file_path, filename):
    pure_filename = os.path.basename(filename)

    # 核心修复：对文件名进行 URL 编码（模拟 Java URLEncoder.encode）
    # 将 "中文.pdf" 转为 "%E4%B8%AD%E6%96%87.pdf"
    encoded_filename = quote_plus(pure_filename, encoding='utf-8')

    data_json = {
        "datasetId": DATASET_ID,
        "parentId": "", #文件夹ID,可以用listV2查询。
        "trainingType": TRAINING_TYPE,
        "chunkSize": 512,
        "chunkSplitter": "",
        "qaPrompt": "",
        "metadata": {}
    }

    for retry in range(RETRY_TIMES):
        try:
            # 关键：with 语句确保文件句柄在请求期间保持打开
            with open(file_path, "rb") as f:
                files = {
                    "file": (encoded_filename, f, "application/pdf"),
                    "data": (None, json.dumps(data_json), "application/json")
                }

                headers = {"Authorization": f"Bearer {API_KEY}"}

                response = requests.post(
                    API_URL,
                    headers=headers,
                    files=files,
                    timeout=60,
                    verify=False
                )

            res_json = response.json()
            # ✅ 关键修改：增加完整响应日志
            if response.status_code == 200 and res_json.get("code") == 200:
                write_log(f"✅ 成功：{pure_filename}（集合 ID：{res_json['data']['collectionId']}）")
                write_log(f"📄 完整响应：{json.dumps(res_json, ensure_ascii=False, indent=2)}")  # 新增行
                return True
            else:
                err_msg = res_json.get("message", str(res_json))
                write_log(f"❌ 失败：{pure_filename}（状态码：{response.status_code}，原因：{err_msg}）")
                write_log(f"📄 完整响应：{json.dumps(res_json, ensure_ascii=False, indent=2)}")  # 新增行
                return False

        except Exception as e:
            write_log(f"❌ 异常：{pure_filename}（错误：{str(e)[:150]}）")
            if retry < RETRY_TIMES - 1:
                write_log(f"🔄 重试第 {retry + 1} 次...")
                time.sleep(REQUEST_INTERVAL * 2)

    write_log(f"❌ 最终失败：{pure_filename}（已重试 {RETRY_TIMES} 次）")
    return False


def batch_import_pdfs():
    if not os.path.exists(LOCAL_PDF_DIR):
        write_log(f"⚠️  错误：文件夹 {LOCAL_PDF_DIR} 不存在，请检查路径！")
        return

    pdf_files = [f for f in os.listdir(LOCAL_PDF_DIR) if f.lower().endswith(".pdf")]
    total = len(pdf_files)
    if total == 0:
        write_log("⚠️  提示：文件夹内未找到 PDF 文件！")
        return

    write_log(f"📢 开始批量导入：共发现 {total} 个 PDF 文件")
    success_count = 0
    fail_count = 0

    for i, filename in enumerate(pdf_files, 1):
        file_path = os.path.join(LOCAL_PDF_DIR, filename)
        write_log(f"\n📤 正在处理第 {i}/{total} 个文件：{filename}")
        result = import_single_pdf(file_path, filename)
        if result:
            success_count += 1
        else:
            fail_count += 1
        if i < total:
            time.sleep(REQUEST_INTERVAL)

    write_log(f"\n📊 导入完成！总计：{total} 个，成功：{success_count} 个，失败：{fail_count} 个")
    write_log("=" * 50 + "\n")


if __name__ == "__main__":
    write_log("=" * 50 + " 开始新的批量导入任务 " + "=" * 50)
    batch_import_pdfs()