130 lines
4.1 KiB
Python
130 lines
4.1 KiB
Python
from pathlib import Path
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from typing import Optional
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.services.training import generate_public_training_case_drafts
|
|
|
|
|
|
DEFAULT_FIXTURE = ROOT / "tests" / "fixtures" / "training_case_generator_seed_v1.json"
|
|
|
|
|
|
class _PublicCliError(Exception):
|
|
def __init__(self, code: str) -> None:
|
|
super().__init__(code)
|
|
self.code = code
|
|
|
|
|
|
def run_generator(input_path: Path, output_path: Optional[Path] = None) -> dict:
|
|
payload = _load_payload(input_path)
|
|
generated = generate_public_training_case_drafts(payload)
|
|
|
|
if output_path is not None:
|
|
_write_output(output_path, generated)
|
|
|
|
event_types = sorted(
|
|
{
|
|
item.get("scenario", {}).get("source_event_type")
|
|
for item in generated.get("items", [])
|
|
if isinstance(item, dict)
|
|
}
|
|
)
|
|
return {
|
|
"passed": True,
|
|
"input": _display_path(input_path),
|
|
"output": _display_path(output_path) if output_path is not None else None,
|
|
"generated": len(generated.get("items", [])),
|
|
"event_types": event_types,
|
|
}
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
parser = _build_parser()
|
|
args = parser.parse_args(argv)
|
|
input_path = Path(args.input)
|
|
output_path = Path(args.output) if args.output else None
|
|
try:
|
|
summary = run_generator(input_path=input_path, output_path=output_path)
|
|
except (_PublicCliError, OSError, ValueError, json.JSONDecodeError) as exc:
|
|
print(json.dumps(_error_summary(input_path, output_path, exc), ensure_ascii=False))
|
|
return 2
|
|
|
|
print(json.dumps(summary, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="scripts/run_training_case_generator.py",
|
|
description="Offline prototype: generate public Training case drafts from trading-derived seed events.",
|
|
epilog=(
|
|
"Reads a local seed JSON package and writes a public draft package when --output is set. "
|
|
"The prototype is read-only and does not write to database-backed Training tables."
|
|
),
|
|
)
|
|
parser.add_argument("--input", default=str(DEFAULT_FIXTURE), help=f"Seed JSON path (default: {DEFAULT_FIXTURE.relative_to(ROOT)}).")
|
|
parser.add_argument("--output", help="Optional output path for generated public draft package JSON.")
|
|
return parser
|
|
|
|
|
|
def _load_payload(input_path: Path) -> dict:
|
|
with input_path.open("r", encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("invalid_json_body")
|
|
return payload
|
|
|
|
|
|
def _write_output(output_path: Path, payload: dict) -> None:
|
|
try:
|
|
if output_path.exists() and output_path.is_dir():
|
|
raise _PublicCliError("output_path_is_directory")
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_text(
|
|
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
except _PublicCliError:
|
|
raise
|
|
except OSError as exc:
|
|
raise _PublicCliError("output_path_unwritable") from exc
|
|
|
|
|
|
def _display_path(path: Optional[Path]) -> Optional[str]:
|
|
if path is None:
|
|
return None
|
|
try:
|
|
return str(path.resolve().relative_to(ROOT))
|
|
except ValueError:
|
|
return str(path)
|
|
|
|
|
|
def _public_error(exc: Exception) -> str:
|
|
if isinstance(exc, _PublicCliError):
|
|
return exc.code
|
|
if isinstance(exc, IsADirectoryError):
|
|
return "input_path_is_directory"
|
|
if isinstance(exc, FileNotFoundError):
|
|
return "input_file_not_found"
|
|
if isinstance(exc, json.JSONDecodeError):
|
|
return "invalid_json_body"
|
|
return str(exc)
|
|
|
|
|
|
def _error_summary(input_path: Path, output_path: Optional[Path], exc: Exception) -> dict:
|
|
return {
|
|
"passed": False,
|
|
"input": _display_path(input_path),
|
|
"output": _display_path(output_path),
|
|
"error": _public_error(exc),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|