spb/wp15_uqo

1"""Raw-file manifest: one row per monthly file, machine-readable (CLAUDE.md §4.2)."""23from __future__ import annotations45import json6from dataclasses import asdict, dataclass, field7from pathlib import Path8910@dataclass11class FileRecord:12    dataset_id: str13    dataset_revision: str14    remote_path: str15    local_path: str16    year: int17    month: int18    expected_bytes: int | None = None19    downloaded_bytes: int | None = None20    sha256: str | None = None21    lfs_sha256: str | None = None22    status: str = "pending"  # pending | downloading | complete | failed23    download_started_at: str | None = None24    download_completed_at: str | None = None25    row_count: int | None = None26    column_names: list[str] = field(default_factory=list)27    column_types: list[str] = field(default_factory=list)28    min_timestamp: str | None = None29    max_timestamp: str | None = None30    unique_symbol_count: int | None = None31    duplicate_key_count: int | None = None32    null_count_by_column: dict = field(default_factory=dict)33    audit_version: str = "0.1"34    code_commit: str | None = None353637class Manifest:38    """JSON-backed manifest with atomic writes; convertible to parquet/csv."""3940    def __init__(self, path: Path):41        self.path = path42        self.records: dict[str, FileRecord] = {}43        if path.exists():44            data = json.loads(path.read_text())45            for r in data["files"]:46                self.records[r["remote_path"]] = FileRecord(**r)4748    def upsert(self, rec: FileRecord) -> None:49        self.records[rec.remote_path] = rec5051    def save(self) -> None:52        tmp = self.path.with_suffix(".tmp")53        tmp.write_text(json.dumps(54            {"files": [asdict(r) for r in sorted(self.records.values(),55                                                 key=lambda x: x.remote_path)]},56            indent=1))57        tmp.replace(self.path)5859    def export_tabular(self) -> None:60        import polars as pl61        rows = []62        for r in self.records.values():63            d = asdict(r)64            # JSON-encode nested fields: empty structs are not writable to Parquet65            d["column_names"] = "|".join(d["column_names"])66            d["column_types"] = "|".join(d["column_types"])67            d["null_count_by_column"] = json.dumps(d["null_count_by_column"])68            rows.append(d)69        df = pl.DataFrame(rows)70        df.write_parquet(self.path.with_suffix(".parquet"))71        df.write_csv(self.path.with_suffix(".csv"))7273    def pending(self) -> list[FileRecord]:74        return [r for r in self.records.values() if r.status != "complete"]75