Merge pull request #80 from govtechmy/feat/SSD-1263-new-field-peringkat

asyiqinasir · web-flow · commit b5ab815915c7 · 2026-04-17T10:25:35.000+08:00
Feat/ssd 1263 new field peringkat
diff --git a/src/models/analitik_sekolah.py b/src/models/analitik_sekolah.py
@@ -13,8 +13,24 @@
 from src.core.time import _utc_now
 
 
-class AnalitikItem(BaseModel):
-    """Individual analytics item with type, percentage, and total."""
+class PeringkatItem(BaseModel):
+    """Breakdown of peringkat distribution within a jenis category."""
+    
+    peringkat: str = Field(..., description="Peringkat (e.g., Rendah, Menengah)")
+    total: int = Field(..., description="Jumlah sekolah untuk peringkat ini dalam jenis ini")
+
+
+class AnalitikJenisItem(BaseModel):
+    """Analytics item for jenisLabel dimension."""
+    
+    jenis: str = Field(..., description="Kategori atau jenis")
+    peratus: float = Field(..., description="Peratusan daripada jumlah keseluruhan")
+    total: int = Field(..., description="Jumlah bilangan untuk kategori ini")
+    peringkatBreakdown: list[PeringkatItem] = Field(default_factory=list, description="Breakdown mengikut peringkat dalam jenis ini")
+
+
+class AnalitikBantuanItem(BaseModel):
+    """Analytics item for bantuan dimension."""
     
     jenis: str = Field(..., description="Kategori atau jenis")
     peratus: float = Field(..., description="Peratusan daripada jumlah keseluruhan")
@@ -24,8 +40,8 @@ class AnalitikItem(BaseModel):
 class AnalitikSekolahData(BaseModel):
     """Container for all analytics dimensions in array format."""
     
-    jenisLabel: list[AnalitikItem] = Field(default_factory=list, description="Analisis mengikut jenis/label sekolah") 
-    bantuan: list[AnalitikItem] = Field(default_factory=list, description="Analisis mengikut jenis bantuan")
+    jenisLabel: list[AnalitikJenisItem] = Field(default_factory=list, description="Analisis mengikut jenis/label sekolah") 
+    bantuan: list[AnalitikBantuanItem] = Field(default_factory=list, description="Analisis mengikut jenis bantuan")
 
 
 _settings = get_settings()
@@ -55,12 +71,16 @@ def from_sekolah_list(cls, sekolah_list: list["Sekolah"]) -> "AnalitikSekolah":
         
         # Initialize analytics containers with defaultdicts for dynamic counting
         jenis_counts = defaultdict(int)
+        jenis_peringkat_counts = defaultdict(lambda: defaultdict(int))
         bantuan_counts = defaultdict(int)
 
         # Process each sekolah to build analytics dynamically
         for sekolah in sekolah_list:
             # Count categories
-            cls._increment_count(jenis_counts, sekolah.jenisLabel)
+            jenis_key = cls._normalize_value(sekolah.jenisLabel)
+            peringkat_key = cls._normalize_peringkat_value(sekolah.peringkat)
+            jenis_counts[jenis_key] += 1
+            jenis_peringkat_counts[jenis_key][peringkat_key] += 1
             cls._increment_count(bantuan_counts, sekolah.bantuan)
             
             # Sum totals
@@ -77,10 +97,14 @@ def from_sekolah_list(cls, sekolah_list: list["Sekolah"]) -> "AnalitikSekolah":
                 student_count += sekolah.enrolmenKhas
             total_pelajar += student_count
 
-        # Convert counts to AnalitikItem arrays
+        # Convert counts to analytics arrays
         data = AnalitikSekolahData(
-            jenisLabel=cls._convert_to_analitik_items(jenis_counts, jumlah_sekolah),
-            bantuan=cls._convert_to_analitik_items(bantuan_counts, jumlah_sekolah),
+            jenisLabel=cls._convert_to_analitik_jenis_items(
+                jenis_counts,
+                jumlah_sekolah,
+                jenis_peringkat_counts=jenis_peringkat_counts,
+            ),
+            bantuan=cls._convert_to_analitik_bantuan_items(bantuan_counts, jumlah_sekolah),
         )
 
         return cls(
@@ -98,8 +122,12 @@ def _increment_count(counter: dict, value: Optional[str]) -> None:
         counter[key] = counter.get(key, 0) + 1
 
     @staticmethod
-    def _convert_to_analitik_items(counter: defaultdict, total: int) -> list[AnalitikItem]:
-        """Convert defaultdict to list of AnalitikItem objects."""
+    def _convert_to_analitik_jenis_items(
+        counter: defaultdict,
+        total: int,
+        jenis_peringkat_counts: Optional[dict[str, dict]] = None,
+    ) -> list[AnalitikJenisItem]:
+        """Convert defaultdict to list of AnalitikJenisItem objects."""
         items = []
         
         # Add existing categories
@@ -109,32 +137,106 @@ def _convert_to_analitik_items(counter: defaultdict, total: int) -> list[Analiti
             else:
                 peratus = 0.0
             
-            items.append(AnalitikItem(
+            # Build peringkat breakdown for this jenis if data is available
+            peringkat_breakdown = []
+            if jenis_peringkat_counts and jenis in jenis_peringkat_counts:
+                peringkat_counts = jenis_peringkat_counts[jenis]
+                for peringkat_key, peringkat_count in peringkat_counts.items():
+                    if peringkat_key != "TIADA MAKLUMAT":  # Only include actual peringkat values
+                        peringkat_breakdown.append(
+                            PeringkatItem(
+                                peringkat=AnalitikSekolah._display_peringkat(peringkat_key),
+                                total=peringkat_count
+                            )
+                        )
+                # Add TIADA MAKLUMAT if it exists
+                if "TIADA MAKLUMAT" in peringkat_counts and peringkat_counts["TIADA MAKLUMAT"] > 0:
+                    peringkat_breakdown.append(
+                        PeringkatItem(
+                            peringkat="TIADA MAKLUMAT",
+                            total=peringkat_counts["TIADA MAKLUMAT"]
+                        )
+                    )
+                # Sort by total descending
+                peringkat_breakdown.sort(key=lambda x: x.total, reverse=True)
+            
+            items.append(AnalitikJenisItem(
                 jenis=jenis,
                 peratus=peratus,
-                total=count
+                total=count,
+                peringkatBreakdown=peringkat_breakdown
             ))
         
         # Ensure "Tiada Maklumat" exists even if count is 0
         tiada_maklumat_exists = any(item.jenis == "TIADA MAKLUMAT" for item in items)
         if not tiada_maklumat_exists:
-            items.append(AnalitikItem(
+            items.append(AnalitikJenisItem(
                 jenis="TIADA MAKLUMAT",
                 peratus=0.0,
-                total=0
+                total=0,
+                peringkatBreakdown=[]
             ))
         
         # Sort by total count descending
         items.sort(key=lambda x: x.total, reverse=True)
         return items
 
+    @staticmethod
+    def _convert_to_analitik_bantuan_items(
+        counter: defaultdict,
+        total: int,
+    ) -> list[AnalitikBantuanItem]:
+        """Convert defaultdict to list of AnalitikBantuanItem objects."""
+        items = []
+
+        for jenis, count in counter.items():
+            if total > 0:
+                peratus = round((count / total) * 100, 1)
+            else:
+                peratus = 0.0
+
+            items.append(AnalitikBantuanItem(
+                jenis=jenis,
+                peratus=peratus,
+                total=count,
+            ))
+
+        tiada_maklumat_exists = any(item.jenis == "TIADA MAKLUMAT" for item in items)
+        if not tiada_maklumat_exists:
+            items.append(AnalitikBantuanItem(
+                jenis="TIADA MAKLUMAT",
+                peratus=0.0,
+                total=0,
+            ))
+
+        items.sort(key=lambda x: x.total, reverse=True)
+        return items
+
     @staticmethod
     def _normalize_value(value: Optional[str]) -> str:
         """Normalize field values for consistent analytics categorization."""
         if value is None or str(value).strip() == "":
             return "TIADA MAKLUMAT"
         return str(value).strip().upper()
 
+    @classmethod
+    def _normalize_peringkat_value(cls, value: Optional[str]) -> str:
+        if value is not None and hasattr(value, "value"):
+            value = getattr(value, "value")
+        normalized = cls._normalize_value(value)
+        if normalized in {"RENDAH", "MENENGAH"}:
+            return normalized
+        return "TIADA MAKLUMAT"
+
+    @classmethod
+    def _display_peringkat(cls, value: str) -> str:
+        normalized = cls._normalize_peringkat_value(value)
+        if normalized == "RENDAH":
+            return "RENDAH"
+        if normalized == "MENENGAH":
+            return "MENENGAH"
+        return "TIADA MAKLUMAT"
+
     def to_document(self) -> dict:
         """Convert the analytics to a Mongo-ready document, omitting None fields."""
         doc = self.model_dump(exclude_none=True, by_alias=True)
@@ -145,5 +247,7 @@ def to_document(self) -> dict:
 __all__ = [
     "AnalitikSekolah",
     "AnalitikSekolahData",
-    "AnalitikItem",
+    "AnalitikJenisItem",
+    "AnalitikBantuanItem",
+    "PeringkatItem",
 ]
diff --git a/src/models/entiti_sekolah.py b/src/models/entiti_sekolah.py
@@ -9,7 +9,7 @@
 from pydantic import ConfigDict
 from typing_extensions import Literal
 from src.config.settings import get_settings
-from src.models.sekolah import SekolahStatus
+from src.models.sekolah import SekolahStatus, PeringkatEnum
 from src.models.negeri_enum import NegeriEnum
 
 
@@ -36,6 +36,7 @@ class InfoPentadbiran(BaseModel):
     negeri: NegeriEnum | None = Field(default=None, description="State the school is located in")
     ppd: Optional[str] = Field(default=None, description="Pejabat Pendidikan Daerah (district office)")
     parlimen: Optional[str] = Field(default=None, description="Parliament constituency")
+    peringkat: PeringkatEnum | None = Field(default=None, description="School tier (RENDAH/MENENGAH)")
     bantuan: Optional[str] = Field(default=None, description="Bantuan classification")
     bilSesi: Optional[str] = Field(default=None, description="Number of school sessions")
     sesi: Optional[str] = Field(default=None, description="School session")
@@ -135,6 +136,7 @@ def from_sekolah(
             negeri=sekolah.negeri,
             ppd=sekolah.ppd,
             parlimen=sekolah.parlimen,
+            peringkat=sekolah.peringkat,
             bantuan=sekolah.bantuan,
             bilSesi=sekolah.bilSesi,
             sesi=sekolah.sesi,
diff --git a/src/models/sekolah.py b/src/models/sekolah.py
@@ -15,6 +15,7 @@
 MISSING_VALUES = {"TIADA", "", "NONE", "-", "--", "BELUM ADA"}
 BOOL_ADA_MAP = {"ADA": True, "TIADA": False, "": None}
 BOOL_YA_MAP = {"YA": True, "TIDAK": False, "": None}
+PERINGKAT_VALUES = {"RENDAH", "MENENGAH"}
 
 _settings = get_settings()
 
@@ -23,14 +24,21 @@ class SekolahStatus(str, Enum):
     ACTIVE = "ACTIVE"
     INACTIVE = "INACTIVE"
 
+
+class PeringkatEnum(str, Enum):
+    """Enum for school tier (Rendah/Menengah)."""
+    RENDAH = "RENDAH"
+    MENENGAH = "MENENGAH"
+
+
 class Sekolah(BaseModel):
     collection_name: ClassVar[str] = _settings.sekolah_collection
 
     negeri: NegeriEnum | None = Field(default=None, alias="NEGERI")
     ppd: Optional[str] = Field(default=None, alias="PPD")
     parlimen: Optional[str] = Field(default=None, alias="PARLIMEN")
     dun: Optional[str] = Field(default=None, alias="DUN")
-    peringkat: Optional[str] = Field(default=None, alias="PERINGKAT")
+    peringkat: PeringkatEnum | None = Field(default=None, alias="PERINGKAT")
     jenisLabel: Optional[str] = Field(default=None, alias="JENIS/LABEL")
     kodSekolah: str = Field(..., alias="KODSEKOLAH")
     namaSekolah: Optional[str] = Field(default=None, alias="NAMASEKOLAH")
@@ -98,6 +106,19 @@ def normalize_dun(cls, value):
         text = str(value).strip()
         return text or None
 
+    @field_validator("peringkat", mode="before")
+    def normalize_peringkat(cls, value):
+        if value is None:
+            return None
+        if isinstance(value, PeringkatEnum):
+            return value
+        text = str(value).strip().upper()
+        if not text or text in MISSING_VALUES:
+            return None
+        if text in {"RENDAH", "MENENGAH"}:
+            return PeringkatEnum(text)
+        return None
+
     @field_validator(
         "poskodSurat",
         "enrolmenPrasekolah",
@@ -182,6 +203,9 @@ def to_document(self) -> dict:
         status_value = data.get("status")
         if isinstance(status_value, SekolahStatus):
             data["status"] = status_value.value
+        peringkat_value = data.get("peringkat")
+        if isinstance(peringkat_value, PeringkatEnum):
+            data["peringkat"] = peringkat_value.value
         if self.koordinatXX is not None and self.koordinatYY is not None:
             data["location"] = {
                 "type": "Point",
@@ -192,4 +216,4 @@ def to_document(self) -> dict:
         return data
 
 
-__all__ = ["Sekolah"]
+__all__ = ["Sekolah", "PeringkatEnum"]
diff --git a/src/statistics/agg_analitik_sekolah.py b/src/statistics/agg_analitik_sekolah.py
