无法上传文件

[HuHaiYang0]

[applewpj]

请问解决了吗，我面临了相同的问题

[waitmap]

相同的问题
看了网盘内，文件实际上已经上传了，但是本地是报错的。

开始上传文件: test.txt
文件大小: 0.00 B
上传进度: 0% - 计算文件哈希...
上传进度: 10% - 发起预上传请求...
预上传返回的bucket: ul-zb
最终使用的bucket: ul-zb
上传进度: 20% - 更新文件哈希...
上传进度: 30% - 开始单分片上传...
上传进度: 35% - 获取上传授权...
上传进度: 50% - 上传文件到OSS...
操作失败: 上传分片 1 失败: 404

NoSuchBucket
The specified bucket does not exist.
68F6EB8922CC2D3835EF491E
ul-zb.oss-cn-shenzhen.aliyuncs.com
ul-zb
0015-00000101
https://api.aliyun.com/troubleshoot?q=0015-00000101

操作失败: 上传分片 1 失败: 404

NoSuchBucket
The specified bucket does not exist.
68F6EB8922CC2D3835EF491E
ul-zb.oss-cn-shenzhen.aliyuncs.com
ul-zb
0015-00000101
https://api.aliyun.com/troubleshoot?q=0015-00000101

是不是没有考虑到秒传到因素？？

这个就可以了

-- coding: utf-8 --

"""
文件上传服务
"""

import hashlib
import json
import mimetypes
import time
import base64
import os
from pathlib import Path
from typing import Any, Callable, Dict, Optional, Tuple
from datetime import datetime, timezone
import httpx

from ..core.api_client import QuarkAPIClient
from ..exceptions import APIError

class FileUploadService:
"""文件上传服务"""

def __init__(self, client: QuarkAPIClient):
    """
    初始化文件上传服务

    Args:
        client: API客户端实例
    """
    self.api_client = client

def upload_file(
    self,
    file_path: str,
    parent_folder_id: str = "0",
    progress_callback: Optional[Callable] = None
) -> Dict[str, Any]:
    """
    上传文件到夸克网盘

    Args:
        file_path: 本地文件路径
        parent_folder_id: 父文件夹ID，默认为根目录
        progress_callback: 进度回调函数

    Returns:
        上传结果字典

    Raises:
        FileNotFoundError: 文件不存在
        APIError: API调用失败
    """
    file_path_obj = Path(file_path)
    if not file_path_obj.exists():
        raise FileNotFoundError(f"文件不存在: {file_path}")

    if not file_path_obj.is_file():
        raise ValueError(f"路径不是文件: {file_path}")

    # 获取文件信息
    file_size = file_path_obj.stat().st_size
    file_name = file_path_obj.name

    # 获取MIME类型
    mime_type, _ = mimetypes.guess_type(str(file_path_obj))
    if not mime_type:
        mime_type = "application/octet-stream"

    # 计算文件哈希
    if progress_callback:
        progress_callback(0, "计算文件哈希...")

    md5_hash, sha1_hash = self._calculate_file_hashes(file_path_obj, progress_callback)

    # 1. 预上传请求
    if progress_callback:
        progress_callback(10, "发起预上传请求...")

    pre_upload_result = self._pre_upload(
        file_name=file_name,
        file_size=file_size,
        parent_folder_id=parent_folder_id,
        mime_type=mime_type
    )

    if not pre_upload_result.get('status'):
        raise APIError(f"预上传失败: {pre_upload_result}")

    # 2. 更新文件哈希
    if progress_callback:
        progress_callback(20, "更新文件哈希...")

    hash_result = self._update_file_hash(
        md5_hash, sha1_hash, pre_upload_result.get('data', {}).get('task_id', '')
    )

    if not hash_result.get('status'):
        raise APIError(f"哈希验证失败: {hash_result}")

    # 检查是否已完成（秒传）
    if hash_result.get('data', {}).get('finish') is True:
        if progress_callback:
            progress_callback(100, "文件秒传完成")
        return {
            'status': 'success',
            'file_name': file_name,
            'file_size': file_size,
            'md5': md5_hash,
            'sha1': sha1_hash,
            'upload_result': {'strategy': 'instant_upload', 'message': '文件已存在，秒传完成'},
            'finish_result': hash_result.get('data', {})
        }

    # 3. 分片上传
    if progress_callback:
        progress_callback(30, "开始分片上传...")

    part_size = pre_upload_result.get('data', {}).get('metadata', {}).get('part_size', 0)
    if part_size == 0:
        part_size = 4 * 1024 * 1024  # 默认4MB分片

    etags = []
    part_number = 1
    uploaded = 0

    try:
        with open(file_path, "rb") as f:
            while True:
                chunk = f.read(part_size)
                if not chunk:
                    break

                # 上传分片
                if progress_callback:
                    progress_callback(30 + int(uploaded / file_size * 50), f"上传分片 {part_number}...")

                etag = self._upload_part(
                    pre_upload_result.get('data', {}),
                    mime_type,
                    part_number,
                    chunk
                )
                etags.append(etag)
                part_number += 1
                uploaded += len(chunk)

                if progress_callback:
                    progress = min(80, 30 + int(uploaded / file_size * 50))
                    progress_callback(progress, f"已上传 {uploaded}/{file_size} 字节")

    except Exception as e:
        raise APIError(f"读取文件分片上传失败: {str(e)}")

    # 4. 提交上传
    if progress_callback:
        progress_callback(85, "提交上传...")

    commit_result = self._commit_upload(pre_upload_result.get('data', {}), etags)
    if not commit_result.get('status'):
        raise APIError(f"提交上传失败: {commit_result}")

    # 5. 完成上传
    if progress_callback:
        progress_callback(95, "完成上传...")

    finish_result = self._finish_upload(pre_upload_result.get('data', {}))
    if not finish_result.get('status'):
        raise APIError(f"完成上传失败: {finish_result}")

    if progress_callback:
        progress_callback(100, "上传完成")

    return {
        'status': 'success',
        'task_id': pre_upload_result.get('data', {}).get('task_id'),
        'file_name': file_name,
        'file_size': file_size,
        'md5': md5_hash,
        'sha1': sha1_hash,
        'upload_result': {
            'strategy': 'multipart_upload',
            'parts': len(etags),
            'etags': etags
        },
        'finish_result': finish_result.get('data', {})
    }

def _calculate_file_hashes(
    self,
    file_path: Path,
    progress_callback: Optional[Callable] = None
) -> Tuple[str, str]:
    """计算文件的MD5和SHA1哈希值"""
    md5_hash = hashlib.md5()
    sha1_hash = hashlib.sha1()

    file_size = file_path.stat().st_size
    bytes_read = 0

    with open(file_path, 'rb') as f:
        while chunk := f.read(8192):
            md5_hash.update(chunk)
            sha1_hash.update(chunk)
            bytes_read += len(chunk)

            if progress_callback and file_size > 0:
                progress = min(10, int((bytes_read / file_size) * 10))
                progress_callback(progress, f"计算哈希: {progress}%")

    return md5_hash.hexdigest(), sha1_hash.hexdigest()

def _pre_upload(
    self,
    file_name: str,
    file_size: int,
    parent_folder_id: str,
    mime_type: str
) -> Dict[str, Any]:
    """发起预上传请求"""
    current_time = int(time.time() * 1000)

    data = {
        "ccp_hash_update": True,
        "dir_name": "",
        "file_name": file_name,
        "format_type": mime_type,
        "l_created_at": current_time,
        "l_updated_at": current_time,
        "pdir_fid": parent_folder_id,
        "size": file_size,
    }

    response = self.api_client.post(
        "file/upload/pre",
        json_data=data
    )

    if not response.get('status'):
        return {'status': False, 'message': response.get('message', '未知错误')}

    if response.get("code") == 0 and response.get("status") == 200:
        return {'status': True, 'data': response.get('data', {})}

    return {'status': False, 'message': response}

def _update_file_hash(
    self,
    md5_hash: str,
    sha1_hash: str,
    task_id: str
) -> Dict[str, Any]:
    """提交文件哈希验证"""
    data = {"md5": md5_hash, "sha1": sha1_hash, "task_id": task_id}

    response = self.api_client.post(
        "file/update/hash",
        json_data=data
    )

    if not response.get('status'):
        return {'status': False, 'message': response.get('message', '未知错误')}

    if response.get("code") == 0 and response.get("status") == 200:
        finish = response.get("data", {}).get("finish", False)
        return {'status': True, 'data': response.get('data', {}), 'finish': finish}
    
    return {'status': False, 'message': response}

def _upload_part(
    self,
    pre_data: Dict[str, Any],
    mime_type: str,
    part_number: int,
    chunk_data: bytes
) -> str:
    """上传文件分片"""
    now = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT")
    auth_meta = f"PUT\n\n{mime_type}\n{now}\nx-oss-date:{now}\nx-oss-user-agent:aliyun-sdk-js/6.6.1 Chrome 98.0.4758.80 on Windows 10 64-bit\n/{pre_data['bucket']}/{pre_data['obj_key']}?partNumber={part_number}&uploadId={pre_data['upload_id']}"
    
    auth_data = {
        "auth_info": pre_data["auth_info"],
        "auth_meta": auth_meta,
        "task_id": pre_data["task_id"],
    }
    
    auth_response = self.api_client.post("file/upload/auth", json_data=auth_data)
    if not auth_response.get('status'):
        raise APIError(f"获取上传授权失败: {auth_response}")
    
    if auth_response.get("code") == 0 and auth_response.get("status") == 200:
        upload_url = f"https://{pre_data['bucket']}.{pre_data['upload_url'][7:]}/{pre_data['obj_key']}"
        headers = {
            "Authorization": auth_response["data"]["auth_key"],
            "Content-Type": mime_type,
            "Referer": "https://pan.quark.cn/",
            "x-oss-date": now,
            "x-oss-user-agent": "aliyun-sdk-js/6.6.1 Chrome 98.0.4758.80 on Windows 10 64-bit",
        }
        params = {"partNumber": str(part_number), "uploadId": pre_data["upload_id"]}
        
        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.put(
                    upload_url, headers=headers, params=params, content=chunk_data
                )
                response.raise_for_status()
                return response.headers.get("ETag", "").strip('"')
        except httpx.RequestError as e:
            raise APIError(f"上传分片 {part_number} 失败: {str(e)}")
    
    raise APIError(f"获取上传授权失败: {auth_response}")

def _commit_upload(
    self,
    pre_data: Dict[str, Any],
    etags: list
) -> Dict[str, Any]:
    """提交分片上传完成"""
    now = datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S GMT")
    xml_parts = []
    for i, etag in enumerate(etags, 1):
        xml_parts.append(
            f"<Part><PartNumber>{i}</PartNumber><ETag>{etag}</ETag></Part>"
        )
    xml_body = f"""<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<CompleteMultipartUpload>\n{"".join(xml_parts)}\n</CompleteMultipartUpload>"""
    
    md5 = hashlib.md5()
    md5.update(xml_body.encode("utf-8"))
    content_md5 = base64.b64encode(md5.digest()).decode("utf-8")
    callback_json = json.dumps(pre_data["callback"])
    callback_b64 = base64.b64encode(callback_json.encode("utf-8")).decode("utf-8")
    auth_meta = f"POST\n{content_md5}\napplication/xml\n{now}\nx-oss-callback:{callback_b64}\nx-oss-date:{now}\nx-oss-user-agent:aliyun-sdk-js/6.6.1 Chrome 98.0.4758.80 on Windows 10 64-bit\n/{pre_data['bucket']}/{pre_data['obj_key']}?uploadId={pre_data['upload_id']}"
    
    auth_data = {
        "auth_info": pre_data["auth_info"],
        "auth_meta": auth_meta,
        "task_id": pre_data["task_id"],
    }
    
    auth_response = self.api_client.post("file/upload/auth", json_data=auth_data)
    if not auth_response.get('status'):
        return {'status': False, 'message': auth_response}
    
    if auth_response.get("code") == 0 and auth_response.get("status") == 200:
        upload_url = f"https://{pre_data['bucket']}.{pre_data['upload_url'][7:]}/{pre_data['obj_key']}"
        headers = {
            "Authorization": auth_response["data"]["auth_key"],
            "Content-MD5": content_md5,
            "Content-Type": "application/xml",
            "Referer": "https://pan.quark.cn/",
            "x-oss-callback": callback_b64,
            "x-oss-date": now,
            "x-oss-user-agent": "aliyun-sdk-js/6.6.1 Chrome 98.0.4758.80 on Windows 10 64-bit",
        }
        params = {"uploadId": pre_data["upload_id"]}
        
        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.post(
                    upload_url, headers=headers, params=params, content=xml_body
                )
                response.raise_for_status()
                return {'status': True, 'data': response.json()}
        except httpx.RequestError as e:
            return {'status': False, 'message': str(e)}
    
    return {'status': False, 'message': auth_response}

def _finish_upload(
    self,
    pre_data: Dict[str, Any]
) -> Dict[str, Any]:
    """完成上传流程"""
    data = {"obj_key": pre_data["obj_key"], "task_id": pre_data["task_id"]}
    
    response = self.api_client.post("file/upload/finish", json_data=data)
    if not response.get('status'):
        return {'status': False, 'message': response.get('message', '未知错误')}
    
    if response.get("code") == 0 and response.get("status") == 200:
        return {'status': True, 'data': response.get('data', {})}

    return {'status': False, 'message': response}