端侧模型缓存版本绑定与安全清理工程指南

#!/usr/bin/env python3
import os
import sys
import hashlib
import json

def load_meta(cache_dir):
    meta_path = os.path.join(cache_dir, "model_version.meta")
    if not os.path.exists(meta_path):
        return None
    try:
        with open(meta_path, "r") as f:
            return json.load(f)
    except (json.JSONDecodeError, IOError):
        return None

def calculate_digest(file_path):
    sha256 = hashlib.sha256()
    try:
        with open(file_path, "rb") as f:
            while True:
                chunk = f.read(8192)
                if not chunk:
                    break
                sha256.update(chunk)
        return sha256.hexdigest()
    except IOError:
        return None

def cleanup_directory(cache_dir):
    files_to_remove = ["model_version.meta", "model.mlmodelc", "model.mlpackage"]
    for filename in files_to_remove:
        full_path = os.path.join(cache_dir, filename)
        if os.path.exists(full_path):
            try:
                os.remove(full_path)
            except OSError:
                pass

def verify_and_clean(cache_dir, expected_version, expected_digest):
    meta = load_meta(cache_dir)
    
    # Case 1: Meta file missing or corrupted
    if meta is None:
        cleanup_directory(cache_dir)
        sys.exit(1)
    
    # Case 2: Version mismatch
    current_version = meta.get("version")
    if expected_version and current_version != expected_version:
        cleanup_directory(cache_dir)
        sys.exit(1)
    
    # Case 3: Model file missing
    model_filename = meta.get("model_file", "model.mlmodelc")
    model_path = os.path.join(cache_dir, model_filename)
    if not os.path.exists(model_path):
        cleanup_directory(cache_dir)
        sys.exit(1)
    
    # Case 4: Reference count > 0 (In use)
    ref_count = meta.get("ref_count", 0)
    if ref_count > 0:
        sys.exit(2)
    
    # Case 5: Digest mismatch
    if expected_digest:
        current_digest = calculate_digest(model_path)
        if current_digest != expected_digest:
            cleanup_directory(cache_dir)
            sys.exit(1)
    
    # All checks passed, safe to keep or clean based on external logic
    sys.exit(0)

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: script.py <cache_dir> [expected_version] [expected_digest]")
        sys.exit(3)
    
    target_dir = sys.argv[1]
    ver_arg = sys.argv[2] if len(sys.argv) > 2 else None
    dig_arg = sys.argv[3] if len(sys.argv) > 3 else None
    
    verify_and_clean(target_dir, ver_arg, dig_arg)