cloudinit-bot
f09c46bde6
---ci--- project: inboxos phase: 1 milestone: v1.4 status: execute wave: 1 task: T4 ---/ci---
71 lines
2.0 KiB
Python
Executable File
71 lines
2.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Dry-run a single .eml through classify+extract; no writes.
|
|
|
|
v1.4 REQ-INB-51.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
def _mock_extract(eml_path: str) -> dict[str, Any]:
|
|
"""Return a mock record when INBOXOS_LLM_MOCK=1 (offline-test path)."""
|
|
return {
|
|
"_mock": True,
|
|
"Message-ID": "mock-mid",
|
|
"module": "intel",
|
|
"sender": "mock@example.com",
|
|
"subject": f"Mock for {os.path.basename(eml_path)}",
|
|
"received_at": "2026-06-08T00:00:00Z",
|
|
"priority": "low",
|
|
}
|
|
|
|
|
|
def dry_run(eml_path: str, dry_run: bool = True) -> dict[str, Any]:
|
|
"""Process one .eml. If dry_run, suppress all side effects."""
|
|
if not os.path.exists(eml_path):
|
|
sys.stderr.write(f"File not found: {eml_path}\n")
|
|
return {"error": "file_not_found", "path": eml_path}
|
|
|
|
if os.environ.get("INBOXOS_LLM_MOCK", "0") == "1":
|
|
record = _mock_extract(eml_path)
|
|
else:
|
|
sys.stderr.write(
|
|
"Live LLM path not implemented in dry_run; set INBOXOS_LLM_MOCK=1 for offline-test mode\n"
|
|
)
|
|
return {"error": "live_path_not_implemented"}
|
|
|
|
if dry_run:
|
|
record["_dry_run"] = True
|
|
record["_write_attempted"] = False
|
|
else:
|
|
record["_write_attempted"] = True
|
|
return record
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="Dry-run one .eml through classify+extract")
|
|
p.add_argument("--eml", required=True, help="Path to .eml file")
|
|
p.add_argument("--no-dry-run", action="store_true", help="Allow writes (default: dry-run)")
|
|
args = p.parse_args()
|
|
|
|
result = dry_run(args.eml, dry_run=not args.no_dry_run)
|
|
if "error" in result:
|
|
return 1
|
|
print(json.dumps(result, indent=2))
|
|
print(
|
|
f"trace: module={result.get('module', '?')} "
|
|
f"sender={result.get('sender', '?')} "
|
|
f"mid={result.get('Message-ID', '?')}",
|
|
file=sys.stderr,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|