diff --git a/src/clean_data_export_api/reports.py b/src/clean_data_export_api/reports.py new file mode 100644 index 0000000..432d9fa --- /dev/null +++ b/src/clean_data_export_api/reports.py @@ -0,0 +1,133 @@ +"""Write deterministic handoff report files for export runs.""" + +import csv +import json +from collections.abc import Iterable +from pathlib import Path +from typing import Any + +from pydantic import TypeAdapter + +from clean_data_export_api.models import CleanJob, ExportSummary, OutputPaths, RejectedRow + +CLEAN_COLUMNS: tuple[str, ...] = ( + "job_id", + "customer_name", + "job_status", + "work_type", + "scheduled_date", + "last_updated_at", + "source_updated_at", +) + +REJECTED_COLUMNS: tuple[str, ...] = ( + "source_row_id", + "source_job_id", + "reason_code", + "reason_detail", + "raw_record", +) + +_CLEAN_JOBS_ADAPTER = TypeAdapter(list[CleanJob]) + + +def output_paths_for(output_dir: Path) -> OutputPaths: + """Return the standard output file paths for one export directory.""" + return OutputPaths( + clean_csv=output_dir / "clean_jobs.csv", + clean_json=output_dir / "clean_jobs.json", + rejected_csv=output_dir / "rejected_jobs.csv", + run_summary=output_dir / "run_summary.md", + ) + + +def write_reports( + *, + clean_jobs: Iterable[CleanJob], + rejected_rows: Iterable[RejectedRow], + summary: ExportSummary, +) -> None: + """Write all handoff files for a completed export run.""" + clean_job_records = tuple(clean_jobs) + rejected_row_records = tuple(rejected_rows) + + output_dir = summary.output_paths.clean_csv.parent + output_dir.mkdir(parents=True, exist_ok=True) + + _write_clean_csv(summary.output_paths.clean_csv, clean_job_records) + _write_clean_json(summary.output_paths.clean_json, clean_job_records) + _write_rejected_csv(summary.output_paths.rejected_csv, rejected_row_records) + _write_run_summary(summary.output_paths.run_summary, summary) + + +def _write_clean_csv(path: Path, clean_jobs: Iterable[CleanJob]) -> None: + with path.open("w", newline="", encoding="utf-8") as file: + writer = csv.DictWriter(file, fieldnames=CLEAN_COLUMNS) + writer.writeheader() + for job in clean_jobs: + writer.writerow(_clean_job_row(job)) + + +def _write_clean_json(path: Path, clean_jobs: Iterable[CleanJob]) -> None: + clean_job_payload = _CLEAN_JOBS_ADAPTER.dump_python(list(clean_jobs), mode="json") + path.write_text(json.dumps(clean_job_payload, indent=2) + "\n", encoding="utf-8") + + +def _write_rejected_csv(path: Path, rejected_rows: Iterable[RejectedRow]) -> None: + with path.open("w", newline="", encoding="utf-8") as file: + writer = csv.DictWriter(file, fieldnames=REJECTED_COLUMNS) + writer.writeheader() + for row in rejected_rows: + writer.writerow(_rejected_row(row)) + + +def _write_run_summary(path: Path, summary: ExportSummary) -> None: + path.write_text( + "\n".join( + ( + "# Export Run Summary", + "", + f"Run ID: {summary.run_id}", + f"Requested date range: {summary.from_date} to {summary.to_date}", + f"Source records fetched: {summary.source_count}", + f"Clean records written: {summary.clean_count}", + f"Duplicates removed: {summary.duplicate_count}", + f"Rejected records written: {summary.rejected_count}", + "", + "## Output File Paths", + "", + f"- Clean CSV: `{summary.output_paths.clean_csv}`", + f"- Clean JSON: `{summary.output_paths.clean_json}`", + f"- Rejected CSV: `{summary.output_paths.rejected_csv}`", + f"- Run summary: `{summary.output_paths.run_summary}`", + "", + "## Known Limits", + "", + "- Uses fictional local fixture data only.", + "- Covers jobs and job_updates only.", + "- Does not connect to a real vendor API.", + "", + "## Suggested Next Steps", + "", + "- Review rejected rows before using the clean export.", + "- Add buyer-specific validation rules before a real integration.", + "- Review vendor API docs before replacing the fixture source.", + "", + ) + ), + encoding="utf-8", + ) + + +def _clean_job_row(job: CleanJob) -> dict[str, Any]: + return job.model_dump(mode="json") + + +def _rejected_row(row: RejectedRow) -> dict[str, str]: + return { + "source_row_id": row.source_row_id or "", + "source_job_id": row.source_job_id or "", + "reason_code": row.reason_code, + "reason_detail": row.reason_detail, + "raw_record": json.dumps(row.raw_record, sort_keys=True), + } diff --git a/src/clean_data_export_api/repository.py b/src/clean_data_export_api/repository.py new file mode 100644 index 0000000..4b3c062 --- /dev/null +++ b/src/clean_data_export_api/repository.py @@ -0,0 +1,127 @@ +"""Persist export run history in SQLite.""" + +import sqlite3 +from contextlib import closing +from datetime import UTC, date, datetime +from pathlib import Path + +from clean_data_export_api.models import OutputPaths + +EXPORT_RUN_COLUMNS: tuple[str, ...] = ( + "id", + "started_at", + "from_date", + "to_date", + "source_count", + "clean_count", + "duplicate_count", + "rejected_count", + "clean_csv_path", + "clean_json_path", + "rejected_csv_path", + "run_summary_path", +) + +_EXPORT_RUN_VALUE_COLUMNS = EXPORT_RUN_COLUMNS[1:] +_EXPORT_RUN_COLUMN_DEFINITIONS: tuple[tuple[str, str], ...] = ( + ("id", "integer primary key autoincrement"), + ("started_at", "text not null"), + ("from_date", "text not null"), + ("to_date", "text not null"), + ("source_count", "integer not null"), + ("clean_count", "integer not null"), + ("duplicate_count", "integer not null"), + ("rejected_count", "integer not null"), + ("clean_csv_path", "text not null"), + ("clean_json_path", "text not null"), + ("rejected_csv_path", "text not null"), + ("run_summary_path", "text not null"), +) + +_CREATE_EXPORT_RUNS_SQL = ( + "create table if not exists export_runs (" + + ", ".join(f"{name} {definition}" for name, definition in _EXPORT_RUN_COLUMN_DEFINITIONS) + + ")" +) +_INSERT_EXPORT_RUN_SQL = ( + f"insert into export_runs ({', '.join(_EXPORT_RUN_VALUE_COLUMNS)}) " + f"values ({', '.join('?' for _ in _EXPORT_RUN_VALUE_COLUMNS)})" +) + + +class RunRepository: + """Small SQLite repository for export run history.""" + + def __init__(self, database_path: Path) -> None: + self._database_path = database_path + + def insert_run( + self, + *, + from_date: date, + to_date: date, + source_count: int, + clean_count: int, + duplicate_count: int, + rejected_count: int, + output_paths: OutputPaths, + started_at: datetime | None = None, + ) -> int: + """Insert one export run and return its run id.""" + self._database_path.parent.mkdir(parents=True, exist_ok=True) + with closing(sqlite3.connect(self._database_path)) as connection: + _ensure_schema(connection) + cursor = connection.execute( + _INSERT_EXPORT_RUN_SQL, + _export_run_values( + from_date=from_date, + to_date=to_date, + source_count=source_count, + clean_count=clean_count, + duplicate_count=duplicate_count, + rejected_count=rejected_count, + output_paths=output_paths, + started_at=started_at or datetime.now(UTC), + ), + ) + run_id = cursor.lastrowid + if run_id is None: + msg = "SQLite did not return an export run id" + raise RuntimeError(msg) + connection.commit() + return run_id + + +def _ensure_schema(connection: sqlite3.Connection) -> None: + connection.execute(_CREATE_EXPORT_RUNS_SQL) + + +def _export_run_values( + *, + from_date: date, + to_date: date, + source_count: int, + clean_count: int, + duplicate_count: int, + rejected_count: int, + output_paths: OutputPaths, + started_at: datetime, +) -> tuple[str, str, str, int, int, int, int, str, str, str, str]: + return ( + _format_timestamp(started_at), + from_date.isoformat(), + to_date.isoformat(), + source_count, + clean_count, + duplicate_count, + rejected_count, + str(output_paths.clean_csv), + str(output_paths.clean_json), + str(output_paths.rejected_csv), + str(output_paths.run_summary), + ) + + +def _format_timestamp(value: datetime) -> str: + utc_value = value if value.tzinfo is not None else value.replace(tzinfo=UTC) + return utc_value.astimezone(UTC).isoformat().replace("+00:00", "Z") diff --git a/tests/test_reports.py b/tests/test_reports.py new file mode 100644 index 0000000..643ac04 --- /dev/null +++ b/tests/test_reports.py @@ -0,0 +1,100 @@ +import csv +import json +from datetime import UTC, date, datetime + +from clean_data_export_api.models import CleanJob, ExportSummary, RejectedRow +from clean_data_export_api.reports import ( + CLEAN_COLUMNS, + REJECTED_COLUMNS, + output_paths_for, + write_reports, +) + + +def test_output_paths_for_uses_documented_filenames(tmp_path) -> None: + paths = output_paths_for(tmp_path) + + assert paths.clean_csv == tmp_path / "clean_jobs.csv" + assert paths.clean_json == tmp_path / "clean_jobs.json" + assert paths.rejected_csv == tmp_path / "rejected_jobs.csv" + assert paths.run_summary == tmp_path / "run_summary.md" + + +def test_write_reports_creates_deterministic_handoff_files(tmp_path) -> None: + paths = output_paths_for(tmp_path / "nested" / "outputs") + clean_jobs = [_clean_job()] + rejected_rows = [_rejected_row()] + summary = ExportSummary( + run_id=1, + from_date=date(2026, 6, 1), + to_date=date(2026, 6, 5), + source_count=9, + clean_count=1, + duplicate_count=0, + rejected_count=1, + output_paths=paths, + ) + + write_reports(clean_jobs=clean_jobs, rejected_rows=rejected_rows, summary=summary) + + with paths.clean_csv.open(newline="", encoding="utf-8") as file: + clean_csv_rows = list(csv.DictReader(file)) + assert clean_csv_rows == [ + { + "job_id": "JOB-1001", + "customer_name": "Avery Plumbing", + "job_status": "scheduled", + "work_type": "repair", + "scheduled_date": "2026-06-02", + "last_updated_at": "2026-06-02T08:45:00Z", + "source_updated_at": "2026-06-01T09:15:00Z", + } + ] + assert tuple(clean_csv_rows[0]) == CLEAN_COLUMNS + + clean_json = json.loads(paths.clean_json.read_text(encoding="utf-8")) + assert clean_json == clean_csv_rows + + with paths.rejected_csv.open(newline="", encoding="utf-8") as file: + rejected_csv_rows = list(csv.DictReader(file)) + assert tuple(rejected_csv_rows[0]) == REJECTED_COLUMNS + assert rejected_csv_rows == [ + { + "source_row_id": "row-004", + "source_job_id": "", + "reason_code": "missing_job_id", + "reason_detail": "job_id is required", + "raw_record": '{"job_id": "", "source_row_id": "row-004"}', + } + ] + + run_summary = paths.run_summary.read_text(encoding="utf-8") + assert "Requested date range: 2026-06-01 to 2026-06-05" in run_summary + assert "Source records fetched: 9" in run_summary + assert "Clean records written: 1" in run_summary + assert "Duplicates removed: 0" in run_summary + assert "Rejected records written: 1" in run_summary + assert f"Clean CSV: `{paths.clean_csv}`" in run_summary + assert "Uses fictional local fixture data only." in run_summary + + +def _clean_job() -> CleanJob: + return CleanJob( + job_id="JOB-1001", + customer_name="Avery Plumbing", + job_status="scheduled", + work_type="repair", + scheduled_date=date(2026, 6, 2), + last_updated_at=datetime(2026, 6, 2, 8, 45, tzinfo=UTC), + source_updated_at=datetime(2026, 6, 1, 9, 15, tzinfo=UTC), + ) + + +def _rejected_row() -> RejectedRow: + return RejectedRow( + source_row_id="row-004", + source_job_id=None, + reason_code="missing_job_id", + reason_detail="job_id is required", + raw_record={"source_row_id": "row-004", "job_id": ""}, + ) diff --git a/tests/test_repository.py b/tests/test_repository.py new file mode 100644 index 0000000..cd783a7 --- /dev/null +++ b/tests/test_repository.py @@ -0,0 +1,86 @@ +import sqlite3 +from contextlib import closing +from datetime import UTC, date, datetime +from pathlib import Path + +from clean_data_export_api.models import OutputPaths +from clean_data_export_api.repository import EXPORT_RUN_COLUMNS, RunRepository + + +def test_insert_run_persists_export_history_with_documented_columns(tmp_path) -> None: + db_path = tmp_path / "db" / "history.sqlite3" + repository = RunRepository(db_path) + output_paths = _output_paths(tmp_path / "outputs") + started_at = datetime(2026, 6, 6, 12, 0, tzinfo=UTC) + + run_id = repository.insert_run( + from_date=date(2026, 6, 1), + to_date=date(2026, 6, 5), + source_count=9, + clean_count=3, + duplicate_count=1, + rejected_count=6, + output_paths=output_paths, + started_at=started_at, + ) + + assert run_id == 1 + with closing(sqlite3.connect(db_path)) as connection: + columns = tuple( + column[1] + for column in connection.execute("pragma table_info(export_runs)").fetchall() + ) + select_export_runs_sql = "select " + ", ".join(EXPORT_RUN_COLUMNS) + " from export_runs" + row = connection.execute(select_export_runs_sql).fetchone() + + assert columns == EXPORT_RUN_COLUMNS + assert row == ( + 1, + "2026-06-06T12:00:00Z", + "2026-06-01", + "2026-06-05", + 9, + 3, + 1, + 6, + str(output_paths.clean_csv), + str(output_paths.clean_json), + str(output_paths.rejected_csv), + str(output_paths.run_summary), + ) + + +def test_insert_run_appends_history_rows(tmp_path) -> None: + repository = RunRepository(tmp_path / "history.sqlite3") + output_paths = _output_paths(tmp_path / "outputs") + + first_run_id = repository.insert_run( + from_date=date(2026, 6, 1), + to_date=date(2026, 6, 5), + source_count=9, + clean_count=3, + duplicate_count=1, + rejected_count=6, + output_paths=output_paths, + ) + second_run_id = repository.insert_run( + from_date=date(2026, 6, 1), + to_date=date(2026, 6, 5), + source_count=9, + clean_count=3, + duplicate_count=1, + rejected_count=6, + output_paths=output_paths, + ) + + assert first_run_id == 1 + assert second_run_id == 2 + + +def _output_paths(output_dir: Path) -> OutputPaths: + return OutputPaths( + clean_csv=output_dir / "clean_jobs.csv", + clean_json=output_dir / "clean_jobs.json", + rejected_csv=output_dir / "rejected_jobs.csv", + run_summary=output_dir / "run_summary.md", + )