# edited by glg import ast import json import sys from dataclasses import dataclass from pathlib import Path from typing import Dict, Iterable, List, Optional, Sequence, Tuple REPO_ROOT = Path(__file__).resolve().parents[2] PENJUALAN_ROOT = REPO_ROOT / "pypos" / "modules" / "penjualan" RUBRIC_PATH = REPO_ROOT / "docs" / "quality" / "penjualan_rubric_v100.json" PYTEST_INI_PATH = REPO_ROOT / "pytest.ini" QUALITY_GATE_PATH = REPO_ROOT / "scripts" / "ci" / "run_quality_gate.py" if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) @dataclass(frozen=True) class FunctionBudget: file_path: str class_name: str function_name: str max_lines: int @dataclass(frozen=True) class KpiResult: kpi_id: str name: str weight: int score: int issues: Tuple[str, ...] FUNCTION_BUDGETS: Tuple[FunctionBudget, ...] = ( FunctionBudget( file_path="pypos/modules/penjualan/models/load_transaksi_model.py", class_name="LoadTransaksiModel", function_name="delete_transaksi_by_id", max_lines=70, ), FunctionBudget( file_path="pypos/modules/penjualan/models/load_transaksi_model.py", class_name="LoadTransaksiModel", function_name="_prepare_delete_payload", max_lines=70, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_counter_service.py", class_name="TransaksiCounterService", function_name="get_and_increment_counter", max_lines=65, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_counter_service.py", class_name="TransaksiCounterService", function_name="_increment_counter_atomic", max_lines=45, ), FunctionBudget( file_path="pypos/modules/penjualan/controllers/transaksi_penjualan_controller.py", class_name="TransaksiPenjualanController", function_name="__init__", max_lines=40, ), FunctionBudget( file_path="pypos/modules/penjualan/controllers/transaksi_penjualan_controller.py", class_name="TransaksiPenjualanController", function_name="_init_penjualan_services", max_lines=85, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_payload_service.py", class_name="TransaksiPayloadService", function_name="build_transaksi_payload", max_lines=60, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_payload_service.py", class_name="TransaksiPayloadService", function_name="_build_financial_context", max_lines=55, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_payload_service.py", class_name="TransaksiPayloadService", function_name="_build_transaksi_data_dict", max_lines=95, ), FunctionBudget( file_path="pypos/modules/penjualan/views/index.py", class_name="TransaksiPenjualanView", function_name="build_barang_table_and_diskon_info", max_lines=50, ), FunctionBudget( file_path="pypos/modules/penjualan/views/index.py", class_name="TransaksiPenjualanView", function_name="_setup_info_pembayaran_panel", max_lines=75, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_lookup_harga_service.py", class_name="TransaksiLookupHargaService", function_name="cari_barang_by_id", max_lines=55, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_lookup_harga_service.py", class_name="TransaksiLookupHargaService", function_name="_apply_diskon_grosir", max_lines=45, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_lookup_harga_service.py", class_name="TransaksiLookupHargaService", function_name="_apply_diskon_free", max_lines=45, ), FunctionBudget( file_path="pypos/modules/penjualan/services/transaksi_simpan_use_case_service.py", class_name="TransaksiSimpanUseCaseService", function_name="execute", max_lines=70, ), FunctionBudget( file_path="pypos/modules/penjualan/controllers/transaksi_penjualan_controller.py", class_name="TransaksiPenjualanController", function_name="_handle_persist_error", max_lines=30, ), FunctionBudget( file_path="pypos/modules/penjualan/models/settlement_model.py", class_name="SettlementModel", function_name="simpan_settlement", max_lines=155, ), FunctionBudget( file_path="pypos/modules/penjualan/models/settlement_model.py", class_name="SettlementModel", function_name="set_settlement", max_lines=210, ), FunctionBudget( file_path="pypos/modules/penjualan/models/settlement_model.py", class_name="SettlementModel", function_name="execute_settlement_atomic", max_lines=165, ), FunctionBudget( file_path="pypos/modules/penjualan/models/return_model.py", class_name="ReturnModel", function_name="insert_return", max_lines=150, ), FunctionBudget( file_path="pypos/modules/penjualan/controllers/return_controller.py", class_name="ReturnController", function_name="proses_return", max_lines=105, ), FunctionBudget( file_path="pypos/modules/penjualan/services/settlement_orchestrator_service.py", class_name="SettlementOrchestratorService", function_name="execute_settlement", max_lines=185, ), ) def _read_text(path: Path) -> str: return path.read_text(encoding="utf-8-sig") def _read_json(path: Path) -> Dict[str, object]: return json.loads(_read_text(path)) def _iter_penjualan_python_files(root: Path = PENJUALAN_ROOT) -> Iterable[Path]: for path in sorted(root.rglob("*.py")): if "__pycache__" in path.parts: continue yield path def _extract_import_target(node: ast.AST) -> str: if isinstance(node, ast.Import): aliases = [str(alias.name or "").strip() for alias in node.names] return ",".join([name for name in aliases if name]) if isinstance(node, ast.ImportFrom): return str(node.module or "").strip() return "" def _parse_python(path: Path) -> ast.AST: return ast.parse(_read_text(path)) def _resolve_function_node( abs_path: Path, class_name: str, function_name: str, ) -> Optional[ast.FunctionDef]: tree = _parse_python(abs_path) for node in ast.walk(tree): if not isinstance(node, ast.ClassDef) or node.name != class_name: continue for child in node.body: if isinstance(child, ast.FunctionDef) and child.name == function_name: return child return None def _resolve_function_length( abs_path: Path, class_name: str, function_name: str, ) -> int: node = _resolve_function_node(abs_path, class_name, function_name) if node is None: return -1 end_lineno = int(getattr(node, "end_lineno", node.lineno)) return int(end_lineno - node.lineno + 1) def _contains_generic_catch(function_node: ast.FunctionDef) -> bool: for node in ast.walk(function_node): if not isinstance(node, ast.ExceptHandler): continue if node.type is None: return True if isinstance(node.type, ast.Name) and str(node.type.id) == "Exception": return True return False def _collect_test_names(test_files: Sequence[Path]) -> List[str]: names: List[str] = [] for file_path in test_files: if not file_path.exists(): continue try: tree = _parse_python(file_path) except SyntaxError: continue for node in ast.walk(tree): if isinstance(node, ast.FunctionDef) and str(node.name).startswith("test_"): names.append(str(node.name)) return names def _to_bool(value) -> bool: return bool(value) def _compute_weighted_score(weight: int, passed_count: int, total_count: int) -> int: if total_count <= 0: return 0 if passed_count >= total_count: return int(weight) ratio = float(passed_count) / float(total_count) return int(round(float(weight) * ratio)) def check_forbidden_imports(root: Path = PENJUALAN_ROOT) -> List[str]: violations: List[str] = [] for path in _iter_penjualan_python_files(root): try: tree = _parse_python(path) except SyntaxError as exc: violations.append(f"{path}: syntax error saat parse import ({exc})") continue rel = path.relative_to(REPO_ROOT).as_posix() is_model = "/models/" in f"/{rel}" is_service = "/services/" in f"/{rel}" is_view = "/views/" in f"/{rel}" for node in ast.walk(tree): target = _extract_import_target(node) target_lower = str(target or "").strip().lower() if not target_lower: continue if "endpoint" in target_lower or target_lower.startswith("endpoint"): violations.append(f"{rel}: import endpoint terdeteksi ({target})") if is_service and ".controllers." in target_lower: violations.append(f"{rel}: service tidak boleh import controller ({target})") if is_service and ".views." in target_lower: violations.append(f"{rel}: service tidak boleh import view ({target})") if is_model and ".controllers." in target_lower: violations.append(f"{rel}: model tidak boleh import controller ({target})") if is_model and ".services." in target_lower: violations.append(f"{rel}: model tidak boleh import service ({target})") if is_model and ".views." in target_lower: violations.append(f"{rel}: model tidak boleh import view ({target})") if is_view and ".controllers." in target_lower: violations.append(f"{rel}: view tidak boleh import controller ({target})") if is_view and ".services." in target_lower: violations.append(f"{rel}: view tidak boleh import service ({target})") if is_view and ".models." in target_lower: violations.append(f"{rel}: view tidak boleh import model langsung ({target})") return violations def check_function_budgets(root: Path = REPO_ROOT) -> List[str]: violations: List[str] = [] for budget in FUNCTION_BUDGETS: abs_path = root / budget.file_path if not abs_path.exists(): violations.append(f"{budget.file_path}: file budget tidak ditemukan") continue length = _resolve_function_length(abs_path, budget.class_name, budget.function_name) if length < 0: violations.append( f"{budget.file_path}: fungsi {budget.class_name}.{budget.function_name} tidak ditemukan" ) continue if length > int(budget.max_lines): violations.append( f"{budget.file_path}: {budget.class_name}.{budget.function_name}={length} " f"melebihi budget {budget.max_lines}" ) return violations def _check_kpi_critical_function_budget(rubric: Dict[str, object]) -> KpiResult: entry = next((k for k in list(rubric.get("kpis") or []) if k.get("id") == "KPI-01"), {}) weight = int(entry.get("weight") or 0) issues: List[str] = [] passed = 0 items = list(rubric.get("critical_functions") or []) for item in items: file_path = str(item.get("file_path") or "").strip() class_name = str(item.get("class_name") or "").strip() function_name = str(item.get("function_name") or "").strip() max_loc = int(item.get("max_loc") or 0) abs_path = REPO_ROOT / file_path if not abs_path.exists(): issues.append(f"KPI-01: file tidak ditemukan ({file_path})") continue length = _resolve_function_length(abs_path, class_name, function_name) if length < 0: issues.append(f"KPI-01: fungsi tidak ditemukan ({class_name}.{function_name})") continue if length > max_loc: issues.append( f"KPI-01: {file_path} {class_name}.{function_name}={length} melebihi max_loc={max_loc}" ) continue passed += 1 score = _compute_weighted_score(weight, passed, len(items)) return KpiResult( kpi_id="KPI-01", name=str(entry.get("name") or "Critical Function Budget"), weight=weight, score=score, issues=tuple(issues), ) def _check_kpi_typed_exception(rubric: Dict[str, object]) -> KpiResult: entry = next((k for k in list(rubric.get("kpis") or []) if k.get("id") == "KPI-02"), {}) weight = int(entry.get("weight") or 0) issues: List[str] = [] passed = 0 items = list(rubric.get("critical_exception_targets") or []) for item in items: file_path = str(item.get("file_path") or "").strip() class_name = str(item.get("class_name") or "").strip() function_name = str(item.get("function_name") or "").strip() abs_path = REPO_ROOT / file_path if not abs_path.exists(): issues.append(f"KPI-02: file tidak ditemukan ({file_path})") continue node = _resolve_function_node(abs_path, class_name, function_name) if node is None: issues.append(f"KPI-02: fungsi tidak ditemukan ({class_name}.{function_name})") continue if _contains_generic_catch(node): issues.append( f"KPI-02: generic catch terdeteksi pada {file_path} {class_name}.{function_name}" ) continue passed += 1 score = _compute_weighted_score(weight, passed, len(items)) return KpiResult( kpi_id="KPI-02", name=str(entry.get("name") or "Typed Exception"), weight=weight, score=score, issues=tuple(issues), ) def _check_kpi_error_envelope(rubric: Dict[str, object]) -> KpiResult: # Local import agar script tetap ringan saat hanya parse AST. from pypos.modules.penjualan.services.error_envelope_service import ErrorEnvelopeService entry = next((k for k in list(rubric.get("kpis") or []) if k.get("id") == "KPI-03"), {}) weight = int(entry.get("weight") or 0) checks: List[Tuple[bool, str]] = [] sample = ErrorEnvelopeService.build_error( status=0, error_code="TRX_SAMPLE", reason="sample_reason", trace_id="trace-sample-01", message="sample", ) checks.append( ( _to_bool(sample.get("error_code")) and _to_bool(sample.get("reason")) and _to_bool(sample.get("trace_id")), "KPI-03: ErrorEnvelopeService.build_error tidak menghasilkan error_code/reason/trace_id", ) ) persist_flow_text = _read_text( REPO_ROOT / "pypos" / "modules" / "penjualan" / "services" / "transaksi_persist_flow_service.py" ) checks.append( ( "ErrorEnvelopeService.build_error" in persist_flow_text and "trace_id" in persist_flow_text and "error_code" in persist_flow_text and "reason" in persist_flow_text, "KPI-03: persist flow belum konsisten memakai error envelope standar", ) ) controller_text = _read_text( REPO_ROOT / "pypos" / "modules" / "penjualan" / "controllers" / "transaksi_penjualan_controller.py" ) checks.append( ( "_handle_persist_error" in controller_text and "error_code" in controller_text and "reason" in controller_text and "trace_id" in controller_text, "KPI-03: controller persist belum membaca contract error envelope lengkap", ) ) issues = tuple([message for ok, message in checks if not ok]) passed = sum(1 for ok, _ in checks if ok) score = _compute_weighted_score(weight, passed, len(checks)) return KpiResult( kpi_id="KPI-03", name=str(entry.get("name") or "Error Envelope Standard"), weight=weight, score=score, issues=issues, ) def _check_kpi_enterprise_controls(rubric: Dict[str, object]) -> KpiResult: entry = next((k for k in list(rubric.get("kpis") or []) if k.get("id") == "KPI-04"), {}) weight = int(entry.get("weight") or 0) checks: List[Tuple[bool, str]] = [] enterprise_path = ( REPO_ROOT / "pypos" / "modules" / "penjualan" / "services" / "transaksi_enterprise_control_service.py" ) enterprise_text = _read_text(enterprise_path) if enterprise_path.exists() else "" checks.append( ( enterprise_path.exists() and "transaksi_idempotency" in enterprise_text and "transaksi_persist_audit" in enterprise_text and "transaksi_approval_trail" in enterprise_text, "KPI-04: tabel enterprise control belum lengkap", ) ) checks.append( ( "immutable_audit_log" in enterprise_text and "immutable_approval_log" in enterprise_text and "run_reconciliation" in enterprise_text, "KPI-04: immutable trigger / reconciliation belum lengkap", ) ) outbox_path = ( REPO_ROOT / "pypos" / "modules" / "penjualan" / "services" / "free_produk_sync_outbox_service.py" ) outbox_text = _read_text(outbox_path) if outbox_path.exists() else "" checks.append( ( "recover_stale_inflight_leases" in outbox_text and "_recover_stale_inflight_rows" in outbox_text, "KPI-04: recovery stale inflight outbox belum aktif", ) ) reconciliation_script = REPO_ROOT / "scripts" / "ops" / "run_penjualan_reconciliation_job.py" checks.append( ( reconciliation_script.exists() and "--fail-on-dead" in _read_text(reconciliation_script), "KPI-04: reconciliation job/flag fail-on-dead belum tersedia", ) ) transaksi_controller_path = ( REPO_ROOT / "pypos" / "modules" / "penjualan" / "controllers" / "transaksi_penjualan_controller.py" ) transaksi_controller_text = _read_text(transaksi_controller_path) hotspot_use_case_path = ( REPO_ROOT / "pypos" / "modules" / "penjualan" / "services" / "transaksi_penjualan_hotspot_use_case_service.py" ) hotspot_use_case_text = _read_text(hotspot_use_case_path) if hotspot_use_case_path.exists() else "" has_controller_hook = "_record_admin_qty_approval_trail" in transaksi_controller_text has_direct_record = "record_approval_trail" in transaksi_controller_text has_service_record = ( "record_admin_qty_approval_trail" in hotspot_use_case_text and "record_approval_trail" in hotspot_use_case_text ) checks.append( ( has_controller_hook and (has_direct_record or has_service_record), "KPI-04: approval trail qty besar belum tercatat", ) ) pembayaran_controller_path = ( REPO_ROOT / "pypos" / "modules" / "penjualan" / "controllers" / "pembayaran_controller.py" ) pembayaran_controller_text = _read_text(pembayaran_controller_path) checks.append( ( "record_approval_trail" in pembayaran_controller_text and "diskon_pembayaran" in pembayaran_controller_text, "KPI-04: approval trail diskon pembayaran belum tercatat", ) ) issues = tuple([message for ok, message in checks if not ok]) passed = sum(1 for ok, _ in checks if ok) score = _compute_weighted_score(weight, passed, len(checks)) return KpiResult( kpi_id="KPI-04", name=str(entry.get("name") or "Enterprise Controls"), weight=weight, score=score, issues=issues, ) def _parse_pytest_markers(pytest_ini_path: Path = PYTEST_INI_PATH) -> List[str]: if not pytest_ini_path.exists(): return [] markers: List[str] = [] in_marker_block = False for raw_line in _read_text(pytest_ini_path).splitlines(): line = str(raw_line or "").rstrip() if not line: if in_marker_block: break continue if line.strip().startswith("#") or line.strip().startswith(";"): continue if line.strip().startswith("markers"): in_marker_block = True continue if in_marker_block: if not line.startswith(" "): break marker_name = str(line.strip().split(":", 1)[0] or "").strip() if marker_name: markers.append(marker_name) return markers def _check_test_matrix_rule(rule: Dict[str, object]) -> Tuple[bool, int, int]: files = [ REPO_ROOT / str(path or "").strip() for path in list(rule.get("files") or []) if str(path or "").strip() ] names = _collect_test_names(files) name_filters = [str(x or "").strip().lower() for x in list(rule.get("name_contains_any") or []) if str(x or "").strip()] if name_filters: names = [ name for name in names if any(frag in name.lower() for frag in name_filters) ] min_cases = int(rule.get("min_cases") or 0) count = len(names) return count >= min_cases, count, min_cases def _check_kpi_test_maturity(rubric: Dict[str, object]) -> KpiResult: entry = next((k for k in list(rubric.get("kpis") or []) if k.get("id") == "KPI-05"), {}) weight = int(entry.get("weight") or 0) issues: List[str] = [] checks: List[bool] = [] required_markers = [str(marker or "").strip() for marker in list(rubric.get("required_markers") or [])] available_markers = set(_parse_pytest_markers()) for marker in required_markers: ok = marker in available_markers checks.append(ok) if not ok: issues.append(f"KPI-05: marker pytest belum terdaftar ({marker})") test_matrix = dict(rubric.get("test_matrix") or {}) for key, rule in test_matrix.items(): if not isinstance(rule, dict): checks.append(False) issues.append(f"KPI-05: rule test matrix tidak valid ({key})") continue ok, count, min_cases = _check_test_matrix_rule(rule) checks.append(ok) if not ok: issues.append(f"KPI-05: {key} kurang skenario (count={count}, min={min_cases})") passed = sum(1 for check in checks if check) score = _compute_weighted_score(weight, passed, len(checks)) return KpiResult( kpi_id="KPI-05", name=str(entry.get("name") or "Test Maturity"), weight=weight, score=score, issues=tuple(issues), ) def _check_kpi_ci_quality_gate(rubric: Dict[str, object]) -> KpiResult: entry = next((k for k in list(rubric.get("kpis") or []) if k.get("id") == "KPI-06"), {}) weight = int(entry.get("weight") or 0) issues: List[str] = [] checks: List[bool] = [] quality_text = _read_text(QUALITY_GATE_PATH) if QUALITY_GATE_PATH.exists() else "" checks.append("run_penjualan_quality_guard.py" in quality_text) if not checks[-1]: issues.append("KPI-06: run_quality_gate belum memanggil run_penjualan_quality_guard.py") checks.append("critical_flow" in quality_text) if not checks[-1]: issues.append("KPI-06: stage critical_flow belum terdaftar di run_quality_gate") checks.append("STAGE_ORDER" in quality_text and "quality" in quality_text) if not checks[-1]: issues.append("KPI-06: stage order quality gate tidak valid") checks.append("compileall" in quality_text) if not checks[-1]: issues.append("KPI-06: compileall belum dijalankan pada quality stage") # Cek final rubric harus ada secara fisik. checks.append(RUBRIC_PATH.exists()) if not checks[-1]: issues.append("KPI-06: rubric final v100 tidak ditemukan") passed = sum(1 for check in checks if check) score = _compute_weighted_score(weight, passed, len(checks)) return KpiResult( kpi_id="KPI-06", name=str(entry.get("name") or "CI Quality Gate Lock"), weight=weight, score=score, issues=tuple(issues), ) def evaluate_penjualan_kpi() -> Dict[str, object]: if not RUBRIC_PATH.exists(): return { "max_score": 0, "pass_score": 0, "score": 0, "kpi_results": [], "issues": [f"Rubric tidak ditemukan: {RUBRIC_PATH}"], } rubric = _read_json(RUBRIC_PATH) score_cfg = dict(rubric.get("score") or {}) max_score = int(score_cfg.get("max") or 100) pass_score = int(score_cfg.get("pass_score") or max_score) results = [ _check_kpi_critical_function_budget(rubric), _check_kpi_typed_exception(rubric), _check_kpi_error_envelope(rubric), _check_kpi_enterprise_controls(rubric), _check_kpi_test_maturity(rubric), _check_kpi_ci_quality_gate(rubric), ] score = sum(int(item.score) for item in results) issues: List[str] = [] for item in results: issues.extend(list(item.issues)) if score < pass_score: issues.append(f"KPI total belum lulus: score={score}, required={pass_score}") return { "max_score": max_score, "pass_score": pass_score, "score": score, "kpi_results": [item.__dict__ for item in results], "issues": issues, } def run_penjualan_quality_guard() -> List[str]: issues: List[str] = [] issues.extend(check_forbidden_imports()) issues.extend(check_function_budgets()) kpi_eval = evaluate_penjualan_kpi() issues.extend(list(kpi_eval.get("issues") or [])) return issues def main() -> int: kpi_eval = evaluate_penjualan_kpi() print( "Penjualan KPI score " f"{int(kpi_eval.get('score') or 0)}/{int(kpi_eval.get('max_score') or 0)} " f"(required={int(kpi_eval.get('pass_score') or 0)})" ) for row in list(kpi_eval.get("kpi_results") or []): print( f"- {row.get('kpi_id')}: score={int(row.get('score') or 0)}/" f"{int(row.get('weight') or 0)} [{row.get('name')}]" ) issues = run_penjualan_quality_guard() if issues: print("Penjualan quality guard gagal.") for item in issues: print(f"- {item}") return 1 print("Penjualan quality guard lulus.") return 0 if __name__ == "__main__": raise SystemExit(main())