#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Build the CKAN release data resources for the Hermosillo COVID-19 mobility dataset. This script consolidates daily source CSV files named YYYY-MM-DD.csv into a single long-format CSV table, creates an equivalent Parquet file when pyarrow is available, preserves the original daily CSV files in a ZIP archive, and writes release-level SHA-256 integrity files. The script intentionally avoids hard-coded local paths. Provide the source and output directories through command-line arguments. Example: python3 build_ckan_mobility_package.py \ --source-dir /path/to/daily_csv_files \ --output-dir /path/to/ckan_release_v1_0_0 Recommended dependencies: python3 -m pip install pandas pyarrow """ from __future__ import annotations import argparse import csv import hashlib import re import sys import zipfile from dataclasses import dataclass from datetime import date, datetime, timezone from pathlib import Path from typing import Iterable, Optional import pandas as pd DATASET_SHORT_NAME = "mobility_hmo" VERSION = "1.0.0" VERSION_TAG = "v" + VERSION.replace(".", "_") DATE_FILENAME_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})\.csv$", re.IGNORECASE) ENCODING_CANDIDATES = ("utf-8-sig", "utf-8", "latin-1") DELIMITER_CANDIDATES = (",", ";", "\t", "|") DEFAULT_CHUNK_SIZE = 100_000 @dataclass(frozen=True) class SourceFile: path: Path file_date: date encoding: str delimiter: str columns: list[str] def sha256_file(path: Path, block_size: int = 1024 * 1024) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(block_size), b""): digest.update(block) return digest.hexdigest() def format_bytes(num_bytes: int) -> str: units = ("B", "KB", "MB", "GB", "TB") size = float(num_bytes) for unit in units: if size < 1024 or unit == units[-1]: return f"{size:.1f} {unit}" if unit != "B" else f"{int(size)} B" size /= 1024 return f"{num_bytes} B" def parse_source_date(path: Path) -> Optional[date]: match = DATE_FILENAME_RE.fullmatch(path.name) if not match: return None return datetime.strptime(match.group(1), "%Y-%m-%d").date() def infer_delimiter(sample: str) -> str: try: dialect = csv.Sniffer().sniff(sample, delimiters="".join(DELIMITER_CANDIDATES)) if dialect.delimiter in DELIMITER_CANDIDATES: return dialect.delimiter except csv.Error: pass lines = [line for line in sample.splitlines() if line.strip()][:25] if not lines: return "," return max(DELIMITER_CANDIDATES, key=lambda delimiter: sum(line.count(delimiter) for line in lines)) def detect_csv_format(path: Path, sample_size: int = 262_144) -> tuple[str, str]: raw = path.read_bytes()[:sample_size] for encoding in ENCODING_CANDIDATES: try: sample = raw.decode(encoding) return encoding, infer_delimiter(sample.lstrip("\ufeff")) except UnicodeDecodeError: continue sample = raw.decode("latin-1", errors="replace") return "latin-1", infer_delimiter(sample) def normalize_columns(columns: Iterable[object]) -> list[str]: normalized: list[str] = [] seen: dict[str, int] = {} for idx, column in enumerate(columns, start=1): name = str(column).replace("\ufeff", "").strip() if not name or name.lower().startswith("unnamed:"): name = f"unnamed_column_{idx}" seen[name] = seen.get(name, 0) + 1 normalized.append(name if seen[name] == 1 else f"{name}__duplicate_{seen[name]}") return normalized def read_header(path: Path, encoding: str, delimiter: str) -> list[str]: frame = pd.read_csv(path, sep=delimiter, encoding=encoding, nrows=0, dtype="string", keep_default_na=False) return normalize_columns(frame.columns) def scan_source_files(source_dir: Path) -> list[SourceFile]: files: list[SourceFile] = [] for path in sorted(source_dir.glob("*.csv")): file_date = parse_source_date(path) if file_date is None: continue encoding, delimiter = detect_csv_format(path) columns = read_header(path, encoding, delimiter) files.append(SourceFile(path=path, file_date=file_date, encoding=encoding, delimiter=delimiter, columns=columns)) if not files: raise FileNotFoundError(f"No source files matching YYYY-MM-DD.csv were found in {source_dir}") return sorted(files, key=lambda item: item.file_date) def union_columns(files: list[SourceFile]) -> list[str]: columns: list[str] = [] seen: set[str] = set() for source in files: for column in source.columns: if column not in seen: seen.add(column) columns.append(column) return columns def read_chunks(source: SourceFile, chunk_size: int): yield from pd.read_csv( source.path, sep=source.delimiter, encoding=source.encoding, dtype="string", keep_default_na=False, chunksize=chunk_size, ) def build_resources(source_dir: Path, output_dir: Path, chunk_size: int) -> None: data_dir = output_dir / "data" integrity_dir = output_dir / "integrity" data_dir.mkdir(parents=True, exist_ok=True) integrity_dir.mkdir(parents=True, exist_ok=True) files = scan_source_files(source_dir) columns = union_columns(files) output_columns = ["date", "source_file"] + columns start_date = files[0].file_date.isoformat() end_date = files[-1].file_date.isoformat() csv_path = data_dir / f"{DATASET_SHORT_NAME}_{VERSION_TAG}.csv" parquet_path = data_dir / f"{DATASET_SHORT_NAME}_{VERSION_TAG}.parquet" zip_path = data_dir / f"raw_daily_csv_{start_date}_{end_date}.zip" if csv_path.exists(): csv_path.unlink() if parquet_path.exists(): parquet_path.unlink() try: import pyarrow as pa import pyarrow.parquet as pq parquet_available = True except Exception: pa = pq = None parquet_available = False parquet_writer = None total_rows = 0 first_chunk = True try: for source in files: for chunk in read_chunks(source, chunk_size): chunk.columns = source.columns for column in columns: if column not in chunk.columns: chunk[column] = "" chunk = chunk[columns] chunk.insert(0, "source_file", source.path.name) chunk.insert(0, "date", source.file_date.isoformat()) chunk = chunk[output_columns] total_rows += len(chunk) chunk.to_csv(csv_path, mode="w" if first_chunk else "a", header=first_chunk, index=False, encoding="utf-8", lineterminator="\n") first_chunk = False if parquet_available: table = pa.Table.from_pandas(chunk.astype("string"), preserve_index=False) if parquet_writer is None: parquet_writer = pq.ParquetWriter(parquet_path, table.schema, compression="snappy") else: table = table.cast(parquet_writer.schema) parquet_writer.write_table(table) finally: if parquet_writer is not None: parquet_writer.close() with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: for source in files: archive.write(source.path, arcname=f"raw_daily_csv/{source.path.name}") manifest_path = integrity_dir / "manifest_files.csv" checksum_path = integrity_dir / "checksums_sha256.txt" rows = [] for path in [csv_path, parquet_path, zip_path]: if path.exists(): rows.append({ "relative_path": str(path.relative_to(output_dir)), "size_bytes": path.stat().st_size, "size_human": format_bytes(path.stat().st_size), "sha256": sha256_file(path), }) with manifest_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=["relative_path", "size_bytes", "size_human", "sha256"]) writer.writeheader() writer.writerows(rows) checksum_lines = [f"{row['sha256']} {row['relative_path']}" for row in rows] checksum_path.write_text("\n".join(checksum_lines) + "\n", encoding="utf-8") summary = output_dir / "integrity" / "run_summary.txt" summary.write_text( f"Dataset resource build completed.\nGenerated: {datetime.now(timezone.utc).isoformat()}\n" f"Source files: {len(files)}\nDate range: {start_date}/{end_date}\nRows consolidated: {total_rows}\n", encoding="utf-8", ) def main() -> int: parser = argparse.ArgumentParser(description="Build CKAN release data resources from daily mobility CSV files.") parser.add_argument("--source-dir", required=True, type=Path, help="Directory containing source files named YYYY-MM-DD.csv.") parser.add_argument("--output-dir", required=True, type=Path, help="Directory where release resources will be written.") parser.add_argument("--chunk-size", type=int, default=DEFAULT_CHUNK_SIZE, help="Number of rows per pandas chunk.") args = parser.parse_args() build_resources(args.source_dir.resolve(), args.output_dir.resolve(), args.chunk_size) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except Exception as exc: print(f"ERROR: {exc}", file=sys.stderr) raise SystemExit(1) from exc