Merge pull request #81 from govtechmy/feat/SSD-1229-fileVersion-field-in-DatasetStatus

Feat/ssd 1229 file version field in dataset status

Author: asyiqinasir

src/core/gsheet.py

@@ -1,10 +1,59 @@
 import requests
+import logging
+import re
+import urllib.parse
 
+logger = logging.getLogger(__name__)
 
-def fetch_csv_data(gsheet_id: str, gid: str) -> bytes:
+def _extract_filename(content_disp: str) -> str | None:
+    if not content_disp:
+        return None
+
+    # RFC 5987 format: filename*=UTF-8''encoded_name.csv
+    match_star = re.search(r"filename\*=UTF-8''([^;]+)(?:;|$)", content_disp, re.IGNORECASE,)
+    if match_star:
+        filename = urllib.parse.unquote(match_star.group(1))
+        logger.info("Extracted filename: %s", filename)
+        return filename
+
+    # Standard format: filename="file.csv" OR filename=file.csv
+    match = re.search(r'filename=(?:"([^"]*)"|([^;]+))(?:;|$)', content_disp, re.IGNORECASE,)
+    if match:
+        filename = match.group(1) if match.group(1) is not None else match.group(2).strip()
+        logger.info("Extracted filename (standard): %s", filename)
+        return filename
+
+    logger.warning("Failed to extract filename from Content-Disposition: %s", content_disp)
+    return None
+
+def _extract_file_version(file_name: str | None) -> str | None:
+    if not file_name:
+        return None
+
+    name = file_name.strip().rsplit(".", 1)[0]
+    name = name.split(" - ")[0]
+    parts = name.split("_")
+
+    if len(parts) >= 2:
+        file_version = parts[-1]
+        logger.info("Extracted fileVersion: %s", file_version)
+        return file_version
+
+    logger.warning("Failed to extract fileVersion from filename: %s", file_name)
+    return None
+
+def fetch_csv_data(gsheet_id: str, gid: str) -> tuple[bytes, str | None]:
     url = f"https://docs.google.com/spreadsheets/d/{gsheet_id}/export?format=csv&gid={gid}"
-    print(f'Fetching CSV data from URL: {url}')
+    logger.info("Fetching CSV data from URL: %s", url)
 
     response = requests.get(url)
     response.raise_for_status()
-    return response.content
+
+    content_type = response.headers.get("Content-Type", "")
+    if "text/csv" not in content_type:
+        logger.warning("Google Sheet not accessible as CSV. Content-Type: %s", content_type)
+
+    content_disp = response.headers.get("Content-Disposition", "")
+    filename = _extract_filename(content_disp)
+
+    return response.content, filename

src/core/s3.py

@@ -6,16 +6,18 @@
 from botocore.exceptions import ClientError, ResponseStreamingError
 
 from src.core.aws import get_s3_client, get_s3_bucket_name
+from src.core.gsheet import _extract_file_version
 
 s3 = get_s3_client()
 logger = logging.getLogger(__name__)
 
-def _upload_to_s3(csv_bytes: bytes, bucket: str, prefix: str) -> str:
+def _upload_to_s3(csv_bytes: bytes, bucket: str, prefix: str, source_filename: str | None = None,) -> str:
     if not bucket:
         bucket = get_s3_bucket_name()
 
     timestamp = int(time.time())
-    s3_key = f"{prefix}/{timestamp}.csv"
+    version = _extract_file_version(source_filename) or "unknown"
+    s3_key = f"{prefix}/{version}/{timestamp}.csv"
 
     s3.put_object(
         Bucket=bucket,

src/models/dataset_status.py

@@ -21,6 +21,7 @@ class DatasetStatus(BaseModel):
 
 	id: str = Field(..., alias="_id", description="Dataset identifier (e.g. sekolah, institusi, analitik)")
 	lastUpdatedAt: datetime = Field(default_factory=_utc_now, description="UTC timestamp of last successful ingestion")
+	fileVersion: str | None = Field(default=None, description="Cleaned version of the file name e.g SenaraiSekolahWeb_Mac2026")
 
 	def to_document(self) -> dict:
 		return self.model_dump(by_alias=True)

src/pipeline/dataset_status.py

@@ -5,24 +5,29 @@
 from src.config.settings import Settings
 from src.core.time import _utc_now
 from src.utils.db.get_db_collection import get_db_collection
+from typing import Optional
 
 logger = logging.getLogger(__name__)
 
 
-def upsert_dataset_status(dataset_name: str, settings: Settings) -> None:
+def upsert_dataset_status(dataset_name: str, settings: Settings, file_version: Optional[str]) -> None:
     """Upsert the lastUpdatedAt timestamp for a dataset into DatasetStatus.
 
     The document shape is:
-    {"_id": <dataset_name>, "lastUpdatedAt": <UTC datetime>}
+    {"_id": <dataset_name>, "lastUpdatedAt": <UTC datetime>, "fileVersion": <file_version>}
+    fileVersion is only set if provided (not None).
     """
     if not dataset_name:
         raise ValueError("dataset_name must be a non-empty string")
 
     try:
         collection = get_db_collection(settings, name=settings.dataset_status_collection)
+        update_fields = {"lastUpdatedAt": _utc_now()}
+        if file_version is not None:
+            update_fields["fileVersion"] = file_version
         collection.update_one(
             {"_id": dataset_name},
-            {"$set": {"lastUpdatedAt": _utc_now()}},
+            {"$set": update_fields},
             upsert=True,
         )
     except Exception:

src/pipeline/ingestion.py

@@ -17,7 +17,7 @@
 
 from src.config import Settings, get_settings
 from src.models import Sekolah
-from src.core.gsheet import fetch_csv_data
+from src.core.gsheet import fetch_csv_data, _extract_file_version
 from src.core.s3 import (_upload_to_s3, _latest_csv_from_s3, _read_csv_from_s3)
 from src.models.sekolah import SekolahStatus
 from src.pipeline.status_sync import sync_entiti_statuses
@@ -91,22 +91,22 @@ def _read_csv(path: str) -> Iterable[Dict[str, Any]]:
 def _read_google_sheet(sheet_id: str, gid: str) -> Iterable[Dict[str, Any]]:
     logger.info("Scraping Google Sheet directly (sheet_id=%s, gid=%s)", sheet_id, gid,)
 
-    csv_bytes = fetch_csv_data(sheet_id, gid)
+    csv_bytes, _ = fetch_csv_data(sheet_id, gid)
     df = pd.read_csv(io.BytesIO(csv_bytes), dtype=str).fillna("")
 
     logger.info("Google Sheet loaded: %d rows, %d columns", df.shape[0], df.shape[1])
     return df.to_dict(orient="records")
 
 
-def _load_rows(settings: Settings) -> Iterable[Dict[str, Any]]:
-    csv_bytes = fetch_csv_data(settings.gsheet_id, settings.gsheet_gid)
+def _load_rows(settings: Settings) -> tuple[Iterable[Dict[str, Any]], str | None]:
+    csv_bytes, file_name = fetch_csv_data(settings.gsheet_id, settings.gsheet_gid)
     logger.info("Uploading CSV data to S3 bucket %s", settings.s3_bucket_dataproc)
-    s3_key = _upload_to_s3(csv_bytes, settings.s3_bucket_dataproc, settings.s3_prefix_sekolah)
+    s3_key = _upload_to_s3(csv_bytes, settings.s3_bucket_dataproc, settings.s3_prefix_sekolah, file_name)
     logger.info("CSV uploaded to S3 at key: %s", s3_key)
 
     df = _read_csv_from_s3(settings.s3_bucket_dataproc, s3_key)
     logger.info("CSV loaded from S3: %d rows, %d columns", df.shape[0], df.shape[1])
-    return df.to_dict(orient="records")
+    return df.to_dict(orient="records"), file_name
 
 
 def _chunked(rows: Iterable[Dict[str, Any]], size: int) -> Iterator[list[Dict[str, Any]]]:
@@ -142,13 +142,14 @@ def _format_validation_messages(exc: ValidationError) -> list[str]:
 def _collect_documents(
     settings: Settings,
     kodSekolah_madani: Set[str],
-) -> tuple[list[dict[str, Any]], list[dict[str, Any]], int, set[Any]]:
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]], int, set[Any], str | None]:
     documents: list[dict[str, Any]] = []
     errors: list[dict[str, Any]] = []
     total = 0
     present_identifiers: set[Any] = set()
 
-    for index, row in enumerate(_load_rows(settings), start=1):
+    rows, file_name = _load_rows(settings)
+    for index, row in enumerate(rows, start=1):
         total += 1
 
         raw_kod = str(row.get("KODSEKOLAH", "")).strip()
@@ -169,7 +170,7 @@ def _collect_documents(
         document["isSekolahAngkatMADANI"] = canonical_kod in kodSekolah_madani
         documents.append(document)
 
-    return documents, errors, total, present_identifiers
+    return documents, errors, total, present_identifiers, file_name
 
 
 def _load_kodSekolah_madani(database, settings: Settings) -> Set[str]:
@@ -326,7 +327,7 @@ def run(settings: Settings) -> dict[str, Any]:
     sekolah_collection = database[Sekolah.collection_name]
     entiti_collection = database[settings.entiti_sekolah_collection]
     kodSekolah_madani = _load_kodSekolah_madani(database, settings)
-    documents, errors, total, present_identifiers = _collect_documents(settings, kodSekolah_madani)
+    documents, errors, total, present_identifiers, file_name = _collect_documents(settings, kodSekolah_madani)
 
     for document in documents:
         # All schools present in raw file are ACTIVE
@@ -378,7 +379,13 @@ def run(settings: Settings) -> dict[str, Any]:
         "inactivated": inactivated,
         "entiti_synced": entiti_synced,
     }
-    upsert_dataset_status("sekolah", settings)
+
+    file_version = _extract_file_version(file_name)
+
+    if not file_version:
+        logger.warning("File name not found or failed to extract fileVersion. filename=%s", file_name)
+
+    upsert_dataset_status("sekolah", settings, file_version)
     return summary
 
 def run_with_overrides(**overrides: Any) -> dict[str, Any]:

src/pipeline/institusi.py

@@ -70,7 +70,7 @@ def _load_rows(settings: Settings) -> Iterable[Dict[str, Any]]:
     if not settings.institusi_gsheet_id or not settings.institusi_gsheet_gid:
         raise RuntimeError("GSHEET_ID and INSTITUSI_GSHEET_GID must be set")
 
-    csv_bytes = fetch_csv_data(settings.institusi_gsheet_id, settings.institusi_gsheet_gid)
+    csv_bytes, _ = fetch_csv_data(settings.institusi_gsheet_id, settings.institusi_gsheet_gid)
     logger.info("Uploading Institusi CSV data to S3 bucket %s", settings.s3_bucket_dataproc)
     s3_key = _upload_to_s3(csv_bytes, settings.s3_bucket_dataproc, settings.s3_prefix_institusi)
     logger.info("Institusi CSV uploaded to S3 at key: %s", s3_key)