import sqlite3 from pypos.core.base_model import BaseModel from pypos.core.utils.db_helper import connect_sqlite from pypos.core.utils.path_utils import get_db_path from pypos.core.utils.sql_identifier_utils import quote_sql_identifier from pypos.core.utils.sql_query_builder import ( build_pragma_table_info_sql, build_sql_with_optional_where, render_sql_template, ) class CustomerSearchModel(BaseModel): def __init__(self, db_path=None): super().__init__() self.db_path = db_path or get_db_path() # edited by glg def _ensure_member_columns(self): # Runtime schema mutation dimatikan untuk menjaga kontrol migrasi terpusat. return False def _table_exists(self, cursor, table_name: str) -> bool: cursor.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,), ) return cursor.fetchone() is not None # edited by glg def _get_table_cols(self, cursor, table_name: str): try: query = build_pragma_table_info_sql(table_name) except ValueError: return [] cursor.execute(query) return [str(row[1]) for row in cursor.fetchall() if row and len(row) > 1 and row[1]] def load_all_customers(self): return self.search_customers(keyword="", limit=0, offset=0) # edited by glg @staticmethod def _build_active_filter_sql(customer_cols): clauses = [] if "status" in customer_cols: clauses.append("COALESCE(c.status, 1) = 1") if "trash" in customer_cols: clauses.append("COALESCE(c.trash, 0) = 0") return " AND ".join(clauses) # edited by glg @staticmethod def _compose_where_sql(active_filter_sql, keyword_filter_sql): clauses = [] if str(active_filter_sql or "").strip(): clauses.append(f"({active_filter_sql})") if str(keyword_filter_sql or "").strip(): clauses.append(f"({keyword_filter_sql})") if not clauses: return "" return " WHERE " + " AND ".join(clauses) # edited by glg @staticmethod def _build_kode_member_expr(customer_cols): candidates = [] if "kode" in customer_cols: candidates.append("NULLIF(c.kode, '')") if "member_id" in customer_cols: candidates.append("NULLIF(c.member_id, '')") if "mid" in customer_cols: candidates.append("NULLIF(c.mid, '')") candidates.append("CAST(c.id AS TEXT)") if len(candidates) == 1: return f"{candidates[0]} AS kode_member" return f"COALESCE({', '.join(candidates)}) AS kode_member" # edited by glg def _build_customer_query_parts(self, cursor, customer_cols): level_join = "" level_fallback_candidates = [] if "level_nama" in customer_cols: level_fallback_candidates.append("NULLIF(c.level_nama, '')") if "level_id" in customer_cols: level_fallback_candidates.append("NULLIF(CAST(c.level_id AS TEXT), '')") if level_fallback_candidates: level_fallback_expr = ", ".join(level_fallback_candidates + ["'-'"]) level_select = f"COALESCE({level_fallback_expr}) AS level_nama" else: level_fallback_expr = "'-'" level_select = "'-' AS level_nama" if self._table_exists(cursor, "per_customer_level") and "level_id" in customer_cols: lvl_cols = self._get_table_cols(cursor, "per_customer_level") level_name_col = "nama" if "nama" in lvl_cols else ("level_nama" if "level_nama" in lvl_cols else None) if level_name_col: level_join = ( "LEFT JOIN per_customer_level pcl " "ON CAST(pcl.id AS TEXT) = CAST(c.level_id AS TEXT)" ) level_select = render_sql_template( "COALESCE(NULLIF(pcl.{level_name_col}, ''), {fallback_expr}) AS level_nama", level_name_col=quote_sql_identifier(level_name_col, strict=True), fallback_expr=level_fallback_expr, ) point_join = "" loyalti_expr = "COALESCE(c.loyalti_point, 0)" if "loyalti_point" in customer_cols else "0" point_select = f"{loyalti_expr} AS point" if self._table_exists(cursor, "_rek_pembantu_customer_cache"): cache_cols = self._get_table_cols(cursor, "_rek_pembantu_customer_cache") point_col = None if "saldo_kredit" in cache_cols: point_col = "saldo_kredit" elif "saldo_debet" in cache_cols: point_col = "saldo_debet" elif "saldo" in cache_cols: point_col = "saldo" join_key = None if "customerID" in cache_cols: join_key = "customerID" elif "extern_id" in cache_cols: join_key = "extern_id" elif "_company_rekening_customerID" in cache_cols: join_key = "_company_rekening_customerID" if point_col and join_key: point_join = render_sql_template( """ LEFT JOIN ( SELECT CAST({join_key} AS TEXT) AS customer_id, MAX(COALESCE({point_col}, 0)) AS point FROM _rek_pembantu_customer_cache GROUP BY CAST({join_key} AS TEXT) ) pc ON pc.customer_id = CAST(c.id AS TEXT) """, join_key=quote_sql_identifier(join_key, strict=True), point_col=quote_sql_identifier(point_col, strict=True), ) point_select = f"COALESCE(pc.point, {loyalti_expr}, 0) AS point" return level_join, level_select, point_join, point_select # edited by glg def _build_keyword_filter(self, keyword, customer_cols): kw = str(keyword or "").strip() if not kw: return "", [] like = f"%{kw}%" clauses = [] params = [] searchable_cols = ["nama", "kode", "member_id", "mid", "email", "tlp_1"] for col_name in searchable_cols: if col_name not in customer_cols: continue clauses.append(f"COALESCE(c.{col_name}, '') LIKE ?") params.append(like) if "id" in customer_cols: clauses.append("CAST(c.id AS TEXT) LIKE ?") params.append(like) if not clauses: return "", [] return " OR ".join(clauses), params def count_customers(self, keyword=""): conn = connect_sqlite(self.db_path) try: conn.row_factory = sqlite3.Row cursor = conn.cursor() if not self._table_exists(cursor, "per_customers"): return 0 customer_cols = self._get_table_cols(cursor, "per_customers") active_filter_sql = self._build_active_filter_sql(customer_cols) keyword_filter_sql, params = self._build_keyword_filter(keyword, customer_cols) where_sql = self._compose_where_sql(active_filter_sql, keyword_filter_sql) query = build_sql_with_optional_where( "SELECT COUNT(1) AS total FROM per_customers c", where_sql, ) cursor.execute(query, tuple(params)) row = cursor.fetchone() return int(row["total"] or 0) if row else 0 finally: conn.close() def search_customers(self, keyword="", limit=100, offset=0): conn = connect_sqlite(self.db_path) try: conn.row_factory = sqlite3.Row cursor = conn.cursor() if not self._table_exists(cursor, "per_customers"): return [] customer_cols = self._get_table_cols(cursor, "per_customers") level_join, level_select, point_join, point_select = self._build_customer_query_parts( cursor, customer_cols, ) active_filter_sql = self._build_active_filter_sql(customer_cols) keyword_filter_sql, params = self._build_keyword_filter(keyword, customer_cols) where_sql = self._compose_where_sql(active_filter_sql, keyword_filter_sql) kode_member_expr = self._build_kode_member_expr(customer_cols) nama_expr = "COALESCE(c.nama, '') AS nama" if "nama" in customer_cols else "'' AS nama" email_expr = "COALESCE(c.email, '') AS email" if "email" in customer_cols else "'' AS email" telp_expr = "COALESCE(c.tlp_1, '') AS telp" if "tlp_1" in customer_cols else "'' AS telp" propinsi_expr = "COALESCE(c.propinsi, '') AS propinsi" if "propinsi" in customer_cols else "'' AS propinsi" paging_sql = "ORDER BY c.id ASC" paging_params = [] try: paging_limit = int(limit or 0) except (TypeError, ValueError): paging_limit = 0 if paging_limit > 0: try: paging_offset = max(0, int(offset or 0)) except (TypeError, ValueError): paging_offset = 0 paging_sql += " LIMIT ? OFFSET ?" paging_params.extend([paging_limit, paging_offset]) query = render_sql_template( """ SELECT c.id, {kode_member_expr}, {level_select}, {nama_expr}, {email_expr}, {telp_expr}, {point_select}, {propinsi_expr} FROM per_customers c {level_join} {point_join} {where_sql} {paging_sql} """, kode_member_expr=kode_member_expr, level_select=level_select, nama_expr=nama_expr, email_expr=email_expr, telp_expr=telp_expr, point_select=point_select, propinsi_expr=propinsi_expr, level_join=level_join, point_join=point_join, where_sql=where_sql, paging_sql=paging_sql, ) cursor.execute(query, tuple(params + paging_params)) results = cursor.fetchall() return [dict(row) for row in results] finally: conn.close()