This commit is contained in:
jxxghp
2025-05-08 09:56:43 +08:00
parent ea31072ae5
commit 570dddc120
2 changed files with 43 additions and 44 deletions
+16 -20
View File
@@ -1,13 +1,13 @@
import base64 import base64
import hashlib import hashlib
import io
import secrets import secrets
import threading import threading
import time import time
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor
from pathlib import Path from pathlib import Path
from queue import Queue from queue import Queue, Empty
from typing import List, Dict, Optional, Tuple, Union from typing import List, Dict, Optional, Tuple, Union
import io
import requests import requests
from tqdm import tqdm from tqdm import tqdm
@@ -591,13 +591,6 @@ class AliPan(StorageBase, metaclass=Singleton):
raise Exception(resp.get("message")) raise Exception(resp.get("message"))
return resp.get('part_info_list', []) return resp.get('part_info_list', [])
@staticmethod
def _upload_part(upload_url: str, data: bytes):
"""
上传单个分片
"""
return requests.put(upload_url, data=data)
def _list_uploaded_parts(self, drive_id: str, file_id: str, upload_id: str) -> dict: def _list_uploaded_parts(self, drive_id: str, file_id: str, upload_id: str) -> dict:
""" """
获取已上传分片列表 获取已上传分片列表
@@ -650,12 +643,14 @@ class AliPan(StorageBase, metaclass=Singleton):
part_size = min(file_size // self.MAX_WORKERS, self.MAX_PART_SIZE) part_size = min(file_size // self.MAX_WORKERS, self.MAX_PART_SIZE)
return part_size, min(file_size // part_size + 1, self.MAX_WORKERS) return part_size, min(file_size // part_size + 1, self.MAX_WORKERS)
def _log_progress(self, desc: str, total: int) -> tqdm: @staticmethod
def _log_progress(desc: str, total: int) -> tqdm:
""" """
创建一个可以输出到日志的进度条 创建一个可以输出到日志的进度条
""" """
class TqdmToLogger(io.StringIO): class TqdmToLogger(io.StringIO):
def write(s, buf): def write(s, buf): # noqa
buf = buf.strip('\r\n\t ') buf = buf.strip('\r\n\t ')
if buf: if buf:
logger.info(buf) logger.info(buf)
@@ -671,8 +666,9 @@ class AliPan(StorageBase, metaclass=Singleton):
miniters=1 miniters=1
) )
def _upload_part(self, upload_url: str, data: bytes, part_num: int, @staticmethod
progress_queue: Queue) -> Tuple[int, str]: def _upload_part(upload_url: str, data: bytes, part_num: int,
progress_queue: Queue) -> Tuple[int, str]:
""" """
上传单个分片 上传单个分片
""" """
@@ -703,10 +699,10 @@ class AliPan(StorageBase, metaclass=Singleton):
# 2. 创建文件并检查秒传 # 2. 创建文件并检查秒传
create_res = self._create_file(drive_id=target_dir.drive_id, create_res = self._create_file(drive_id=target_dir.drive_id,
parent_file_id=target_dir.fileid, parent_file_id=target_dir.fileid,
file_name=target_name, file_name=target_name,
file_path=local_path, file_path=local_path,
chunk_size=part_size) chunk_size=part_size)
if create_res.get('rapid_upload', False): if create_res.get('rapid_upload', False):
logger.info(f"【阿里云盘】{target_name} 秒传完成!") logger.info(f"【阿里云盘】{target_name} 秒传完成!")
return self.get_item(target_path) return self.get_item(target_path)
@@ -731,7 +727,7 @@ class AliPan(StorageBase, metaclass=Singleton):
# 5. 初始化进度条 # 5. 初始化进度条
logger.info(f"【阿里云盘】开始上传: {local_path} -> {target_path}" logger.info(f"【阿里云盘】开始上传: {local_path} -> {target_path}"
f"分片大小:{StringUtils.str_filesize(part_size)},线程数:{workers}") f"分片大小:{StringUtils.str_filesize(part_size)},线程数:{workers}")
progress_bar = self._log_progress(f"【阿里云盘】{target_name} 上传进度", file_size) progress_bar = self._log_progress(f"【阿里云盘】{target_name} 上传进度", file_size)
# 6. 创建进度队列 # 6. 创建进度队列
@@ -774,7 +770,7 @@ class AliPan(StorageBase, metaclass=Singleton):
try: try:
uploaded = progress_queue.get(timeout=1) uploaded = progress_queue.get(timeout=1)
progress_bar.update(uploaded) progress_bar.update(uploaded)
except: except Empty:
pass pass
# 等待所有任务完成 # 等待所有任务完成
+18 -15
View File
@@ -1,18 +1,17 @@
import base64 import base64
import hashlib import hashlib
import io
import json import json
import secrets import secrets
import threading import threading
import time import time
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path from pathlib import Path
from queue import Queue from queue import Queue, Empty
from typing import List, Dict, Optional, Tuple, Union from typing import List, Dict, Optional, Tuple, Union
import io
import oss2 import oss2
import requests import requests
from oss2 import SizedFileAdapter, determine_part_size
from oss2.models import PartInfo from oss2.models import PartInfo
from tqdm import tqdm from tqdm import tqdm
@@ -421,8 +420,9 @@ class U115Pan(StorageBase, metaclass=Singleton):
part_size = min(file_size // self.MAX_WORKERS, self.MAX_PART_SIZE) part_size = min(file_size // self.MAX_WORKERS, self.MAX_PART_SIZE)
return part_size, min(file_size // part_size + 1, self.MAX_WORKERS) return part_size, min(file_size // part_size + 1, self.MAX_WORKERS)
def _upload_part(self, bucket: oss2.Bucket, object_name: str, upload_id: str, @staticmethod
part_number: int, part_data: bytes, progress_queue: Queue) -> PartInfo: def _upload_part(bucket: oss2.Bucket, object_name: str, upload_id: str,
part_number: int, part_data: bytes, progress_queue: Queue) -> PartInfo:
""" """
上传单个分片 上传单个分片
""" """
@@ -437,12 +437,14 @@ class U115Pan(StorageBase, metaclass=Singleton):
logger.error(f"【115】分片 {part_number} 上传失败: {str(e)}") logger.error(f"【115】分片 {part_number} 上传失败: {str(e)}")
raise raise
def _log_progress(self, desc: str, total: int) -> tqdm: @staticmethod
def _log_progress(desc: str, total: int) -> tqdm:
""" """
创建一个可以输出到日志的进度条 创建一个可以输出到日志的进度条
""" """
class TqdmToLogger(io.StringIO): class TqdmToLogger(io.StringIO):
def write(s, buf): def write(s, buf): # noqa
buf = buf.strip('\r\n\t ') buf = buf.strip('\r\n\t ')
if buf: if buf:
logger.info(buf) logger.info(buf)
@@ -463,6 +465,7 @@ class U115Pan(StorageBase, metaclass=Singleton):
""" """
实现带秒传、断点续传和多线程并发上传 实现带秒传、断点续传和多线程并发上传
""" """
def encode_callback(cb: dict): def encode_callback(cb: dict):
return oss2.utils.b64encode_as_string(json.dumps(cb).strip()) return oss2.utils.b64encode_as_string(json.dumps(cb).strip())
@@ -584,22 +587,22 @@ class U115Pan(StorageBase, metaclass=Singleton):
access_key_secret=AccessKeySecret, access_key_secret=AccessKeySecret,
security_token=SecurityToken security_token=SecurityToken
) )
bucket = oss2.Bucket(auth, endpoint, bucket_name) bucket = oss2.Bucket(auth, endpoint, bucket_name) # noqa
# 计算分片大小和线程数 # 计算分片大小和线程数
part_size, workers = self._calc_parts(file_size) part_size, workers = self._calc_parts(file_size)
logger.info(f"【115】开始上传: {local_path} -> {target_path}" logger.info(f"【115】开始上传: {local_path} -> {target_path}"
f"分片大小:{StringUtils.str_filesize(part_size)},线程数:{workers}") f"分片大小:{StringUtils.str_filesize(part_size)},线程数:{workers}")
# 初始化进度条 # 初始化进度条
progress_bar = self._log_progress(f"【115】{target_name} 上传进度", file_size) progress_bar = self._log_progress(f"【115】{target_name} 上传进度", file_size)
# 初始化分片上传 # 初始化分片上传
upload_id = bucket.init_multipart_upload(object_name, upload_id = bucket.init_multipart_upload(object_name,
params={ params={
"encoding-type": "url", "encoding-type": "url",
"sequential": "" "sequential": ""
}).upload_id }).upload_id
# 创建进度队列 # 创建进度队列
progress_queue = Queue() progress_queue = Queue()
@@ -635,7 +638,7 @@ class U115Pan(StorageBase, metaclass=Singleton):
try: try:
uploaded = progress_queue.get(timeout=1) uploaded = progress_queue.get(timeout=1)
progress_bar.update(uploaded) progress_bar.update(uploaded)
except: except Empty:
pass pass
# 等待所有任务完成 # 等待所有任务完成
@@ -660,7 +663,7 @@ class U115Pan(StorageBase, metaclass=Singleton):
try: try:
result = bucket.complete_multipart_upload(object_name, upload_id, parts, result = bucket.complete_multipart_upload(object_name, upload_id, parts,
headers=headers) headers=headers)
if result.status == 200: if result.status == 200:
logger.debug(f"【115】上传 Step 6 回调结果:{result.resp.response.json()}") logger.debug(f"【115】上传 Step 6 回调结果:{result.resp.response.json()}")
logger.info(f"【115】{target_name} 上传成功") logger.info(f"【115】{target_name} 上传成功")