python -m venv venv
source venv/bin/activate  # Windows 使用 venv\Scripts\activate
pip install fastapi uvicorn pydantic[dotenv] python-multipart

.
├─ server.py              # FastAPI 服務(wù)端
├─ upload.db              # SQLite 元數(shù)據(jù)庫（運(yùn)行后生成）
└─ uploads/
   └─ {upload_id}/
      ├─ chunks/         # 分片臨時(shí)目錄
      ├─ meta.json       # 元信息（冗余存儲(chǔ)，輔助排查）
      └─ final/          # 最終文件目錄

# server.py
import hashlib
import json
import os
import sqlite3
import threading
from contextlib import contextmanager
from math import ceil
from pathlib import Path
from typing import List, Optional

from fastapi import FastAPI, UploadFile, File, HTTPException, Query, Form
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uvicorn

BASE_DIR = Path(__file__).parent.resolve()
DATA_DIR = BASE_DIR / "uploads"
DB_PATH = BASE_DIR / "upload.db"
DATA_DIR.mkdir(exist_ok=True)

app = FastAPI(title="Chunked Upload (FastAPI)")

# ---------- 并發(fā)鎖（同一 upload_id 合并上鎖） ----------
_merge_locks = {}
_merge_locks_global = threading.Lock()

def _get_lock(upload_id: str) -> threading.Lock:
    with _merge_locks_global:
        if upload_id not in _merge_locks:
            _merge_locks[upload_id] = threading.Lock()
        return _merge_locks[upload_id]

# ---------- SQLite 簡單封裝 ----------
def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        c = conn.cursor()
        c.execute("""CREATE TABLE IF NOT EXISTS uploads(
            upload_id TEXT PRIMARY KEY,
            filename TEXT,
            size INTEGER,
            chunk_size INTEGER,
            total_chunks INTEGER,
            file_hash TEXT
        )""")
        c.execute("""CREATE TABLE IF NOT EXISTS chunks(
            upload_id TEXT,
            idx INTEGER,
            size INTEGER,
            hash TEXT,
            PRIMARY KEY(upload_id, idx)
        )""")
        conn.commit()

@contextmanager
def db() :
    conn = sqlite3.connect(DB_PATH)
    try:
        yield conn
    finally:
        conn.close()

# ---------- 工具 ----------
def sha256_of_bytes(b: bytes) -> str:
    h = hashlib.sha256()
    h.update(b)
    return h.hexdigest()

def md5_of_bytes(b: bytes) -> str:
    h = hashlib.md5()
    h.update(b)
    return h.hexdigest()

def ensure_upload_folders(upload_id: str) -> Path:
    root = DATA_DIR / upload_id
    (root / "chunks").mkdir(parents=True, exist_ok=True)
    (root / "final").mkdir(parents=True, exist_ok=True)
    return root

def write_meta(root: Path, meta: dict):
    (root / "meta.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2))

# ---------- 請求/響應(yīng)模型 ----------
class InitReq(BaseModel):
    filename: str
    size: int
    chunk_size: int
    file_hash: Optional[str] = None   # 可選：客戶端傳文件整體 SHA256/MD5

class InitResp(BaseModel):
    upload_id: str
    total_chunks: int

class StatusResp(BaseModel):
    upload_id: str
    uploaded: List[int]
    total_chunks: int

class MergeReq(BaseModel):
    upload_id: str
    expected_hash: Optional[str] = None  # 可選：合并后對整文件校驗(yàn)

# ---------- 路由 ----------
@app.post("/upload/init", response_model=InitResp)
def init_upload(req: InitReq):
    if req.size <= 0 or req.chunk_size <= 0:
        raise HTTPException(400, "size/chunk_size 必須為正整數(shù)")
    total = ceil(req.size / req.chunk_size)

    upload_id = hashlib.sha1(f"{req.filename}:{req.size}:{req.chunk_size}:{req.file_hash or ''}".encode()).hexdigest()
    root = ensure_upload_folders(upload_id)

    with db() as conn:
        c = conn.cursor()
        # UPSERT
        c.execute("INSERT OR IGNORE INTO uploads(upload_id, filename, size, chunk_size, total_chunks, file_hash) VALUES(?,?,?,?,?,?)",
                  (upload_id, req.filename, req.size, req.chunk_size, total, req.file_hash))
        conn.commit()

    write_meta(root, {
        "upload_id": upload_id,
        "filename": req.filename,
        "size": req.size,
        "chunk_size": req.chunk_size,
        "total_chunks": total,
        "file_hash": req.file_hash
    })
    return InitResp(upload_id=upload_id, total_chunks=total)

@app.post("/upload/chunk")
async def upload_chunk(
    upload_id: str = Query(..., description="init 返回的 upload_id"),
    index: int = Query(..., ge=1, description="分片序號，從1開始"),
    content_hash: Optional[str] = Query(None, description="可選：分片內(nèi)容校驗(yàn)（sha256或md5，客戶端自定義）"),
    file: UploadFile = File(...)
):
    # 檢查 upload 是否存在
    with db() as conn:
        c = conn.cursor()
        row = c.execute("SELECT total_chunks FROM uploads WHERE upload_id=?", (upload_id,)).fetchone()
        if not row:
            raise HTTPException(404, "upload_id 不存在，請先 /upload/init")
        total_chunks = row[0]

    if index > total_chunks:
        raise HTTPException(400, f"index 超出總分片數(shù)：{total_chunks}")

    root = ensure_upload_folders(upload_id)
    chunk_path = root / "chunks" / f"{index}.part"

    # 將分片寫入臨時(shí)文件
    data = await file.read()
    if content_hash:
        # 這里兼容 md5/sha256：長度32大概率md5，長度64大概率sha256，也可自定義一個(gè) query param 指定算法
        calc = md5_of_bytes(data) if len(content_hash) == 32 else sha256_of_bytes(data)
        if calc != content_hash:
            raise HTTPException(400, "content_hash 校驗(yàn)失敗")

    with open(chunk_path, "wb") as f:
        f.write(data)

    # 記錄 DB
    with db() as conn:
        c = conn.cursor()
        c.execute("INSERT OR REPLACE INTO chunks(upload_id, idx, size, hash) VALUES(?,?,?,?)",
                  (upload_id, index, len(data), content_hash))
        conn.commit()

    return JSONResponse({"message": "ok", "index": index, "size": len(data)})

@app.get("/upload/status", response_model=StatusResp)
def upload_status(upload_id: str):
    with db() as conn:
        c = conn.cursor()
        up = c.execute("SELECT total_chunks FROM uploads WHERE upload_id=?", (upload_id,)).fetchone()
        if not up:
            raise HTTPException(404, "upload_id 不存在")
        total = up[0]
        rows = c.execute("SELECT idx FROM chunks WHERE upload_id=? ORDER BY idx", (upload_id,)).fetchall()
        uploaded = [r[0] for r in rows]
    return StatusResp(upload_id=upload_id, uploaded=uploaded, total_chunks=total)

@app.post("/upload/merge")
def upload_merge(req: MergeReq):
    # 防止并發(fā)重復(fù)合并
    lock = _get_lock(req.upload_id)
    with lock:
        with db() as conn:
            c = conn.cursor()
            up = c.execute("SELECT filename, total_chunks FROM uploads WHERE upload_id=?", (req.upload_id,)).fetchone()
            if not up:
                raise HTTPException(404, "upload_id 不存在")
            filename, total = up

            # 校驗(yàn)分片是否齊全
            rows = c.execute("SELECT idx FROM chunks WHERE upload_id=? ORDER BY idx", (req.upload_id,)).fetchall()
            uploaded = [r[0] for r in rows]
            if len(uploaded) != total or uploaded != list(range(1, total + 1)):
                raise HTTPException(400, f"分片不完整，已上傳：{uploaded}，應(yīng)為 1..{total}")

        root = ensure_upload_folders(req.upload_id)
        final_path = root / "final" / filename
        # 合并
        with open(final_path, "wb") as dst:
            for i in range(1, total + 1):
                chunk_path = root / "chunks" / f"{i}.part"
                with open(chunk_path, "rb") as src:
                    dst.write(src.read())

        # 可選：合并后哈希校驗(yàn)
        if req.expected_hash:
            with open(final_path, "rb") as f:
                data = f.read()
            calc = md5_of_bytes(data) if len(req.expected_hash) == 32 else sha256_of_bytes(data)
            if calc != req.expected_hash:
                raise HTTPException(400, "合并后文件校驗(yàn)失敗")

        # 也可以在此處清理分片文件，或異步清理
        return JSONResponse({"message": "merged", "final_path": str(final_path)})
        
# ---------- 啟動(dòng) ----------
if __name__ == "__main__":
    init_db()
    uvicorn.run(app, host="0.0.0.0", port=8000)

pip install requests tqdm

# client.py
import math
import os
import time
import hashlib
import requests
from pathlib import Path
from tqdm import tqdm

SERVER = "http://localhost:8000"

def sha256_of_file(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()

def md5_of_bytes(b: bytes) -> str:
    import hashlib
    h = hashlib.md5()
    h.update(b)
    return h.hexdigest()

def upload_file(path: str, chunk_size: int = 5 * 1024 * 1024, use_md5_for_chunk=False, retries=3, sleep=1):
    p = Path(path)
    size = p.stat().st_size
    file_hash = sha256_of_file(p)  # 也可改為 MD5
    total_chunks = math.ceil(size / chunk_size)

    # 1) init
    r = requests.post(f"{SERVER}/upload/init", json={
        "filename": p.name,
        "size": size,
        "chunk_size": chunk_size,
        "file_hash": file_hash
    })
    r.raise_for_status()
    j = r.json()
    upload_id = j["upload_id"]
    total_chunks = j["total_chunks"]
    print("upload_id:", upload_id, "total_chunks:", total_chunks)

    # 2) 查詢已上傳
    def get_status():
        rr = requests.get(f"{SERVER}/upload/status", params={"upload_id": upload_id})
        rr.raise_for_status()
        return rr.json()["uploaded"]

    uploaded = set(get_status())

    # 3) 逐片上傳（可按需并發(fā)）
    with open(p, "rb") as f:
        for index in tqdm(range(1, total_chunks + 1), desc="Uploading"):
            if index in uploaded:
                continue
            start = (index - 1) * chunk_size
            end = min(size, index * chunk_size)
            f.seek(start)
            data = f.read(end - start)

            content_hash = md5_of_bytes(data) if use_md5_for_chunk else None
            for attempt in range(retries):
                try:
                    files = {"file": (f"{index}.part", data, "application/octet-stream")}
                    params = {"upload_id": upload_id, "index": index}
                    if content_hash:
                        params["content_hash"] = content_hash
                    rr = requests.post(f"{SERVER}/upload/chunk", files=files, params=params, timeout=30)
                    rr.raise_for_status()
                    break
                except Exception as e:
                    print(f"[{index}] retry {attempt+1}/{retries} because {e}")
                    time.sleep(sleep)
            else:
                raise RuntimeError(f"upload chunk {index} failed after retries")

    # 4) 合并
    r = requests.post(f"{SERVER}/upload/merge", json={
        "upload_id": upload_id,
        "expected_hash": file_hash  # 合并后校驗(yàn)整文件
    })
    r.raise_for_status()
    print("merge result:", r.json())

if __name__ == "__main__":
    # 示例：python client.py
    upload_file("bigfile.bin", chunk_size=5 * 1024 * 1024, use_md5_for_chunk=True)

async function upload(file) {
  const chunkSize = 5 * 1024 * 1024;
  const totalChunks = Math.ceil(file.size / chunkSize);

  // 1) init
  const initResp = await fetch('/upload/init', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json'},
    body: JSON.stringify({ filename: file.name, size: file.size, chunk_size: chunkSize })
  }).then(r => r.json());
  const { upload_id } = initResp;

  // 2) 查詢已上傳
  const status = await fetch(`/upload/status?upload_id=${upload_id}`).then(r => r.json());
  const uploaded = new Set(status.uploaded);

  // 3) 順序或并發(fā)上傳
  for (let i = 1; i <= totalChunks; i++) {
    if (uploaded.has(i)) continue;
    const start = (i - 1) * chunkSize;
    const end = Math.min(file.size, i * chunkSize);
    const blob = file.slice(start, end);
    const form = new FormData();
    form.append('file', blob, `${i}.part`);

    await fetch(`/upload/chunk?upload_id=${upload_id}&index=${i}`, { method: 'POST', body: form });
  }

  // 4) merge
  await fetch('/upload/merge', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json'},
    body: JSON.stringify({ upload_id })
  });
}

# pip install minio
from minio import Minio

client = Minio("127.0.0.1:9000", access_key="minioadmin", secret_key="minioadmin", secure=False)

# 1) 為每個(gè)分片生成預(yù)簽名 PUT URL，前端直接 PUT 到 MinIO
url = client.presigned_put_object("my-bucket", f"{upload_id}/chunks/{index}.part", expires=timedelta(hours=1))

# 2) 合并階段：服務(wù)端從 MinIO 流式讀取所有分片，寫入最終對象