一、系统要求

  • 操作系统： Ubuntu / Debian / Linux
  • Python 版本： Python 3.6+
  • 网络： 可访问 api.aliyundrive.com

二、安装必要软件和依赖

1. 更新系统并安装基础工具

sudo apt update && sudo apt upgrade -y
sudo apt install -y curl wget vim git jq
  • curl: 用于测试 API 请求。
  • jq: JSON 格式化查看响应内容（可选但推荐）。
  • python3 和 pip3: 运行脚本所需。

2. 安装 Python 依赖

sudo apt install -y python3 python3-pip

安装 requests 库（HTTP 请求库）：

pip3 install requests

✅ 验证是否安装成功：

python3 -c "import requests; print(requests.__version__)"

三、获取 Access Token

Access Token 是调用阿里云盘 API 的身份凭证。

方法：浏览器抓包获取（推荐）

步骤如下：

  1. 打开 阿里云盘网页版。
  2. 使用手机号 1382768456 登录（昵称：大爷）。
  3. 打开浏览器开发者工具（F12 → Network）。
  4. 刷新页面。
  5. 在请求列表中找到任意一个 XHR 请求，例如： 
    https://api.aliyundrive.com/v2/file/list
  6. 查看该请求的 Request Headers。
  7. 找到字段： 
    Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
  8. 复制 Bearer 后面的部分（即 eyJ...），这就是您的 ACCESS_TOKEN。

⚠️ 注意：

  • 不要复制 refresh_token。
  • Token 有效期约 2 小时，需定期更新。

四、获取 DRIVE_ID（存储空间 ID）

DRIVE_ID 是您默认网盘的唯一标识，必须正确设置。

使用 curl 命令获取：

curl -s -X POST "https://api.aliyundrive.com/v2/user/get" \
 -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
 -H "Content-Type: application/json" \
  -d '{}' | jq

📌 请将 YOUR_ACCESS_TOKEN 替换为真实的 token。

✅ 成功响应示例：

{
 "user_id": "fe6e600ac900435e8ba8989f6128706e",
 "nick_name": "ethnicity",
 "default_drive_id": "5808324",
 "domain_id": "bj29"
}

✅ 正确的 DRIVE_ID = 5808324 （即 default_drive_id）
❌ 不要用 domain_id （如 bj29）

五、配置环境变量（安全方式）

建议通过环境变量传递敏感信息，避免写入代码。

编辑 /etc/profile 全局设置：

sudo nano /etc/profile

在文件末尾添加以下内容：

export ACCESS_TOKEN="eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.完颜振江.Y0IiOlGkytlzDjw_xN5pFpOUevGk73masg0SF5R1lEc84GUJQPM3B_SQrUuiY73GiVkF5ETYw_luXwIz1gSl_16ajqlaus95vtt6yPE24c0nUOKEQ7SiGH-完颜振江"
export DRIVE_ID="完颜振江"

保存后生效：

source /etc/profile

✅ 验证是否设置成功：

echo $ACCESS_TOKEN | head -c 30 && echo "..."
echo $DRIVE_ID

六、上传脚本：aliyun_upload.py

脚本功能说明

功能 描述
✅ 自动创建远程目录 如 csdn/backups 不存在则逐级创建
✅ 秒传支持 使用 SHA1 + Proof Code 实现快速上传
✅ 小文件兼容 文件 ≥15 字节即可上传
✅ 错误重试机制 无
✅ 目录递归上传 支持上传整个文件夹

完整脚本代码（小文件）

#!/usr/bin/env python3
# aliyun_upload.py - 阿里云盘上传工具（最终完整修复版）

import os
import sys
import json
import hashlib
import requests


# ========== 配置区 ==========
# 推荐通过环境变量设置
ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
if not ACCESS_TOKEN:
    print("❌ 错误：请设置环境变量 ACCESS_TOKEN")
    sys.exit(1)

DRIVE_ID = os.getenv("DRIVE_ID", "5808324")  # 你的默认 drive_id
API_BASE = "https://api.aliyundrive.com"  # ⚠️ 不能有多余空格！
# =============================================


# 请求头
headers = {
    "Authorization": f"Bearer {ACCESS_TOKEN}",
    "Content-Type": "application/json"
}


def sha1(data: bytes) -> str:
    """计算字节数据的 SHA1 大写十六进制"""
    return hashlib.sha1(data).hexdigest().upper()


def get_content_hash(filepath: str) -> str:
    """获取文件内容哈希"""
    with open(filepath, 'rb') as f:
        data = f.read()
    return sha1(data)


def get_proof_code(filepath: str) -> str:
    """
    获取 proof_code：从第 8 字节开始（偏移 7），读 8 字节
    必须满足 file_size >= 15 才能提取 [7,15)
    """
    file_size = os.path.getsize(filepath)
    if file_size < 15:
        print(f"📎 文件 '{os.path.basename(filepath)}' 太小（{file_size}B），无法生成 proof_code")
        return ""

    with open(filepath, 'rb') as f:
        f.seek(7)
        raw_bytes = f.read(8)

    if len(raw_bytes) != 8:
        print("⚠️ 未能读取完整的 8 字节用于 proof_code")
        return ""

    # 调试信息
    print(f"🔍 Proof Bytes (hex): {raw_bytes.hex()}")
    print(f"🔍 Proof Bytes (ascii): {repr(raw_bytes)}")

    return sha1(raw_bytes)


def list_files(parent_file_id: str):
    """列出指定目录下的文件"""
    url = f"{API_BASE}/v2/file/list"
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_file_id
    }
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=10)
        return resp.json()
    except Exception as e:
        print(f"❌ 列出文件失败: {e}")
        return {}


def create_folder(name: str, parent_id: str):
    """创建文件夹"""
    url = f"{API_BASE}/v2/file/create"
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_id,
        "name": name,
        "type": "folder"
    }
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=10)
        return resp.json()
    except Exception as e:
        print(f"❌ 创建目录失败: {e}")
        return {}


def ensure_remote_dir(path: str):
    """确保远程路径存在，自动创建缺失目录"""
    parent_id = "root"
    if not path or path.strip() == "/":
        return parent_id

    parts = [p for p in path.split('/') if p]
    for part in parts:
        items = list_files(parent_id).get("items", [])
        folder = next((x for x in items if x["name"] == part and x["type"] == "folder"), None)

        if not folder:
            print(f"📁 创建目录: {part}")
            res = create_folder(part, parent_id)
            if "file_id" in res:
                parent_id = res["file_id"]
            else:
                print(f"❌ 创建失败: {res}")
                return None
        else:
            parent_id = folder["file_id"]

    return parent_id


def upload_single_file(local_file: str, parent_folder_id: str):
    """上传单个文件"""
    filename = os.path.basename(local_file)
    file_size = os.path.getsize(local_file)

    try:
        content_hash = get_content_hash(local_file)
        proof_code = get_proof_code(local_file)
    except Exception as e:
        print(f"❌ 读取文件失败 {filename}: {e}")
        return

    print(f"📤 上传: {filename} ({file_size} 字节)")
    print(f"🔍 Content Hash: {content_hash}")

    if not proof_code:
        print("❌ 无法生成有效的 proof_code，上传被拒绝")
        return

    print(f"🔑 Proof Code: {proof_code}")

    # Step 1: 创建上传任务（必须包含 proof）
    url = f"{API_BASE}/v2/file/create_with_proof"
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_folder_id,
        "name": filename,
        "type": "file",
        "content_hash_name": "sha1",
        "content_hash": content_hash,
        "proof_version": "v1",
        "proof_code": proof_code,
        "size": file_size
    }

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=10)
        result = resp.json()
    except Exception as e:
        print(f"❌ 请求失败: {e}")
        return

    if "code" in result:
        print(f"❌ 创建上传任务失败: {result}")
        return

    upload_url = result.get("part_info_list", [{}])[0].get("upload_url")
    file_id = result.get("file_id")
    upload_id = result.get("upload_id")

    if not upload_url:
        print("❌ 获取 upload_url 失败")
        return

    # Step 2: 上传文件数据
    try:
        with open(local_file, 'rb') as f:
            data = f.read()
        put_headers = {"Content-Type": ""}  # 关键：不要发送默认 Content-Type
        response = requests.put(upload_url, data=data, headers=put_headers, timeout=30)
        if response.status_code != 200:
            print(f"❌ 文件上传失败: {response.status_code} {response.text}")
            return
    except Exception as e:
        print(f"❌ 上传异常: {e}")
        return

    print("✅ 文件数据已上传")

    # Step 3: 完成上传
    complete_url = f"{API_BASE}/v2/file/complete"
    complete_payload = {
        "drive_id": DRIVE_ID,
        "file_id": file_id,
        "upload_id": upload_id
    }

    try:
        resp = requests.post(complete_url, headers=headers, json=complete_payload, timeout=10)
        result = resp.json()
        if "file_id" in result:
            print(f"🎉 成功！{filename} 已上传到阿里云盘")
        else:
            print(f"❌ 完成失败: {result}")
    except Exception as e:
        print(f"❌ 完成请求失败: {e}")


def main():
    if len(sys.argv) < 2:
        print("❌ 用法: python3 aliyun_upload.py <本地文件/目录> [远程路径]")
        print("   示例:")
        print("     python3 aliyun_upload.py test.txt")
        print("     python3 aliyun_upload.py logs/ csdn/backups")
        sys.exit(1)

    local_path = sys.argv[1]
    remote_path = sys.argv[2] if len(sys.argv) > 2 else ""

    if not os.path.exists(local_path):
        print(f"❌ 本地路径不存在: {local_path}")
        sys.exit(1)

    # 确保远程目录存在
    target_folder_id = ensure_remote_dir(remote_path)
    if not target_folder_id:
        print("❌ 目标目录获取失败，请检查 token 或网络")
        sys.exit(1)

    print(f"📁 目标目录 ID: {target_folder_id}")

    if os.path.isfile(local_path):
        upload_single_file(local_path, target_folder_id)
    elif os.path.isdir(local_path):
        print(f"📦 正在上传目录: {local_path}")
        for root, dirs, files in os.walk(local_path):
            for file in files:
                filepath = os.path.join(root, file)
                upload_single_file(filepath, target_folder_id)
    else:
        print("❌ 不支持的类型")


if __name__ == "__main__":
     main()

支持大文件上传的版本（分片上传）

#!/usr/bin/env python3
# aliyun_upload.py - Aliyun Pan Upload Tool (Chunked Upload for Large Files)

import os
import sys
import json
import hashlib
import requests
from pathlib import Path


# ========== Configuration ==========
ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
if not ACCESS_TOKEN:
    print("Error: ACCESS_TOKEN environment variable is not set.")
    sys.exit(1)

DRIVE_ID = os.getenv("DRIVE_ID", "5808324")  # Default from your info
API_BASE = "https://api.aliyundrive.com"
UPLOAD_CHUNK_SIZE = 10 * 1024 * 1024  # 10MB per chunk
# ===================================


headers = {
    "Authorization": f"Bearer {ACCESS_TOKEN}",
    "Content-Type": "application/json"
}


def sha1(data: bytes) -> str:
    """Calculate SHA1 hash in uppercase hex"""
    return hashlib.sha1(data).hexdigest().upper()


def get_content_hash(filepath: str) -> str:
    """Stream-calculate SHA1 for large files"""
    hash_sha1 = hashlib.sha1()
    with open(filepath, 'rb') as f:
        for chunk in iter(lambda: f.read(UPLOAD_CHUNK_SIZE), b""):
            hash_sha1.update(chunk)
    return hash_sha1.hexdigest().upper()


def list_files(parent_file_id: str):
    """List files under parent ID"""
    url = f"{API_BASE}/v2/file/list"
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_file_id
    }
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=10)
        return resp.json()
    except Exception as e:
        print(f"Failed to list files: {e}")
        return {}


def create_folder(name: str, parent_id: str):
    """Create folder"""
    url = f"{API_BASE}/v2/file/create"
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_id,
        "name": name,
        "type": "folder"
    }
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=10)
        return resp.json()
    except Exception as e:
        print(f"Failed to create folder: {e}")
        return {}


def ensure_remote_dir(path: str):
    """Ensure remote directory exists, create if missing"""
    parent_id = "root"
    if not path or path.strip() == "/":
        return parent_id

    parts = [p for p in path.split('/') if p]
    for part in parts:
        items = list_files(parent_id).get("items", [])
        folder = next((x for x in items if x["name"] == part and x["type"] == "folder"), None)

        if not folder:
            print(f"Creating directory: {part}")
            res = create_folder(part, parent_id)
            if "file_id" in res:
                parent_id = res["file_id"]
            else:
                print(f"Failed to create dir: {res}")
                return None
        else:
            parent_id = folder["file_id"]

    return parent_id


def upload_single_file_chunked(local_file: str, parent_folder_id: str):
    """Upload file in chunks"""
    filename = os.path.basename(local_file)
    file_size = os.path.getsize(local_file)

    try:
        content_hash = get_content_hash(local_file)
    except Exception as e:
        print(f"Failed to read file {filename}: {e}")
        return

    print(f"Uploading: {filename} ({file_size} bytes)")
    print(f"Content Hash (SHA1): {content_hash}")

    # Step 1: Create upload session
    url = f"{API_BASE}/v2/file/create"
    num_parts = (file_size + UPLOAD_CHUNK_SIZE - 1) // UPLOAD_CHUNK_SIZE
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_folder_id,
        "name": filename,
        "type": "file",
        "content_hash_name": "sha1",
        "content_hash": content_hash,
        "size": file_size,
        "part_info_list": [{"part_number": i + 1} for i in range(num_parts)]
    }

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=30)
        result = resp.json()
    except Exception as e:
        print(f"Create request failed: {e}")
        return

    if "code" in result:
        if result["code"] in ["RapidProofNeed", "PreHashMatched"]:
            pass  # Ignore
        else:
            print(f"Create failed: {result}")
            return

    upload_id = result.get("upload_id")
    file_id = result.get("file_id")
    part_info_list = result.get("part_info_list", [])

    if not upload_id or not file_id or not part_info_list:
        print(f"Missing upload params: {result}")
        return

    print(f"Total chunks: {len(part_info_list)}")

    # Step 2: Upload each chunk
    uploaded_parts = []
    with open(local_file, 'rb') as f:
        for part_info in part_info_list:
            part_num = part_info["part_number"]
            start = (part_num - 1) * UPLOAD_CHUNK_SIZE
            end = min(start + UPLOAD_CHUNK_SIZE, file_size)
            chunk_size = end - start

            f.seek(start)
            chunk_data = f.read(chunk_size)

            upload_url = part_info.get("upload_url")
            if not upload_url:
                print(f"No upload URL for part {part_num}")
                return

            print(f"Uploading chunk {part_num}/{len(part_info_list)} [{start}-{end}]")

            try:
                r = requests.put(
                    upload_url,
                    data=chunk_data,
                    headers={"Content-Type": ""},
                    timeout=60
                )
                if r.status_code != 200:
                    print(f"Chunk {part_num} failed: {r.status_code}, {r.text}")
                    return
            except Exception as e:
                print(f"Chunk {part_num} error: {e}")
                return

            uploaded_parts.append({
                "part_number": part_num,
                "etag": "*"
            })

    print("All chunks uploaded.")

    # Step 3: Complete upload
    complete_url = f"{API_BASE}/v2/file/complete"
    complete_payload = {
        "drive_id": DRIVE_ID,
        "file_id": file_id,
        "upload_id": upload_id,
        "part_info_list": uploaded_parts
    }

    try:
        resp = requests.post(complete_url, headers=headers, json=complete_payload, timeout=30)
        result = resp.json()
        if "file_id" in result:
            print(f"Success! File '{filename}' uploaded.")
        else:
            print(f"Complete failed: {result}")
    except Exception as e:
        print(f"Complete request failed: {e}")


def main():
    if len(sys.argv) < 2:
        print("Usage: python3 aliyun_upload.py <local_file> [remote_path]")
        print("Example:")
        print("  python3 aliyun_upload.py test.txt")
        print("  python3 aliyun_upload.py 2025-10-10-v4.zip bocheng-k8s")
        sys.exit(1)

    local_path = sys.argv[1]
    remote_path = sys.argv[2] if len(sys.argv) > 2 else ""

    if not os.path.exists(local_path):
        print(f"Local file not found: {local_path}")
        sys.exit(1)

    if not os.path.isfile(local_path):
        print("Only single file upload is supported.")
        sys.exit(1)

    target_folder_id = ensure_remote_dir(remote_path)
    if not target_folder_id:
        print("Failed to get target directory ID.")
        sys.exit(1)

    print(f"Target folder ID: {target_folder_id}")

    upload_single_file_chunked(local_path, target_folder_id)


if __name__ == "__main__":
     main()

更大文件上传（增强版）

准备虚拟环境：

# 创建虚拟环境
python3 -m venv /alipan/venv

# 激活环境
source /alipan/venv/bin/activate

# 安装依赖
pip install requests urllib3<2.0.0 chardet==3.0.4 --force-reinstall

脚本代码：

#!/usr/bin/env python3

import os
import sys
import json
import hashlib
import requests
from time import sleep
from requests.adapters import HTTPAdapter
from urllib3.poolmanager import PoolManager
from urllib3.util.ssl_ import create_urllib3_context
import ssl


# Configuration
ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
if not ACCESS_TOKEN:
    print("Error: ACCESS_TOKEN environment variable is not set.")
    sys.exit(1)

DRIVE_ID = os.getenv("DRIVE_ID", "5808324")
API_BASE = "https://api.aliyundrive.com"
UPLOAD_CHUNK_SIZE = 160 * 1024 * 1024  # 16MB per chunk
MAX_RETRIES = 5
TIMEOUT = 600  # 10 minutes per upload


# Custom TLS Adapter (optional, only if you face handshake issues)
class TLSAdapter(HTTPAdapter):
    def init_poolmanager(self, connections, maxsize, block=False, *args, **kwargs):
        ctx = create_urllib3_context()
        ctx.set_ciphers('DEFAULT:@SECLEVEL=1')
        # Remove the following line if TLS 1.3 is supported
        # ctx.options |= ssl.OP_NO_TLSv1_3
        kwargs['ssl_context'] = ctx
        return super().init_poolmanager(connections, maxsize, block, *args, **kwargs)


# Create session with custom adapter
session = requests.Session()
session.mount('https://', TLSAdapter())
session.headers.update({
    "Authorization": f"Bearer {ACCESS_TOKEN}",
    "Content-Type": "application/json"
})


def get_content_hash(filepath):
    hash_sha1 = hashlib.sha1()
    with open(filepath, 'rb') as f:
        for chunk in iter(lambda: f.read(UPLOAD_CHUNK_SIZE), b""):
            hash_sha1.update(chunk)
    return hash_sha1.hexdigest().upper()


def list_files(parent_file_id):
    url = f"{API_BASE}/v2/file/list"
    payload = {"drive_id": DRIVE_ID, "parent_file_id": parent_file_id}
    try:
        resp = session.post(url, json=payload, timeout=30)
        return resp.json()
    except Exception as e:
        print(f"Failed to list files: {e}")
        return {}


def create_folder(name, parent_id):
    url = f"{API_BASE}/v2/file/create"
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_id,
        "name": name,
        "type": "folder"
    }
    try:
        resp = session.post(url, json=payload, timeout=30)
        return resp.json()
    except Exception as e:
        print(f"Failed to create folder: {e}")
        return {}


def ensure_remote_dir(path):
    parent_id = "root"
    if not path or path.strip() == "/":
        return parent_id
    parts = [p for p in path.split('/') if p]
    for part in parts:
        items = list_files(parent_id).get("items", [])
        folder = next((x for x in items if x["name"] == part and x["type"] == "folder"), None)
        if not folder:
            print(f"Creating directory: {part}")
            res = create_folder(part, parent_id)
            if "file_id" in res:
                parent_id = res["file_id"]
            else:
                print(f"Failed to create dir: {res}")
                return None
        else:
            parent_id = folder["file_id"]
    return parent_id


def upload_single_file_chunked(local_file, parent_folder_id):
    filename = os.path.basename(local_file)
    file_size = os.path.getsize(local_file)

    try:
        content_hash = get_content_hash(local_file)
    except Exception as e:
        print(f"Failed to read file {filename}: {e}")
        return

    print(f"Uploading: {filename} ({file_size} bytes)")
    print(f"Content Hash (SHA1): {content_hash}")

    num_parts = (file_size + UPLOAD_CHUNK_SIZE - 1) // UPLOAD_CHUNK_SIZE
    payload = {
        "drive_id": DRIVE_ID,
        "parent_file_id": parent_folder_id,
        "name": filename,
        "type": "file",
        "content_hash_name": "sha1",
        "content_hash": content_hash,
        "size": file_size,
        "part_info_list": [{"part_number": i + 1} for i in range(num_parts)]
    }

    try:
        resp = session.post(f"{API_BASE}/v2/file/create", json=payload, timeout=30)
        result = resp.json()
    except Exception as e:
        print(f"Create request failed: {e}")
        return

    upload_id = result.get("upload_id")
    file_id = result.get("file_id")
    part_info_list = result.get("part_info_list", [])

    if not upload_id or not file_id or not part_info_list:
        print(f"Missing upload params: {result}")
        return

    print(f"Total chunks: {len(part_info_list)}")

    uploaded_parts = []
    with open(local_file, 'rb') as f:
        for part_info in part_info_list:
            part_num = part_info["part_number"]
            start = (part_num - 1) * UPLOAD_CHUNK_SIZE
            end = min(start + UPLOAD_CHUNK_SIZE, file_size)
            chunk_size = end - start

            f.seek(start)
            chunk_data = f.read(chunk_size)

            if len(chunk_data) != chunk_size:
                print(f"Error: read size mismatch for part {part_num}")
                return

            # Calculate local MD5 of this chunk
            local_md5 = hashlib.md5(chunk_data).hexdigest().upper()
            print(f"Local MD5 for part {part_num}: {local_md5}")

            upload_url = part_info.get("upload_url")
            if not upload_url:
                print(f"No upload URL for part {part_num}")
                return

            success = False
            etag = None
            for attempt in range(1, MAX_RETRIES + 1):
                print(f"Uploading chunk {part_num}/{len(part_info_list)} [{start}-{end}] (attempt {attempt})")
                try:
                    r = session.put(
                        upload_url,
                        data=chunk_data,
                        headers={"Content-Type": ""},
                        timeout=TIMEOUT
                    )
                    if r.status_code == 200:
                        raw_etag = r.headers.get('ETag') or r.headers.get('etag')
                        if raw_etag:
                            etag = raw_etag.strip('" \t').upper()
                        else:
                            print(f"Warning: No ETag received from server for part {part_num}")
                            continue

                        # Verify ETag matches local MD5
                        if etag == local_md5:
                            print(f"Got ETag for part {part_num}: {etag}")
                            success = True
                            break
                        else:
                            print(f"ETag mismatch for part {part_num}: expected {local_md5}, got {etag}")
                            print("Retrying...")
                            sleep(2 ** attempt)
                    else:
                        print(f"Chunk {part_num} failed: {r.status_code}, {r.text}")
                        sleep(2 ** attempt)
                except Exception as e:
                    print(f"Chunk {part_num} error (attempt {attempt}): {e}")
                    if attempt < MAX_RETRIES:
                        sleep(2 ** attempt)
                    else:
                        print(f"Chunk {part_num} failed after {MAX_RETRIES} attempts.")

            if not success:
                print(f"Aborting upload due to failure in chunk {part_num}.")
                return

            uploaded_parts.append({"part_number": part_num, "etag": etag})

    print("All chunks uploaded.")

    complete_payload = {
        "drive_id": DRIVE_ID,
        "file_id": file_id,
        "upload_id": upload_id,
        "part_info_list": uploaded_parts
    }

    try:
        resp = session.post(f"{API_BASE}/v2/file/complete", json=complete_payload, timeout=30)
        result = resp.json()
        if "file_id" in result:
            print(f"Upload completed successfully: {filename}")
        else:
            print(f"Complete failed: {result}")
    except Exception as e:
        print(f"Complete request failed: {e}")


def main():
    if len(sys.argv) < 2:
        print("Usage: python3 aliyun_upload_path.py <local_file> [remote_path]")
        print("Example:")
        print("  python3 aliyun_upload_path.py test.txt")
        print("  python3 aliyun_upload_path.py 2025-10-10-v4.zip bocheng-k8s")
        sys.exit(1)

    local_path = sys.argv[1]
    remote_path = sys.argv[2] if len(sys.argv) > 2 else ""

    if not os.path.exists(local_path):
        print(f"Local file not found: {local_path}")
        sys.exit(1)

    if not os.path.isfile(local_path):
        print("Only single file upload is supported.")
        sys.exit(1)

    target_folder_id = ensure_remote_dir(remote_path)
    if not target_folder_id:
        print("Failed to get target directory ID.")
        sys.exit(1)

    print(f"Target folder ID: {target_folder_id}")
    upload_single_file_chunked(local_path, target_folder_id)


if __name__ == "__main__":
     main()

七、使用方法

1. 保存脚本

nano aliyun_upload.py
# 粘贴上面代码，Ctrl+O 保存，Ctrl+X 退出
chmod +x aliyun_upload.py

2. 上传文件

# 上传单个文件到根目录
python3 aliyun_upload.py test.txt

# 上传文件到指定目录（自动创建 csdn）
python3 aliyun_upload.py test.txt csdn

# 上传整个目录
python3 aliyun_upload.py ./logs csdn/backups

八、常见问题与解决

问题 原因 解决方案
AccessTokenInvalid Token 过期或格式错误 重新抓包获取有效 token
NotFound.Drive DRIVE_ID 错误 使用 default_drive_id，不是 domain_id
InvalidRapidProof proof_code 计算错误 确保偏移 7 字节，读 8 字节
NameError: local_file 变量名拼写错误 已修复
URL 包含空格 API_BASE 有空格 删除 https://api.aliyundrive.com 后的空格

九、安全建议

  • 🔐 不要在代码中硬编码 ACCESS_TOKEN。
  • 🔄 定期刷新 token（每 1~2 小时）。
  • 📁 敏感脚本权限设为 600。
  • 🛡️ 生产环境建议使用服务账户 + refresh_token 自动续期机制。

十、结语

您现在拥有了一个稳定、可靠、可维护的阿里云盘上传工具。只要确保：

  • ✅ ACCESS_TOKEN 有效
  • ✅ DRIVE_ID = 5808324
  • ✅ API_BASE 无多余空格
  • ✅ 环境变量已加载
# python # 开发语言
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

基于BP神经网络的语音特征信号分类研究（Matlab代码实现）

语音信号分类是语音信号处理领域的核心研究方向，广泛应用于语音识别、声纹认证、语音情感分析、智能人机交互等诸多领域。由于语音信号具有非线性、时变性、随机性的复杂特征，传统分类方法难以精准挖掘语音特征与信号类别之间的关联关系，存在分类精度低、抗干扰能力弱、泛化性差等问题。BP神经网络凭借其强大的非线性映射、自适应学习和容错能力，能够有效适配复杂语音特征数据的分类任务。

avatar AI Agent技术社区

基于BP神经网络的语音特征信号分类研究（Matlab代码实现）

语音信号分类是语音信号处理领域的核心研究方向，广泛应用于语音识别、声纹认证、语音情感分析、智能人机交互等诸多领域。由于语音信号具有非线性、时变性、随机性的复杂特征，传统分类方法难以精准挖掘语音特征与信号类别之间的关联关系，存在分类精度低、抗干扰能力弱、泛化性差等问题。BP神经网络凭借其强大的非线性映射、自适应学习和容错能力，能够有效适配复杂语音特征数据的分类任务。

avatar AI Agent技术社区
cover

【安心陪诊 Agent】医疗 AI 的安全边界：不诊断、不剂量、不替代医生

avatar AI Agent技术社区