# edited by glg import argparse import json import sys from pathlib import Path from typing import Any, Dict REPO_ROOT = Path(__file__).resolve().parents[2] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from pypos.core.utils.path_utils import get_app_data_dir, get_db_path from pypos.modules.penjualan.services.transaksi_enterprise_control_service import ( TransaksiEnterpriseControlService, ) from pypos.modules.platform_ops.services.dr_backup_policy_service import DrBackupPolicyService from pypos.modules.platform_ops.services.dr_restore_drill_service import DrRestoreDrillService from pypos.modules.platform_ops.services.fleet_rollout_orchestrator_service import ( FleetRolloutOrchestratorService, ) def _to_bool_flag(value: Any, default: bool = False) -> bool: if isinstance(value, bool): return bool(value) if value is None: return bool(default) if isinstance(value, (int, float)): return float(value) > 0.0 text = str(value).strip().lower() if text in {"1", "true", "yes", "on"}: return True if text in {"0", "false", "no", "off"}: return False return bool(default) def parse_args(): parser = argparse.ArgumentParser( description=( "Pipeline otomatis DR + reconciliation dedup + fleet gate. " "Exit code non-zero jika ada stage kritikal gagal." ) ) parser.add_argument("--backup-root", default=str(Path(get_app_data_dir()) / "backups")) parser.add_argument("--restore-root", default=str(Path(get_app_data_dir()) / "restore_drill")) parser.add_argument("--db-path", default=str(get_db_path())) parser.add_argument("--tag", default="ops_pipeline") parser.add_argument("--keep-last", type=int, default=14) parser.add_argument("--max-age-days", type=int, default=30) parser.add_argument( "--extra-file", action="append", default=[], help="File ekstra untuk backup. Dapat dipakai berulang.", ) parser.add_argument("--run-reconciliation", type=int, default=1) parser.add_argument("--stale-inprogress-seconds", type=int, default=300) parser.add_argument("--run-retention", type=int, default=1) parser.add_argument("--retention-idempotency-days", type=int, default=30) parser.add_argument("--retention-audit-days", type=int, default=90) parser.add_argument("--retention-approval-days", type=int, default=90) parser.add_argument("--retention-outbox-days", type=int, default=30) parser.add_argument("--retention-limit", type=int, default=500) parser.add_argument("--fail-on-dead", type=int, default=1) parser.add_argument("--fleet-plan-file", default="") parser.add_argument("--fleet-wave-name", default="canary") parser.add_argument("--fleet-metrics-file", default="") parser.add_argument("--fleet-fail-threshold-pct", type=float, default=2.0) parser.add_argument("--fleet-latency-threshold-ms", type=float, default=2000.0) parser.add_argument("--output", default="") return parser.parse_args() def run_pipeline(args) -> Dict[str, Any]: db_path = str(args.db_path or "").strip() or str(get_db_path()) backup_root = str(args.backup_root or "").strip() restore_root = str(args.restore_root or "").strip() source_files = [db_path, *[str(v) for v in (args.extra_file or []) if str(v or "").strip()]] result: Dict[str, Any] = { "ok": False, "backup": {}, "restore_drill": {}, "reconciliation": {}, "fleet_gate": {}, "checks": { "backup_ok": False, "restore_ok": False, "reconciliation_ok": True, "fleet_gate_ok": True, }, "error": "", } try: backup_service = DrBackupPolicyService(backup_root) backup_payload = backup_service.run_backup(source_paths=source_files, tag=args.tag) prune_payload = backup_service.prune_backups( keep_last=max(1, int(args.keep_last or 1)), max_age_days=max(1, int(args.max_age_days or 1)), ) backup_payload["prune"] = prune_payload result["backup"] = backup_payload copied_files = list(backup_payload.get("copied_files") or []) missing_files = list(backup_payload.get("missing_files") or []) db_required_missing = db_path in missing_files result["checks"]["backup_ok"] = bool(copied_files) and not db_required_missing manifest_path = str(backup_payload.get("manifest_path") or "").strip() restore_service = DrRestoreDrillService(work_root=restore_root) restore_payload = restore_service.run_restore_drill( manifest_path=manifest_path, restore_root=restore_root, ) result["restore_drill"] = restore_payload result["checks"]["restore_ok"] = bool(restore_payload.get("ok")) if _to_bool_flag(args.run_reconciliation, default=True): retention_policy = { "enabled": _to_bool_flag(args.run_retention, default=True), "idempotency_days": max(1, int(args.retention_idempotency_days or 1)), "audit_days": max(1, int(args.retention_audit_days or 1)), "approval_days": max(1, int(args.retention_approval_days or 1)), "outbox_days": max(1, int(args.retention_outbox_days or 1)), "limit": max(1, int(args.retention_limit or 1)), } reconcile_service = TransaksiEnterpriseControlService( db_path=db_path, stale_inprogress_seconds=max(60, int(args.stale_inprogress_seconds or 60)), ) reconciliation_payload = reconcile_service.run_reconciliation( retention_policy=retention_policy, ) outbox_dead = int(reconciliation_payload.get("outbox_dead") or 0) fail_on_dead = _to_bool_flag(args.fail_on_dead, default=True) reconciliation_payload["fail_on_dead"] = bool(fail_on_dead) reconciliation_payload["gate_reason"] = ( "outbox_dead_detected" if (fail_on_dead and outbox_dead > 0) else "ok" ) result["reconciliation"] = reconciliation_payload result["checks"]["reconciliation_ok"] = not (fail_on_dead and outbox_dead > 0) else: result["reconciliation"] = {"skipped": True, "reason": "run_reconciliation_disabled"} result["checks"]["reconciliation_ok"] = True fleet_plan_file = str(args.fleet_plan_file or "").strip() fleet_metrics_file = str(args.fleet_metrics_file or "").strip() if fleet_plan_file and fleet_metrics_file: orchestrator = FleetRolloutOrchestratorService() fleet_payload = orchestrator.evaluate_wave_files( plan_file=fleet_plan_file, wave_name=str(args.fleet_wave_name or "canary").strip() or "canary", metrics_file=fleet_metrics_file, fail_threshold_pct=float(args.fleet_fail_threshold_pct or 2.0), latency_threshold_ms=float(args.fleet_latency_threshold_ms or 2000.0), ) result["fleet_gate"] = fleet_payload result["checks"]["fleet_gate_ok"] = bool(fleet_payload.get("healthy")) and ( str(fleet_payload.get("decision") or "").strip() == "continue" ) else: result["fleet_gate"] = { "skipped": True, "reason": "missing_plan_or_metrics", "plan_file": fleet_plan_file, "metrics_file": fleet_metrics_file, } result["checks"]["fleet_gate_ok"] = True result["ok"] = all(bool(v) for v in (result.get("checks") or {}).values()) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError) as exc: result["ok"] = False result["error"] = str(exc) return result def main() -> int: args = parse_args() payload = run_pipeline(args) serialized = json.dumps(payload, ensure_ascii=False, indent=2) if args.output: Path(str(args.output)).write_text(serialized, encoding="utf-8") print(serialized) return 0 if bool(payload.get("ok")) else 2 if __name__ == "__main__": raise SystemExit(main())