Automating Dublin Core Enrichment from Raw CSV: Precision Mapping, Validation, and FAIR Compliance in Pandas Pipelines #

import asyncio
import csv
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, List, Dict, Any, AsyncIterator
import pandas as pd
from pydantic import BaseModel, field_validator, ValidationError, ConfigDict

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("dc_enrichment_pipeline")

# Controlled vocabulary for institutional compliance
SUBJECT_VOCABULARY = {"genomics", "proteomics", "climate", "materials", "neuroscience"}

class DublinCoreRecord(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    
    identifier: str
    title: str
    creator: str
    date: datetime
    description: Optional[str] = None
    subject: Optional[List[str]] = None
    license: Optional[str] = None

    @field_validator("date", mode="before")
    @classmethod
    def parse_iso8601(cls, v: Any) -> datetime:
        if isinstance(v, str):
            cleaned = v.strip().replace("Z", "+00:00")
            try:
                return datetime.fromisoformat(cleaned)
            except ValueError as e:
                raise ValueError(f"Invalid ISO-8601 date format: {v}") from e
        if isinstance(v, (int, float)):
            return datetime.fromtimestamp(v, tz=timezone.utc)
        raise ValueError("Date must be ISO-8601 string or numeric timestamp")

    @field_validator("subject", mode="before")
    @classmethod
    def normalize_subjects(cls, v: Any) -> Optional[List[str]]:
        if v is None:
            return None
        if isinstance(v, str):
            terms = [s.strip().lower() for s in v.split(";") if s.strip()]
            invalid = [t for t in terms if t not in SUBJECT_VOCABULARY]
            if invalid:
                raise ValueError(f"Non-compliant subjects: {invalid}")
            return terms
        if isinstance(v, list):
            return [str(s).strip().lower() for s in v]
        raise ValueError("Subject must be string or list")

# Header alias mapping
HEADER_MAP = {
    "id": "identifier", "uid": "identifier", "record_id": "identifier",
    "title": "title", "name": "title", "dataset_name": "title",
    "author": "creator", "auth": "creator", "pi": "creator", "creator": "creator",
    "date": "date", "dt": "date", "created_at": "date", "timestamp": "date",
    "desc": "description", "notes": "description", "summary": "description",
    "tags": "subject", "keywords": "subject", "category": "subject",
    "rights": "license", "license": "license", "access": "license"
}

def process_chunk(chunk: pd.DataFrame, chunk_idx: int) -> Dict[str, Any]:
    """Validate chunk, isolate errors, return compliant records and error metrics."""
    compliant_records = []
    errors = []
    
    # Normalize headers
    normalized_cols = []
    for col in chunk.columns:
        clean = col.strip().lower()
        normalized_cols.append(HEADER_MAP.get(clean, clean))
    chunk.columns = normalized_cols
    
    required_cols = {"identifier", "title", "creator", "date"}
    missing = required_cols - set(chunk.columns)
    if missing:
        raise ValueError(f"Missing required DC columns: {missing}")
    
    # Retain only canonical DC fields so unmapped instrument columns do not
    # trip the model's extra="forbid" guard.
    dc_fields = set(DublinCoreRecord.model_fields)
    dc_cols = [c for c in chunk.columns if c in dc_fields]
    
    for idx, row in chunk[dc_cols].iterrows():
        try:
            record = DublinCoreRecord.model_validate(row.to_dict())
            compliant_records.append(record.model_dump(mode="json"))
        except ValidationError as e:
            first_error = e.errors()[0] if e.errors() else {}
            errors.append({
                "chunk_index": chunk_idx,
                "row_index": int(idx),
                "error_type": first_error.get("type", "validation_failure"),
                "message": first_error.get("msg", str(e)),
                "raw_data": row.to_dict()
            })
            
    return {"records": compliant_records, "errors": errors}

async def stream_csv_chunks(filepath: Path, chunk_size: int = 10000) -> AsyncIterator[pd.DataFrame]:
    """Memory-bounded CSV ingestion with strict quoting and dtype hints."""
    reader = pd.read_csv(
        filepath,
        chunksize=chunk_size,
        quoting=csv.QUOTE_ALL,
        keep_default_na=False,
        dtype=str,
        low_memory=True
    )
    for chunk in reader:
        yield chunk

async def run_enrichment_pipeline(input_path: Path, output_path: Path) -> None:
    """Orchestrate async batch processing, validation, and drift logging."""
    logger.info(f"Initializing pipeline for {input_path}")
    
    all_records = []
    all_errors = []
    chunk_idx = 0
    baseline_null_ratios = {}
    
    async for chunk in stream_csv_chunks(input_path):
        # Drift detection: track null ratios per column
        null_ratios = (chunk.isna().sum() / len(chunk)).to_dict()
        for col, ratio in null_ratios.items():
            baseline_null_ratios.setdefault(col, []).append(ratio)
            
        try:
            result = await asyncio.to_thread(process_chunk, chunk, chunk_idx)
            all_records.extend(result["records"])
            all_errors.extend(result["errors"])
        except Exception as e:
            logger.error(f"Chunk {chunk_idx} failed catastrophically: {e}")
            all_errors.append({"chunk_index": chunk_idx, "error_type": "chunk_parse_failure", "message": str(e)})
            
        chunk_idx += 1
        if chunk_idx % 5 == 0:
            logger.info(f"Processed {chunk_idx} chunks. Compliant: {len(all_records)}, Errors: {len(all_errors)}")

    # Persist compliant records (blocking file I/O offloaded to a worker thread)
    def write_output():
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(all_records, f, indent=2, ensure_ascii=False)
            
    await asyncio.to_thread(write_output)
    
    # Log validation failures and drift metrics
    logger.warning(f"Pipeline complete. {len(all_errors)} records quarantined.")
    if all_errors:
        error_log_path = output_path.with_suffix(".validation_errors.json")
        with open(error_log_path, "w") as ef:
            json.dump(all_errors, ef, indent=2)
            
    # Drift summary
    drift_summary = {
        col: {"mean_null_ratio": sum(ratios)/len(ratios), "max_null_ratio": max(ratios)}
        for col, ratios in baseline_null_ratios.items()
    }
    logger.info(f"Metadata drift metrics: {json.dumps(drift_summary)}")

if __name__ == "__main__":
    INPUT_CSV = Path("raw_lab_export.csv")
    OUTPUT_JSON = Path("dublin_core_enriched.json")
    asyncio.run(run_enrichment_pipeline(INPUT_CSV, OUTPUT_JSON))