使用MSGraph SharePoint API Python时遇403权限错误求助

尝试使用MS Graph的patch方法将SharePoint文件移动到目标文件夹时，执行代码行result = await self.user_client.drives.by_drive_id(parent_drive_id).items.by_drive_item_id(src_item_id).patch(request_body)持续返回403权限错误：

msgraph.generated.models.o_data_errors.o_data_error.ODataError:
        APIError
        Code: 403
        message: None
        error: MainError(additional_data={}, code='accessDenied', details=None, inner_error=InnerError(additional_data={}, client_request_id='XXXX', date=DateTime(2024, 5, 23, 8, 5, 34, tzinfo=Timezone('UTC')), odata_type=None, request_id='XXXX'), message='Access denied', target=None)

已确认拥有对应SharePoint站点的读写移动权限，且相同认证方式下的文件下载功能正常，怀疑问题出在request_body设置或drive/item ID定义上，尝试多种ID组合仍无法解决。请问：

1. 可能的错误点在哪里？
2. 若目标文件夹ID有误，如何获取正确的ID？

async def _move_sharepoint_file(self, url, site_name, src_drive_path, file_name, dst_site_name, dst_drive_path, new_file_name=None):
    # Check permissions and verify authentication - ensure proper scopes are included in token
    
    # Get source site and its drive items
        sites = await self.user_client.sites.with_url(url).get()
        src_site_id = None
        dst_site_id = None
        
        for site in sites.value:
            print(site.name)
            if site.name == site_name:
                src_site_id = site.id
            if site.name == dst_site_name:
                dst_site_id = site.id
        
        # if src_site_id is None:
        #     raise ValueError(f"Source site with name '{site_name}' not found")
        # if dst_site_id is None:
        #     raise ValueError(f"Destination site with name '{dst_site_name}' not found")

        query_params = ItemsRequestBuilder.ItemsRequestBuilderGetQueryParameters(
            filter = True
        )

        request_configuration = ItemsRequestBuilder.ItemsRequestBuilderGetRequestConfiguration(
            query_parameters = query_params,
        )
        
        path_list = src_drive_path.replace("\\", "/").split('/')
        print(path_list)
        drive_items = await self.user_client.drives.by_drive_id(src_site_id).items.get(request_configuration=request_configuration)
        drive_items = drive_items.value
        
        partial_dir = ""
        print(f"path_list: {path_list}")
        for dir in path_list:
            if len(path_list) == 1 and path_list[0] == "":
                
                correct_drive_items = drive_items
                
                break
            else:
                partial_dir += f"/{dir}"
            print(f"partial_dir: {partial_dir}") if self.verbose else None
            [print(f"drive: {drive.name}") for drive in drive_items] if self.verbose else None
            print() if self.verbose else None
            for drive_item in drive_items:
                if drive_item.name == dir:
                    break
            else:
                raise ValueError(f"Drive with name '{partial_dir}' not found")
            
            drive_id = drive_item.id
            drive_items = await self.user_client.drives.by_drive_id(src_site_id).items.by_drive_item_id(drive_id).children.get()
            drive_items = drive_items.value
            
            correct_drive_items = []
            for drive_item in drive_items:
                print(f"drive_item.name: {drive_item.name}") if self.verbose else None
                parent_path = drive_item.parent_reference.path
                print(f"drive_item.parent_reference.path: {drive_item.parent_reference.path}") if self.verbose else None
                if parent_path.endswith(partial_dir):
                    correct_drive_items.append(drive_item)
                    print(f"YES") if self.verbose else None
        drives = correct_drive_items
        drives = [drive for drive in drives if drive.name in file_name]
        if len(drives) == 0:    
            raise ValueError(f"Drive with name '{file_name}' not found")
        
        src_item_id = drives[0].id
        parent_drive_id = drives[0].parent_reference.drive_id
        print(parent_drive_id)

        # Find the source item
        #src_item_id = await self.get_item_id(src_site_id, src_drive_path, file_name)

        # Get destination folder ID
        dst_drive_items = await self.user_client.drives.by_drive_id(dst_site_id).items.get(request_configuration=request_configuration)
        dst_folder_id = None
        for drive_item in dst_drive_items.value:
            if drive_item.name == dst_drive_path.split('/')[-1]:
                dst_folder_id = drive_item.id
                
                parent_dst_id = drive_item.parent_reference.drive_id
                print("Not none")
                break
        
        
        if src_item_id is None:
            raise ValueError(f"Item with name '{file_name}' in path '{src_drive_path}' not found")
        if dst_folder_id is None:
            raise ValueError(f"Destination folder '{dst_drive_path}' not found")

        # Get source drive ID
        src_drive_items = await self.user_client.drives.by_drive_id(src_site_id).items.get(request_configuration=request_configuration)
        src_drive_id = None
        for drive_item in src_drive_items.value:
            if drive_item.name == src_drive_path:
                src_drive_id = drive_item.id
                break
        
        if src_drive_id is None:
            raise ValueError(f"Source drive '{src_drive_path}' not found")

       
       
        request_body = DriveItem(
            parent_reference = ItemReference(
                site_id= dst_site_id,
                id = dst_folder_id,
            ),
            name ="new-item-name.pdf",
        )
        
        result = await self.user_client.drives.by_drive_id(parent_drive_id).items.by_drive_item_id(src_item_id).patch(request_body)
        
        if result.is_success:
            print("File moved successfully.")
            return result.content
        else:
            print(f"Failed to move file. Status code: {result.status_code}")
            return None

一、可能的错误点 #

1. 权限范围不足
   文件下载仅需Files.Read或Sites.Read范围，但移动文件需要更高权限。确保认证令牌包含以下至少一个范围：

   • Files.ReadWrite.All
   • Sites.ReadWrite.All
   • 目标站点专属的Sites.ReadWrite范围
     可通过解析令牌的scope字段确认权限是否达标。

2. Request Body参数缺失
   跨站点移动时，parent_reference需同时指定site_id、drive_id和文件夹id，当前代码缺少目标文件夹所在的drive_id。修改后的request_body示例：

   request_body = DriveItem(
       parent_reference = ItemReference(
           site_id= dst_site_id,
           drive_id= parent_dst_id,  # 目标文件夹所属Drive ID
           id = dst_folder_id,
       ),
       name = new_file_name if new_file_name else file_name,  # 避免硬编码文件名
   )
   

3. 文件夹ID获取逻辑缺陷
   获取目标文件夹ID时，代码仅匹配路径最后一级文件夹名，若存在同名文件夹会取到错误ID。需参照源文件路径的遍历逻辑，逐级验证完整路径的每一级文件夹。

4. 跨站点移动限制
   跨站点移动文件需满足：两个站点属于同一租户，且当前用户在目标站点拥有添加项目的权限（不仅是读写权限）。

二、正确获取目标文件夹ID的方法 #

1. 遍历完整路径获取
   实现递归或循环遍历目标路径的每一级文件夹，确保拿到对应完整路径的ID：

   async def get_folder_id(self, site_id, folder_path):
       path_list = folder_path.replace("\\", "/").split('/')
       current_items = await self.user_client.drives.by_drive_id(site_id).items.get()
       current_items = current_items.value
       target_item = None
       for dir_name in path_list:
           if not dir_name:
               continue
           target_item = None
           for item in current_items:
               if item.name == dir_name and item.folder is not None:
                   target_item = item
                   break
           if not target_item:
               raise ValueError(f"Folder '{dir_name}' not found in path")
           current_items = await self.user_client.drives.by_drive_id(site_id).items.by_drive_item_id(target_item.id).children.get()
           current_items = current_items.value
       return target_item.id, target_item.parent_reference.drive_id
   

   调用该函数可同时获取文件夹ID和所属Drive ID。

2. Graph Explorer验证
   在Graph Explorer中执行以下查询，手动获取目标文件夹信息：

   GET /sites/{dst_site_id}/drive/root:/{完整目标文件夹路径}?select=id,driveId
   

   返回结果中的id即为文件夹ID，driveId是对应的Drive ID。

内容的提问来源于stack exchange，提问作者soulsbornefan