"""CSV catalog of a dataset.
One row per raw file: original and pseudonymized values, paths, hashes,
status and dates. The catalog is the re-identification key of the dataset.
"""
from __future__ import annotations
import csv
import os
import threading
from pathlib import Path
COLUMNS = [
"status",
"risk",
"risk_reason",
"raw_path",
"anonymized_path",
"patient_folder",
"format",
"original_patient_id",
"anonymized_patient_id",
"file_patient_id",
"original_ecg_id",
"anonymized_ecg_id",
"original_last_name",
"original_first_name",
"original_file_name",
"anonymized_file_name",
"folder_anonymized",
"companions",
"duplicate",
"raw_size",
"raw_mtime_ns",
"raw_sha256",
"anonymized_sha256",
"detected_at",
"anonymized_at",
"last_change_at",
"message",
"tool_version",
]
ACTIVE = {"anonymized", "copied", "changed"}
"""Statuses of rows whose anonymized file exists."""
SEPARATOR = " | "
[docs]
class Catalog:
"""Rows keyed by raw path (relative to the dataset folder).
Writes go to a temporary file renamed over the catalog, so the CSV is
never left half written.
"""
[docs]
def __init__(self, path: Path) -> None:
self.path = Path(path)
self.rows: dict[str, dict[str, str]] = {}
self.lock = threading.RLock()
if self.path.exists():
with open(self.path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
if "patient_code" in (reader.fieldnames or []):
raise ValueError(
f"{self.path} was written by an older version (patient_code "
f"column): delete it and the ANONYMIZED folder to start again")
for row in reader:
self.rows[row["raw_path"]] = {c: row.get(c, "") or "" for c in COLUMNS}
[docs]
def get(self, raw_path: str) -> dict[str, str] | None:
with self.lock:
return self.rows.get(raw_path)
[docs]
def put(self, row: dict[str, str]) -> None:
with self.lock:
self.rows[row["raw_path"]] = {c: str(row.get(c, "") or "") for c in COLUMNS}
[docs]
def active(self) -> list[dict[str, str]]:
with self.lock:
return [r for r in self.rows.values() if r["status"] in ACTIVE]
[docs]
def used_codes(self) -> set[str]:
with self.lock:
return {r[c] for r in self.rows.values() for c in ("anonymized_patient_id", "anonymized_ecg_id") if r[c]}
[docs]
def patient_codes(self) -> dict[str, str]:
"""Patient code by patient folder (every row ever written keeps its code)."""
with self.lock:
return {r["patient_folder"]: r["anonymized_patient_id"] for r in self.rows.values()
if r["patient_folder"] and r["anonymized_patient_id"]}
[docs]
def save(self) -> None:
with self.lock:
rows = sorted(self.rows.values(), key=lambda r: r["raw_path"])
tmp = self.path.with_name(self.path.name + ".tmp")
with open(tmp, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=COLUMNS)
writer.writeheader()
writer.writerows(rows)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, self.path)
