数据版本与迭代：DVC数据版本控制

import hashlibimport jsonimport osfrom dataclasses import dataclass, fieldfrom typing import Dict, List, Optional, Tuple  class ContentAddressableStore:    """内容寻址存储，按文件内容哈希存储与去重"""     def __init__(self):        self._store: Dict[str, bytes] = {}        self._refcount: Dict[str, int] = {}     def _hash(self, data: bytes) -> str:        return hashlib.sha256(data).hexdigest()     def put(self, path: str) -> str:        with open(path, "rb") as f:            data = f.read()        digest = self._hash(data)        if digest not in self._store:            self._store[digest] = data        self._refcount[digest] = self._refcount.get(digest, 0) + 1        return digest     def has(self, digest: str) -> bool:        return digest in self._store     def get(self, digest: str) -> bytes:        return self._store.get(digest, b"")  @dataclassclass PointerFile:    """DVC指针文件，记录数据文件的哈希与元信息"""     path: str    md5: str    size: int    isdir: bool = False     def serialize(self) -> str:        return json.dumps({            "outs": [{                "path": self.path,                "md5": self.md5,                "size": self.size,                "isdir": self.isdir,            }]        }, ensure_ascii=False, indent=2)     @classmethod    def deserialize(cls, text: str) -> "PointerFile":        d = json.loads(text)        out = d["outs"][0]        return cls(out["path"], out["md5"], out["size"], out.get("isdir", False))  class DVCTracker:    """DVC追踪器，管理指针文件与工作区数据"""     def __init__(self, store: ContentAddressableStore):        self.store = store        self.pointers: Dict[str, PointerFile] = {}     def add(self, path: str) -> PointerFile:        digest = self.store.put(path)        size = os.path.getsize(path)        ptr = PointerFile(path=path, md5=digest, size=size)        self.pointers[path] = ptr        return ptr     def checkout(self, path: str) -> bool:        ptr = self.pointers.get(path)        if not ptr or not self.store.has(ptr.md5):            return False        data = self.store.get(ptr.md5)        with open(path, "wb") as f:            f.write(data)        return True     def status(self) -> List[str]:        changed = []        for path, ptr in self.pointers.items():            if not os.path.exists(path):                changed.append(f"{path}: deleted")                continue            with open(path, "rb") as f:                cur = hashlib.sha256(f.read()).hexdigest()            if cur != ptr.md5:                changed.append(f"{path}: modified")        return changed  @dataclassclass PipelineStage:    """数据管道阶段定义"""     name: str    cmd: str    deps: List[str] = field(default_factory=list)    outs: List[str] = field(default_factory=list)    params: Dict[str, str] = field(default_factory=dict)    deps_md5: Dict[str, str] = field(default_factory=dict)    outs_md5: Dict[str, str] = field(default_factory=dict)  class Pipeline:    """可复现管道，按依赖哈希增量执行"""     def __init__(self):        self.stages: Dict[str, PipelineStage] = {}     def add_stage(self, stage: PipelineStage):        self.stages[stage.name] = stage     def _file_md5(self, path: str) -> str:        if not os.path.exists(path):            return ""        with open(path, "rb") as f:            return hashlib.sha256(f.read()).hexdigest()     def _params_signature(self, params: Dict[str, str]) -> str:        clean = {k: v for k, v in params.items() if k != "__sig__"}        return hashlib.sha256(            json.dumps(clean, sort_keys=True).encode()        ).hexdigest()     def repro(self, name: str, runner) -> str:        stage = self.stages.get(name)        if not stage:            return f"stage {name} not found"        changed = False        for dep in stage.deps:            cur = self._file_md5(dep)            if cur != stage.deps_md5.get(dep, ""):                changed = True                stage.deps_md5[dep] = cur        cur_params = self._params_signature(stage.params)        if cur_params != stage.params.get("__sig__", ""):            changed = True            stage.params["__sig__"] = cur_params        if not changed and all(            self._file_md5(o) == stage.outs_md5.get(o, "")            for o in stage.outs        ):            return f"stage {name}: up to date, skipped"        runner(stage.cmd)        for o in stage.outs:            stage.outs_md5[o] = self._file_md5(o)        return f"stage {name}: reproduced"  if __name__ == "__main__":    store = ContentAddressableStore()    tracker = DVCTracker(store)     data_path = "/tmp/dvc_demo_data.csv"    with open(data_path, "w") as f:        f.write("user_id,click_count\n1,42\n2,17\n")     ptr = tracker.add(data_path)    print(f"add {data_path}: md5={ptr.md5[:12]}, size={ptr.size}")     os.remove(data_path)    ok = tracker.checkout(data_path)    print(f"checkout restored: {ok}")     st = tracker.status()    print(f"status: {st if st else 'clean'}")     pipe = Pipeline()    stage = PipelineStage(        name="featurize",        cmd="python featurize.py",        deps=["raw.csv"],        outs=["features.parquet"],        params={"window": "7d", "threshold": "0.5"},    )    pipe.add_stage(stage)     def fake_runner(cmd):        print(f"  running: {cmd}")     res = pipe.repro("featurize", fake_runner)    print(res)    res2 = pipe.repro("featurize", fake_runner)    print(res2)