CRC32 比較を使って HTTP 経由で Web インターフェースディレクトリを ESP32 LittleFS に同期する方法

import urllib.parse
import zlib
from pathlib import Path

import requests

BASE_URL = "http://10.56.11.162"

def compute_crc32(path: Path) -> int:
    crc = 0
    with path.open("rb") as f:
        while chunk := f.read(65536):
            crc = zlib.crc32(chunk, crc)
    return crc & 0xFFFFFFFF

def get_device_file_crc(path: str):
    url = f"{BASE_URL}/api/file-crc?path={urllib.parse.quote(path, safe='')}"
    r = requests.get(url, timeout=10)
    if r.status_code == 200:
        j = r.json()
        return j.get("crc32_u32"), j.get("size")
    return None, None

def upload_file(path: str, local_path: Path) -> bool:
    url = f"{BASE_URL}/api/files/put?path={urllib.parse.quote(path, safe='')}"
    # 完全にメモリに読み込む：requests は Content-Length を送る必要があり、
    # chunked Transfer-Encoding ではだめ（esp_http_server が拒否する）。
    r = requests.put(url, data=local_path.read_bytes(), timeout=30)
    return r.status_code in (200, 201)

def list_device_files() -> list[str]:
    r = requests.get(f"{BASE_URL}/api/files", timeout=10)
    return [f["path"] for f in r.json().get("files", [])]

def delete_device_file(path: str) -> bool:
    url = f"{BASE_URL}/api/files/delete?path={urllib.parse.quote(path, safe='')}"
    return requests.delete(url, timeout=10).status_code == 200

def sync(local_dir: Path):
    local_files = {
        p.relative_to(local_dir).as_posix(): p
        for p in local_dir.rglob("*") if p.is_file()
    }

    for rel, full in local_files.items():
        local_crc = compute_crc32(full)
        dev_crc, _ = get_device_file_crc(rel)
        if dev_crc is None:
            print(f"{rel}: missing on device, uploading")
            upload_file(rel, full)
        elif dev_crc != local_crc:
            print(f"{rel}: CRC mismatch (local=0x{local_crc:08X} "
                  f"device=0x{dev_crc:08X}), uploading")
            upload_file(rel, full)
        else:
            print(f"{rel}: up-to-date")

    for dev in list_device_files():
        if dev not in local_files:
            if dev == "nodes.json":
                continue  # 保護されたシステムファイル
            print(f"{dev}: deleted on device")
            delete_device_file(dev)