合并

etl_billiards/tasks/members_dwd_task.py

@@ -1,13 +1,14 @@
-# -*- coding: utf-8 -*-
+# -*- coding: utf-8 -*-
 from .base_dwd_task import BaseDwdTask
 from loaders.dimensions.member import MemberLoader
 from models.parsers import TypeParser
 import json
+from utils.windowing import build_window_segments
 
 class MembersDwdTask(BaseDwdTask):
     """
     DWD Task: Process Member Records from ODS to Dimension Table
-    Source: billiards_ods.ods_member_profile
+    Source: billiards_ods.member_profiles
     Target: billiards.dim_member
     """
 
@@ -17,53 +18,72 @@ class MembersDwdTask(BaseDwdTask):
     def execute(self) -> dict:
         self.logger.info(f"Starting {self.get_task_code()} task")
 
-        window_start, window_end, _ = self._get_time_window()
-        self.logger.info(f"Processing window: {window_start} to {window_end}")
+        base_start, base_end, _ = self._get_time_window()
+        segments = build_window_segments(
+            self.config,
+            base_start,
+            base_end,
+            tz=self.tz,
+            override_only=True,
+        )
+        if not segments:
+            segments = [(base_start, base_end)]
+
+        total_segments = len(segments)
+        if total_segments > 1:
+            self.logger.info(f"{self.get_task_code()}: ????? {total_segments} ?")
 
         loader = MemberLoader(self.db)
         store_id = self.config.get("app.store_id")
 
         total_inserted = 0
         total_updated = 0
-        total_errors = 0
 
-        # Iterate ODS Data
-        batches = self.iter_ods_rows(
-            table_name="billiards_ods.ods_member_profile",
-            columns=["site_id", "member_id", "payload", "fetched_at"],
-            start_time=window_start,
-            end_time=window_end
+        for idx, (window_start, window_end) in enumerate(segments, start=1):
+            self.logger.info(
+                f"Processing window {idx}/{total_segments}: {window_start} to {window_end}"
+            )
+            batches = self.iter_ods_rows(
+                table_name="billiards_ods.member_profiles",
+                columns=["site_id", "member_id", "payload", "fetched_at"],
+                start_time=window_start,
+                end_time=window_end
+            )
+
+            for batch in batches:
+                if not batch:
+                    continue
+
+                parsed_rows = []
+                for row in batch:
+                    payload = self.parse_payload(row)
+                    if not payload:
+                        continue
+
+                    parsed = self._parse_member(payload, store_id)
+                    if parsed:
+                        parsed_rows.append(parsed)
+
+                if parsed_rows:
+                    inserted, updated, skipped = loader.upsert_members(parsed_rows, store_id)
+                    total_inserted += inserted
+                    total_updated += updated
+
+            self.db.commit()
+
+        overall_start = segments[0][0]
+        overall_end = segments[-1][1]
+
+        self.logger.info(
+            f"Task {self.get_task_code()} completed. Inserted: {total_inserted}, Updated: {total_updated}"
         )
 
-        for batch in batches:
-            if not batch:
-                continue
-            
-            parsed_rows = []
-            for row in batch:
-                payload = self.parse_payload(row)
-                if not payload:
-                    continue
-                
-                parsed = self._parse_member(payload, store_id)
-                if parsed:
-                    parsed_rows.append(parsed)
-            
-            if parsed_rows:
-                inserted, updated, skipped = loader.upsert_members(parsed_rows, store_id)
-                total_inserted += inserted
-                total_updated += updated
-
-        self.db.commit()
-        
-        self.logger.info(f"Task {self.get_task_code()} completed. Inserted: {total_inserted}, Updated: {total_updated}")
-        
         return {
             "status": "success",
             "inserted": total_inserted,
             "updated": total_updated,
-            "window_start": window_start.isoformat(),
-            "window_end": window_end.isoformat()
+            "window_start": overall_start.isoformat(),
+            "window_end": overall_end.isoformat()
         }
 
     def _parse_member(self, raw: dict, store_id: int) -> dict:
@@ -87,3 +107,4 @@ class MembersDwdTask(BaseDwdTask):
         except Exception as e:
             self.logger.warning(f"Error parsing member: {e}")
             return None
+