123pan网盘云盘插件

创建插件

打开扣子开发平台的工作空间-资源库

点击右上角的 +资源 选择 插件

插件名称和插件介绍可以任意填写

插件工具创建方式选择：云侧插件-在Coze IDE中创建

IDE运行时选择：Python3

接下来点击 【在IDE中创建工具】

点击 【创建工具】

随意填写即可.

上传文件代码

运行流程：

在工具中填写以下代码：

from runtime import Args
from typings.uploadFile.uploadFile import Input, Output

"""
上传文件到123网盘并创建分享链接的工具

每个文件需要导出一个名为 `handler` 的函数，作为工具的入口点。

参数:
args: 入口函数的参数
args.input - 输入参数，可以通过 args.input.xxx 获取测试输入值
args.logger - 日志实例，用于打印日志，由运行时注入

返回值:
函数的返回值数据，应与声明的输出参数匹配
"""

import requests
import hashlib
import os
import time
from typing import Dict, Any, Optional

class Pan123Client:
    """123网盘客户端类，用于文件上传和分享操作"""
    
    def __init__(self, client_id: str, client_secret: str, dir_id: str = "") -> None:
        """
        初始化123网盘客户端
        
        Args:
            client_id: 客户端ID
            client_secret: 客户端密钥
            dir_id: 目录ID，默认为根目录
        """
        self.client_id = client_id
        self.client_secret = client_secret
        self.dir_id = dir_id or "0"  # 默认根目录
        self.base_url = "https://open-api.123pan.com"
        
        # 根据提供的凭据获取访问令牌
        if client_id and client_secret:
            self._get_private_token()
        else:
            self._get_public_token()
            
        # 设置请求头
        self.headers = {
            "Platform": "open_platform",
            "Authorization": self.access_token
        }
    
    def _get_private_token(self) -> Dict[str, Any]:
        """使用私有凭据获取访问令牌"""
        try:
            response = requests.post(
                url=f"{self.base_url}/api/v1/access_token",
                headers={"Platform": "open_platform"},
                data={
                    "clientID": self.client_id,
                    "clientSecret": self.client_secret
                }
            )
            response.raise_for_status()
            data = response.json()
            self.access_token = data["data"]["accessToken"]
            return data["data"]
        except requests.exceptions.RequestException as e:
            raise Exception(f"获取私有令牌失败: {e}")
    
    def _get_public_token(self) -> Dict[str, Any]:
        """使用公共凭据获取访问令牌"""
        try:
            response = requests.get("公共密钥，请联系润雨（qrecyc）获取")
            response.raise_for_status()
            data = response.json()
            self.access_token = data["accessToken"]
            self.dir_id = str(data["dirID"])
            return data
        except requests.exceptions.RequestException as e:
            raise Exception(f"获取公共令牌失败: {e}")
    
    def create_file(self, filename: str, etag: str, size: int) -> Dict[str, Any]:
        """
        创建文件上传任务
        
        Args:
            filename: 文件名
            etag: 文件MD5值
            size: 文件大小（字节）
            
        Returns:
            创建文件任务的响应数据
        """
        body = {
            "parentFileID": self.dir_id,
            "fileName": filename,
            "etag": etag,
            "size": size,
            "type": 0,  # 0表示普通文件
        }
        
        try:
            response = requests.post(
                url=f"{self.base_url}/upload/v2/file/create",
                headers=self.headers,
                data=body
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"创建文件失败: {e}")
    
    def upload_slice(self, base_url: str, preupload_id: str, slice_size: int, 
                    slice_no: int, slice_md5: str, slice_data: bytes) -> Dict[str, Any]:
        """
        上传文件分片
        
        Args:
            base_url: 上传服务器地址
            preupload_id: 预上传ID
            slice_size: 分片大小
            slice_no: 分片序号（从1开始）
            slice_md5: 分片MD5值
            slice_data: 分片二进制数据
            
        Returns:
            上传分片的响应数据
        """
        files = {
            'slice': ('slice', slice_data)
        }
        data = {
            'preuploadID': preupload_id,
            'sliceNo': slice_no,
            'sliceMD5': slice_md5
        }
        
        try:
            response = requests.post(
                url=f"{base_url}/upload/v2/file/slice",
                headers=self.headers,
                files=files,
                data=data
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"上传分片失败: {e}")
    
    def complete_upload(self, preupload_id: str) -> Dict[str, Any]:
        """
        确认上传完成
        
        Args:
            preupload_id: 预上传ID
            
        Returns:
            包含文件ID的响应数据
        """
        data = {"preuploadID": preupload_id}
        
        try:
            response = requests.post(
                url=f"{self.base_url}/upload/v2/file/upload_complete",
                headers=self.headers,
                json=data
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"确认上传失败: {e}")
    
    def create_share(self, file_id_list: str, share_name: str = "") -> Dict[str, Any]:
        """
        创建文件分享链接
        
        Args:
            file_id_list: 文件ID列表，多个用逗号分隔
            share_name: 分享名称（可选）
            
        Returns:
            分享链接信息
        """
        data = {
            "fileIDList": file_id_list,
            "shareName": share_name,
            "shareExpire": 0,  # 0表示永久有效
            "sharePwd": "",    # 空字符串表示无提取码
            "shareAuto": 0     # 0表示关闭免登录
        }
        
        try:
            response = requests.post(
                url=f"{self.base_url}/api/v1/share/create",
                headers=self.headers,
                json=data
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"创建分享失败: {e}")
    
    def get_download_url(self, file_id: str) -> str:
        """
        获取文件下载链接
        
        Args:
            file_id: 文件ID
            
        Returns:
            文件下载链接
        """
        params = {"fileId": file_id}
        
        try:
            response = requests.get(
                url=f"{self.base_url}/api/v1/file/download_info",
                headers=self.headers,
                params=params
            )
            response.raise_for_status()
            data = response.json()
            return data['data']['downloadUrl']
        except requests.exceptions.RequestException as e:
            raise Exception(f"获取下载链接失败: {e}")

def handler(args: Args[Input]) -> Output:
    """
    文件上传处理函数
    
    从指定URL下载文件，上传到123网盘，并创建分享链接
    
    Args:
        args: 包含输入参数的对象
        
    Returns:
        包含文件信息的响应数据
    """
    try:
        # 初始化123网盘客户端
        client = Pan123Client(
            args.input.clientID,
            args.input.clientSecret,
            args.input.dirID
        )
        
        # 从URL下载文件
        args.logger.info(f"开始下载文件: {args.input.url}")
        response = requests.get(args.input.url, stream=True)
        if response.status_code != 200:
            raise Exception(f"文件下载失败，状态码: {response.status_code}")
        
        # 获取文件内容
        file_content = response.content
        file_size = len(file_content)
        
        # 生成文件名（添加时间戳避免重复）
        original_name = args.input.url.split('/')[-1]
        name, ext = os.path.splitext(original_name)
        timestamp = str(int(time.time()))
        file_name = f"{name}_{timestamp}{ext}"
        
        # 计算文件MD5值
        file_md5 = hashlib.md5(file_content).hexdigest()
        
        args.logger.info(f"文件信息 - 名称: {file_name}, 大小: {file_size} bytes, MD5: {file_md5}")
        
        # 创建文件上传任务
        upload_result = client.create_file(file_name, file_md5, file_size)
        args.logger.info(f"创建文件任务结果: {upload_result}")
        
        # 检查是否需要分片上传
        if upload_result["data"].get('preuploadID'):
            # 分片上传
            base_url = upload_result['data']['servers'][0]
            preupload_id = upload_result['data']['preuploadID']
            slice_size = upload_result['data']['sliceSize']
            
            # 计算分片数量
            slices = [
                file_content[i:i+slice_size] 
                for i in range(0, len(file_content), slice_size)
            ]
            
            args.logger.info(f"开始分片上传，共 {len(slices)} 个分片")
            
            # 上传每个分片
            for i, slice_data in enumerate(slices):
                slice_md5 = hashlib.md5(slice_data).hexdigest()
                slice_result = client.upload_slice(
                    base_url=base_url,
                    preupload_id=preupload_id,
                    slice_size=len(slice_data),
                    slice_no=i+1,
                    slice_md5=slice_md5,
                    slice_data=slice_data
                )
                args.logger.info(f"分片 {i+1}/{len(slices)} 上传完成")
            
            # 确认上传完成
            complete_result = client.complete_upload(preupload_id)
            args.logger.info(f"上传完成确认结果: {complete_result}")
            file_id = complete_result['data']['fileID']
        else:
            # 无需分片上传，直接获取文件ID
            file_id = upload_result["data"]['fileID']
        
        args.logger.info(f"文件上传成功，文件ID: {file_id}")
        
        # 创建分享链接
        share_result = client.create_share(file_id, share_name=file_name)
        args.logger.info(f"分享创建结果: {share_result}")
        
        # 构建分享链接
        share_key = share_result.get('data', {}).get('shareKey', '')
        share_url = f"https://www.123pan.com/s/{share_key}" if share_key else ""
        
        # 获取下载链接
        download_url = client.get_download_url(file_id)
        
        return {
            "fileName": file_name,
            "fileSize": file_size,
            "fileID": file_id,
            "shareUrl": share_url,
            "downUrl": download_url,
            "msg":"上传成功",
            "tips": "咨询VX：qrecyc"
        }
        
    except Exception as e:
        args.logger.error(f"文件上传失败: {str(e)}")
        return {
            "fileName": "",
            "fileSize": "",
            "fileID": "",
            "shareUrl": "",
            "downUrl": "",
            "msg":f"上传失败: {str(e)}",
            "tips": "咨询VX：qrecyc"
        }

修改输入参数

点击【元数据】，填写出以下内容

url：文件链接 （必选）

clientID：密钥id，不填使用公共密钥

clientSecret：密钥内容，不填使用公共密钥

dirID：存储目录ID

修改输出参数

downUrl：文件下载url（有效期较短）

fileID：文件ID

fileName：文件名称

fileSize：文件大小（字节）

msg：消息提示

shareUrl：文件的分享链接（长期有效）

tips：运行提示

运行测试

填入相关信息，点击 运行

注意需要填入你自己的clientID和clientSecret或联系润雨获取公共密钥

{
  "url": "https://youke1.picui.cn/s1/2025/07/15/6876348e8ac0a.png",
  "clientID": "",
  "clientSecret": "",
  "dirID":""
}

可以看到已经上传成功了

示例输出：

{"downUrl":"https://download-cdn.cjjd19.com/123-556/044aa8c0/1658012-0/044aa8c0c7acacca36165c6828491e2c/c-m90?v=5&t=1753109139&s=175310913916e208f7a64bfab49ac363a652813ceb&r=H2KTSF&bzc=1&bzs=1658012&bzp=0&bi=2150452067&filename=6876348e8ac0a_1753022736.png&cache_type=1&ndcp=1","fileID":22820788,"fileName":"6876348e8ac0a_1753022736.png","fileSize":2084511,"msg":"上传成功","shareUrl":"https://www.123pan.com/s/QJyA-F7H8","tips":"咨询VX：qrecyc"}

如果出现错误，请及时联系润雨（qrecyc）排查原因

发布插件

测试通过后，点击右上角的 【发布】按钮

插件已经配置完成，接下来可以使用了~