lou
50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
"""
|
|
分卷接收模块 (receiver)
|
|
模块: 服务端 / 分卷接收
|
|
输入: 卷数据 (transfer_id, index)
|
|
输出: 落盘卷 + ok/409
|
|
|
|
逐卷校验 SHA-256 (对照 init 登记的规格); 重复卷直接 ok (幂等, 断点续传重发不炸)。
|
|
"""
|
|
|
|
import hashlib
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from task_manager import TaskManager
|
|
|
|
|
|
class ReceiverError(RuntimeError):
|
|
"""卷校验失败 (哈希不一致/规格缺失)"""
|
|
|
|
|
|
class Receiver:
|
|
def __init__(self, tasks: TaskManager, tmp_root: str | Path) -> None:
|
|
self.tasks = tasks
|
|
self.tmp_root = Path(tmp_root)
|
|
|
|
def receive(self, transfer_id: str, idx: int, data: bytes) -> None:
|
|
"""接收并校验一卷。失败抛 ReceiverError (api 转 409)。"""
|
|
if self.tasks.is_chunk_received(transfer_id, idx):
|
|
return # 幂等: 重复卷直接 ok
|
|
spec = self.tasks.chunk_spec(transfer_id, idx)
|
|
if spec is None:
|
|
raise ReceiverError(f"任务 {transfer_id} 卷 {idx} 规格不存在")
|
|
|
|
# 校验大小 + SHA-256
|
|
if len(data) != spec["size"]:
|
|
raise ReceiverError(f"卷 {idx} 大小不符: 期望 {spec['size']}, 实际 {len(data)}")
|
|
sha256 = hashlib.sha256(data).hexdigest()
|
|
if sha256 != spec["sha256"]:
|
|
raise ReceiverError(f"卷 {idx} SHA-256 不一致 (数据损坏或串卷)")
|
|
|
|
# 落盘临时目录
|
|
chunk_dir = self.tmp_root / transfer_id
|
|
chunk_dir.mkdir(parents=True, exist_ok=True)
|
|
chunk_path = chunk_dir / f"chunk_{idx:04d}"
|
|
tmp_path = chunk_path.with_suffix(".tmp")
|
|
tmp_path.write_bytes(data)
|
|
shutil.move(str(tmp_path), str(chunk_path))
|
|
|
|
self.tasks.mark_chunk_received(transfer_id, idx)
|