[Bug] Onedrive的驱动内部复制和移动无法作用于文件夹

请确认以下事项

• 我已确认阅读并同意 AGPL-3.0 第15条 。
  本程序不提供任何明示或暗示的担保，使用风险由您自行承担。

• 我已确认阅读并同意 AGPL-3.0 第16条 。
  无论何种情况，版权持有人或其他分发者均不对使用本程序所造成的任何损失承担责任。

• 我确认我的描述清晰，语法礼貌，能帮助开发者快速定位问题，并符合社区规则。

• 我已确认阅读了OpenList文档。

• 我已确认没有重复的问题或讨论。

• 我认为此问题必须由OpenList处理，而非第三方。

• 我已确认此功能尚未被实现。

• 我已确认此功能是合理的，且有普遍需求，并非我个人需要。

需求描述

我在移动文件夹时，发现目前同一个site下的子号移动文件夹需要调用openlist作为中转，这很浪费时间和流量，而用网页复制和移动会有文件大小和数目的限制

实现思路

依此需求就用ai写了一个加以实现，希望openlist以后在移动文件夹时添加是否在同一site下的选项。
以下是代码实现和需要的api权限：

    "Files.ReadWrite.All",
    "Sites.ReadWrite.All",
    "User.Read.All",
    "Sites.FullControl.All",
    "Directory.Read.All"

import requests
import time
import urllib.parse

# --- 配置信息 ---
TENANT_ID = "租户ID"
CLIENT_ID = "应用ID"
CLIENT_SECRET = "应用密钥"
SOURCE_USER_EMAIL = "源文件文件夹用户的邮箱"
TARGET_USER_EMAIL = "目标用户的邮箱"
START_FOLDER_NAME = "需要复制的文件夹名称"

GRAPH_BASE_URL = "https://graph.microsoft.com/v1.0"

# 全局计数器
total_files_copied = 0
total_folders_created = 0


def get_access_token():
    url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
    headers = {'Content-Type': 'application/x-www-form-urlencoded'}
    data = {
        'client_id': CLIENT_ID,
        'scope': 'https://graph.microsoft.com/.default',
        'client_secret': CLIENT_SECRET,
        'grant_type': 'client_credentials'
    }
    resp = requests.post(url, headers=headers, data=data)
    resp.raise_for_status()
    return resp.json()['access_token']


def get_drive_id(access_token, user_email):
    headers = {'Authorization': f'Bearer {access_token}'}
    url = f"{GRAPH_BASE_URL}/users/{user_email}/drive"
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    return resp.json()['id']


def get_item_id_by_path(access_token, drive_id, path):
    headers = {'Authorization': f'Bearer {access_token}'}
    # 对路径进行 URL 编码
    encoded_path = urllib.parse.quote(path)
    url = f"{GRAPH_BASE_URL}/drives/{drive_id}/root:/{encoded_path}"
    resp = requests.get(url, headers=headers)
    if resp.status_code == 404:
        return None
    resp.raise_for_status()
    return resp.json()['id']


def list_children(access_token, drive_id, item_id):
    """列出所有子项，处理分页"""
    headers = {'Authorization': f'Bearer {access_token}'}
    url = f"{GRAPH_BASE_URL}/drives/{drive_id}/items/{item_id}/children"
    items = []
    while url:
        resp = requests.get(url, headers=headers)
        if resp.status_code != 200:
            print(f"读取目录失败: {resp.text}")
            return []
        data = resp.json()
        items.extend(data['value'])
        url = data.get('@odata.nextLink')
    return items


def get_or_create_folder(access_token, drive_id, parent_id, folder_name):
    """
    在目标 parent_id 下寻找名为 folder_name 的文件夹。
    如果存在，返回其 ID；如果不存在，创建并返回 ID。
    """
    headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'}

    # 1. 尝试创建 (如果存在会报错，除非用 conflictBehavior)
    # 这里我们使用 fail，然后通过捕获错误或者先查询来处理，
    # 但更简单的做法是：先创建，设定 conflictBehavior 为 'fail'，如果返回 409 (Conflict)，则去查询 ID

    url = f"{GRAPH_BASE_URL}/drives/{drive_id}/items/{parent_id}/children"
    payload = {
        "name": folder_name,
        "folder": {},
        "@microsoft.graph.conflictBehavior": "fail"
    }

    resp = requests.post(url, headers=headers, json=payload)

    if resp.status_code == 201:
        # 创建成功
        print(f"   [新建文件夹] {folder_name}")
        return resp.json()['id']
    elif resp.status_code == 409:
        # 已存在，我们需要获取它的 ID
        # 我们可以通过 filter 查，或者遍历，这里用 search 比较麻烦，
        # 最稳妥是列出 parent 的子项找到它，或者直接构造路径访问（如果有 path）。
        # 由于我们只知道 ID 链，没法轻易构造 path。
        # 这里用 filter 查询：
        filter_url = f"{GRAPH_BASE_URL}/drives/{drive_id}/items/{parent_id}/children?$filter=name eq '{folder_name}'"
        search_resp = requests.get(filter_url, headers=headers)
        if search_resp.status_code == 200:
            val = search_resp.json().get('value')
            if val:
                # print(f"   [进入现有文件夹] {folder_name}")
                return val[0]['id']

    print(f"无法获取或创建文件夹 {folder_name}: {resp.status_code} {resp.text}")
    return None


def copy_file(access_token, source_drive_id, file_id, target_drive_id, target_parent_id, file_name):
    """复制单个文件"""
    url = f"{GRAPH_BASE_URL}/drives/{source_drive_id}/items/{file_id}/copy"
    headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'}
    payload = {
        "parentReference": {"driveId": target_drive_id, "id": target_parent_id},
        "name": file_name
    }
    resp = requests.post(url, headers=headers, json=payload)
    if resp.status_code == 202:
        return True
    else:
        print(f"   [文件复制失败] {file_name}: {resp.status_code}")
        return False


def recursive_copy(token, source_drive_id, source_item_id, target_drive_id, target_parent_id, current_path=""):
    """
    核心递归函数
    """
    global total_files_copied, total_folders_created

    # 1. 获取源文件夹下的所有东西
    items = list_children(token, source_drive_id, source_item_id)

    print(f"正在扫描: {current_path}/ (包含 {len(items)} 个项目)")

    for item in items:
        item_name = item['name']
        item_id = item['id']

        if 'folder' in item:
            # --- 处理文件夹 ---
            # 在目标端确保该文件夹存在
            new_target_folder_id = get_or_create_folder(token, target_drive_id, target_parent_id, item_name)
            if new_target_folder_id:
                total_folders_created += 1
                # 递归调用！进入下一层
                recursive_copy(token, source_drive_id, item_id, target_drive_id, new_target_folder_id,
                               current_path + "/" + item_name)

        elif 'file' in item:
            # --- 处理文件 ---
            # print(f"   -> 复制文件: {item_name}")
            success = copy_file(token, source_drive_id, item_id, target_drive_id, target_parent_id, item_name)
            if success:
                total_files_copied += 1
                print(f"已复制: {total_files_copied} | 文件夹: {total_folders_created}", end="\r")

                # 简单的流控，每50个文件稍微停顿一下
                if total_files_copied % 50 == 0:
                    time.sleep(1)


def main():
    try:
        print("1. 初始化...")
        token = get_access_token()
        s_drive = get_drive_id(token, SOURCE_USER_EMAIL)
        t_drive = get_drive_id(token, TARGET_USER_EMAIL)

        print(f"2. 定位源根目录 '{START_FOLDER_NAME}'...")
        s_root_item = get_item_id_by_path(token, s_drive, START_FOLDER_NAME)
        if not s_root_item:
            print("源文件夹未找到。")
            return

        print(f"3. 准备目标根目录 '{START_FOLDER_NAME}'...")
        # 获取目标 Drive 的根 ID
        t_root_resp = requests.get(f"{GRAPH_BASE_URL}/drives/{t_drive}/root",
                                   headers={'Authorization': f'Bearer {token}'})
        t_drive_root_id = t_root_resp.json()['id']

        # 在目标根目录下创建/获取起始文件夹
        t_start_folder_id = get_or_create_folder(token, t_drive, t_drive_root_id, START_FOLDER_NAME)

        print("4. 开始深度递归复制 (这可能需要较长时间)...")
        recursive_copy(token, s_drive, s_root_item, t_drive, t_start_folder_id, START_FOLDER_NAME)

        print(f"\n\n=== 任务完成 ===")
        print(f"共触发文件复制: {total_files_copied}")
        print(f"共处理子文件夹: {total_folders_created}")

    except Exception as e:
        print(f"\n错误: {str(e)}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    main()

附加信息

No response

The payload names no repository files or tests. Start by locating the OneDrive driver and its existing folder copy and move entry points, then compare their behavior with the supplied Microsoft Graph Python example. Done means same-site folder copies and moves no longer require OpenList as an intermediary and handle large folders without the stated web limits.