Parsing EDI 852 files with Python pandas

import pandas as pd
import re
import logging
import json
from pathlib import Path
from typing import Iterator, Dict, List, Optional
from datetime import datetime, timezone
from dataclasses import dataclass, asdict

# HIPAA/DEA-compliant structured logging configuration
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler()]
)

@dataclass
class AuditEvent:
    timestamp_utc: str
    segment_id: str
    action: str
    compliance_flag: Optional[str] = None
    details: Optional[str] = None

def _log_audit(event: AuditEvent) -> None:
    """Serialize audit events to JSON for immutable compliance logging."""
    logging.info(json.dumps(asdict(event), default=str))

def normalize_ndc(raw_ndc: str) -> str:
    """Strip trailing check digit, remove hyphens, enforce 10-digit FDA standard."""
    cleaned = raw_ndc.replace("-", "").strip()
    if len(cleaned) == 11:
        return cleaned[:10]
    if len(cleaned) == 10:
        return cleaned
    raise ValueError(f"Non-compliant NDC length: {len(cleaned)} digits")

def normalize_timestamp(raw_ts: str, qualifier: str) -> str:
    """Convert X12 DTM (CCYYMMDD or CCYYMMDDHHMM) to ISO8601 UTC with ms precision."""
    if len(raw_ts) == 8:
        dt = datetime.strptime(raw_ts, "%Y%m%d")
    elif len(raw_ts) == 12:
        dt = datetime.strptime(raw_ts, "%Y%m%d%H%M")
    else:
        raise ValueError(f"Invalid DTM format: {raw_ts}")
    return dt.replace(tzinfo=timezone.utc).isoformat(timespec="milliseconds")

UOM_TO_BASE = {"EA": 1.0, "PK": 10.0, "BX": 100.0, "CS": 500.0}

def parse_edi_852_segments(
    file_path: str, 
    chunk_size: int = 5000,
    quarantine_path: str = "quarantine_852.json"
) -> Iterator[pd.DataFrame]:
    """
    Generator yielding pandas DataFrames from EDI 852 files.
    Implements deterministic normalization, UOM conversion, and audit logging.
    """
    segment_pattern = re.compile(r"^([A-Z]{2,3})\*(.+?)\*?~$")
    buffer: List[Dict] = []
    quarantine: List[Dict] = []
    
    _log_audit(AuditEvent(
        timestamp_utc=datetime.now(timezone.utc).isoformat(),
        segment_id="INIT",
        action="FILE_OPENED",
        details=file_path
    ))

    with open(file_path, "r", encoding="utf-8-sig") as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
                
            match = segment_pattern.match(line)
            if not match:
                quarantine.append({"line": line_num, "raw": line, "error": "MALFORMED_SEGMENT"})
                continue

            seg_id, payload = match.groups()
            elements = payload.split("*")
            
            if seg_id == "LIN":
                # Reset transaction buffer on new item
                if buffer:
                    yield pd.DataFrame(buffer)
                    buffer.clear()
                try:
                    ndc = normalize_ndc(elements[2])
                except ValueError as e:
                    quarantine.append({"line": line_num, "raw": line, "error": str(e)})
                    continue
                buffer.append({"ndc": ndc, "line": line_num})
                
            elif seg_id == "QTY" and buffer:
                try:
                    qty_type = elements[1]
                    raw_qty = float(elements[2])
                    uom = elements[3] if len(elements) > 3 else "EA"
                    base_multiplier = UOM_TO_BASE.get(uom, 1.0)
                    buffer[-1].update({
                        f"qty_{qty_type.lower()}": raw_qty,
                        f"base_{qty_type.lower()}": round(raw_qty * base_multiplier, 4),
                        "uom": uom
                    })
                except (IndexError, ValueError) as e:
                    quarantine.append({"line": line_num, "raw": line, "error": str(e)})
                    
            elif seg_id == "DTM" and buffer:
                try:
                    qualifier = elements[1]
                    ts = normalize_timestamp(elements[2], qualifier)
                    buffer[-1].update({f"dtm_{qualifier}": ts})
                except (IndexError, ValueError) as e:
                    quarantine.append({"line": line_num, "raw": line, "error": str(e)})

        if buffer:
            yield pd.DataFrame(buffer)

    if quarantine:
        Path(quarantine_path).write_text(json.dumps(quarantine, indent=2), encoding="utf-8")
        _log_audit(AuditEvent(
            timestamp_utc=datetime.now(timezone.utc).isoformat(),
            segment_id="QUARANTINE",
            action="MALFORMED_RECORDS_ROUTED",
            details=f"{len(quarantine)} records quarantined to {quarantine_path}"
        ))

from datetime import datetime, timezone
import pandas as pd

# Defined elsewhere on this page (see the surrounding blocks):
# - AuditEvent
# - _log_audit

def validate_compliance_chunk(df: pd.DataFrame) -> pd.DataFrame:
    """
    Enforce DEA/FDA validation rules on ingested chunks.
    Flags discrepancies, enforces non-negative base units, and logs violations.
    """
    required_cols = {"ndc", "qty_onhand", "base_onhand"}
    missing = required_cols - set(df.columns)
    if missing:
        raise RuntimeError(f"Missing compliance-critical columns: {missing}")

    # DEA 21 CFR 1304: Base units must be >= 0 for active inventory
    df["compliance_flag"] = None
    invalid_mask = df["base_onhand"] < 0
    df.loc[invalid_mask, "compliance_flag"] = "NEGATIVE_BASE_UNIT_DEA_FLAG"

    # FDA DSCSA: NDC format validation
    df.loc[~df["ndc"].str.match(r"^\d{10}$"), "compliance_flag"] = "NDC_FORMAT_MISMATCH"

    # Log violations for audit trail
    violations = df[df["compliance_flag"].notna()]
    if not violations.empty:
        _log_audit(AuditEvent(
            timestamp_utc=datetime.now(timezone.utc).isoformat(),
            segment_id="VALIDATION",
            action="COMPLIANCE_VIOLATIONS_DETECTED",
            compliance_flag="MULTI",
            details=f"{len(violations)} records failed validation"
        ))
        
    return df.dropna(subset=["ndc", "base_onhand"])