File size: 3,069 Bytes
b75c637 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """
Export module — writes final artifacts (JSON summary, per-stage outputs).
Engine contract:
run(EngineInput) -> EngineOutput
The main report generation (HTML, CSV, JSONL) is handled by
``engine.reporting``. This module produces a supplementary JSON
summary artifact in the run directory.
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Any, Dict
from engine.io_contract import (
Artifact,
EngineInput,
EngineOutput,
StageStatus,
)
logger = logging.getLogger("modules.export")
def run(engine_input: EngineInput) -> EngineOutput:
"""
Export a JSON summary of the pipeline run.
"""
try:
run_dir = Path(engine_input.run_dir)
exports_dir = run_dir / "exports"
exports_dir.mkdir(parents=True, exist_ok=True)
# Build summary
summary_data = {
"run_id": engine_input.run_id,
"total_records": len(engine_input.records),
"input_files": [
{
"name": f.path.name,
"type": f.file_type.value,
"size": f.size_bytes,
}
for f in engine_input.input_spec.files
],
"record_sample": [
{
"row_id": r.row_id,
"source": r.source_file,
"entity_name": r.entity_name,
"entity_email": r.entity_email,
"entity_ip": r.entity_ip,
}
for r in engine_input.records[:20]
],
}
out_path = exports_dir / "summary.json"
out_path.write_text(
json.dumps(summary_data, indent=2, ensure_ascii=False, default=str),
encoding="utf-8",
)
logger.info("Exported summary to %s", out_path)
return EngineOutput(
stage="export",
status=StageStatus.SUCCESS,
records=engine_input.records, # pass through
artifacts=[
Artifact(
name="summary.json",
path=out_path,
mime_type="application/json",
description="Pipeline run summary",
)
],
summary=f"Exported summary.json ({len(engine_input.records)} records)",
)
except Exception as exc:
logger.error("Export failed: %s", exc, exc_info=True)
return EngineOutput(
stage="export",
status=StageStatus.FAILED,
error=str(exc),
)
# ---------------------------------------------------------------------------
# Legacy compatibility
# ---------------------------------------------------------------------------
def export(data: Any, outpath: str) -> str:
"""Legacy wrapper (deprecated). Use ``run()`` instead."""
with open(outpath, "w") as f:
f.write(str(data))
return outpath
if __name__ == "__main__":
print(export({"result": 123}, "output.txt"))
|