scripts/record_product_flows.py

from __future__ import annotations

import json
import os
import sys
import time
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path

BASE_URL = os.environ.get("KINK_AUDIT_BASE_URL", "http://127.0.0.1:8011").rstrip("/")
ARTIFACT_DIR = Path(os.environ.get("KINK_AUDIT_ARTIFACT_DIR", "/tmp/kink_product_recordings"))
ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)


def load_product_flow_module():
    path = Path(__file__).resolve().parent / "playwright_product_flow.py"
    spec = spec_from_file_location("playwright_product_flow", path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Could not load {path}")
    module = module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def main() -> int:
    flow = load_product_flow_module()
    flow.BASE_URL = BASE_URL
    results = flow.run_selected_scenarios(["all"], headless=True)
    payload = flow.results_to_json(results)
    payload["generated_at"] = time.time()
    payload["base_url"] = BASE_URL
    out_path = ARTIFACT_DIR / "record_product_flows.json"
    out_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
    print(json.dumps({"recording_json": str(out_path), "ok": payload["ok"], "summary": payload["summary"]}, indent=2))
    return 0 if payload["ok"] else 1


if __name__ == "__main__":
    raise SystemExit(main())