catalog-extract/scripts/csvio.py · py · download
"""Canonical schemas + an append-only, header-guarded, per-row-flushing CSV writer.
Both product and store files are append-only with a `snapshot_date` column:
one block of rows per run. The history IS the file. Brand-specific attributes
that have no canonical column go into `extra_attributes` (JSON) so the header
never widens and never drifts.
"""
from __future__ import annotations
import csv
import json
import os
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
PRODUCT_FIELDS = [
"snapshot_date", "brand", "sku", "name", "family", "collection", "category", "gender",
"price_value", "price_currency", "list_price_value", "is_discounted", "stock_status",
"movement_type", "caliber", "power_reserve",
"case_material", "case_diameter_mm", "case_thickness", "case_shape", "crystal",
"water_resistance", "dial_color", "bracelet_material", "bracelet_color", "buckle",
"is_limited_edition", "rating", "review_count",
"description", "image_url", "image_count", "product_url", "extra_attributes",
]
STORE_FIELDS = [
"snapshot_date", "brand", "store_name", "store_type", "address", "city",
"state_region", "postal_code", "country_code", "phone", "email",
"latitude", "longitude", "website", "store_url",
]
RETAILER_FIELDS = [
"snapshot_date", "retailer", "brand", "matched_sku", "match_method", "match_score",
"retailer_product_id", "manufacturer_ref", "name", "gender",
"price_value", "list_price_value", "is_on_sale", "price_currency",
"availability", "category", "collection",
"image_url", "product_url", "raw_identifiers",
]
def today_snapshot() -> str:
return datetime.now(timezone.utc).date().isoformat()
def _clean(v):
if v is None:
return ""
if isinstance(v, bool):
return "true" if v else "false"
if isinstance(v, float):
return "" if v != v else v # NaN -> ""
if isinstance(v, (dict, list)):
return json.dumps(v, ensure_ascii=False)
return str(v).strip()
def make_row(fields: list[str], brand: str, data: dict, *, extra_key: str | None = "extra_attributes") -> dict:
"""Fill every declared field (missing -> ""). Unknown keys in `data` are
folded into `extra_attributes` (if the schema has it) rather than dropped."""
row = {f: "" for f in fields}
row["snapshot_date"] = today_snapshot()
if "brand" in row:
row["brand"] = brand
extra = {}
if extra_key and extra_key in row and data.get(extra_key):
e = data[extra_key]
extra.update(e if isinstance(e, dict) else json.loads(e))
for k, v in data.items():
if k == extra_key:
continue
if k in row:
row[k] = _clean(v)
elif extra_key and v not in (None, "", [], {}):
extra[k] = v
if extra_key and extra_key in row:
row[extra_key] = json.dumps(extra, ensure_ascii=False) if extra else ""
return row
class HeaderMismatch(RuntimeError):
pass
class IncrementalCsv:
"""Open once, write header if new, flush + fsync after every row. Fails
loud if an existing file's header differs from the declared schema."""
def __init__(self, path: str | os.PathLike, fieldnames: list[str]):
self.path = Path(path)
self.fieldnames = list(fieldnames)
self.path.parent.mkdir(parents=True, exist_ok=True)
existing = self.path.exists() and self.path.stat().st_size > 0
if existing:
with self.path.open("r", encoding="utf-8-sig", newline="") as fh:
header = next(csv.reader(fh), None)
if header is not None and header != self.fieldnames:
raise HeaderMismatch(
f"{self.path}: header has {len(header)} cols, schema has {len(self.fieldnames)}; "
f"first difference at {next((i for i, (a, b) in enumerate(zip(header, self.fieldnames)) if a != b), min(len(header), len(self.fieldnames)))}. "
"Move the old file aside rather than letting columns silently reorder.")
self._fh = self.path.open("a", encoding="utf-8-sig", newline="")
self._w = csv.DictWriter(self._fh, fieldnames=self.fieldnames, extrasaction="ignore")
if not existing:
self._w.writeheader()
self._flush()
self.rows_written = 0
def write(self, row: dict) -> None:
self._w.writerow(row)
self._flush()
self.rows_written += 1
def _flush(self) -> None:
self._fh.flush()
try:
os.fsync(self._fh.fileno())
except OSError:
pass
def close(self) -> None:
try:
self._fh.close()
except Exception: # noqa: BLE001
pass
def __enter__(self):
return self
def __exit__(self, *exc):
self.close()
def read_rows(path: str | os.PathLike) -> list[dict]:
p = Path(path)
if not p.exists() or p.stat().st_size == 0:
return []
with p.open("r", encoding="utf-8-sig", newline="") as fh:
return list(csv.DictReader(fh))
def existing_keys(path, key: str = "sku") -> set[str]:
return {r.get(key, "").strip() for r in read_rows(path) if r.get(key, "").strip()}
def previous_snapshot_count(path) -> tuple[str | None, int]:
"""(newest snapshot_date, row count in it) from an existing file."""
counts = Counter(r.get("snapshot_date", "") for r in read_rows(path))
counts.pop("", None)
if not counts:
return None, 0
d = max(counts)
return d, counts[d]