Refactor to checkpoint based mirror sync

Signed-off-by: Samk <sampurnapyne1710@gmail.com>

Author: Samk1710

.github/workflows/sync.yml

@@ -24,8 +24,8 @@ jobs:
       - name: Install required packages
         run: pip install -r requirements.txt
 
-      - name: Run sync (daily)
-        run: python sync_catalog.py daily
+      - name: Run sync
+        run: python sync_catalog.py
 
       - name: Commit and push if it changed
         run: |-

sync_catalog.py

@@ -5,231 +5,154 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
-import argparse
 import json
+import math
 import sys
-from datetime import date, datetime, timedelta, timezone
+from datetime import date, datetime, timezone
 from pathlib import Path
-from typing import Any, Dict
 
 import requests
-
 from aboutcode.pipeline import BasePipeline, LoopProgress
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 
 ROOT_PATH = Path(__file__).parent
-CATALOG_PATH = ROOT_PATH / "catalog"
-PAGE_DIRECTORY = CATALOG_PATH / "pages"
-
-API_URL = "https://euvdservices.enisa.europa.eu/api/search"
+ADVISORY_PATH = ROOT_PATH / "advisory"
+CHECKPOINT_FILE = ROOT_PATH / "checkpoint.json"
 
 HEADERS = {
     "User-Agent": "Vulnerablecode",
     "Accept": "application/json",
 }
 
 PAGE_SIZE = 100
-DEFAULT_START_YEAR = 1970
 REQUEST_TIMEOUT = 10
 
 
-class EUVDCatalogMirror(BasePipeline):
+class EUVDAdvisoryMirror(BasePipeline):
+
+    url = "https://euvdservices.enisa.europa.eu/api/search"
+
     @classmethod
     def steps(cls):
-        return (cls.collect_catalog,)
+        return (
+            cls.load_checkpoint,
+            cls.create_session,
+            cls.collect_new_advisory,
+            cls.save_checkpoint,
+        )
 
-    def collect_catalog(self) -> None:
-        if getattr(self, "mode", "backfill") == "daily":
-            return self.sync_yesterday()
-        self.backfill_from_year(DEFAULT_START_YEAR)
+    def log(self, message):
+        now_local = datetime.now(timezone.utc).astimezone()
+        timestamp = now_local.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+        print(f"{timestamp} {message}")
 
-    def backfill_from_year(self, start_year: int) -> None:
+    def load_checkpoint(self):
         """
-        Backfill EUVD catalog data starting from the given year(DEFAULT_START_YEAR) up to today. Data is collected month by month and stored in year/month directories as JSON files.
+        - Load the ``last run`` date from checkpoint.json to fetch only new advisories.
+        - If the checkpoint.json does not exist, fetch all advisories.
         """
-        today = date.today()
-        backfill_start = date(start_year, 1, 1)
-        backfill_end = today
-
-        months: list[tuple[int, int]] = []
-        current = backfill_start
-
-        while current <= backfill_end:
-            months.append((current.year, current.month))
-            if current.month == 12:
-                current = date(current.year + 1, 1, 1)
-                continue
-            current = date(current.year, current.month + 1, 1)
-
-        self.log(f"Starting backfill for {len(months)} months")
-
-        progress = LoopProgress(total_iterations=len(months), logger=self.log)
-
-        for year, month in progress.iter(months):
-            month_start = date(year, month, 1)
-            day_token = month_start.isoformat()
-
-            if year == backfill_end.year and month == backfill_end.month:
-                self._collect_paginated(
-                    start=month_start,
-                    end=backfill_end,
-                    year=year,
-                    month=month,
-                    day_token=day_token,
-                )
-                continue
-
-            if month == 12:
-                next_month = date(year + 1, 1, 1)
-            else:
-                next_month = date(year, month + 1, 1)
-            month_end = next_month - timedelta(days=1)
-
-            self._collect_paginated(
-                start=month_start,
-                end=month_end,
-                year=year,
-                month=month,
-                day_token=day_token,
-            )
-
-        self.log("Backfill completed")
+        self.fetch_params = {}
+        if not CHECKPOINT_FILE.exists():
+            return
+        with CHECKPOINT_FILE.open() as f:
+            checkpoint = json.load(f)
+        last_run = checkpoint.get("last_run")
+        if last_run:
+            self.fetch_params["fromUpdatedDate"] = last_run
+
+    def create_session(self):
+        retry = Retry(
+            total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504]
+        )
+        adapter = HTTPAdapter(max_retries=retry)
+        self.session = requests.Session()
+        self.session.headers.update(HEADERS)
+        self.session.mount("https://", adapter)
 
-    def sync_yesterday(self) -> None:
-        target_date = date.today() - timedelta(days=1)
+    def collect_new_advisory(self):
+        """
+        Fetch new advisories from the EUVD API with paginated requests.
 
-        first_page = self.fetch_page(
-            {
-                "fromUpdatedDate": target_date.isoformat(),
-                "toUpdatedDate": target_date.isoformat(),
-                "size": 1,
-                "page": 0,
-            }
-        )
+        - Fetch the ``total`` advisories  and determine the number of pages to iterate over.
+        - Iterate through all pages, fetching up to PAGE_SIZE advisories per request.
+        - Save each advisory as a JSON file at ``/advisory/{year}/{month}/{EUVD_ID}.json``.
+        - Advisories with missing publication dates are stored as at ``/advisory/unpublished/{EUVD_ID}.json``.
+        """
+        count_page = self.fetch_page({**self.fetch_params, "size": 1, "page": 0})
+        total = count_page.get("total", 0)
+        if not total:
+            self.log("No new advisories found")
+            return
 
-        total = first_page.get("total", 0)
-        total_pages = (total + PAGE_SIZE - 1) // PAGE_SIZE
+        total_pages = math.ceil(total / PAGE_SIZE)
+        self.log(f"Collecting {total} advisories across {total_pages} pages")
 
+        self.has_unpublished = (ADVISORY_PATH / "unpublished").exists()
         progress = LoopProgress(total_iterations=total_pages, logger=self.log)
 
         for page in progress.iter(range(total_pages)):
             data = self.fetch_page(
-                {
-                    "fromUpdatedDate": target_date.isoformat(),
-                    "toUpdatedDate": target_date.isoformat(),
-                    "size": PAGE_SIZE,
-                    "page": page,
-                }
+                {**self.fetch_params, "size": PAGE_SIZE, "page": page}
             )
+            for advisory in data.get("items", []):
+                self.save_advisory(advisory)
 
-            self.write_page_file(
-                year=target_date.year,
-                month=target_date.month,
-                day_token=target_date.isoformat(),
-                page_number=page + 1,
-                payload=data,
-            )
-
-    def _collect_paginated(
-        self,
-        start: date,
-        end: date,
-        year: int,
-        month: int,
-        day_token: str,
-    ) -> None:
-        """
-        Fetch all EUVD results for the given date range using paginated requests.The total number of results is read from the API response and used to determine how many pages to fetch. Each page is written as a separate JSON file under the corresponding year/month directory.
-        """
-
-        first_page = self.fetch_page(
-            {
-                "fromUpdatedDate": start.isoformat(),
-                "toUpdatedDate": end.isoformat(),
-                "size": 1,
-                "page": 0,
-            }
-        )
-
-        total = first_page.get("total", 0)
-        total_pages = (total + PAGE_SIZE - 1) // PAGE_SIZE
+    def save_advisory(self, advisory):
+        euvd_id = advisory.get("id")
+        if not euvd_id:
+            self.log(f"Advisory missing id, skipping: {advisory}")
+            return
 
-        for page in range(total_pages):
-            data = self.fetch_page(
-                {
-                    "fromUpdatedDate": start.isoformat(),
-                    "toUpdatedDate": end.isoformat(),
-                    "size": PAGE_SIZE,
-                    "page": page,
-                }
-            )
+        date_published = advisory.get("datePublished", "")
+        dir_path = self.advisory_dir(date_published)
 
-            self.write_page_file(
-                year=year,
-                month=month,
-                day_token=day_token,
-                page_number=page + 1,
-                payload=data,
-            )
+        if dir_path is None:
+            dir_path = ADVISORY_PATH / "unpublished"
+            self.has_unpublished = True
 
-    def write_page_file(
-        self,
-        year: int,
-        month: int,
-        day_token: str,
-        page_number: int,
-        payload: Dict[str, Any],
-    ) -> None:
-        year_str = f"{year:04d}"
-        month_str = f"{month:02d}"
-        page_str = f"{page_number:04d}"
-
-        dir_path = PAGE_DIRECTORY / year_str / month_str
         dir_path.mkdir(parents=True, exist_ok=True)
 
-        filename = f"page{day_token}-{page_str}.json"
-        path = dir_path / filename
+        # If an existing unpublished advisory is published now, remove the stale advisory from unpublished directory.
+        if self.has_unpublished and dir_path != ADVISORY_PATH / "unpublished":
+            stale_advisory = ADVISORY_PATH / "unpublished" / f"{euvd_id}.json"
+            if stale_advisory.exists():
+                stale_advisory.unlink()
 
-        if path.exists():
-            return
+        # If old advisory is updated, the new data overwrites the existing file.
+        with (dir_path / f"{euvd_id}.json").open("w", encoding="utf-8") as f:
+            json.dump(advisory, f, indent=2)
 
-        with path.open("w", encoding="utf-8") as output:
-            json.dump(payload, output, indent=2)
+    def advisory_dir(self, date_published):
+        """
+        Returns the directory path for an advisory based on its publication date.
+        """
+        try:
+            published_at = datetime.strptime(date_published, "%b %d, %Y, %I:%M:%S %p")
+            return (
+                ADVISORY_PATH / f"{published_at.year:04d}" / f"{published_at.month:02d}"
+            )
+        except (ValueError, TypeError):
+            return None
 
-    def fetch_page(self, params: Dict[str, Any]) -> Dict[str, Any]:
-        response = requests.get(
-            API_URL,
-            params=params,
-            headers=HEADERS,
-            timeout=REQUEST_TIMEOUT,
-        )
+    def save_checkpoint(self):
+        with CHECKPOINT_FILE.open("w") as f:
+            json.dump({"last_run": date.today().isoformat()}, f, indent=2)
+
+    def fetch_page(self, params):
+        response = self.session.get(self.url, params=params, timeout=REQUEST_TIMEOUT)
         response.raise_for_status()
-        data: Any = response.json()
+        data = response.json()
         if not isinstance(data, dict):
-            return {}
+            raise ValueError(
+                f"Unexpected response type {type(data).__name__} for params {params}"
+            )
         return data
 
-    def log(self, message: str) -> None:
-        now = datetime.now(timezone.utc).astimezone()
-        stamp = now.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
-        print(f"{stamp} {message}")
-
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="EUVD Catalog Mirror")
-    parser.add_argument(
-        "mode",
-        nargs="?",
-        default="daily",
-        choices=["backfill", "daily"],
-        help="Sync mode: 'backfill' for full history, 'daily' for yesterday only",
-    )
-    args = parser.parse_args()
-
-    mirror = EUVDCatalogMirror()
-    mirror.mode = args.mode
-
+    mirror = EUVDAdvisoryMirror()
     status_code, error_message = mirror.execute()
     if error_message:
         print(error_message)
     sys.exit(status_code)
-