105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
"""Per-run output layout: one JSON file per job plus a daily run report.
|
|
|
|
The single-field CLI wrote `output/<date>_field<id>.json`, which collides as
|
|
soon as one field has two diseases. The batch instead writes one file per
|
|
(field, disease) pair under a per-day directory, plus a `_run_report.json`
|
|
that is the actual operational artifact: it says whether the morning's run
|
|
succeeded, and if not, exactly which jobs failed and why.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from pipeline.job import JobResult, json_default
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Keys always present in the report's totals, even at zero, so a dashboard or
|
|
# script reading the report does not need to guess which statuses exist.
|
|
_STATUSES = (
|
|
"ok",
|
|
"no_risk",
|
|
"dry_run",
|
|
"failed",
|
|
"skipped_existing",
|
|
"skipped_deadline",
|
|
)
|
|
|
|
|
|
def run_dir(output_dir: Path, as_of: date) -> Path:
|
|
return output_dir / as_of.isoformat()
|
|
|
|
|
|
def write_job_output(output_dir: Path, as_of: date, payload: dict[str, Any]) -> Path:
|
|
"""Write one job's full payload to output/<as_of>/<output_stem>.json."""
|
|
target_dir = run_dir(output_dir, as_of)
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
path = target_dir / f"{payload['output_stem']}.json"
|
|
with path.open("w", encoding="utf-8") as fh:
|
|
json.dump(payload, fh, ensure_ascii=False, indent=2, default=json_default)
|
|
return path
|
|
|
|
|
|
def _job_report_entry(result: JobResult) -> dict[str, Any]:
|
|
job = result.job
|
|
return {
|
|
"field_id": job.field_id,
|
|
"crop": job.crop_english,
|
|
"disease": job.disease_english,
|
|
"organic": job.organic,
|
|
"station": job.station,
|
|
"status": result.status,
|
|
"duration_s": round(result.duration_s, 2),
|
|
"recommended_count": result.recommended_count,
|
|
"allowed_products_empty": result.allowed_products_empty,
|
|
"message": result.message,
|
|
"error_type": result.error_type,
|
|
"error_message": result.error_message,
|
|
}
|
|
|
|
|
|
def write_run_report(
|
|
output_dir: Path,
|
|
as_of: date,
|
|
results: list[JobResult],
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Write output/<as_of>/_run_report.json: totals by status plus per-job
|
|
diagnostics.
|
|
|
|
This is the operational artifact for the daily run: it names exactly
|
|
which jobs succeeded, which were skipped and why, and which failed and
|
|
with what error, so the exit code alone never has to carry that
|
|
information. `ok_with_empty_allowed_products` is tracked separately from
|
|
a bare count of `ok` jobs because, once the `products` table has organic
|
|
entries for every crop, a sudden jump in that number is the fastest
|
|
signal that the prefilter or the Weaviate allowlist filter has regressed
|
|
(see pipeline/stages/vector.py).
|
|
"""
|
|
totals = {status: 0 for status in _STATUSES}
|
|
for result in results:
|
|
totals[result.status] = totals.get(result.status, 0) + 1
|
|
|
|
empty_allowed_products = sum(1 for r in results if r.allowed_products_empty)
|
|
|
|
report: dict[str, Any] = {
|
|
"as_of": as_of.isoformat(),
|
|
"job_count": len(results),
|
|
"totals": totals,
|
|
"ok_with_empty_allowed_products": empty_allowed_products,
|
|
"jobs": [_job_report_entry(r) for r in results],
|
|
}
|
|
|
|
target_dir = run_dir(output_dir, as_of)
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
path = target_dir / "_run_report.json"
|
|
with path.open("w", encoding="utf-8") as fh:
|
|
json.dump(report, fh, ensure_ascii=False, indent=2)
|
|
logger.info("Wrote run report -> %s", path)
|
|
return report
|