import hashlib import json # edited by glg class DashboardValueUtils: @staticmethod def safe_int(value, default=0): try: return int(value) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): return int(default or 0) @staticmethod def to_non_negative_int(value, default=0): try: parsed = int(value) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): parsed = int(default or 0) return parsed if parsed >= 0 else int(default or 0) @staticmethod def format_int_local(value): try: parsed = int(value or 0) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): parsed = 0 return f"{max(0, parsed):,}".replace(",", ".") @staticmethod def safe_float(value, default=None): if value is None: return default if isinstance(value, (int, float)): return float(value) text = str(value).strip() if not text: return default normalized = text.replace("Rp", "").replace("rp", "").replace(" ", "") if "," in normalized and "." in normalized: normalized = normalized.replace(".", "").replace(",", ".") elif "." in normalized: parts = normalized.split(".") if len(parts) > 2: normalized = "".join(parts) elif len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit() and len(parts[1]) == 3: normalized = "".join(parts) else: normalized = normalized.replace(",", "") try: return float(normalized) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): return default @staticmethod def format_rupiah_local(value): amount = DashboardValueUtils.safe_float(value, default=None) if amount is None: return "" return f"Rp {DashboardValueUtils.format_int_local(round(amount))}" @staticmethod def friendly_sync_table_name(table_name): key = str(table_name or "").strip().lower() mapping = { "produk": "Produk", "price": "Harga Jual", "price_per_area": "Harga Per Area", "diskon": "Diskon", "diskon_customer": "Diskon Customer", "per_customers": "Data Customer", "per_employee": "Data Karyawan", "setting_struk": "Pengaturan Struk", "bank": "Metode Pembayaran", } if key in mapping: return mapping[key] if not key: return "Data Master" return key.replace("_", " ").title() @staticmethod def extract_check_update_samples(raw_value): if not isinstance(raw_value, dict): return [] raw_samples = raw_value.get("samples") if not isinstance(raw_samples, list): return [] samples = [] for raw in raw_samples[:3]: if not isinstance(raw, dict): continue changes_payload = raw.get("changes") if isinstance(raw.get("changes"), dict) else {} entity_name = str( raw.get("entity_name") or raw.get("name") or raw.get("nama") or "" ).strip() changed_field = str( raw.get("changed_field") or raw.get("field") or raw.get("kolom") or "" ).strip() entity_id = str( raw.get("entity_id") or raw.get("id") or "" ).strip() if not entity_name and not entity_id: continue before_value = ( raw.get("value_before") if "value_before" in raw else raw.get("before") ) if before_value is None: before_value = ( raw.get("old_value") if "old_value" in raw else raw.get("old") ) if before_value is None: before_value = ( raw.get("harga_lama") if "harga_lama" in raw else raw.get("price_before") ) if before_value is None and isinstance(changes_payload, dict): before_value = ( changes_payload.get("before") if "before" in changes_payload else changes_payload.get("old") ) after_value = ( raw.get("value_after") if "value_after" in raw else raw.get("after") ) if after_value is None: after_value = ( raw.get("new_value") if "new_value" in raw else raw.get("new") ) if after_value is None: after_value = ( raw.get("harga_baru") if "harga_baru" in raw else raw.get("price_after") ) if after_value is None and isinstance(changes_payload, dict): after_value = ( changes_payload.get("after") if "after" in changes_payload else changes_payload.get("new") ) samples.append( { "entity_name": entity_name, "entity_id": entity_id, "changed_field": changed_field, "before_value": before_value, "after_value": after_value, } ) return samples @staticmethod def extract_changed_tables_from_check(check_response, requested_tables=None): data = check_response.get("data") if isinstance(check_response, dict) else {} if not isinstance(data, dict): return [] requested = { str(name or "").strip().lower() for name in (requested_tables or []) if str(name or "").strip() } rows_by_table = {} samples_by_table = {} for raw_name, raw_value in data.items(): table_name = str(raw_name or "").strip() if not table_name: continue key = table_name.lower() if requested and key not in requested: continue count = 0 if isinstance(raw_value, dict): for count_key in ("row", "rows", "count", "jml", "total", "updated", "new"): if count_key in raw_value: count = max(count, DashboardValueUtils.safe_int(raw_value.get(count_key), 0)) else: count = DashboardValueUtils.safe_int(raw_value, 0) if count > 0: rows_by_table[key] = max(rows_by_table.get(key, 0), count) if isinstance(raw_value, dict): samples_by_table[key] = DashboardValueUtils.extract_check_update_samples(raw_value) items = [] for key, count in rows_by_table.items(): items.append( { "table": key, "rows": int(count), "label": DashboardValueUtils.friendly_sync_table_name(key), "samples": list(samples_by_table.get(key) or []), } ) items.sort(key=lambda row: int(row.get("rows") or 0), reverse=True) return items @staticmethod def extract_row_total_from_check(check_response, changed_tables): fallback_total = int( sum( DashboardValueUtils.safe_int((item or {}).get("rows"), 0) for item in (changed_tables or []) ) ) if isinstance(check_response, dict): raw_total = check_response.get("row") total = DashboardValueUtils.safe_int(raw_total, -1) if total > 0: return int(total) if total == 0 and fallback_total > 0: return fallback_total if total == 0: return 0 return fallback_total @staticmethod def build_auto_sync_update_signature(row_total, changed_tables): normalized = { "row_total": int(row_total or 0), "tables": [ { "table": str(item.get("table") or ""), "rows": int(item.get("rows") or 0), "samples": [ { "entity_name": str(sample.get("entity_name") or ""), "entity_id": str(sample.get("entity_id") or ""), "changed_field": str(sample.get("changed_field") or ""), "before_value": sample.get("before_value"), "after_value": sample.get("after_value"), } for sample in list(item.get("samples") or [])[:3] ], } for item in (changed_tables or []) ], } raw = json.dumps(normalized, sort_keys=True, ensure_ascii=True) # edited by glg # Gunakan SHA-256 untuk fingerprint non-kredensial agar lolos policy scanner modern. return hashlib.sha256(raw.encode("utf-8")).hexdigest() @staticmethod def build_auto_sync_prompt_message(row_total, changed_tables): _ = row_total if changed_tables: top_tables = changed_tables[:4] produk_names = [] seen_names = set() for item in top_tables: for sample in list(item.get("samples") or [])[:4]: entity_name = str(sample.get("entity_name") or "").strip() if not entity_name: continue name_key = entity_name.lower() if name_key in seen_names: continue seen_names.add(name_key) produk_names.append(entity_name) if len(produk_names) >= 6: break if len(produk_names) >= 6: break if produk_names: tampil_names = produk_names[:5] extra_line = "" if len(produk_names) > 5: extra_line = "\n- Dan produk lainnya" daftar_produk = "\n".join(f"- {name}" for name in tampil_names) return ( "Ada perubahan pada produk dari server.\n\n" "Produk yang berubah:\n" f"{daftar_produk}" f"{extra_line}\n\n" "Apakah Anda ingin mengupdate data sekarang?" ) return ( "Ada perubahan pada data produk dari server.\n\n" "Terdapat perubahan data produk.\n\n" "Apakah Anda ingin mengupdate data sekarang?" ) return ( "Ada perubahan pada data produk dari server.\n\n" "Terdapat perubahan data produk.\n\n" "Apakah Anda ingin mengupdate data sekarang?" )