INNER CODE UNIT · Python

_load_history_csv

simonlin1212/Vibe-Research · calc/cli.py:98

def _load_history_csv(spec: dict, run_dir: str | None) -> tuple[list, dict]:
    """从 <run_dir>/raw/ 确定性加载一列数值:返回 (values, 解析记录)。realpath 校验,拒绝越出 raw/。"""
    if not run_dir:
        raise ValueError("history_csv 需要 --run-dir")
    if not isinstance(spec, dict):
        raise ValueError("history_csv 必须是对象 {raw_ref, column, where}")
    rel = str(spec.get("raw_ref") or "")
    col = spec.get("column")
    where = spec.get("where") or {}
    if not rel or not col:
        raise ValueError("history_csv 需要 raw_ref 与 column")
    target = _raw_target(rel, run_dir, "history_csv")
    data = target.read_bytes()
    sha = hashlib.sha256(data).hexdigest()
    rows = list(csv.DictReader(data.decode("utf-8").splitlines()))
    if not rows or col not in rows[0]:
        raise ValueError(f"CSV 无列 {col!r}")
    selected = [r for r in rows if all(str(r.get(k)) == str(v) for k, v in where.items())]

View source record →